Differential D6334 Diff 23169 swh/provenance/tests/conftest.py

Changeset View

Standalone View

swh/provenance/tests/conftest.py

	# Copyright (C) 2021 The Software Heritage developers			# Copyright (C) 2021 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 datetime, timedelta, timezone			from datetime import datetime, timedelta, timezone
	from os import path			from os import path
	from typing import Any, Dict, Iterable, Iterator			from typing import Any, Dict, Generator, Iterable, Iterator

	from _pytest.fixtures import SubRequest			from _pytest.fixtures import SubRequest
				import mongomock.database
	import msgpack			import msgpack
	import psycopg2.extensions			import psycopg2.extensions
	import pymongo.database
	import pytest			import pytest
	from pytest_postgresql.factories import postgresql			from pytest_postgresql.factories import postgresql

	from swh.journal.serializers import msgpack_ext_hook			from swh.journal.serializers import msgpack_ext_hook
	from swh.provenance import get_provenance, get_provenance_storage			from swh.provenance import get_provenance, get_provenance_storage
	from swh.provenance.api.client import ProvenanceStorageRPCClient			from swh.provenance.api.client import ProvenanceStorageRPCClient
	import swh.provenance.api.server as server			import swh.provenance.api.server as server
	from swh.provenance.archive import ArchiveInterface			from swh.provenance.archive import ArchiveInterface
	Show All 28 Lines
	@pytest.fixture			@pytest.fixture
	def app(			def app(
	provenance_postgresqldb: Dict[str, str]			provenance_postgresqldb: Dict[str, str]
	) -> Iterator[server.ProvenanceStorageRPCServerApp]:			) -> Iterator[server.ProvenanceStorageRPCServerApp]:
	assert hasattr(server, "storage")			assert hasattr(server, "storage")
	server.storage = get_provenance_storage(			server.storage = get_provenance_storage(
	cls="postgresql", db=provenance_postgresqldb			cls="postgresql", db=provenance_postgresqldb
	)			)
				server.storage.open()
	yield server.app			yield server.app
				server.storage.close()


	# the RPCClient class used as client used in these tests			# the RPCClient class used as client used in these tests
	@pytest.fixture			@pytest.fixture
	def swh_rpc_client_class() -> type:			def swh_rpc_client_class() -> type:
	return ProvenanceStorageRPCClient			return ProvenanceStorageRPCClient


	@pytest.fixture(params=["mongodb", "postgresql", "rpcapi"])			@pytest.fixture(params=["mongodb", "postgresql", "rpcapi"])
	def provenance_storage(			def provenance_storage(
	request: SubRequest,			request: SubRequest,
	provenance_postgresqldb: Dict[str, str],			provenance_postgresqldb: Dict[str, str],
	mongodb: pymongo.database.Database,			mongodb: mongomock.database.Database,
	swh_rpc_client: ProvenanceStorageRPCClient,			swh_rpc_client: ProvenanceStorageRPCClient,
	) -> ProvenanceStorageInterface:			) -> Generator[ProvenanceStorageInterface, None, None]:
	"""Return a working and initialized ProvenanceStorageInterface object"""			"""Return a working and initialized ProvenanceStorageInterface object"""

	if request.param == "rpcapi":			if request.param == "rpcapi":
	assert isinstance(swh_rpc_client, ProvenanceStorageInterface)			assert isinstance(swh_rpc_client, ProvenanceStorageInterface)
	return swh_rpc_client			swh_rpc_client.open()
				yield swh_rpc_client
				swh_rpc_client.close()

	elif request.param == "mongodb":			elif request.param == "mongodb":
	from swh.provenance.mongo.backend import ProvenanceStorageMongoDb			mongodb_params = {
				"host": mongodb.client.address[0],
	return ProvenanceStorageMongoDb(mongodb)			"port": mongodb.client.address[1],
				"dbname": mongodb.name,
				}
				mongodb_storage = get_provenance_storage(
				cls=request.param, db=mongodb_params, engine="mongomock"
				)
				mongodb_storage.open()
				yield mongodb_storage
				mongodb_storage.close()

	else:			else:
	# in test sessions, we DO want to raise any exception occurring at commit time			# in test sessions, we DO want to raise any exception occurring at commit time
	return get_provenance_storage(			storage = get_provenance_storage(
	cls=request.param, db=provenance_postgresqldb, raise_on_commit=True			cls=request.param, db=provenance_postgresqldb, raise_on_commit=True
	)			)
				storage.open()
				yield storage
				storage.close()


	provenance_postgresql = postgresql("postgresql_proc", dbname="provenance_tests")			provenance_postgresql = postgresql("postgresql_proc", dbname="provenance_tests")


	@pytest.fixture			@pytest.fixture
	def provenance(			def provenance(
	provenance_postgresql: psycopg2.extensions.connection,			provenance_postgresql: psycopg2.extensions.connection,
	) -> ProvenanceInterface:			) -> Generator[ProvenanceInterface, None, None]:
	"""Return a working and initialized ProvenanceInterface object"""			"""Return a working and initialized ProvenanceInterface object"""

	from swh.core.cli.db import populate_database_for_package			from swh.core.cli.db import populate_database_for_package

	populate_database_for_package(			populate_database_for_package(
	"swh.provenance", provenance_postgresql.dsn, flavor="with-path"			"swh.provenance", provenance_postgresql.dsn, flavor="with-path"
	)			)
	# in test sessions, we DO want to raise any exception occurring at commit time			# in test sessions, we DO want to raise any exception occurring at commit time
	return get_provenance(			provenance = get_provenance(
	cls="postgresql",			cls="postgresql",
	db=provenance_postgresql.get_dsn_parameters(),			db=provenance_postgresql.get_dsn_parameters(),
	raise_on_commit=True,			raise_on_commit=True,
	)			)
				provenance.open()
				yield provenance
				provenance.close()


	@pytest.fixture			@pytest.fixture
	def archive(swh_storage: StorageInterface) -> ArchiveInterface:			def archive(swh_storage: StorageInterface) -> ArchiveInterface:
	"""Return an ArchiveStorage-based ArchiveInterface object"""			"""Return an ArchiveStorage-based ArchiveInterface object"""
	return ArchiveStorage(swh_storage)			return ArchiveStorage(swh_storage)


	Show All 34 Lines