Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/npm/tests/test_npm.py
| Show First 20 Lines • Show All 312 Lines • ▼ Show 20 Lines | assert actual_load_status == { | ||||
| "status": "eventful", | "status": "eventful", | ||||
| "snapshot_id": expected_snapshot_id.hex(), | "snapshot_id": expected_snapshot_id.hex(), | ||||
| } | } | ||||
| assert_last_visit_matches( | assert_last_visit_matches( | ||||
| swh_storage, url, status="full", type="npm", snapshot=expected_snapshot_id | swh_storage, url, status="full", type="npm", snapshot=expected_snapshot_id | ||||
| ) | ) | ||||
| stats = get_stats(swh_storage) | |||||
| 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 | |||||
| contents = swh_storage.content_get(_expected_new_contents_first_visit) | |||||
| count = sum(0 if content is None else 1 for content in contents) | |||||
| assert count == len(_expected_new_contents_first_visit) | |||||
| assert ( | |||||
| list(swh_storage.directory_missing(_expected_new_directories_first_visit)) == [] | |||||
| ) | |||||
| assert list(swh_storage.revision_missing(_expected_new_revisions_first_visit)) == [] | |||||
| versions = [ | versions = [ | ||||
| ("0.0.2", "d8a1c7474d2956ac598a19f0f27d52f7015f117e"), | ("0.0.2", "d8a1c7474d2956ac598a19f0f27d52f7015f117e"), | ||||
| ("0.0.3", "5f9eb78af37ffd12949f235e86fac04898f9f72a"), | ("0.0.3", "5f9eb78af37ffd12949f235e86fac04898f9f72a"), | ||||
| ("0.0.4", "ba019b192bdb94bd0b5bd68b3a5f92b5acc2239a"), | ("0.0.4", "ba019b192bdb94bd0b5bd68b3a5f92b5acc2239a"), | ||||
| ] | ] | ||||
| expected_snapshot = Snapshot( | expected_snapshot = Snapshot( | ||||
| id=expected_snapshot_id, | id=expected_snapshot_id, | ||||
| branches={ | branches={ | ||||
| b"HEAD": SnapshotBranch( | b"HEAD": SnapshotBranch( | ||||
| target=b"releases/0.0.4", target_type=TargetType.ALIAS | target=b"releases/0.0.4", target_type=TargetType.ALIAS | ||||
| ), | ), | ||||
| **{ | **{ | ||||
| b"releases/" | b"releases/" | ||||
| + version_name.encode(): SnapshotBranch( | + version_name.encode(): SnapshotBranch( | ||||
| target=hash_to_bytes(version_id), target_type=TargetType.REVISION, | target=hash_to_bytes(version_id), target_type=TargetType.REVISION, | ||||
| ) | ) | ||||
| for (version_name, version_id) in versions | for (version_name, version_id) in versions | ||||
| }, | }, | ||||
| }, | }, | ||||
| ) | ) | ||||
| check_snapshot(expected_snapshot, swh_storage) | check_snapshot(expected_snapshot, swh_storage) | ||||
| contents = swh_storage.content_get(_expected_new_contents_first_visit) | |||||
| count = sum(0 if content is None else 1 for content in contents) | |||||
| assert count == len(_expected_new_contents_first_visit) | |||||
| assert ( | |||||
| list(swh_storage.directory_missing(_expected_new_directories_first_visit)) == [] | |||||
| ) | |||||
| assert list(swh_storage.revision_missing(_expected_new_revisions_first_visit)) == [] | |||||
| metadata_authority = MetadataAuthority( | metadata_authority = MetadataAuthority( | ||||
| type=MetadataAuthorityType.FORGE, url="https://npmjs.com/", | type=MetadataAuthorityType.FORGE, url="https://npmjs.com/", | ||||
| ) | ) | ||||
| for (version_name, revision_id) in versions: | for (version_name, revision_id) in versions: | ||||
| revision = swh_storage.revision_get([hash_to_bytes(revision_id)])[0] | revision = swh_storage.revision_get([hash_to_bytes(revision_id)])[0] | ||||
| directory_id = revision.directory | directory_id = revision.directory | ||||
| directory_swhid = ExtendedSWHID( | directory_swhid = ExtendedSWHID( | ||||
| Show All 17 Lines | for (version_name, revision_id) in versions: | ||||
| origin="https://www.npmjs.com/package/org", | origin="https://www.npmjs.com/package/org", | ||||
| revision=revision_swhid, | revision=revision_swhid, | ||||
| ) | ) | ||||
| ] | ] | ||||
| assert swh_storage.raw_extrinsic_metadata_get( | assert swh_storage.raw_extrinsic_metadata_get( | ||||
| directory_swhid, metadata_authority, | directory_swhid, metadata_authority, | ||||
| ) == PagedResult(next_page_token=None, results=expected_metadata,) | ) == PagedResult(next_page_token=None, results=expected_metadata,) | ||||
| stats = get_stats(swh_storage) | |||||
| 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 | |||||
| def test_npm_loader_incremental_visit(swh_storage, requests_mock_datadir_visits): | def test_npm_loader_incremental_visit(swh_storage, requests_mock_datadir_visits): | ||||
| package = "org" | package = "org" | ||||
| url = package_url(package) | url = package_url(package) | ||||
| loader = NpmLoader(swh_storage, url) | loader = NpmLoader(swh_storage, url) | ||||
| expected_snapshot_id = hash_to_bytes("d0587e1195aed5a8800411a008f2f2d627f18e2d") | expected_snapshot_id = hash_to_bytes("d0587e1195aed5a8800411a008f2f2d627f18e2d") | ||||
| actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
| ▲ Show 20 Lines • Show All 62 Lines • ▼ Show 20 Lines | def test_npm_loader_version_divergence(swh_storage): | ||||
| assert actual_load_status == { | assert actual_load_status == { | ||||
| "status": "eventful", | "status": "eventful", | ||||
| "snapshot_id": expected_snapshot_id.hex(), | "snapshot_id": expected_snapshot_id.hex(), | ||||
| } | } | ||||
| assert_last_visit_matches( | assert_last_visit_matches( | ||||
| swh_storage, url, status="full", type="npm", snapshot=expected_snapshot_id | swh_storage, url, status="full", type="npm", snapshot=expected_snapshot_id | ||||
| ) | ) | ||||
| stats = get_stats(swh_storage) | |||||
| assert { # 1 new releases artifacts | |||||
| "content": 534, | |||||
| "directory": 153, | |||||
| "origin": 1, | |||||
| "origin_visit": 1, | |||||
| "release": 0, | |||||
| "revision": 2, | |||||
| "skipped_content": 0, | |||||
| "snapshot": 1, | |||||
| } == stats | |||||
| expected_snapshot = Snapshot( | expected_snapshot = Snapshot( | ||||
| id=expected_snapshot_id, | id=expected_snapshot_id, | ||||
| branches={ | branches={ | ||||
| b"HEAD": SnapshotBranch( | b"HEAD": SnapshotBranch( | ||||
| target_type=TargetType.ALIAS, target=b"releases/0.1.0" | target_type=TargetType.ALIAS, target=b"releases/0.1.0" | ||||
| ), | ), | ||||
| b"releases/0.1.0": SnapshotBranch( | b"releases/0.1.0": SnapshotBranch( | ||||
| target_type=TargetType.REVISION, | target_type=TargetType.REVISION, | ||||
| target=hash_to_bytes("845673bfe8cbd31b1eaf757745a964137e6f9116"), | target=hash_to_bytes("845673bfe8cbd31b1eaf757745a964137e6f9116"), | ||||
| ), | ), | ||||
| b"releases/0.1.1-alpha.14": SnapshotBranch( | b"releases/0.1.1-alpha.14": SnapshotBranch( | ||||
| target_type=TargetType.REVISION, | target_type=TargetType.REVISION, | ||||
| target=hash_to_bytes("05181c12cd8c22035dd31155656826b85745da37"), | target=hash_to_bytes("05181c12cd8c22035dd31155656826b85745da37"), | ||||
| ), | ), | ||||
| }, | }, | ||||
| ) | ) | ||||
| check_snapshot(expected_snapshot, swh_storage) | check_snapshot(expected_snapshot, swh_storage) | ||||
| stats = get_stats(swh_storage) | |||||
| assert { # 1 new releases artifacts | |||||
| "content": 534, | |||||
| "directory": 153, | |||||
| "origin": 1, | |||||
| "origin_visit": 1, | |||||
| "release": 0, | |||||
| "revision": 2, | |||||
| "skipped_content": 0, | |||||
| "snapshot": 1, | |||||
| } == stats | |||||
| def test_npm_artifact_with_no_intrinsic_metadata(swh_storage, requests_mock_datadir): | def test_npm_artifact_with_no_intrinsic_metadata(swh_storage, 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(swh_storage, url) | loader = NpmLoader(swh_storage, url) | ||||
| actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
| # no branch as one artifact without any intrinsic metadata | # no branch as one artifact without any intrinsic metadata | ||||
| expected_snapshot = Snapshot( | expected_snapshot = Snapshot( | ||||
| id=hash_to_bytes("1a8893e6a86f444e8be8e7bda6cb34fb1735a00e"), branches={}, | id=hash_to_bytes("1a8893e6a86f444e8be8e7bda6cb34fb1735a00e"), branches={}, | ||||
| ) | ) | ||||
| assert actual_load_status == { | assert actual_load_status == { | ||||
| "status": "eventful", | "status": "eventful", | ||||
| "snapshot_id": expected_snapshot.id.hex(), | "snapshot_id": expected_snapshot.id.hex(), | ||||
| } | } | ||||
| check_snapshot(expected_snapshot, swh_storage) | |||||
| assert_last_visit_matches( | assert_last_visit_matches( | ||||
| swh_storage, url, status="full", type="npm", snapshot=expected_snapshot.id | swh_storage, url, status="full", type="npm", snapshot=expected_snapshot.id | ||||
| ) | ) | ||||
| check_snapshot(expected_snapshot, swh_storage) | |||||
| def test_npm_artifact_with_no_upload_time(swh_storage, requests_mock_datadir): | def test_npm_artifact_with_no_upload_time(swh_storage, 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(swh_storage, url) | loader = NpmLoader(swh_storage, url) | ||||
| actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
| # no branch as one artifact without any intrinsic metadata | # no branch as one artifact without any intrinsic metadata | ||||
| expected_snapshot = Snapshot( | expected_snapshot = Snapshot( | ||||
| id=hash_to_bytes("1a8893e6a86f444e8be8e7bda6cb34fb1735a00e"), branches={}, | id=hash_to_bytes("1a8893e6a86f444e8be8e7bda6cb34fb1735a00e"), branches={}, | ||||
| ) | ) | ||||
| assert actual_load_status == { | assert actual_load_status == { | ||||
| "status": "uneventful", | "status": "uneventful", | ||||
| "snapshot_id": expected_snapshot.id.hex(), | "snapshot_id": expected_snapshot.id.hex(), | ||||
| } | } | ||||
| check_snapshot(expected_snapshot, swh_storage) | |||||
| assert_last_visit_matches( | assert_last_visit_matches( | ||||
| swh_storage, url, status="partial", type="npm", snapshot=expected_snapshot.id | swh_storage, url, status="partial", type="npm", snapshot=expected_snapshot.id | ||||
| ) | ) | ||||
| check_snapshot(expected_snapshot, swh_storage) | |||||
| def test_npm_artifact_use_mtime_if_no_time(swh_storage, requests_mock_datadir): | def test_npm_artifact_use_mtime_if_no_time(swh_storage, 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(swh_storage, url) | loader = NpmLoader(swh_storage, url) | ||||
| Show All 14 Lines | expected_snapshot = Snapshot( | ||||
| target_type=TargetType.ALIAS, target=b"releases/0.0.1" | target_type=TargetType.ALIAS, target=b"releases/0.0.1" | ||||
| ), | ), | ||||
| b"releases/0.0.1": SnapshotBranch( | b"releases/0.0.1": SnapshotBranch( | ||||
| target_type=TargetType.REVISION, | target_type=TargetType.REVISION, | ||||
| target=hash_to_bytes("9e4dd2b40d1b46b70917c0949aa2195c823a648e"), | target=hash_to_bytes("9e4dd2b40d1b46b70917c0949aa2195c823a648e"), | ||||
| ), | ), | ||||
| }, | }, | ||||
| ) | ) | ||||
| check_snapshot(expected_snapshot, swh_storage) | |||||
| assert_last_visit_matches( | assert_last_visit_matches( | ||||
| swh_storage, url, status="full", type="npm", snapshot=expected_snapshot.id | swh_storage, url, status="full", type="npm", snapshot=expected_snapshot.id | ||||
| ) | ) | ||||
| check_snapshot(expected_snapshot, swh_storage) | |||||
| def test_npm_no_artifact(swh_storage, requests_mock_datadir): | def test_npm_no_artifact(swh_storage, 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(swh_storage, url) | loader = NpmLoader(swh_storage, url) | ||||
| Show All 17 Lines | |||||