Changeset View
Changeset View
Standalone View
Standalone View
swh/indexer/tests/test_indexer.py
| Show All 11 Lines | from swh.indexer.indexer import ( | ||||
| ContentIndexer, | ContentIndexer, | ||||
| ContentPartitionIndexer, | ContentPartitionIndexer, | ||||
| OriginIndexer, | OriginIndexer, | ||||
| RevisionIndexer, | 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 | from .utils import BASE_TEST_CONFIG, REVISION | ||||
| class _TestException(Exception): | class _TestException(Exception): | ||||
| pass | pass | ||||
| class CrashingIndexerMixin: | class CrashingIndexerMixin: | ||||
| USE_TOOLS = False | USE_TOOLS = False | ||||
| ▲ Show 20 Lines • Show All 55 Lines • ▼ Show 20 Lines | def test_content_indexer_catch_exceptions(): | ||||
| with pytest.raises(_TestException): | with pytest.raises(_TestException): | ||||
| indexer.run([b"foo"]) | indexer.run([b"foo"]) | ||||
| def test_revision_indexer_catch_exceptions(): | def test_revision_indexer_catch_exceptions(): | ||||
| indexer = CrashingRevisionIndexer(config=BASE_TEST_CONFIG) | indexer = CrashingRevisionIndexer(config=BASE_TEST_CONFIG) | ||||
| indexer.storage = Mock() | indexer.storage = Mock() | ||||
| indexer.storage.revision_get.return_value = ["rev"] | indexer.storage.revision_get.return_value = [REVISION] | ||||
| 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"]) | ||||
| ▲ Show 20 Lines • Show All 55 Lines • Show Last 20 Lines | |||||