Differential D3616 Diff 12747 swh/loader/package/tests/test_loader_metadata.py

Changeset View

Standalone View

swh/loader/package/tests/test_loader_metadata.py

	# Copyright (C) 2019-2020 The Software Heritage developers			# Copyright (C) 2019-2020 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			import datetime
	from typing import Iterator, Optional, Sequence, Tuple			from typing import Iterator, List, Optional, Sequence, Tuple

	import attr			import attr

	from swh.loader.package.loader import (			from swh.loader.package.loader import (
	BasePackageInfo,			BasePackageInfo,
	PackageLoader,			PackageLoader,
	RawExtrinsicMetadataCore,			RawExtrinsicMetadataCore,
	)			)
	Show All 23 Lines
	REVISION_SWHID = SWHID(object_type="revision", object_id=REVISION_ID)			REVISION_SWHID = SWHID(object_type="revision", object_id=REVISION_ID)


	FETCHER = MetadataFetcher(			FETCHER = MetadataFetcher(
	name="swh.loader.package.tests.test_loader_metadata.MetadataTestLoader",			name="swh.loader.package.tests.test_loader_metadata.MetadataTestLoader",
	version=__version__,			version=__version__,
	)			)

	METADATA = [			REVISION_METADATA = [
	RawExtrinsicMetadata(			RawExtrinsicMetadata(
	type=MetadataTargetType.REVISION,			type=MetadataTargetType.REVISION,
	id=REVISION_SWHID,			id=REVISION_SWHID,
	discovery_date=datetime.datetime.now(),			discovery_date=datetime.datetime.now(),
	authority=AUTHORITY,			authority=AUTHORITY,
	fetcher=FETCHER,			fetcher=FETCHER,
	format="test-format1",			format="test-format1",
	metadata=b"foo bar",			metadata=b"foo bar",
	origin=ORIGIN_URL,			origin=ORIGIN_URL,
	),			),
	RawExtrinsicMetadata(			RawExtrinsicMetadata(
	type=MetadataTargetType.REVISION,			type=MetadataTargetType.REVISION,
	id=REVISION_SWHID,			id=REVISION_SWHID,
	discovery_date=datetime.datetime.now() + datetime.timedelta(seconds=1),			discovery_date=datetime.datetime.now() + datetime.timedelta(seconds=1),
	authority=AUTHORITY,			authority=AUTHORITY,
	fetcher=FETCHER,			fetcher=FETCHER,
	format="test-format2",			format="test-format2",
	metadata=b"bar baz",			metadata=b"bar baz",
	origin=ORIGIN_URL,			origin=ORIGIN_URL,
	),			),
	]			]

				ORIGIN_METADATA = [
				RawExtrinsicMetadata(
				type=MetadataTargetType.ORIGIN,
				id=ORIGIN_URL,
				discovery_date=datetime.datetime.now(),
				authority=AUTHORITY,
				fetcher=FETCHER,
				format="test-format3",
				metadata=b"baz qux",
				),
				]


	class MetadataTestLoader(PackageLoader[BasePackageInfo]):			class MetadataTestLoader(PackageLoader[BasePackageInfo]):
	def get_versions(self) -> Sequence[str]:			def get_versions(self) -> Sequence[str]:
	return ["v1.0.0"]			return ["v1.0.0"]

	def _load_revision(self, p_info: BasePackageInfo, origin) -> Optional[Sha1Git]:			def _load_revision(self, p_info: BasePackageInfo, origin) -> Optional[Sha1Git]:
	return REVISION_ID			return REVISION_ID

	def get_metadata_authority(self):			def get_metadata_authority(self):
	return attr.evolve(AUTHORITY, metadata={})			return attr.evolve(AUTHORITY, metadata={})

	def get_package_info(self, version: str) -> Iterator[Tuple[str, BasePackageInfo]]:			def get_package_info(self, version: str) -> Iterator[Tuple[str, BasePackageInfo]]:
	m0 = METADATA[0]			m0 = REVISION_METADATA[0]
	m1 = METADATA[1]			m1 = REVISION_METADATA[1]
	p_info = BasePackageInfo(			p_info = BasePackageInfo(
	url=ORIGIN_URL,			url=ORIGIN_URL,
	filename="archive.tgz",			filename="archive.tgz",
	revision_extrinsic_metadata=[			revision_extrinsic_metadata=[
	RawExtrinsicMetadataCore(m0.format, m0.metadata, m0.discovery_date),			RawExtrinsicMetadataCore(m0.format, m0.metadata, m0.discovery_date),
	RawExtrinsicMetadataCore(m1.format, m1.metadata, m1.discovery_date),			RawExtrinsicMetadataCore(m1.format, m1.metadata, m1.discovery_date),
	],			],
	)			)

	yield (version, p_info)			yield (version, p_info)

				def get_extrinsic_origin_metadata(self) -> List[RawExtrinsicMetadataCore]:
				m = ORIGIN_METADATA[0]
				return [RawExtrinsicMetadataCore(m.format, m.metadata, m.discovery_date)]


	def test_load_revision_metadata(swh_config, caplog):			def test_load_metadata(swh_config, caplog):
	storage = get_storage("memory")			storage = get_storage("memory")

	loader = MetadataTestLoader(ORIGIN_URL)			loader = MetadataTestLoader(ORIGIN_URL)
	loader.storage = storage			loader.storage = storage

	load_status = loader.load()			load_status = loader.load()
	assert load_status == {			assert load_status == {
	"status": "eventful",			"status": "eventful",
	"snapshot_id": FULL_SNAPSHOT_ID,			"snapshot_id": FULL_SNAPSHOT_ID,
	}			}

	result = storage.object_metadata_get(			result = storage.object_metadata_get(
	MetadataTargetType.REVISION, REVISION_SWHID, AUTHORITY,			MetadataTargetType.REVISION, REVISION_SWHID, AUTHORITY,
	)			)
	assert result["next_page_token"] is None			assert result["next_page_token"] is None
	assert result["results"] == METADATA			assert result["results"] == REVISION_METADATA

				result = storage.object_metadata_get(
				MetadataTargetType.ORIGIN, ORIGIN_URL, AUTHORITY,
				)
				assert result["next_page_token"] is None
				assert result["results"] == ORIGIN_METADATA

	assert caplog.text == ""			assert caplog.text == ""


	def test_existing_authority(swh_config, caplog):			def test_existing_authority(swh_config, caplog):
	storage = get_storage("memory")			storage = get_storage("memory")

	loader = MetadataTestLoader(ORIGIN_URL)			loader = MetadataTestLoader(ORIGIN_URL)
	loader.storage = storage			loader.storage = storage
	loader.config["create_authorities"] = False			loader.config["create_authorities"] = False

	storage.metadata_authority_add([attr.evolve(AUTHORITY, metadata={})])			storage.metadata_authority_add([attr.evolve(AUTHORITY, metadata={})])

	load_status = loader.load()			load_status = loader.load()
	assert load_status == {			assert load_status == {
	"status": "eventful",			"status": "eventful",
	"snapshot_id": FULL_SNAPSHOT_ID,			"snapshot_id": FULL_SNAPSHOT_ID,
	}			}

	result = storage.object_metadata_get(			result = storage.object_metadata_get(
	MetadataTargetType.REVISION, REVISION_SWHID, AUTHORITY,			MetadataTargetType.REVISION, REVISION_SWHID, AUTHORITY,
	)			)
	assert result["next_page_token"] is None			assert result["next_page_token"] is None
	assert result["results"] == METADATA			assert result["results"] == REVISION_METADATA

	assert caplog.text == ""			assert caplog.text == ""


	def test_existing_fetcher(swh_config, caplog):			def test_existing_fetcher(swh_config, caplog):
	storage = get_storage("memory")			storage = get_storage("memory")

	loader = MetadataTestLoader(ORIGIN_URL)			loader = MetadataTestLoader(ORIGIN_URL)
	loader.storage = storage			loader.storage = storage
	loader.config["create_fetchers"] = False			loader.config["create_fetchers"] = False

	storage.metadata_fetcher_add([attr.evolve(FETCHER, metadata={})])			storage.metadata_fetcher_add([attr.evolve(FETCHER, metadata={})])

	load_status = loader.load()			load_status = loader.load()
	assert load_status == {			assert load_status == {
	"status": "eventful",			"status": "eventful",
	"snapshot_id": FULL_SNAPSHOT_ID,			"snapshot_id": FULL_SNAPSHOT_ID,
	}			}

	result = storage.object_metadata_get(			result = storage.object_metadata_get(
	MetadataTargetType.REVISION, REVISION_SWHID, AUTHORITY,			MetadataTargetType.REVISION, REVISION_SWHID, AUTHORITY,
	)			)
	assert result["next_page_token"] is None			assert result["next_page_token"] is None
	assert result["results"] == METADATA			assert result["results"] == REVISION_METADATA

	assert caplog.text == ""			assert caplog.text == ""