Differential D8002 Diff 28836 swh/indexer/tests/conftest.py

Changeset View

Standalone View

swh/indexer/tests/conftest.py

	# Copyright (C) 2019-2020 The Software Heritage developers			# Copyright (C) 2019-2022 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 datetime import timedelta			from datetime import timedelta
	from functools import partial			from functools import partial
	import os			import os
	from typing import List, Tuple			from typing import List, Tuple
	from unittest.mock import patch			from unittest.mock import patch

	import pytest			import pytest
	from pytest_postgresql import factories			from pytest_postgresql import factories
	import yaml			import yaml

	from swh.core.db.pytest_plugin import initialize_database_for_module, postgresql_fact			from swh.core.db.pytest_plugin import initialize_database_for_module
	from swh.indexer.storage import get_indexer_storage			from swh.indexer.storage import IndexerStorage, get_indexer_storage
	from swh.indexer.storage.db import Db as IndexerDb
	from swh.objstorage.factory import get_objstorage			from swh.objstorage.factory import get_objstorage
	from swh.storage import get_storage			from swh.storage import get_storage

	from .utils import fill_obj_storage, fill_storage			from .utils import fill_obj_storage, fill_storage

	TASK_NAMES: List[Tuple[str, str]] = [			TASK_NAMES: List[Tuple[str, str]] = [
	# (scheduler-task-type, task-class-test-name)			# (scheduler-task-type, task-class-test-name)
	("index-revision-metadata", "revision_intrinsic_metadata"),			("index-directory-metadata", "directory_intrinsic_metadata"),
	("index-origin-metadata", "origin_intrinsic_metadata"),			("index-origin-metadata", "origin_intrinsic_metadata"),
	]			]


	idx_postgresql_proc = factories.postgresql_proc(			idx_postgresql_proc = factories.postgresql_proc(
	dbname="indexer_storage",
	load=[			load=[
	partial(			partial(
	initialize_database_for_module,			initialize_database_for_module,
	modname="indexer",			modname="indexer",
	version=IndexerDb.current_version,			version=IndexerStorage.current_version,
	)			)
	],			],
	)			)

	idx_storage_postgresql = postgresql_fact("idx_postgresql_proc")			idx_storage_postgresql = factories.postgresql("idx_postgresql_proc")


	@pytest.fixture			@pytest.fixture
	def indexer_scheduler(swh_scheduler):			def indexer_scheduler(swh_scheduler):
	# Insert the expected task types within the scheduler			# Insert the expected task types within the scheduler
	for task_name, task_class_name in TASK_NAMES:			for task_name, task_class_name in TASK_NAMES:
	swh_scheduler.create_task_type(			swh_scheduler.create_task_type(
	{			{
	▲ Show 20 Lines • Show All 84 Lines • Show Last 20 Lines