Changeset View
Changeset View
Standalone View
Standalone View
swh/indexer/tests/test_indexer.py
# Copyright (C) 2020 The Software Heritage developers | # Copyright (C) 2020 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
from typing import Any, Dict, Iterable, List, Optional | from typing import Any, Dict, Iterable, List, Optional | ||||
from unittest.mock import Mock | from unittest.mock import Mock | ||||
import pytest | import pytest | ||||
from swh.indexer.indexer import ( | from swh.indexer.indexer import ( | ||||
ContentIndexer, | ContentIndexer, | ||||
ContentPartitionIndexer, | ContentPartitionIndexer, | ||||
DirectoryIndexer, | |||||
OriginIndexer, | OriginIndexer, | ||||
RevisionIndexer, | |||||
) | ) | ||||
from swh.indexer.storage import PagedResult, Sha1 | from swh.indexer.storage import PagedResult, Sha1 | ||||
from swh.model.model import Content | from swh.model.model import Content | ||||
from .utils import BASE_TEST_CONFIG, REVISION | from .utils import BASE_TEST_CONFIG, DIRECTORY2 | ||||
class _TestException(Exception): | class _TestException(Exception): | ||||
pass | pass | ||||
class CrashingIndexerMixin: | class CrashingIndexerMixin: | ||||
USE_TOOLS = False | USE_TOOLS = False | ||||
Show All 15 Lines | |||||
class CrashingContentIndexer(CrashingIndexerMixin, ContentIndexer): | class CrashingContentIndexer(CrashingIndexerMixin, ContentIndexer): | ||||
pass | pass | ||||
class CrashingContentPartitionIndexer(CrashingIndexerMixin, ContentPartitionIndexer): | class CrashingContentPartitionIndexer(CrashingIndexerMixin, ContentPartitionIndexer): | ||||
pass | pass | ||||
class CrashingRevisionIndexer(CrashingIndexerMixin, RevisionIndexer): | class CrashingDirectoryIndexer(CrashingIndexerMixin, DirectoryIndexer): | ||||
pass | pass | ||||
class CrashingOriginIndexer(CrashingIndexerMixin, OriginIndexer): | class CrashingOriginIndexer(CrashingIndexerMixin, OriginIndexer): | ||||
pass | pass | ||||
class TrivialContentPartitionIndexer(ContentPartitionIndexer[str]): | class TrivialContentPartitionIndexer(ContentPartitionIndexer[str]): | ||||
Show All 20 Lines | def test_content_indexer_catch_exceptions(): | ||||
assert indexer.run([b"foo"]) == {"status": "failed"} | assert indexer.run([b"foo"]) == {"status": "failed"} | ||||
indexer.catch_exceptions = False | indexer.catch_exceptions = False | ||||
with pytest.raises(_TestException): | with pytest.raises(_TestException): | ||||
indexer.run([b"foo"]) | indexer.run([b"foo"]) | ||||
def test_revision_indexer_catch_exceptions(): | def test_directory_indexer_catch_exceptions(): | ||||
indexer = CrashingRevisionIndexer(config=BASE_TEST_CONFIG) | indexer = CrashingDirectoryIndexer(config=BASE_TEST_CONFIG) | ||||
indexer.storage = Mock() | indexer.storage = Mock() | ||||
indexer.storage.revision_get.return_value = [REVISION] | indexer.storage.directory_get.return_value = [DIRECTORY2] | ||||
assert indexer.run([b"foo"]) == {"status": "failed"} | assert indexer.run([b"foo"]) == {"status": "failed"} | ||||
assert indexer.process_journal_objects({"revision": [REVISION.to_dict()]}) == { | assert indexer.process_journal_objects({"directory": [DIRECTORY2.to_dict()]}) == { | ||||
"status": "failed" | "status": "failed" | ||||
} | } | ||||
indexer.catch_exceptions = False | indexer.catch_exceptions = False | ||||
with pytest.raises(_TestException): | with pytest.raises(_TestException): | ||||
indexer.run([b"foo"]) | indexer.run([b"foo"]) | ||||
with pytest.raises(_TestException): | with pytest.raises(_TestException): | ||||
indexer.process_journal_objects({"revision": [REVISION.to_dict()]}) | indexer.process_journal_objects({"directory": [DIRECTORY2.to_dict()]}) | ||||
def test_origin_indexer_catch_exceptions(): | def test_origin_indexer_catch_exceptions(): | ||||
indexer = CrashingOriginIndexer(config=BASE_TEST_CONFIG) | indexer = CrashingOriginIndexer(config=BASE_TEST_CONFIG) | ||||
assert indexer.run(["http://example.org"]) == {"status": "failed"} | assert indexer.run(["http://example.org"]) == {"status": "failed"} | ||||
assert indexer.process_journal_objects( | assert indexer.process_journal_objects( | ||||
▲ Show 20 Lines • Show All 55 Lines • Show Last 20 Lines |