Differential D5962 Diff 21451 swh/provenance/tests/test_provenance_db.py

Changeset View

Standalone View

swh/provenance/tests/test_provenance_db.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

	import datetime			from datetime import datetime, timedelta, timezone
				from typing import Type

				from swh.model.model import OriginVisitStatus
	from swh.model.tests.swh_model_data import TEST_OBJECTS			from swh.model.tests.swh_model_data import TEST_OBJECTS
	from swh.provenance.model import OriginEntry			from swh.provenance.model import OriginEntry
	from swh.provenance.origin import origin_add			from swh.provenance.origin import origin_add
				from swh.provenance.postgresql.provenancedb_base import ProvenanceDBBase
	from swh.provenance.postgresql.provenancedb_with_path import ProvenanceWithPathDB			from swh.provenance.postgresql.provenancedb_with_path import ProvenanceWithPathDB
	from swh.provenance.postgresql.provenancedb_without_path import ProvenanceWithoutPathDB			from swh.provenance.postgresql.provenancedb_without_path import ProvenanceWithoutPathDB
				from swh.provenance.provenance import ProvenanceInterface, ProvenanceStorageInterface
	from swh.provenance.storage.archive import ArchiveStorage			from swh.provenance.storage.archive import ArchiveStorage
				from swh.storage.postgresql.storage import Storage


	def ts2dt(ts: dict) -> datetime.datetime:			# TODO: remove this function in favour of TimestampWithTimezone.to_datetime
	timestamp = datetime.datetime.fromtimestamp(			# from swh.model.model
	ts["timestamp"]["seconds"],			def ts2dt(ts: dict) -> datetime:
	datetime.timezone(datetime.timedelta(minutes=ts["offset"])),			timestamp = datetime.fromtimestamp(
				ts["timestamp"]["seconds"], timezone(timedelta(minutes=ts["offset"]))
	)			)
	return timestamp.replace(microsecond=ts["timestamp"]["microseconds"])			return timestamp.replace(microsecond=ts["timestamp"]["microseconds"])


	def test_provenance_origin_add(provenance, swh_storage_with_objects):			def test_provenance_origin_add(
				provenance: ProvenanceInterface, swh_storage_with_objects: Storage
				) -> None:
	"""Test the origin_add function"""			"""Test the origin_add function"""
	archive = ArchiveStorage(swh_storage_with_objects)			archive = ArchiveStorage(swh_storage_with_objects)
	for status in TEST_OBJECTS["origin_visit_status"]:			for status in TEST_OBJECTS["origin_visit_status"]:
				assert isinstance(status, OriginVisitStatus)
	if status.snapshot is not None:			if status.snapshot is not None:
				vlorentzUnsubmitted Done Inline Actions hmm, we should probably make TEST_OBJECTS a TypedDict in swh-model instead... vlorentz: hmm, we should probably make TEST_OBJECTS a TypedDict in swh-model instead...
				aevisoAuthorUnsubmitted Done Inline Actions I don't know the impact of such change, but I agree it is a good idea aeviso: I don't know the impact of such change, but I agree it is a good idea
	entry = OriginEntry(url=status.origin, snapshot=status.snapshot)			entry = OriginEntry(url=status.origin, snapshot=status.snapshot)
	origin_add(provenance, archive, [entry])			origin_add(provenance, archive, [entry])
	# TODO: check some facts here			# TODO: check some facts here


	def test_provenance_flavor(provenance):			def test_provenance_flavor(provenance: ProvenanceInterface) -> None:
				assert isinstance(provenance.storage, ProvenanceDBBase)
	assert provenance.storage.flavor in ("with-path", "without-path")			assert provenance.storage.flavor in ("with-path", "without-path")
				backend_class: Type[ProvenanceStorageInterface]
	if provenance.storage.flavor == "with-path":			if provenance.storage.flavor == "with-path":
	backend_class = ProvenanceWithPathDB			backend_class = ProvenanceWithPathDB
	else:			else:
	backend_class = ProvenanceWithoutPathDB			backend_class = ProvenanceWithoutPathDB
	assert isinstance(provenance.storage, backend_class)			assert isinstance(provenance.storage, backend_class)