Differential D6607 Diff 24003 swh/loader/package/nixguix/tests/test_nixguix.py

Changeset View

Standalone View

swh/loader/package/nixguix/tests/test_nixguix.py

Show First 20 Lines • Show All 530 Lines • ▼ Show 20 Lines	def test_raise_exception(swh_storage, requests_mock_datadir, mocker):
loader = NixGuixLoader(swh_storage, sources_url)		loader = NixGuixLoader(swh_storage, sources_url)
res = loader.load()		res = loader.load()

assert res == {		assert res == {
"status": "eventful",		"status": "eventful",
"snapshot_id": SNAPSHOT1.id.hex(),		"snapshot_id": SNAPSHOT1.id.hex(),
}		}

check_snapshot(SNAPSHOT1, storage=swh_storage)

assert len(mock_download.mock_calls) == 2

# The visit is partial because some artifact downloads failed		# The visit is partial because some artifact downloads failed
assert_last_visit_matches(		assert_last_visit_matches(
swh_storage, sources_url, status="partial", type="nixguix"		swh_storage,
		sources_url,
		status="partial",
		type="nixguix",
		snapshot=SNAPSHOT1.id,
)		)

		check_snapshot(SNAPSHOT1, storage=swh_storage)

		assert len(mock_download.mock_calls) == 2


def test_load_nixguix_one_common_artifact_from_other_loader(		def test_load_nixguix_one_common_artifact_from_other_loader(
swh_storage, datadir, requests_mock_datadir_visits, caplog		swh_storage, datadir, requests_mock_datadir_visits, caplog
):		):
"""Misformatted revision should be caught and logged, then loading continues		"""Misformatted revision should be caught and logged, then loading continues

"""		"""
caplog.set_level(logging.ERROR, "swh.loader.package.nixguix.loader")		caplog.set_level(logging.ERROR, "swh.loader.package.nixguix.loader")
Show All 13 Lines	):
]		]
archive_loader = ArchiveLoader(swh_storage, url=gnu_url, artifacts=gnu_artifacts)		archive_loader = ArchiveLoader(swh_storage, url=gnu_url, artifacts=gnu_artifacts)
actual_load_status = archive_loader.load()		actual_load_status = archive_loader.load()
expected_snapshot_id = "c419397fd912039825ebdbea378bc6283f006bf5"		expected_snapshot_id = "c419397fd912039825ebdbea378bc6283f006bf5"
assert actual_load_status["status"] == "eventful"		assert actual_load_status["status"] == "eventful"
assert actual_load_status["snapshot_id"] == expected_snapshot_id # noqa		assert actual_load_status["snapshot_id"] == expected_snapshot_id # noqa

assert_last_visit_matches(		assert_last_visit_matches(
archive_loader.storage, gnu_url, status="full", type="tar"		archive_loader.storage,
		gnu_url,
		status="full",
		type="tar",
		snapshot=hash_to_bytes(expected_snapshot_id),
)		)

# 2. Then ingest with the nixguix loader which lists the same artifact within its		# 2. Then ingest with the nixguix loader which lists the same artifact within its
# sources.json		# sources.json

# ensure test setup is ok		# ensure test setup is ok
data_sources = os.path.join(		data_sources = os.path.join(
datadir, "https_nix-community.github.io", "nixpkgs-swh_sources_special.json"		datadir, "https_nix-community.github.io", "nixpkgs-swh_sources_special.json"
)		)
all_sources = json.loads(open(data_sources).read())		all_sources = json.loads(open(data_sources).read())
found = False		found = False
for source in all_sources["sources"]:		for source in all_sources["sources"]:
if source["urls"][0] == artifact_url:		if source["urls"][0] == artifact_url:
found = True		found = True
assert (		assert (
found is True		found is True
), f"test setup error: {artifact_url} must be in {data_sources}"		), f"test setup error: {artifact_url} must be in {data_sources}"

# first visit with a snapshot, ok		# first visit with a snapshot, ok
sources_url = "https://nix-community.github.io/nixpkgs-swh/sources_special.json"		sources_url = "https://nix-community.github.io/nixpkgs-swh/sources_special.json"
loader = NixGuixLoader(swh_storage, sources_url)		loader = NixGuixLoader(swh_storage, sources_url)
actual_load_status2 = loader.load()		actual_load_status2 = loader.load()
assert actual_load_status2["status"] == "eventful"		assert actual_load_status2["status"] == "eventful"

assert_last_visit_matches(swh_storage, sources_url, status="full", type="nixguix")

snapshot_id = actual_load_status2["snapshot_id"]		snapshot_id = actual_load_status2["snapshot_id"]

		assert_last_visit_matches(
		swh_storage,
		sources_url,
		status="full",
		type="nixguix",
		snapshot=hash_to_bytes(snapshot_id),
		)

snapshot = snapshot_get_all_branches(swh_storage, hash_to_bytes(snapshot_id))		snapshot = snapshot_get_all_branches(swh_storage, hash_to_bytes(snapshot_id))
assert snapshot		assert snapshot