Changeset View
Changeset View
Standalone View
Standalone View
swh/indexer/tests/conftest.py
| Show All 13 Lines | |||||
| from swh.indexer.storage import get_indexer_storage | from swh.indexer.storage import get_indexer_storage | ||||
| from .utils import fill_storage, fill_obj_storage | from .utils import fill_storage, fill_obj_storage | ||||
| TASK_NAMES = ['revision_intrinsic_metadata', 'origin_intrinsic_metadata'] | TASK_NAMES = ['revision_intrinsic_metadata', 'origin_intrinsic_metadata'] | ||||
| storage_config = { | |||||
| 'cls': 'pipeline', | |||||
| 'steps': [ | |||||
| {'cls': 'validate'}, | |||||
| {'cls': 'memory'}, | |||||
| ] | |||||
| } | |||||
| @pytest.fixture | @pytest.fixture | ||||
| def indexer_scheduler(swh_scheduler): | def indexer_scheduler(swh_scheduler): | ||||
| for taskname in TASK_NAMES: | for taskname in TASK_NAMES: | ||||
| swh_scheduler.create_task_type({ | swh_scheduler.create_task_type({ | ||||
| 'type': taskname, | 'type': taskname, | ||||
| 'description': 'The {} indexer testing task'.format(taskname), | 'description': 'The {} indexer testing task'.format(taskname), | ||||
| 'backend_name': 'swh.indexer.tests.tasks.{}'.format(taskname), | 'backend_name': 'swh.indexer.tests.tasks.{}'.format(taskname), | ||||
| 'default_interval': timedelta(days=1), | 'default_interval': timedelta(days=1), | ||||
| Show All 18 Lines | |||||
| @pytest.fixture | @pytest.fixture | ||||
| def storage(): | def storage(): | ||||
| """An instance of in-memory storage that gets injected into all indexers | """An instance of in-memory storage that gets injected into all indexers | ||||
| classes. | classes. | ||||
| """ | """ | ||||
| storage = get_storage('memory') | storage = get_storage(**storage_config) | ||||
| fill_storage(storage) | fill_storage(storage) | ||||
| with patch('swh.storage.in_memory.InMemoryStorage') as storage_mock: | with patch('swh.storage.in_memory.InMemoryStorage') as storage_mock: | ||||
| storage_mock.return_value = storage | storage_mock.return_value = storage | ||||
| yield storage | yield storage | ||||
| @pytest.fixture | @pytest.fixture | ||||
| def obj_storage(): | def obj_storage(): | ||||
| Show All 17 Lines | |||||