Differential D3667 Diff 12903 swh/loader/package/nixguix/tests/test_nixguix.py

Changeset View

Standalone View

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

Show All 9 Lines
import pytest		import pytest

from json.decoder import JSONDecodeError		from json.decoder import JSONDecodeError
from swh.storage.interface import StorageInterface		from swh.storage.interface import StorageInterface
from typing import Dict, Optional, Tuple		from typing import Dict, Optional, Tuple

from unittest.mock import patch		from unittest.mock import patch

from swh.model.model import Snapshot, SnapshotBranch, TargetType		from swh.model.identifiers import SWHID
		from swh.model.model import (
		MetadataAuthority,
		MetadataAuthorityType,
		MetadataFetcher,
		MetadataTargetType,
		RawExtrinsicMetadata,
		Snapshot,
		SnapshotBranch,
		TargetType,
		)
from swh.loader.package.archive.loader import ArchiveLoader		from swh.loader.package.archive.loader import ArchiveLoader
from swh.loader.package.nixguix.loader import (		from swh.loader.package.nixguix.loader import (
NixGuixPackageInfo,		NixGuixPackageInfo,
NixGuixLoader,		NixGuixLoader,
		parse_sources,
retrieve_sources,		retrieve_sources,
clean_sources,		clean_sources,
)		)

from swh.loader.package.utils import download		from swh.loader.package.utils import download
from swh.model.hashutil import hash_to_bytes, hash_to_hex		from swh.model.hashutil import hash_to_bytes, hash_to_hex
from swh.storage.exc import HashCollision		from swh.storage.exc import HashCollision
		from swh.storage.algos.origin import origin_get_latest_visit_status
		from swh.storage.interface import PagedResult

		from swh.loader.package import __version__

from swh.loader.tests import (		from swh.loader.tests import (
assert_last_visit_matches,		assert_last_visit_matches,
get_stats,		get_stats,
check_snapshot as check_snapshot_full,		check_snapshot as check_snapshot_full,
)		)


sources_url = "https://nix-community.github.io/nixpkgs-swh/sources.json"		sources_url = "https://nix-community.github.io/nixpkgs-swh/sources.json"


		@pytest.fixture
		def raw_sources(datadir) -> bytes:
		with open(
		os.path.join(
		datadir, "https_nix-community.github.io", "nixpkgs-swh_sources.json"
		),
		"rb",
		) as f:
		return f.read()


SNAPSHOT1 = Snapshot(		SNAPSHOT1 = Snapshot(
id=hash_to_bytes("0c5881c74283793ebe9a09a105a9381e41380383"),		id=hash_to_bytes("0c5881c74283793ebe9a09a105a9381e41380383"),
branches={		branches={
b"evaluation": SnapshotBranch(		b"evaluation": SnapshotBranch(
target=hash_to_bytes("cc4e04c26672dd74e5fd0fecb78b435fb55368f7"),		target=hash_to_bytes("cc4e04c26672dd74e5fd0fecb78b435fb55368f7"),
target_type=TargetType.REVISION,		target_type=TargetType.REVISION,
),		),
b"https://github.com/owner-1/repository-1/revision-1.tgz": SnapshotBranch(		b"https://github.com/owner-1/repository-1/revision-1.tgz": SnapshotBranch(
Show All 26 Lines	def check_snapshot(snapshot: Snapshot, storage: StorageInterface):
for rev in revisions:		for rev in revisions:
metadata = rev["metadata"]		metadata = rev["metadata"]
raw = metadata["extrinsic"]["raw"]		raw = metadata["extrinsic"]["raw"]
assert "url" in raw		assert "url" in raw
assert "integrity" in raw		assert "integrity" in raw


def test_retrieve_sources(swh_config, requests_mock_datadir):		def test_retrieve_sources(swh_config, requests_mock_datadir):
j = retrieve_sources(sources_url)		j = parse_sources(retrieve_sources(sources_url))
assert "sources" in j.keys()		assert "sources" in j.keys()
assert len(j["sources"]) == 2		assert len(j["sources"]) == 2


def test_retrieve_non_existing(swh_config, requests_mock_datadir):		def test_retrieve_non_existing(swh_config, requests_mock_datadir):
with pytest.raises(ValueError):		with pytest.raises(ValueError):
NixGuixLoader("https://non-existing-url")		NixGuixLoader("https://non-existing-url")

Show All 39 Lines	sources = {
],		],
"revision": "my-revision",		"revision": "my-revision",
}		}
clean = clean_sources(sources)		clean = clean_sources(sources)

assert len(clean["sources"]) == 1		assert len(clean["sources"]) == 1


def test_loader_one_visit(swh_config, requests_mock_datadir):		def test_loader_one_visit(swh_config, requests_mock_datadir, raw_sources):
loader = NixGuixLoader(sources_url)		loader = NixGuixLoader(sources_url)
res = loader.load()		res = loader.load()
assert res["status"] == "eventful"		assert res["status"] == "eventful"

stats = get_stats(loader.storage)		stats = get_stats(loader.storage)
assert {		assert {
"content": 1,		"content": 1,
"directory": 3,		"directory": 3,
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"person": 1,		"person": 1,
"release": 0,		"release": 0,
"revision": 1,		"revision": 1,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 1,		"snapshot": 1,
} == stats		} == stats

# The visit is partial because urls pointing to non tarball file		# The visit is partial because urls pointing to non tarball file
# are not handled yet		# are not handled yet
assert_last_visit_matches(		assert_last_visit_matches(
loader.storage, sources_url, status="partial", type="nixguix"		loader.storage, sources_url, status="partial", type="nixguix"
)		)

		(_, visit_status) = origin_get_latest_visit_status(loader.storage, sources_url)
		snapshot_swhid = SWHID(
		object_type="snapshot", object_id=hash_to_hex(visit_status.snapshot)
		)
		metadata_authority = MetadataAuthority(
		type=MetadataAuthorityType.FORGE, url=sources_url,
		)
		expected_metadata = [
		RawExtrinsicMetadata(
		type=MetadataTargetType.SNAPSHOT,
		id=snapshot_swhid,
		authority=metadata_authority,
		fetcher=MetadataFetcher(
		name="swh.loader.package.nixguix.loader.NixGuixLoader",
		version=__version__,
		),
		discovery_date=loader.visit_date,
		format="nixguix-sources-json",
		metadata=raw_sources,
		origin=sources_url,
		)
		]
		assert loader.storage.raw_extrinsic_metadata_get(
		type=MetadataTargetType.SNAPSHOT,
		id=snapshot_swhid,
		authority=metadata_authority,
		) == PagedResult(next_page_token=None, results=expected_metadata,)


def test_uncompress_failure(swh_config, requests_mock_datadir):		def test_uncompress_failure(swh_config, requests_mock_datadir):
"""Non tarball files are currently not supported and the uncompress		"""Non tarball files are currently not supported and the uncompress
function fails on such kind of files.		function fails on such kind of files.

However, even in this case of failure (because of the url		However, even in this case of failure (because of the url
https://example.com/file.txt), a snapshot and a visit has to be		https://example.com/file.txt), a snapshot and a visit has to be
created (with a status partial since all files are not archived).		created (with a status partial since all files are not archived).
▲ Show 20 Lines • Show All 129 Lines • ▼ Show 20 Lines	assert {
"person": 1,		"person": 1,
"release": 0,		"release": 0,
"revision": 2,		"revision": 2,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 2,		"snapshot": 2,
} == stats		} == stats


def test_resolve_revision_from(swh_config, requests_mock_datadir):		def test_resolve_revision_from(swh_config, requests_mock_datadir, datadir):
loader = NixGuixLoader(sources_url)		loader = NixGuixLoader(sources_url)

known_artifacts = {		known_artifacts = {
"id1": {"extrinsic": {"raw": {"url": "url1", "integrity": "integrity1"}}},		"id1": {"extrinsic": {"raw": {"url": "url1", "integrity": "integrity1"}}},
"id2": {"extrinsic": {"raw": {"url": "url2", "integrity": "integrity2"}}},		"id2": {"extrinsic": {"raw": {"url": "url2", "integrity": "integrity2"}}},
}		}

p_info = NixGuixPackageInfo.from_metadata(		p_info = NixGuixPackageInfo.from_metadata(
▲ Show 20 Lines • Show All 221 Lines • Show Last 20 Lines