Page MenuHomeSoftware Heritage
Differential D6158 Diff 22591 swh/loader/package/jar/tests/test_jar.py

Changeset View

Standalone View

View Options

swh/loader/package/jar/tests/test_jar.py

  • This file was added.
# Copyright (C) 2019-2021 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
import hashlib
import json
from pathlib import Path
import string
import pytest
from swh.loader.package import __version__
from swh.loader.package.jar.loader import JarLoader, JarPackageInfo
from swh.loader.tests import assert_last_visit_matches, check_snapshot, get_stats
from swh.model.hashutil import hash_to_bytes
from swh.model.identifiers import (
CoreSWHID,
ExtendedObjectType,
ExtendedSWHID,
ObjectType,
)
from swh.model.model import (
MetadataAuthority,
MetadataAuthorityType,
MetadataFetcher,
Person,
RawExtrinsicMetadata,
RevisionType,
Snapshot,
SnapshotBranch,
TargetType,
Timestamp,
TimestampWithTimezone,
)
from swh.storage.algos.snapshot import snapshot_get_all_branches
from swh.storage.interface import PagedResult
URL = "https://repo1.maven.org/maven2/"
MVN_ARTIFACTS = [
{
"time": 1626109619335,
"url": "https://repo1.maven.org/maven2/al/aldi/sprova4j/0.1.0/"
+ "sprova4j-0.1.0.jar",
"gid": "al.aldi",
"aid": "sprova4j",
"filename": "sprova4j-0.1.0.jar",
"version": "0.1.0",
},
{
"time": 1626111425534,
"url": "https://repo1.maven.org/maven2/al/aldi/sprova4j/0.1.1/"
+ "sprova4j-0.1.1.jar",
"gid": "al.aldi",
"aid": "sprova4j",
"filename": "sprova4j-0.1.1.jar",
"version": "0.1.1",
},
]
SWH_PERSON = Person(
name=b"Software Heritage",
fullname=b"Software Heritage",
email=b"robot@softwareheritage.org",
)
REVISION_MESSAGE = b"swh-loader-package: synthetic revision message"
REVISION_DATE = TimestampWithTimezone(
timestamp=Timestamp(seconds=1626109619, microseconds=335000),
offset=0,
negative_utc=False,
)
_expected_new_contents_first_visit = [
"cd807364cd7730022b3849f90ccf4bababbada84",
"79e33dd52ebdf615e6696ae69add91cb990d81e2",
"8002bd514156f05a0940ae14ef86eb0179cbd510",
"23479553a6ccec30d377dee0496123a65d23fd8c",
"07ffbebb933bc1660e448f07d8196c2b083797f9",
"abf021b581f80035b56153c9aa27195b8d7ebbb8",
"eec70ba80a6862ed2619727663b17eb0d9dfe131",
"81a493dacb44dedf623f29ecf62c0e035bf698de",
"bda85ed0bbecf8cddfea04234bee16f476f64fe4",
"1ec91d561f5bdf59acb417086e04c54ead94e94e",
"d517b423da707fa21378623f35facebff53cb59d",
"3f0f21a764972d79e583908991c893c999613354",
"a2dd4d7dfe6043baf9619081e4e29966989211af",
"f62685cf0c6825a4097c949280b584cf0e16d047",
"56afc1ea60cef6548ce0a34f44e91b0e4b063835",
"cf7c740926e7ebc9ac8978a5c4f0e1e7a0e9e3af",
"86ff828bea1c22ca3d50ed82569b9c59ce2c41a1",
"1d0fa04454d9fec31d8ee3f35b58158ca1e28b15",
"e90239a2c8d9ede61a29671a8b397a743e18fa34",
"ce8851005d084aea089bcd8cf01052f4b234a823",
"2c34ce622aa7fa68d104900840f66671718e6249",
"e6a6fec32dcb3bee93c34fc11b0174a6b0b0ec6d",
"405d3e1be4b658bf26de37f2c90c597b2796b9d7",
"d0d2f5848721e04300e537826ef7d2d6d9441df0",
"399c67e33e38c475fd724d283dd340f6a2e8dc91",
"dea10c1111cc61ac1809fb7e88857e3db054959f",
]
_expected_new_directories_first_visit = [
"6c9de41e4cebb91a8368da1d89ae9873bd540ec3",
"c1a2ee97fc47426d0179f94d223405336b5cd075",
"9e1bdca292765a9528af18743bd793b80362c768",
"193a7af634592ef27fb341762806f61e8fb8eab3",
"a297aa21e3dbf138b370be3aae7a852dd403bbbb",
"da84026119ae04022f007d5b3362e98d46d09045",
"75bb915942a9c441ca62aeffc3b634f1ec9ce5e2",
"0851d359283b2ad82b116c8d1b55ab14b1ec219c",
"2bcbb8b723a025ee9a36b719cea229ed38c37e46",
]
_expected_new_revisions_first_visit = {"67949049e900211db06cfd1c474c6ad40d23c186"}
@pytest.fixture
def data_jar_1(datadir):
content = Path(
datadir, "https_maven.org", "sprova4j-0.1.0-sources.jar"
).read_bytes()
return content
@pytest.fixture
def data_jar_2(datadir):
content = Path(
datadir, "https_maven.org", "sprova4j-0.1.1-sources.jar"
).read_bytes()
return content
def test_jar_visit_with_no_artifact_found(swh_storage, requests_mock_datadir):
unknown_artifact_url = "https://ftp.g.o/unknown/8sync-0.1.0.tar.gz"
loader = JarLoader(
swh_storage,
unknown_artifact_url,
artifacts=[
{
"time": 1626111944729,
"url": unknown_artifact_url, # unknown artifact
"filename": "8sync-0.1.0.tar.gz",
"gid": "al/aldi",
"aid": "sprova4j",
"version": "0.1.0",
}
],
)
actual_load_status = loader.load()
assert actual_load_status["status"] == "uneventful"
assert actual_load_status["snapshot_id"] is not None
stats = get_stats(swh_storage)
assert {
"content": 0,
"directory": 0,
"origin": 1,
"origin_visit": 1,
"release": 0,
"revision": 0,
"skipped_content": 0,
"snapshot": 1,
} == stats
assert_last_visit_matches(
swh_storage, unknown_artifact_url, status="partial", type="jar"
)
def test_jar_visit_with_release_artifact_no_prior_visit(
swh_storage, requests_mock, data_jar_1
):
"""With no prior visit, loading a jar ends up with 1 snapshot
"""
requests_mock.get(MVN_ARTIFACTS[0]["url"], content=data_jar_1)
loader = JarLoader(
swh_storage, MVN_ARTIFACTS[0]["url"], artifacts=[MVN_ARTIFACTS[0]]
)
actual_load_status = loader.load()
assert actual_load_status["status"] == "eventful"
expected_snapshot_first_visit_id = hash_to_bytes(
"ae8f76fe879ea040312182c02dae324a11252f83"
)
expected_snapshot = Snapshot(
id=expected_snapshot_first_visit_id,
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("67949049e900211db06cfd1c474c6ad40d23c186"),
),
},
)
actual_snapshot = snapshot_get_all_branches(
swh_storage, hash_to_bytes(actual_load_status["snapshot_id"])
)
assert actual_snapshot == expected_snapshot
check_snapshot(expected_snapshot, swh_storage)
assert (
hash_to_bytes(actual_load_status["snapshot_id"])
== expected_snapshot_first_visit_id
)
stats = get_stats(swh_storage)
assert_last_visit_matches(
swh_storage, MVN_ARTIFACTS[0]["url"], status="full", type="jar"
)
assert {
"content": len(_expected_new_contents_first_visit),
"directory": len(_expected_new_directories_first_visit),
"origin": 1,
"origin_visit": 1,
"release": 0,
"revision": len(_expected_new_revisions_first_visit),
"skipped_content": 0,
"snapshot": 1,
} == stats
expected_contents = map(hash_to_bytes, _expected_new_contents_first_visit)
assert list(swh_storage.content_missing_per_sha1(expected_contents)) == []
expected_dirs = map(hash_to_bytes, _expected_new_directories_first_visit)
assert list(swh_storage.directory_missing(expected_dirs)) == []
expected_revs = map(hash_to_bytes, _expected_new_revisions_first_visit)
assert list(swh_storage.revision_missing(expected_revs)) == []
rev_id = actual_snapshot.branches[b"releases/0.1.0"].target
(rev,) = swh_storage.revision_get([rev_id])
assert rev.message == b"swh-loader-package: synthetic revision message"
assert rev.author == SWH_PERSON
assert rev.committer == SWH_PERSON
assert rev.date == REVISION_DATE
assert rev.committer_date == REVISION_DATE
assert rev.type == RevisionType.TAR
assert rev.directory == hash_to_bytes("6c9de41e4cebb91a8368da1d89ae9873bd540ec3")
assert rev.synthetic
def test_jar_2_visits_without_change(
swh_storage, requests_mock_datadir, requests_mock, data_jar_2
):
"""With no prior visit, load a gnu project ends up with 1 snapshot
"""
requests_mock.get(MVN_ARTIFACTS[1]["url"], content=data_jar_2)
loader = JarLoader(
swh_storage, MVN_ARTIFACTS[1]["url"], artifacts=[MVN_ARTIFACTS[1]]
)
actual_load_status = loader.load()
assert actual_load_status["status"] == "eventful"
expected_snapshot_first_visit_id = hash_to_bytes(
"e085b3b8b21b5aca029a8f4fabc65893173fcf20"
)
assert (
hash_to_bytes(actual_load_status["snapshot_id"])
== expected_snapshot_first_visit_id
)
assert_last_visit_matches(
swh_storage, MVN_ARTIFACTS[1]["url"], status="full", type="jar"
)
actual_load_status2 = loader.load()
assert actual_load_status2["status"] == "uneventful"
assert actual_load_status2["snapshot_id"] is not None
assert actual_load_status["snapshot_id"] == actual_load_status2["snapshot_id"]
assert_last_visit_matches(
swh_storage, MVN_ARTIFACTS[1]["url"], status="full", type="jar"
)
urls = [
m.url
for m in requests_mock_datadir.request_history
if m.url.startswith("https://repo1.maven.org/maven2")
]
assert len(urls) == 1
def test_metadatata(swh_storage, requests_mock, data_jar_1):
"""With no prior visit, loading a jar ends up with 1 snapshot
"""
requests_mock.get(MVN_ARTIFACTS[0]["url"], content=data_jar_1)
loader = JarLoader(
swh_storage, MVN_ARTIFACTS[0]["url"], artifacts=[MVN_ARTIFACTS[0]]
)
actual_load_status = loader.load()
assert actual_load_status["status"] == "eventful"
actual_snapshot = snapshot_get_all_branches(
swh_storage, hash_to_bytes(actual_load_status["snapshot_id"])
)
print(f"revision is {actual_snapshot.branches[b'releases/0.1.0'].target.hex()}.")
expected_revision_id = hash_to_bytes("67949049e900211db06cfd1c474c6ad40d23c186")
revision = swh_storage.revision_get([expected_revision_id])[0]
assert revision is not None
# assert actual_load_status["snapshot_id"] is not None
revision_swhid = CoreSWHID(
object_type=ObjectType.REVISION, object_id=expected_revision_id
)
directory_swhid = ExtendedSWHID(
object_type=ExtendedObjectType.DIRECTORY, object_id=revision.directory
)
metadata_authority = MetadataAuthority(
type=MetadataAuthorityType.FORGE, url="https://repo1.maven.org/",
)
expected_metadata = [
RawExtrinsicMetadata(
target=directory_swhid,
authority=metadata_authority,
fetcher=MetadataFetcher(
name="swh.loader.package.jar.loader.JarLoader", version=__version__,
),
discovery_date=loader.visit_date,
format="maven-pom-json",
metadata=json.dumps(
[{"gid": "al.aldi", "aid": "sprova4j", "version": "0.1.0"}]
).encode(),
origin=MVN_ARTIFACTS[0]["url"],
revision=revision_swhid,
)
]
assert swh_storage.raw_extrinsic_metadata_get(
directory_swhid, metadata_authority,
) == PagedResult(next_page_token=None, results=expected_metadata,)
def test_jar_extid():
"""Compute primary key should return the right identity
"""
metadata = MVN_ARTIFACTS[0]
p_info = JarPackageInfo(raw_info={**metadata}, **metadata)
expected_manifest = b"al.aldi sprova4j 0.1.0"
for manifest_format in [
string.Template("$aid $gid $version"),
string.Template("$gid $aid"),
]:
actual_id = p_info.extid(manifest_format=manifest_format)
assert actual_id != ("maven-jar", hashlib.sha256(expected_manifest).digest(),)
for manifest_format, expected_manifest in [
(None, "{gid} {aid} {version}".format(**metadata).encode()),
]:
actual_id = p_info.extid(manifest_format=manifest_format)
assert actual_id == ("maven-jar", hashlib.sha256(expected_manifest).digest(),)
with pytest.raises(KeyError):
p_info.extid(manifest_format=string.Template("$a $unknown_key"))
def test_jar_snapshot_append(
swh_storage, requests_mock_datadir, requests_mock, data_jar_1, data_jar_2
):
# first loading with a first artifact
artifact1 = MVN_ARTIFACTS[0]
url1 = artifact1["url"]
requests_mock.get(url1, content=data_jar_1)
loader = JarLoader(swh_storage, url1, [artifact1], snapshot_append=True)
actual_load_status = loader.load()
assert actual_load_status["status"] == "eventful"
assert actual_load_status["snapshot_id"] is not None
assert_last_visit_matches(swh_storage, url1, status="full", type="jar")
# check expected snapshot
snapshot = loader.last_snapshot()
assert len(snapshot.branches) == 2
branch_artifact1_name = f"releases/{artifact1['version']}".encode()
assert b"HEAD" in snapshot.branches
assert branch_artifact1_name in snapshot.branches
assert snapshot.branches[b"HEAD"].target == branch_artifact1_name
# second loading with a second artifact
artifact2 = MVN_ARTIFACTS[1]
url2 = artifact2["url"]
requests_mock.get(url2, content=data_jar_2)
loader = JarLoader(swh_storage, url2, [artifact2], snapshot_append=True)
actual_load_status = loader.load()
assert actual_load_status["status"] == "eventful"
assert actual_load_status["snapshot_id"] is not None
assert_last_visit_matches(swh_storage, url2, status="full", type="jar")
# check expected snapshot, should contain a new branch and the
# branch for the first artifact
snapshot = loader.last_snapshot()
assert len(snapshot.branches) == 2
branch_artifact2_name = f"releases/{artifact2['version']}".encode()
assert b"HEAD" in snapshot.branches
assert branch_artifact2_name in snapshot.branches
assert branch_artifact1_name not in snapshot.branches
assert snapshot.branches[b"HEAD"].target == branch_artifact2_name
About · Developer information · Legal