Differential D5071 Diff 18179 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-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		import datetime
from typing import Iterator, List, Sequence, Tuple		from typing import Iterator, List, Sequence, Tuple

import attr		import attr
Show All 12 Lines	from swh.model.model import (
MetadataFetcher,		MetadataFetcher,
MetadataTargetType,		MetadataTargetType,
Person,		Person,
RawExtrinsicMetadata,		RawExtrinsicMetadata,
Revision,		Revision,
RevisionType,		RevisionType,
Sha1Git,		Sha1Git,
)		)
from swh.storage import get_storage

EMPTY_SNAPSHOT_ID = "1a8893e6a86f444e8be8e7bda6cb34fb1735a00e"		EMPTY_SNAPSHOT_ID = "1a8893e6a86f444e8be8e7bda6cb34fb1735a00e"
FULL_SNAPSHOT_ID = "4a9b608c9f01860a627237dd2409d1d50ec4b054"		FULL_SNAPSHOT_ID = "4a9b608c9f01860a627237dd2409d1d50ec4b054"

AUTHORITY = MetadataAuthority(		AUTHORITY = MetadataAuthority(
type=MetadataAuthorityType.FORGE, url="http://example.org/",		type=MetadataAuthorityType.FORGE, url="http://example.org/",
)		)
ORIGIN_URL = "http://example.org/archive.tgz"		ORIGIN_URL = "http://example.org/archive.tgz"
▲ Show 20 Lines • Show All 94 Lines • ▼ Show 20 Lines	def get_package_info(self, version: str) -> Iterator[Tuple[str, BasePackageInfo]]:

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

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


def test_load_artifact_metadata(swh_config, caplog):		def test_load_artifact_metadata(swh_storage, caplog):
storage = get_storage("memory")		loader = MetadataTestLoader(swh_storage, ORIGIN_URL)

loader = MetadataTestLoader(ORIGIN_URL)
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,
}		}

authority = MetadataAuthority(		authority = MetadataAuthority(
type=MetadataAuthorityType.REGISTRY, url="https://softwareheritage.org/",		type=MetadataAuthorityType.REGISTRY, url="https://softwareheritage.org/",
)		)

result = storage.raw_extrinsic_metadata_get(		result = swh_storage.raw_extrinsic_metadata_get(
MetadataTargetType.DIRECTORY, DIRECTORY_SWHID, authority,		MetadataTargetType.DIRECTORY, DIRECTORY_SWHID, authority,
)		)
assert result.next_page_token is None		assert result.next_page_token is None
assert len(result.results) == 1		assert len(result.results) == 1
assert result.results[0] == RawExtrinsicMetadata(		assert result.results[0] == RawExtrinsicMetadata(
type=MetadataTargetType.DIRECTORY,		type=MetadataTargetType.DIRECTORY,
target=DIRECTORY_SWHID,		target=DIRECTORY_SWHID,
discovery_date=result.results[0].discovery_date,		discovery_date=result.results[0].discovery_date,
authority=authority,		authority=authority,
fetcher=FETCHER,		fetcher=FETCHER,
format="original-artifacts-json",		format="original-artifacts-json",
metadata=b'[{"artifact_key": "value", "length": 0}]',		metadata=b'[{"artifact_key": "value", "length": 0}]',
origin=ORIGIN_URL,		origin=ORIGIN_URL,
revision=REVISION_SWHID,		revision=REVISION_SWHID,
)		)


def test_load_metadata(swh_config, caplog):		def test_load_metadata(swh_storage, caplog):
storage = get_storage("memory")		loader = MetadataTestLoader(swh_storage, ORIGIN_URL)

loader = MetadataTestLoader(ORIGIN_URL)
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.raw_extrinsic_metadata_get(		result = swh_storage.raw_extrinsic_metadata_get(
MetadataTargetType.DIRECTORY, DIRECTORY_SWHID, AUTHORITY,		MetadataTargetType.DIRECTORY, DIRECTORY_SWHID, AUTHORITY,
)		)
assert result.next_page_token is None		assert result.next_page_token is None
assert result.results == DIRECTORY_METADATA		assert result.results == DIRECTORY_METADATA

result = storage.raw_extrinsic_metadata_get(		result = swh_storage.raw_extrinsic_metadata_get(
MetadataTargetType.ORIGIN, ORIGIN_URL, AUTHORITY,		MetadataTargetType.ORIGIN, ORIGIN_URL, AUTHORITY,
)		)
assert result.next_page_token is None		assert result.next_page_token is None
assert result.results == ORIGIN_METADATA		assert result.results == ORIGIN_METADATA

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


def test_existing_authority(swh_config, caplog):		def test_existing_authority(swh_storage, caplog):
storage = get_storage("memory")		loader = MetadataTestLoader(swh_storage, ORIGIN_URL)

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

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.raw_extrinsic_metadata_get(		result = swh_storage.raw_extrinsic_metadata_get(
MetadataTargetType.DIRECTORY, DIRECTORY_SWHID, AUTHORITY,		MetadataTargetType.DIRECTORY, DIRECTORY_SWHID, AUTHORITY,
)		)
assert result.next_page_token is None		assert result.next_page_token is None
assert result.results == DIRECTORY_METADATA		assert result.results == DIRECTORY_METADATA

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


def test_existing_fetcher(swh_config, caplog):		def test_existing_fetcher(swh_storage, caplog):
storage = get_storage("memory")		loader = MetadataTestLoader(swh_storage, ORIGIN_URL)

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

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.raw_extrinsic_metadata_get(		result = swh_storage.raw_extrinsic_metadata_get(
MetadataTargetType.DIRECTORY, DIRECTORY_SWHID, AUTHORITY,		MetadataTargetType.DIRECTORY, DIRECTORY_SWHID, AUTHORITY,
)		)
assert result.next_page_token is None		assert result.next_page_token is None
assert result.results == DIRECTORY_METADATA		assert result.results == DIRECTORY_METADATA

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