Differential D1106 Diff 3479 swh/indexer/tests/conftest.py

Changeset View

Standalone View

swh/indexer/tests/conftest.py

	import pytest
	from datetime import timedelta			from datetime import timedelta
				from unittest.mock import patch

				import pytest

				from swh.objstorage.objstorage_in_memory import InMemoryObjStorage
	from swh.scheduler.tests.conftest import * # noqa			from swh.scheduler.tests.conftest import * # noqa
				from swh.storage.in_memory import Storage

				from swh.indexer.storage.in_memory import IndexerStorage

				from .utils import fill_storage, fill_obj_storage


	TASK_NAMES = ['revision_metadata', 'origin_intrinsic_metadata']			TASK_NAMES = ['revision_metadata', 'origin_intrinsic_metadata']


	@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),
	'min_interval': timedelta(hours=6),			'min_interval': timedelta(hours=6),
	'max_interval': timedelta(days=12),			'max_interval': timedelta(days=12),
	'num_retries': 3,			'num_retries': 3,
	})			})
	return swh_scheduler			return swh_scheduler


				@pytest.fixture
				def idx_storage():
				"""An instance of swh.indexer.storage.in_memory.IndexerStorage that
				gets injected into all indexers classes."""
				idx_storage = IndexerStorage()
				with patch('swh.indexer.storage.in_memory.IndexerStorage') \
				as idx_storage_mock:
				idx_storage_mock.return_value = idx_storage
				yield idx_storage


				@pytest.fixture
				def storage():
				"""An instance of swh.storage.in_memory.Storage that gets injected
				into all indexers classes."""
				storage = Storage()
				fill_storage(storage)
				with patch('swh.storage.in_memory.Storage') as storage_mock:
				storage_mock.return_value = storage
				yield storage


				@pytest.fixture
				def obj_storage():
				"""An instance of swh.objstorage.objstorage_in_memory.InMemoryObjStorage
				that gets injected into all indexers classes."""
				objstorage = InMemoryObjStorage()
				fill_obj_storage(objstorage)
				with patch.dict('swh.objstorage._STORAGE_CLASSES',
				{'memory': lambda: objstorage}):
				yield objstorage


	@pytest.fixture(scope='session')			@pytest.fixture(scope='session')
	def celery_includes():			def celery_includes():
	return [			return [
	'swh.indexer.tests.tasks',			'swh.indexer.tests.tasks',
	'swh.indexer.tasks',			'swh.indexer.tasks',
	]			]