Differential D3533 Diff 12485 swh/loader/package/archive/tests/test_archive.py

Changeset View

Standalone View

swh/loader/package/archive/tests/test_archive.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

from swh.model.hashutil import hash_to_bytes		from swh.model.hashutil import hash_to_bytes
		from swh.model.model import Snapshot, SnapshotBranch, TargetType

from swh.loader.package.archive.loader import ArchiveLoader		from swh.loader.package.archive.loader import ArchiveLoader
from swh.loader.package.tests.common import check_metadata_paths		from swh.loader.package.tests.common import check_metadata_paths
from swh.loader.tests import (		from swh.loader.tests import (
assert_last_visit_matches,		assert_last_visit_matches,
check_snapshot,		check_snapshot,
get_stats,		get_stats,
)		)
▲ Show 20 Lines • Show All 52 Lines • ▼ Show 20 Lines
]		]

_expected_new_revisions_first_visit = {		_expected_new_revisions_first_visit = {
"44183488c0774ce3c957fa19ba695cf18a4a42b3": (		"44183488c0774ce3c957fa19ba695cf18a4a42b3": (
"3aebc29ed1fccc4a6f2f2010fb8e57882406b528"		"3aebc29ed1fccc4a6f2f2010fb8e57882406b528"
)		)
}		}

_expected_branches_first_visit = {
b"HEAD": {"target_type": "alias", "target": b"releases/0.1.0",},
b"releases/0.1.0": {
"target_type": "revision",
"target": hash_to_bytes("44183488c0774ce3c957fa19ba695cf18a4a42b3"),
},
}

# hash is different then before as we changed the snapshot
# gnu used to use `release/` (singular) instead of plural
_expected_new_snapshot_first_visit_id = hash_to_bytes(
"c419397fd912039825ebdbea378bc6283f006bf5"
)


def visit_with_no_artifact_found(swh_config, requests_mock_datadir):		def visit_with_no_artifact_found(swh_config, requests_mock_datadir):
url = URL		url = URL
unknown_artifact_url = "https://ftp.g.o/unknown/8sync-0.1.0.tar.gz"		unknown_artifact_url = "https://ftp.g.o/unknown/8sync-0.1.0.tar.gz"
loader = ArchiveLoader(		loader = ArchiveLoader(
url,		url,
artifacts=[		artifacts=[
{		{
▲ Show 20 Lines • Show All 61 Lines • ▼ Show 20 Lines
def test_visit_with_release_artifact_no_prior_visit(swh_config, requests_mock_datadir):		def test_visit_with_release_artifact_no_prior_visit(swh_config, requests_mock_datadir):
"""With no prior visit, load a gnu project ends up with 1 snapshot		"""With no prior visit, load a gnu project ends up with 1 snapshot

"""		"""
loader = ArchiveLoader(url=URL, artifacts=GNU_ARTIFACTS)		loader = ArchiveLoader(url=URL, artifacts=GNU_ARTIFACTS)

actual_load_status = loader.load()		actual_load_status = loader.load()
assert actual_load_status["status"] == "eventful"		assert actual_load_status["status"] == "eventful"

		expected_snapshot_first_visit_id = hash_to_bytes(
		"c419397fd912039825ebdbea378bc6283f006bf5"
		)

assert (		assert (
hash_to_bytes(actual_load_status["snapshot_id"])		hash_to_bytes(actual_load_status["snapshot_id"])
== _expected_new_snapshot_first_visit_id		== expected_snapshot_first_visit_id
) # noqa		)

assert_last_visit_matches(loader.storage, URL, status="full", type="tar")		assert_last_visit_matches(loader.storage, URL, status="full", type="tar")

stats = get_stats(loader.storage)		stats = get_stats(loader.storage)
assert {		assert {
"content": len(_expected_new_contents_first_visit),		"content": len(_expected_new_contents_first_visit),
"directory": len(_expected_new_directories_first_visit),		"directory": len(_expected_new_directories_first_visit),
"origin": 1,		"origin": 1,
Show All 9 Lines	def test_visit_with_release_artifact_no_prior_visit(swh_config, requests_mock_datadir):
assert list(loader.storage.content_missing_per_sha1(expected_contents)) == []		assert list(loader.storage.content_missing_per_sha1(expected_contents)) == []

expected_dirs = map(hash_to_bytes, _expected_new_directories_first_visit)		expected_dirs = map(hash_to_bytes, _expected_new_directories_first_visit)
assert list(loader.storage.directory_missing(expected_dirs)) == []		assert list(loader.storage.directory_missing(expected_dirs)) == []

expected_revs = map(hash_to_bytes, _expected_new_revisions_first_visit)		expected_revs = map(hash_to_bytes, _expected_new_revisions_first_visit)
assert list(loader.storage.revision_missing(expected_revs)) == []		assert list(loader.storage.revision_missing(expected_revs)) == []

expected_snapshot = {		expected_snapshot = Snapshot(
"id": _expected_new_snapshot_first_visit_id,		id=expected_snapshot_first_visit_id,
"branches": _expected_branches_first_visit,		branches={
}		b"HEAD": SnapshotBranch(
		target_type=TargetType.ALIAS, target=b"releases/0.1.0",
		),
		b"releases/0.1.0": SnapshotBranch(
		target_type=TargetType.REVISION,
		target=hash_to_bytes("44183488c0774ce3c957fa19ba695cf18a4a42b3"),
		),
		},
		)

check_snapshot(expected_snapshot, loader.storage)		check_snapshot(expected_snapshot, loader.storage)


def test_2_visits_without_change(swh_config, requests_mock_datadir):		def test_2_visits_without_change(swh_config, requests_mock_datadir):
"""With no prior visit, load a gnu project ends up with 1 snapshot		"""With no prior visit, load a gnu project ends up with 1 snapshot

"""		"""
▲ Show 20 Lines • Show All 139 Lines • Show Last 20 Lines