Differential D6396 Diff 23253 swh/loader/package/maven/tests/test_maven.py

Changeset View

Standalone View

swh/loader/package/maven/tests/test_maven.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.maven.loader import MavenLoader, MavenPackageInfo
				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 = MavenLoader(
				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"
				vlorentzUnsubmitted Done Inline Actions please check the snapshot id vlorentz: please check the snapshot id
				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="maven"
				)


				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 = MavenLoader(
				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
				ardumontUnsubmitted Done Inline Actions please drop the print statements. ardumont: please drop the print statements.
				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="maven"
				)
				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,
				vlorentzUnsubmitted Done Inline Actions check this at the end of the test (this assertion is harder to debug than the other ones when it fails). Same comment above, to a lesser extent vlorentz: check this at the end of the test (this assertion is harder to debug than the other ones when…
				"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
				vlorentzUnsubmitted Done Inline Actions check the whole release in a single assertion. It makes sure the test isn't missing some checks, and it consistent with the other loaders' tests. vlorentz: check the whole release in a single assertion. It makes sure the test isn't missing some checks…
				borisbaldassariAuthorUnsubmitted Done Inline Actions Good point, thanks. borisbaldassari: Good point, thanks.
				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 = MavenLoader(
				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="maven"
				)

				actual_load_status2 = loader.load()
				ardumontUnsubmitted Done Inline Actions same about dropping print statements. ardumont: same about dropping print statements.
				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="maven"
				)

				urls = [
				m.url
				for m in requests_mock_datadir.request_history
				vlorentzUnsubmitted Done Inline Actions Why not an equality? eg. `assert requests_mock_datadir.request_history == ["https://repo1.maven.org/maven2/..."]` vlorentz: Why not an equality? eg. `assert requests_mock_datadir.request_history == ["https://repo1.
				borisbaldassariAuthorUnsubmitted Done Inline Actions Because `requests_mock_datadir.request_history` is a `requests_mock.request._RequestObjectProxy` object that cannot be compared to a list. It's ugly, though, so I've rewritten these lines for readability. Thanks. borisbaldassari: Because `requests_mock_datadir.request_history` is a `requests_mock.request.
				vlorentzUnsubmitted Done Inline Actions ping (btw, the reason is that, if there is an extra item at the beginning of the list, pytest makes it obvious. with the current code, it only shows the items don't match) vlorentz: ping (btw, the reason is that, if there is an extra item at the beginning of the list, pytest…
				borisbaldassariAuthorUnsubmitted Done Inline Actions This has been addressed in the lines below: assert requests_mock_datadir.request_history[0].url == MVN_ARTIFACTS[1]["url"] assert len(requests_mock_datadir.request_history) == 1 I don't know what I can improve there. As said above the equality just isn't possible as I cannot compare `requests_mock.request._RequestObjectProxy` to a list. But the second line ensures that equality is satisfied (and that there is no other item). Any idea? borisbaldassari: This has been addressed in the lines below: ``` assert requests_mock_datadir.
				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 = MavenLoader(
				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.maven.loader.MavenLoader", 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,)

				vlorentzUnsubmitted Done Inline Actions Sorry to be annoying with this, but I'd really like the metadata to be the original POM instead of this custom dictionary format, that is missing most of the original metadata (`<parent>`, `<dependencies>`, etc.) vlorentz: Sorry to be annoying with this, but I'd really like the metadata to be the original POM instead…
				borisbaldassariAuthorUnsubmitted Done Inline Actions @vlorentz you're not annoying. What is annoying though is the fact that some of my answers are lost (when rebasing I guess). :-) Anyway, I've done as suggested. borisbaldassari: @vlorentz you're not annoying. What is annoying though is the fact that some of my answers are…
				vlorentzUnsubmitted Done Inline Actions thanks! vlorentz: thanks!

				def test_jar_extid():
				"""Compute primary key should return the right identity

				"""

				metadata = MVN_ARTIFACTS[0]

				p_info = MavenPackageInfo(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 = MavenLoader(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="maven")

				# 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 = MavenLoader(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="maven")

				# 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