Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/npm/tests/test_npm.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 | ||||
| import json | import json | ||||
| import os | import os | ||||
| import pytest | import pytest | ||||
| from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
| from swh.model.model import Person | from swh.model.model import Person, Snapshot, SnapshotBranch, TargetType | ||||
| from swh.loader.package.npm.loader import ( | from swh.loader.package.npm.loader import ( | ||||
| NpmLoader, | NpmLoader, | ||||
| extract_npm_package_author, | extract_npm_package_author, | ||||
| artifact_to_revision_id, | artifact_to_revision_id, | ||||
| ) | ) | ||||
| 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 ( | ||||
| ▲ Show 20 Lines • Show All 259 Lines • ▼ Show 20 Lines | def test_revision_metadata_structure(swh_config, requests_mock_datadir): | ||||
| for original_artifact in revision["metadata"]["original_artifact"]: | for original_artifact in revision["metadata"]["original_artifact"]: | ||||
| check_metadata_paths( | check_metadata_paths( | ||||
| original_artifact, | original_artifact, | ||||
| paths=[("filename", str), ("length", int), ("checksums", dict),], | paths=[("filename", str), ("length", int), ("checksums", dict),], | ||||
| ) | ) | ||||
| def test_npm_loader_first_visit(swh_config, requests_mock_datadir): | def test_npm_loader_first_visit(swh_config, requests_mock_datadir): | ||||
| package = "org" | package = "org" | ||||
| loader = NpmLoader(package_url(package)) | url = package_url(package) | ||||
| loader = NpmLoader(url) | |||||
| actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
| expected_snapshot_id = "d0587e1195aed5a8800411a008f2f2d627f18e2d" | expected_snapshot_id = hash_to_bytes("d0587e1195aed5a8800411a008f2f2d627f18e2d") | ||||
| assert actual_load_status == { | assert actual_load_status == { | ||||
| "status": "eventful", | "status": "eventful", | ||||
| "snapshot_id": expected_snapshot_id, | "snapshot_id": expected_snapshot_id.hex(), | ||||
| } | } | ||||
| assert_last_visit_matches( | |||||
| loader.storage, url, status="full", type="npm", snapshot=expected_snapshot_id | |||||
| ) | |||||
| 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, | ||||
| "origin_visit": 1, | "origin_visit": 1, | ||||
| "person": 2, | "person": 2, | ||||
| Show All 11 Lines | assert ( | ||||
| list(loader.storage.directory_missing(_expected_new_directories_first_visit)) | list(loader.storage.directory_missing(_expected_new_directories_first_visit)) | ||||
| == [] | == [] | ||||
| ) | ) | ||||
| assert ( | assert ( | ||||
| list(loader.storage.revision_missing(_expected_new_revisions_first_visit)) == [] | list(loader.storage.revision_missing(_expected_new_revisions_first_visit)) == [] | ||||
| ) | ) | ||||
| expected_snapshot = { | expected_snapshot = Snapshot( | ||||
| "id": hash_to_bytes(expected_snapshot_id), | id=expected_snapshot_id, | ||||
| "branches": { | branches={ | ||||
| b"HEAD": {"target": b"releases/0.0.4", "target_type": "alias"}, | b"HEAD": SnapshotBranch( | ||||
| b"releases/0.0.2": { | target=b"releases/0.0.4", target_type=TargetType.ALIAS | ||||
| "target": hash_to_bytes("d8a1c7474d2956ac598a19f0f27d52f7015f117e"), | ), | ||||
| "target_type": "revision", | b"releases/0.0.2": SnapshotBranch( | ||||
| }, | target=hash_to_bytes("d8a1c7474d2956ac598a19f0f27d52f7015f117e"), | ||||
| b"releases/0.0.3": { | target_type=TargetType.REVISION, | ||||
| "target": hash_to_bytes("5f9eb78af37ffd12949f235e86fac04898f9f72a"), | ), | ||||
| "target_type": "revision", | b"releases/0.0.3": SnapshotBranch( | ||||
| }, | target=hash_to_bytes("5f9eb78af37ffd12949f235e86fac04898f9f72a"), | ||||
| b"releases/0.0.4": { | target_type=TargetType.REVISION, | ||||
| "target": hash_to_bytes("ba019b192bdb94bd0b5bd68b3a5f92b5acc2239a"), | ), | ||||
| "target_type": "revision", | b"releases/0.0.4": SnapshotBranch( | ||||
| }, | target=hash_to_bytes("ba019b192bdb94bd0b5bd68b3a5f92b5acc2239a"), | ||||
| target_type=TargetType.REVISION, | |||||
| ), | |||||
| }, | }, | ||||
| } | ) | ||||
| check_snapshot(expected_snapshot, loader.storage) | check_snapshot(expected_snapshot, loader.storage) | ||||
| def test_npm_loader_incremental_visit(swh_config, requests_mock_datadir_visits): | def test_npm_loader_incremental_visit(swh_config, requests_mock_datadir_visits): | ||||
| package = "org" | package = "org" | ||||
| url = package_url(package) | url = package_url(package) | ||||
| loader = NpmLoader(url) | loader = NpmLoader(url) | ||||
| expected_snapshot_id = hash_to_bytes("d0587e1195aed5a8800411a008f2f2d627f18e2d") | |||||
| actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
| assert actual_load_status["status"] == "eventful" | assert actual_load_status == { | ||||
| assert actual_load_status["status"] is not None | "status": "eventful", | ||||
| assert_last_visit_matches(loader.storage, url, status="full", type="npm") | "snapshot_id": expected_snapshot_id.hex(), | ||||
| } | |||||
| assert_last_visit_matches( | |||||
| loader.storage, url, status="full", type="npm", snapshot=expected_snapshot_id | |||||
| ) | |||||
| 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, | ||||
| "origin_visit": 1, | "origin_visit": 1, | ||||
| Show All 37 Lines | |||||
| @pytest.mark.usefixtures("requests_mock_datadir") | @pytest.mark.usefixtures("requests_mock_datadir") | ||||
| def test_npm_loader_version_divergence(swh_config): | def test_npm_loader_version_divergence(swh_config): | ||||
| package = "@aller_shared" | package = "@aller_shared" | ||||
| url = package_url(package) | url = package_url(package) | ||||
| loader = NpmLoader(url) | loader = NpmLoader(url) | ||||
| actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
| assert actual_load_status["status"] == "eventful" | expected_snapshot_id = hash_to_bytes("b11ebac8c9d0c9e5063a2df693a18e3aba4b2f92") | ||||
| assert actual_load_status["status"] is not None | assert actual_load_status == { | ||||
| assert_last_visit_matches(loader.storage, url, status="full", type="npm") | "status": "eventful", | ||||
| "snapshot_id": expected_snapshot_id.hex(), | |||||
| } | |||||
| assert_last_visit_matches( | |||||
| loader.storage, url, status="full", type="npm", snapshot=expected_snapshot_id | |||||
| ) | |||||
| stats = get_stats(loader.storage) | stats = get_stats(loader.storage) | ||||
| assert { # 1 new releases artifacts | assert { # 1 new releases artifacts | ||||
| "content": 534, | "content": 534, | ||||
| "directory": 153, | "directory": 153, | ||||
| "origin": 1, | "origin": 1, | ||||
| "origin_visit": 1, | "origin_visit": 1, | ||||
| "person": 1, | "person": 1, | ||||
| "release": 0, | "release": 0, | ||||
| "revision": 2, | "revision": 2, | ||||
| "skipped_content": 0, | "skipped_content": 0, | ||||
| "snapshot": 1, | "snapshot": 1, | ||||
| } == stats | } == stats | ||||
| expected_snapshot = { | expected_snapshot = Snapshot( | ||||
| "id": hash_to_bytes("b11ebac8c9d0c9e5063a2df693a18e3aba4b2f92"), | id=expected_snapshot_id, | ||||
| "branches": { | branches={ | ||||
| b"HEAD": {"target_type": "alias", "target": b"releases/0.1.0"}, | b"HEAD": SnapshotBranch( | ||||
| b"releases/0.1.0": { | target_type=TargetType.ALIAS, target=b"releases/0.1.0" | ||||
| "target_type": "revision", | ), | ||||
| "target": hash_to_bytes("845673bfe8cbd31b1eaf757745a964137e6f9116"), | b"releases/0.1.0": SnapshotBranch( | ||||
| }, | target_type=TargetType.REVISION, | ||||
| b"releases/0.1.1-alpha.14": { | target=hash_to_bytes("845673bfe8cbd31b1eaf757745a964137e6f9116"), | ||||
| "target_type": "revision", | ), | ||||
| "target": hash_to_bytes("05181c12cd8c22035dd31155656826b85745da37"), | b"releases/0.1.1-alpha.14": SnapshotBranch( | ||||
| }, | target_type=TargetType.REVISION, | ||||
| target=hash_to_bytes("05181c12cd8c22035dd31155656826b85745da37"), | |||||
| ), | |||||
| }, | }, | ||||
| } | ) | ||||
| check_snapshot(expected_snapshot, loader.storage) | check_snapshot(expected_snapshot, loader.storage) | ||||
| def test_npm_artifact_to_revision_id_none(): | def test_npm_artifact_to_revision_id_none(): | ||||
| """Current loader version should stop soon if nothing can be found | """Current loader version should stop soon if nothing can be found | ||||
| """ | """ | ||||
| artifact_metadata = { | artifact_metadata = { | ||||
| ▲ Show 20 Lines • Show All 57 Lines • ▼ Show 20 Lines | def test_npm_artifact_with_no_intrinsic_metadata(swh_config, requests_mock_datadir): | ||||
| """Skip artifact with no intrinsic metadata during ingestion | """Skip artifact with no intrinsic metadata during ingestion | ||||
| """ | """ | ||||
| package = "nativescript-telerik-analytics" | package = "nativescript-telerik-analytics" | ||||
| url = package_url(package) | url = package_url(package) | ||||
| loader = NpmLoader(url) | loader = NpmLoader(url) | ||||
| actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
| assert actual_load_status["status"] == "eventful" | |||||
| # no branch as one artifact without any intrinsic metadata | # no branch as one artifact without any intrinsic metadata | ||||
| expected_snapshot = { | expected_snapshot = Snapshot( | ||||
| "id": hash_to_bytes("1a8893e6a86f444e8be8e7bda6cb34fb1735a00e"), | id=hash_to_bytes("1a8893e6a86f444e8be8e7bda6cb34fb1735a00e"), branches={}, | ||||
| "branches": {}, | ) | ||||
| assert actual_load_status == { | |||||
| "status": "eventful", | |||||
| "snapshot_id": expected_snapshot.id.hex(), | |||||
| } | } | ||||
| check_snapshot(expected_snapshot, loader.storage) | check_snapshot(expected_snapshot, loader.storage) | ||||
| assert_last_visit_matches(loader.storage, url, status="full", type="npm") | assert_last_visit_matches( | ||||
| loader.storage, url, status="full", type="npm", snapshot=expected_snapshot.id | |||||
| ) | |||||
| def test_npm_artifact_with_no_upload_time(swh_config, requests_mock_datadir): | def test_npm_artifact_with_no_upload_time(swh_config, requests_mock_datadir): | ||||
| """With no time upload, artifact is skipped | """With no time upload, artifact is skipped | ||||
| """ | """ | ||||
| package = "jammit-no-time" | package = "jammit-no-time" | ||||
| url = package_url(package) | url = package_url(package) | ||||
| loader = NpmLoader(url) | loader = NpmLoader(url) | ||||
| actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
| assert actual_load_status["status"] == "uneventful" | |||||
| # no branch as one artifact without any intrinsic metadata | # no branch as one artifact without any intrinsic metadata | ||||
| expected_snapshot = { | expected_snapshot = Snapshot( | ||||
| "id": hash_to_bytes("1a8893e6a86f444e8be8e7bda6cb34fb1735a00e"), | id=hash_to_bytes("1a8893e6a86f444e8be8e7bda6cb34fb1735a00e"), branches={}, | ||||
| "branches": {}, | ) | ||||
| assert actual_load_status == { | |||||
| "status": "uneventful", | |||||
| "snapshot_id": expected_snapshot.id.hex(), | |||||
| } | } | ||||
| check_snapshot(expected_snapshot, loader.storage) | check_snapshot(expected_snapshot, loader.storage) | ||||
| assert_last_visit_matches(loader.storage, url, status="partial", type="npm") | assert_last_visit_matches( | ||||
| loader.storage, url, status="partial", type="npm", snapshot=expected_snapshot.id | |||||
| ) | |||||
| def test_npm_artifact_use_mtime_if_no_time(swh_config, requests_mock_datadir): | def test_npm_artifact_use_mtime_if_no_time(swh_config, requests_mock_datadir): | ||||
| """With no time upload, artifact is skipped | """With no time upload, artifact is skipped | ||||
| """ | """ | ||||
| package = "jammit-express" | package = "jammit-express" | ||||
| url = package_url(package) | url = package_url(package) | ||||
| loader = NpmLoader(url) | loader = NpmLoader(url) | ||||
| actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
| assert actual_load_status["status"] == "eventful" | expected_snapshot_id = hash_to_bytes("d6e08e19159f77983242877c373c75222d5ae9dd") | ||||
| assert actual_load_status == { | |||||
| "status": "eventful", | |||||
| "snapshot_id": expected_snapshot_id.hex(), | |||||
| } | |||||
| # artifact is used | # artifact is used | ||||
| expected_snapshot = { | expected_snapshot = Snapshot( | ||||
| "id": hash_to_bytes("d6e08e19159f77983242877c373c75222d5ae9dd"), | id=expected_snapshot_id, | ||||
| "branches": { | branches={ | ||||
| b"HEAD": {"target_type": "alias", "target": b"releases/0.0.1"}, | b"HEAD": SnapshotBranch( | ||||
| b"releases/0.0.1": { | target_type=TargetType.ALIAS, target=b"releases/0.0.1" | ||||
| "target_type": "revision", | ), | ||||
| "target": hash_to_bytes("9e4dd2b40d1b46b70917c0949aa2195c823a648e"), | b"releases/0.0.1": SnapshotBranch( | ||||
| }, | target_type=TargetType.REVISION, | ||||
| target=hash_to_bytes("9e4dd2b40d1b46b70917c0949aa2195c823a648e"), | |||||
| ), | |||||
| }, | }, | ||||
| } | ) | ||||
| check_snapshot(expected_snapshot, loader.storage) | check_snapshot(expected_snapshot, loader.storage) | ||||
| assert_last_visit_matches(loader.storage, url, status="full", type="npm") | assert_last_visit_matches( | ||||
| loader.storage, url, status="full", type="npm", snapshot=expected_snapshot.id | |||||
| ) | |||||
| def test_npm_no_artifact(swh_config, requests_mock_datadir): | def test_npm_no_artifact(swh_config, requests_mock_datadir): | ||||
| """If no artifacts at all is found for origin, the visit fails completely | """If no artifacts at all is found for origin, the visit fails completely | ||||
| """ | """ | ||||
| package = "catify" | package = "catify" | ||||
| url = package_url(package) | url = package_url(package) | ||||
| loader = NpmLoader(url) | loader = NpmLoader(url) | ||||
| actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
| assert actual_load_status == { | assert actual_load_status == { | ||||
| "status": "failed", | "status": "failed", | ||||
| } | } | ||||
| assert_last_visit_matches(loader.storage, url, status="partial", type="npm") | assert_last_visit_matches(loader.storage, url, status="partial", type="npm") | ||||