Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/npm/tests/test_npm.py
Show First 20 Lines • Show All 60 Lines • ▼ Show 20 Lines | for author, expected_author in [ | ||||
{"name": ["Susan McSween", "William H. Bonney", "Doc Scurlock",]}, | {"name": ["Susan McSween", "William H. Bonney", "Doc Scurlock",]}, | ||||
"Susan McSween", | "Susan McSween", | ||||
), | ), | ||||
(None, None), | (None, None), | ||||
]: | ]: | ||||
assert _author_str(author) == expected_author | assert _author_str(author) == expected_author | ||||
def test_extract_npm_package_author(datadir): | def test_npm_extract_npm_package_author(datadir): | ||||
package_metadata_filepath = os.path.join( | package_metadata_filepath = os.path.join( | ||||
datadir, "https_replicate.npmjs.com", "org_visit1" | datadir, "https_replicate.npmjs.com", "org_visit1" | ||||
) | ) | ||||
with open(package_metadata_filepath) as json_file: | with open(package_metadata_filepath) as json_file: | ||||
package_metadata = json.load(json_file) | package_metadata = json.load(json_file) | ||||
extract_npm_package_author(package_metadata["versions"]["0.0.2"]) == Person( | extract_npm_package_author(package_metadata["versions"]["0.0.2"]) == Person( | ||||
▲ Show 20 Lines • Show All 222 Lines • ▼ Show 20 Lines | |||||
def package_url(package): | def package_url(package): | ||||
return "https://www.npmjs.com/package/%s" % package | return "https://www.npmjs.com/package/%s" % package | ||||
def package_metadata_url(package): | def package_metadata_url(package): | ||||
return "https://replicate.npmjs.com/%s/" % package | return "https://replicate.npmjs.com/%s/" % package | ||||
def test_revision_metadata_structure(swh_config, requests_mock_datadir): | def test_npm_revision_metadata_structure(swh_storage, requests_mock_datadir): | ||||
package = "org" | package = "org" | ||||
loader = NpmLoader(package_url(package)) | loader = NpmLoader(swh_storage, package_url(package)) | ||||
actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
assert actual_load_status["status"] == "eventful" | assert actual_load_status["status"] == "eventful" | ||||
assert actual_load_status["snapshot_id"] is not None | assert actual_load_status["snapshot_id"] is not None | ||||
expected_revision_id = hash_to_bytes("d8a1c7474d2956ac598a19f0f27d52f7015f117e") | expected_revision_id = hash_to_bytes("d8a1c7474d2956ac598a19f0f27d52f7015f117e") | ||||
revision = loader.storage.revision_get([expected_revision_id])[0] | revision = swh_storage.revision_get([expected_revision_id])[0] | ||||
assert revision is not None | assert revision is not None | ||||
check_metadata_paths( | check_metadata_paths( | ||||
revision.metadata, | revision.metadata, | ||||
paths=[ | paths=[ | ||||
("intrinsic.tool", str), | ("intrinsic.tool", str), | ||||
("intrinsic.raw", dict), | ("intrinsic.raw", dict), | ||||
("extrinsic.provider", str), | ("extrinsic.provider", str), | ||||
("extrinsic.when", str), | ("extrinsic.when", str), | ||||
("extrinsic.raw", dict), | ("extrinsic.raw", dict), | ||||
("original_artifact", list), | ("original_artifact", list), | ||||
], | ], | ||||
) | ) | ||||
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, org_api_info): | def test_npm_loader_first_visit(swh_storage, requests_mock_datadir, org_api_info): | ||||
package = "org" | package = "org" | ||||
url = package_url(package) | url = package_url(package) | ||||
loader = NpmLoader(url) | loader = NpmLoader(swh_storage, url) | ||||
actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
expected_snapshot_id = hash_to_bytes("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.hex(), | "snapshot_id": expected_snapshot_id.hex(), | ||||
} | } | ||||
assert_last_visit_matches( | assert_last_visit_matches( | ||||
loader.storage, url, status="full", type="npm", snapshot=expected_snapshot_id | swh_storage, url, status="full", type="npm", snapshot=expected_snapshot_id | ||||
) | ) | ||||
stats = get_stats(loader.storage) | stats = get_stats(swh_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, | ||||
"release": 0, | "release": 0, | ||||
"revision": len(_expected_new_revisions_first_visit), | "revision": len(_expected_new_revisions_first_visit), | ||||
"skipped_content": 0, | "skipped_content": 0, | ||||
"snapshot": 1, | "snapshot": 1, | ||||
} == stats | } == stats | ||||
contents = loader.storage.content_get(_expected_new_contents_first_visit) | contents = swh_storage.content_get(_expected_new_contents_first_visit) | ||||
count = sum(0 if content is None else 1 for content in contents) | count = sum(0 if content is None else 1 for content in contents) | ||||
assert count == len(_expected_new_contents_first_visit) | assert count == len(_expected_new_contents_first_visit) | ||||
assert ( | assert ( | ||||
list(loader.storage.directory_missing(_expected_new_directories_first_visit)) | list(swh_storage.directory_missing(_expected_new_directories_first_visit)) == [] | ||||
== [] | |||||
) | ) | ||||
assert ( | assert list(swh_storage.revision_missing(_expected_new_revisions_first_visit)) == [] | ||||
list(loader.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, loader.storage) | check_snapshot(expected_snapshot, swh_storage) | ||||
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 = loader.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 = SWHID(object_type="directory", object_id=directory_id,) | directory_swhid = SWHID(object_type="directory", object_id=directory_id,) | ||||
revision_swhid = SWHID(object_type="revision", object_id=revision_id,) | revision_swhid = SWHID(object_type="revision", object_id=revision_id,) | ||||
expected_metadata = [ | expected_metadata = [ | ||||
RawExtrinsicMetadata( | RawExtrinsicMetadata( | ||||
type=MetadataTargetType.DIRECTORY, | type=MetadataTargetType.DIRECTORY, | ||||
target=directory_swhid, | target=directory_swhid, | ||||
authority=metadata_authority, | authority=metadata_authority, | ||||
fetcher=MetadataFetcher( | fetcher=MetadataFetcher( | ||||
name="swh.loader.package.npm.loader.NpmLoader", version=__version__, | name="swh.loader.package.npm.loader.NpmLoader", version=__version__, | ||||
), | ), | ||||
discovery_date=loader.visit_date, | discovery_date=loader.visit_date, | ||||
format="replicate-npm-package-json", | format="replicate-npm-package-json", | ||||
metadata=json.dumps( | metadata=json.dumps( | ||||
json.loads(org_api_info)["versions"][version_name] | json.loads(org_api_info)["versions"][version_name] | ||||
).encode(), | ).encode(), | ||||
origin="https://www.npmjs.com/package/org", | origin="https://www.npmjs.com/package/org", | ||||
revision=revision_swhid, | revision=revision_swhid, | ||||
) | ) | ||||
] | ] | ||||
assert loader.storage.raw_extrinsic_metadata_get( | assert swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.DIRECTORY, directory_swhid, metadata_authority, | MetadataTargetType.DIRECTORY, directory_swhid, metadata_authority, | ||||
) == PagedResult(next_page_token=None, results=expected_metadata,) | ) == PagedResult(next_page_token=None, results=expected_metadata,) | ||||
def test_npm_loader_incremental_visit(swh_config, 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(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() | ||||
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( | ||||
loader.storage, url, status="full", type="npm", snapshot=expected_snapshot_id | swh_storage, url, status="full", type="npm", snapshot=expected_snapshot_id | ||||
) | ) | ||||
stats = get_stats(loader.storage) | stats = get_stats(swh_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, | ||||
"release": 0, | "release": 0, | ||||
"revision": len(_expected_new_revisions_first_visit), | "revision": len(_expected_new_revisions_first_visit), | ||||
"skipped_content": 0, | "skipped_content": 0, | ||||
"snapshot": 1, | "snapshot": 1, | ||||
} == stats | } == stats | ||||
# reset loader internal state | # reset loader internal state | ||||
del loader._cached_info | del loader._cached_info | ||||
del loader._cached__raw_info | del loader._cached__raw_info | ||||
actual_load_status2 = loader.load() | actual_load_status2 = loader.load() | ||||
assert actual_load_status2["status"] == "eventful" | assert actual_load_status2["status"] == "eventful" | ||||
snap_id2 = actual_load_status2["snapshot_id"] | snap_id2 = actual_load_status2["snapshot_id"] | ||||
assert snap_id2 is not None | assert snap_id2 is not None | ||||
assert snap_id2 != actual_load_status["snapshot_id"] | assert snap_id2 != actual_load_status["snapshot_id"] | ||||
assert_last_visit_matches(loader.storage, url, status="full", type="npm") | assert_last_visit_matches(swh_storage, url, status="full", type="npm") | ||||
stats = get_stats(loader.storage) | stats = get_stats(swh_storage) | ||||
assert { # 3 new releases artifacts | assert { # 3 new releases artifacts | ||||
"content": len(_expected_new_contents_first_visit) + 14, | "content": len(_expected_new_contents_first_visit) + 14, | ||||
"directory": len(_expected_new_directories_first_visit) + 15, | "directory": len(_expected_new_directories_first_visit) + 15, | ||||
"origin": 1, | "origin": 1, | ||||
"origin_visit": 2, | "origin_visit": 2, | ||||
"release": 0, | "release": 0, | ||||
"revision": len(_expected_new_revisions_first_visit) + 3, | "revision": len(_expected_new_revisions_first_visit) + 3, | ||||
"skipped_content": 0, | "skipped_content": 0, | ||||
"snapshot": 2, | "snapshot": 2, | ||||
} == stats | } == stats | ||||
urls = [ | urls = [ | ||||
m.url | m.url | ||||
for m in requests_mock_datadir_visits.request_history | for m in requests_mock_datadir_visits.request_history | ||||
if m.url.startswith("https://registry.npmjs.org") | if m.url.startswith("https://registry.npmjs.org") | ||||
] | ] | ||||
assert len(urls) == len(set(urls)) # we visited each artifact once across | assert len(urls) == len(set(urls)) # we visited each artifact once across | ||||
@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_storage): | ||||
package = "@aller_shared" | package = "@aller_shared" | ||||
url = package_url(package) | url = package_url(package) | ||||
loader = NpmLoader(url) | loader = NpmLoader(swh_storage, url) | ||||
actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
expected_snapshot_id = hash_to_bytes("b11ebac8c9d0c9e5063a2df693a18e3aba4b2f92") | expected_snapshot_id = hash_to_bytes("b11ebac8c9d0c9e5063a2df693a18e3aba4b2f92") | ||||
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( | ||||
loader.storage, url, status="full", type="npm", snapshot=expected_snapshot_id | swh_storage, url, status="full", type="npm", snapshot=expected_snapshot_id | ||||
) | ) | ||||
stats = get_stats(loader.storage) | stats = get_stats(swh_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, | ||||
"release": 0, | "release": 0, | ||||
"revision": 2, | "revision": 2, | ||||
Show All 12 Lines | expected_snapshot = Snapshot( | ||||
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, loader.storage) | check_snapshot(expected_snapshot, swh_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 | ||||
""" | """ | ||||
class artifact_metadata: | class artifact_metadata: | ||||
▲ Show 20 Lines • Show All 47 Lines • ▼ Show 20 Lines | known_artifacts = { | ||||
}, | }, | ||||
} | } | ||||
assert artifact_to_revision_id(known_artifacts, artifact_metadata) == hash_to_bytes( | assert artifact_to_revision_id(known_artifacts, artifact_metadata) == hash_to_bytes( | ||||
"b11ebac8c9d0c9e5063a2df693a18e3aba4b2f92" | "b11ebac8c9d0c9e5063a2df693a18e3aba4b2f92" | ||||
) | ) | ||||
def test_npm_artifact_with_no_intrinsic_metadata(swh_config, 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(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, loader.storage) | check_snapshot(expected_snapshot, swh_storage) | ||||
assert_last_visit_matches( | assert_last_visit_matches( | ||||
loader.storage, url, status="full", type="npm", snapshot=expected_snapshot.id | swh_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_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(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, loader.storage) | check_snapshot(expected_snapshot, swh_storage) | ||||
assert_last_visit_matches( | assert_last_visit_matches( | ||||
loader.storage, url, status="partial", type="npm", snapshot=expected_snapshot.id | swh_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_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(url) | loader = NpmLoader(swh_storage, url) | ||||
actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
expected_snapshot_id = hash_to_bytes("d6e08e19159f77983242877c373c75222d5ae9dd") | expected_snapshot_id = hash_to_bytes("d6e08e19159f77983242877c373c75222d5ae9dd") | ||||
assert actual_load_status == { | assert actual_load_status == { | ||||
"status": "eventful", | "status": "eventful", | ||||
"snapshot_id": expected_snapshot_id.hex(), | "snapshot_id": expected_snapshot_id.hex(), | ||||
} | } | ||||
# artifact is used | # artifact is used | ||||
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.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, loader.storage) | check_snapshot(expected_snapshot, swh_storage) | ||||
assert_last_visit_matches( | assert_last_visit_matches( | ||||
loader.storage, url, status="full", type="npm", snapshot=expected_snapshot.id | swh_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_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(url) | loader = NpmLoader(swh_storage, 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="failed", type="npm") | assert_last_visit_matches(swh_storage, url, status="failed", type="npm") | ||||
def test_npm_origin_not_found(swh_config, requests_mock_datadir): | def test_npm_origin_not_found(swh_storage, requests_mock_datadir): | ||||
url = package_url("non-existent-url") | url = package_url("non-existent-url") | ||||
loader = NpmLoader(url) | loader = NpmLoader(swh_storage, url) | ||||
assert loader.load() == {"status": "failed"} | assert loader.load() == {"status": "failed"} | ||||
assert_last_visit_matches( | assert_last_visit_matches( | ||||
loader.storage, url, status="not_found", type="npm", snapshot=None | swh_storage, url, status="not_found", type="npm", snapshot=None | ||||
) | ) |