Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/pypi/tests/test_pypi.py
| Show First 20 Lines • Show All 254 Lines • ▼ Show 20 Lines | ): | ||||
| stats = get_stats(loader.storage) | stats = get_stats(loader.storage) | ||||
| assert { | assert { | ||||
| "content": 6, | "content": 6, | ||||
| "directory": 4, | "directory": 4, | ||||
| "origin": 1, | "origin": 1, | ||||
| "origin_visit": 1, | "origin_visit": 1, | ||||
| "release": 0, | "release": 2, | ||||
| "revision": 2, | "revision": 0, | ||||
| "skipped_content": 0, | "skipped_content": 0, | ||||
| "snapshot": 0, | "snapshot": 0, | ||||
| } == stats | } == stats | ||||
| # problem during loading: | # problem during loading: | ||||
| # {visit: partial, status: uneventful, no snapshot} | # {visit: partial, status: uneventful, no snapshot} | ||||
| ▲ Show 20 Lines • Show All 41 Lines • ▼ Show 20 Lines | ignore_urls=[ | ||||
| "https://files.pythonhosted.org/packages/ec/65/c0116953c9a3f47de89e71964d6c7b0c783b01f29fa3390584dbf3046b4d/0805nexter-1.1.0.zip", # noqa | "https://files.pythonhosted.org/packages/ec/65/c0116953c9a3f47de89e71964d6c7b0c783b01f29fa3390584dbf3046b4d/0805nexter-1.1.0.zip", # noqa | ||||
| ] | ] | ||||
| ) | ) | ||||
| # some missing release artifacts: | # some missing release artifacts: | ||||
| # {visit partial, status: eventful, 1 snapshot} | # {visit partial, status: eventful, 1 snapshot} | ||||
| def test_pypi_revision_metadata_structure( | def test_pypi_release_metadata_structure( | ||||
| swh_storage, requests_mock_datadir, _0805nexter_api_info | swh_storage, requests_mock_datadir, _0805nexter_api_info | ||||
| ): | ): | ||||
| url = "https://pypi.org/project/0805nexter" | url = "https://pypi.org/project/0805nexter" | ||||
| loader = PyPILoader(swh_storage, url) | loader = PyPILoader(swh_storage, url) | ||||
| 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("e445da4da22b31bfebb6ffc4383dbf839a074d21") | expected_release_id = hash_to_bytes("a1e10745d375be66c1b65e55c0c15fe98776b53c") | ||||
| revision = swh_storage.revision_get([expected_revision_id])[0] | |||||
| assert revision is not None | expected_snapshot = Snapshot( | ||||
| id=hash_to_bytes(actual_load_status["snapshot_id"]), | |||||
| branches={ | |||||
| b"HEAD": SnapshotBranch( | |||||
ardumont: this looks weird.
Should be one of the other branch name instead, ain't it? | |||||
Done Inline Actionsmy bad vlorentz: my bad | |||||
Not Done Inline Actionsclearer ;) ardumont: clearer ;)
Thanks. | |||||
| target=b"releases/1.2.0", target_type=TargetType.ALIAS, | |||||
| ), | |||||
| b"releases/1.1.0": SnapshotBranch( | |||||
| target=hash_to_bytes("9478c9981887fdf5ada3f1fcb20c81069cdf4c44"), | |||||
| target_type=TargetType.RELEASE, | |||||
| ), | |||||
| b"releases/1.2.0": SnapshotBranch( | |||||
| target=hash_to_bytes("a1e10745d375be66c1b65e55c0c15fe98776b53c"), | |||||
| target_type=TargetType.RELEASE, | |||||
| ), | |||||
| }, | |||||
| ) | |||||
| revision_swhid = CoreSWHID( | assert_last_visit_matches( | ||||
| object_type=ObjectType.REVISION, object_id=expected_revision_id | swh_storage, url, status="full", type="pypi", snapshot=expected_snapshot.id | ||||
| ) | |||||
| check_snapshot(expected_snapshot, swh_storage) | |||||
| release = swh_storage.release_get([expected_release_id])[0] | |||||
| assert release is not None | |||||
| release_swhid = CoreSWHID( | |||||
| object_type=ObjectType.RELEASE, object_id=expected_release_id | |||||
| ) | ) | ||||
| directory_swhid = ExtendedSWHID( | directory_swhid = ExtendedSWHID( | ||||
| object_type=ExtendedObjectType.DIRECTORY, object_id=revision.directory | object_type=ExtendedObjectType.DIRECTORY, object_id=release.target | ||||
| ) | ) | ||||
| metadata_authority = MetadataAuthority( | metadata_authority = MetadataAuthority( | ||||
| type=MetadataAuthorityType.FORGE, url="https://pypi.org/", | type=MetadataAuthorityType.FORGE, url="https://pypi.org/", | ||||
| ) | ) | ||||
| expected_metadata = [ | expected_metadata = [ | ||||
| RawExtrinsicMetadata( | RawExtrinsicMetadata( | ||||
| target=directory_swhid, | target=directory_swhid, | ||||
| authority=metadata_authority, | authority=metadata_authority, | ||||
| fetcher=MetadataFetcher( | fetcher=MetadataFetcher( | ||||
| name="swh.loader.package.pypi.loader.PyPILoader", version=__version__, | name="swh.loader.package.pypi.loader.PyPILoader", version=__version__, | ||||
| ), | ), | ||||
| discovery_date=loader.visit_date, | discovery_date=loader.visit_date, | ||||
| format="pypi-project-json", | format="pypi-project-json", | ||||
| metadata=json.dumps( | metadata=json.dumps( | ||||
| json.loads(_0805nexter_api_info)["releases"]["1.2.0"][0] | json.loads(_0805nexter_api_info)["releases"]["1.2.0"][0] | ||||
| ).encode(), | ).encode(), | ||||
| origin=url, | origin=url, | ||||
| revision=revision_swhid, | release=release_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,) | ||||
| def test_pypi_visit_with_missing_artifact( | def test_pypi_visit_with_missing_artifact( | ||||
| swh_storage, requests_mock_datadir_missing_one | swh_storage, requests_mock_datadir_missing_one | ||||
| ): | ): | ||||
| """Load a pypi project with some missing artifacts ends up with 1 snapshot | """Load a pypi project with some missing artifacts ends up with 1 snapshot | ||||
| """ | """ | ||||
| url = "https://pypi.org/project/0805nexter" | url = "https://pypi.org/project/0805nexter" | ||||
| loader = PyPILoader(swh_storage, url) | loader = PyPILoader(swh_storage, url) | ||||
| actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
| expected_snapshot_id = hash_to_bytes("dd0e4201a232b1c104433741dbf45895b8ac9355") | expected_snapshot_id = hash_to_bytes("eee24d5b0c156ebb4ece0c810c9dce636ebe881f") | ||||
| 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="partial", type="pypi", snapshot=expected_snapshot_id, | swh_storage, url, status="partial", type="pypi", snapshot=expected_snapshot_id, | ||||
| ) | ) | ||||
| expected_snapshot = Snapshot( | expected_snapshot = Snapshot( | ||||
| id=hash_to_bytes(expected_snapshot_id), | id=hash_to_bytes(expected_snapshot_id), | ||||
| branches={ | branches={ | ||||
| b"releases/1.2.0": SnapshotBranch( | b"releases/1.2.0": SnapshotBranch( | ||||
| target=hash_to_bytes("e445da4da22b31bfebb6ffc4383dbf839a074d21"), | target=hash_to_bytes("a1e10745d375be66c1b65e55c0c15fe98776b53c"), | ||||
| target_type=TargetType.REVISION, | target_type=TargetType.RELEASE, | ||||
| ), | ), | ||||
| b"HEAD": SnapshotBranch( | b"HEAD": SnapshotBranch( | ||||
| target=b"releases/1.2.0", target_type=TargetType.ALIAS, | target=b"releases/1.2.0", target_type=TargetType.ALIAS, | ||||
| ), | ), | ||||
| }, | }, | ||||
| ) | ) | ||||
| check_snapshot(expected_snapshot, storage=swh_storage) | check_snapshot(expected_snapshot, storage=swh_storage) | ||||
| stats = get_stats(swh_storage) | stats = get_stats(swh_storage) | ||||
| assert { | assert { | ||||
| "content": 3, | "content": 3, | ||||
| "directory": 2, | "directory": 2, | ||||
| "origin": 1, | "origin": 1, | ||||
| "origin_visit": 1, | "origin_visit": 1, | ||||
| "release": 0, | "release": 1, | ||||
| "revision": 1, | "revision": 0, | ||||
| "skipped_content": 0, | "skipped_content": 0, | ||||
| "snapshot": 1, | "snapshot": 1, | ||||
| } == stats | } == stats | ||||
| def test_pypi_visit_with_1_release_artifact(swh_storage, requests_mock_datadir): | def test_pypi_visit_with_1_release_artifact(swh_storage, requests_mock_datadir): | ||||
| """With no prior visit, load a pypi project ends up with 1 snapshot | """With no prior visit, load a pypi project ends up with 1 snapshot | ||||
| """ | """ | ||||
| url = "https://pypi.org/project/0805nexter" | url = "https://pypi.org/project/0805nexter" | ||||
| loader = PyPILoader(swh_storage, url) | loader = PyPILoader(swh_storage, url) | ||||
| actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
| expected_snapshot_id = hash_to_bytes("ba6e158ada75d0b3cfb209ffdf6daa4ed34a227a") | expected_snapshot_id = hash_to_bytes("62d957f2b5cdc515bea0a46252a3ab29ee271636") | ||||
| 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="pypi", snapshot=expected_snapshot_id | swh_storage, url, status="full", type="pypi", snapshot=expected_snapshot_id | ||||
| ) | ) | ||||
| expected_snapshot = Snapshot( | expected_snapshot = Snapshot( | ||||
| id=expected_snapshot_id, | id=expected_snapshot_id, | ||||
| branches={ | branches={ | ||||
| b"releases/1.1.0": SnapshotBranch( | b"releases/1.1.0": SnapshotBranch( | ||||
| target=hash_to_bytes("4c99891f93b81450385777235a37b5e966dd1571"), | target=hash_to_bytes("9478c9981887fdf5ada3f1fcb20c81069cdf4c44"), | ||||
| target_type=TargetType.REVISION, | target_type=TargetType.RELEASE, | ||||
| ), | ), | ||||
| b"releases/1.2.0": SnapshotBranch( | b"releases/1.2.0": SnapshotBranch( | ||||
| target=hash_to_bytes("e445da4da22b31bfebb6ffc4383dbf839a074d21"), | target=hash_to_bytes("a1e10745d375be66c1b65e55c0c15fe98776b53c"), | ||||
| target_type=TargetType.REVISION, | target_type=TargetType.RELEASE, | ||||
| ), | ), | ||||
| b"HEAD": SnapshotBranch( | b"HEAD": SnapshotBranch( | ||||
| target=b"releases/1.2.0", target_type=TargetType.ALIAS, | target=b"releases/1.2.0", target_type=TargetType.ALIAS, | ||||
| ), | ), | ||||
| }, | }, | ||||
| ) | ) | ||||
| check_snapshot(expected_snapshot, swh_storage) | check_snapshot(expected_snapshot, swh_storage) | ||||
| stats = get_stats(swh_storage) | stats = get_stats(swh_storage) | ||||
| assert { | assert { | ||||
| "content": 6, | "content": 6, | ||||
| "directory": 4, | "directory": 4, | ||||
| "origin": 1, | "origin": 1, | ||||
| "origin_visit": 1, | "origin_visit": 1, | ||||
| "release": 0, | "release": 2, | ||||
| "revision": 2, | "revision": 0, | ||||
| "skipped_content": 0, | "skipped_content": 0, | ||||
| "snapshot": 1, | "snapshot": 1, | ||||
| } == stats | } == stats | ||||
| def test_pypi_multiple_visits_with_no_change(swh_storage, requests_mock_datadir): | def test_pypi_multiple_visits_with_no_change(swh_storage, requests_mock_datadir): | ||||
| """Multiple visits with no changes results in 1 same snapshot | """Multiple visits with no changes results in 1 same snapshot | ||||
| """ | """ | ||||
| url = "https://pypi.org/project/0805nexter" | url = "https://pypi.org/project/0805nexter" | ||||
| loader = PyPILoader(swh_storage, url) | loader = PyPILoader(swh_storage, url) | ||||
| actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
| snapshot_id = hash_to_bytes("ba6e158ada75d0b3cfb209ffdf6daa4ed34a227a") | snapshot_id = hash_to_bytes("62d957f2b5cdc515bea0a46252a3ab29ee271636") | ||||
| assert actual_load_status == { | assert actual_load_status == { | ||||
| "status": "eventful", | "status": "eventful", | ||||
| "snapshot_id": snapshot_id.hex(), | "snapshot_id": snapshot_id.hex(), | ||||
| } | } | ||||
| assert_last_visit_matches( | assert_last_visit_matches( | ||||
| swh_storage, url, status="full", type="pypi", snapshot=snapshot_id | swh_storage, url, status="full", type="pypi", snapshot=snapshot_id | ||||
| ) | ) | ||||
| expected_snapshot = Snapshot( | expected_snapshot = Snapshot( | ||||
| id=snapshot_id, | id=snapshot_id, | ||||
| branches={ | branches={ | ||||
| b"releases/1.1.0": SnapshotBranch( | b"releases/1.1.0": SnapshotBranch( | ||||
| target=hash_to_bytes("4c99891f93b81450385777235a37b5e966dd1571"), | target=hash_to_bytes("9478c9981887fdf5ada3f1fcb20c81069cdf4c44"), | ||||
| target_type=TargetType.REVISION, | target_type=TargetType.RELEASE, | ||||
| ), | ), | ||||
| b"releases/1.2.0": SnapshotBranch( | b"releases/1.2.0": SnapshotBranch( | ||||
| target=hash_to_bytes("e445da4da22b31bfebb6ffc4383dbf839a074d21"), | target=hash_to_bytes("a1e10745d375be66c1b65e55c0c15fe98776b53c"), | ||||
| target_type=TargetType.REVISION, | target_type=TargetType.RELEASE, | ||||
| ), | ), | ||||
| b"HEAD": SnapshotBranch( | b"HEAD": SnapshotBranch( | ||||
| target=b"releases/1.2.0", target_type=TargetType.ALIAS, | target=b"releases/1.2.0", target_type=TargetType.ALIAS, | ||||
| ), | ), | ||||
| }, | }, | ||||
| ) | ) | ||||
| check_snapshot(expected_snapshot, swh_storage) | check_snapshot(expected_snapshot, swh_storage) | ||||
| stats = get_stats(swh_storage) | stats = get_stats(swh_storage) | ||||
| assert { | assert { | ||||
| "content": 6, | "content": 6, | ||||
| "directory": 4, | "directory": 4, | ||||
| "origin": 1, | "origin": 1, | ||||
| "origin_visit": 1, | "origin_visit": 1, | ||||
| "release": 0, | "release": 2, | ||||
| "revision": 2, | "revision": 0, | ||||
| "skipped_content": 0, | "skipped_content": 0, | ||||
| "snapshot": 1, | "snapshot": 1, | ||||
| } == stats | } == stats | ||||
| actual_load_status2 = loader.load() | actual_load_status2 = loader.load() | ||||
| assert actual_load_status2 == { | assert actual_load_status2 == { | ||||
| "status": "uneventful", | "status": "uneventful", | ||||
| "snapshot_id": actual_load_status2["snapshot_id"], | "snapshot_id": actual_load_status2["snapshot_id"], | ||||
| Show All 16 Lines | def test_pypi_incremental_visit(swh_storage, requests_mock_datadir_visits): | ||||
| """With prior visit, 2nd load will result with a different snapshot | """With prior visit, 2nd load will result with a different snapshot | ||||
| """ | """ | ||||
| url = "https://pypi.org/project/0805nexter" | url = "https://pypi.org/project/0805nexter" | ||||
| loader = PyPILoader(swh_storage, url) | loader = PyPILoader(swh_storage, url) | ||||
| visit1_actual_load_status = loader.load() | visit1_actual_load_status = loader.load() | ||||
| visit1_stats = get_stats(swh_storage) | visit1_stats = get_stats(swh_storage) | ||||
| expected_snapshot_id = hash_to_bytes("ba6e158ada75d0b3cfb209ffdf6daa4ed34a227a") | expected_snapshot_id = hash_to_bytes("62d957f2b5cdc515bea0a46252a3ab29ee271636") | ||||
| assert visit1_actual_load_status == { | assert visit1_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="pypi", snapshot=expected_snapshot_id | swh_storage, url, status="full", type="pypi", snapshot=expected_snapshot_id | ||||
| ) | ) | ||||
| assert { | assert { | ||||
| "content": 6, | "content": 6, | ||||
| "directory": 4, | "directory": 4, | ||||
| "origin": 1, | "origin": 1, | ||||
| "origin_visit": 1, | "origin_visit": 1, | ||||
| "release": 0, | "release": 2, | ||||
| "revision": 2, | "revision": 0, | ||||
| "skipped_content": 0, | "skipped_content": 0, | ||||
| "snapshot": 1, | "snapshot": 1, | ||||
| } == visit1_stats | } == visit1_stats | ||||
| # Reset internal state | # Reset internal state | ||||
| del loader._cached__raw_info | del loader._cached__raw_info | ||||
| del loader._cached_info | del loader._cached_info | ||||
| visit2_actual_load_status = loader.load() | visit2_actual_load_status = loader.load() | ||||
| visit2_stats = get_stats(swh_storage) | visit2_stats = get_stats(swh_storage) | ||||
| assert visit2_actual_load_status["status"] == "eventful", visit2_actual_load_status | assert visit2_actual_load_status["status"] == "eventful", visit2_actual_load_status | ||||
| expected_snapshot_id2 = hash_to_bytes("2e5149a7b0725d18231a37b342e9b7c4e121f283") | expected_snapshot_id2 = hash_to_bytes("6a8a84e7f765bed4362315fb054adb2466598636") | ||||
| assert visit2_actual_load_status == { | assert visit2_actual_load_status == { | ||||
| "status": "eventful", | "status": "eventful", | ||||
| "snapshot_id": expected_snapshot_id2.hex(), | "snapshot_id": expected_snapshot_id2.hex(), | ||||
| } | } | ||||
| assert_last_visit_matches( | assert_last_visit_matches( | ||||
| swh_storage, url, status="full", type="pypi", snapshot=expected_snapshot_id2 | swh_storage, url, status="full", type="pypi", snapshot=expected_snapshot_id2 | ||||
| ) | ) | ||||
| expected_snapshot = Snapshot( | expected_snapshot = Snapshot( | ||||
| id=expected_snapshot_id2, | id=expected_snapshot_id2, | ||||
| branches={ | branches={ | ||||
| b"releases/1.1.0": SnapshotBranch( | b"releases/1.1.0": SnapshotBranch( | ||||
| target=hash_to_bytes("4c99891f93b81450385777235a37b5e966dd1571"), | target=hash_to_bytes("9478c9981887fdf5ada3f1fcb20c81069cdf4c44"), | ||||
| target_type=TargetType.REVISION, | target_type=TargetType.RELEASE, | ||||
| ), | ), | ||||
| b"releases/1.2.0": SnapshotBranch( | b"releases/1.2.0": SnapshotBranch( | ||||
| target=hash_to_bytes("e445da4da22b31bfebb6ffc4383dbf839a074d21"), | target=hash_to_bytes("a1e10745d375be66c1b65e55c0c15fe98776b53c"), | ||||
| target_type=TargetType.REVISION, | target_type=TargetType.RELEASE, | ||||
| ), | ), | ||||
| b"releases/1.3.0": SnapshotBranch( | b"releases/1.3.0": SnapshotBranch( | ||||
| target=hash_to_bytes("51247143b01445c9348afa9edfae31bf7c5d86b1"), | target=hash_to_bytes("d46442e99bb6e05df5f75a7f0f7f61a4f2098147"), | ||||
| target_type=TargetType.REVISION, | target_type=TargetType.RELEASE, | ||||
| ), | ), | ||||
| b"HEAD": SnapshotBranch( | b"HEAD": SnapshotBranch( | ||||
| target=b"releases/1.3.0", target_type=TargetType.ALIAS, | target=b"releases/1.3.0", target_type=TargetType.ALIAS, | ||||
| ), | ), | ||||
| }, | }, | ||||
| ) | ) | ||||
| assert_last_visit_matches( | assert_last_visit_matches( | ||||
| swh_storage, url, status="full", type="pypi", snapshot=expected_snapshot.id | swh_storage, url, status="full", type="pypi", snapshot=expected_snapshot.id | ||||
| ) | ) | ||||
| check_snapshot(expected_snapshot, swh_storage) | check_snapshot(expected_snapshot, swh_storage) | ||||
| assert { | assert { | ||||
| "content": 6 + 1, # 1 more content | "content": 6 + 1, # 1 more content | ||||
| "directory": 4 + 2, # 2 more directories | "directory": 4 + 2, # 2 more directories | ||||
| "origin": 1, | "origin": 1, | ||||
| "origin_visit": 1 + 1, | "origin_visit": 1 + 1, | ||||
| "release": 0, | "release": 2 + 1, # 1 more release | ||||
| "revision": 2 + 1, # 1 more revision | "revision": 0, | ||||
| "skipped_content": 0, | "skipped_content": 0, | ||||
| "snapshot": 1 + 1, # 1 more snapshot | "snapshot": 1 + 1, # 1 more snapshot | ||||
| } == visit2_stats | } == visit2_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://files.pythonhosted.org") | if m.url.startswith("https://files.pythonhosted.org") | ||||
| Show All 16 Lines | |||||
| def test_pypi_visit_1_release_with_2_artifacts(swh_storage, requests_mock_datadir): | def test_pypi_visit_1_release_with_2_artifacts(swh_storage, requests_mock_datadir): | ||||
| """With no prior visit, load a pypi project ends up with 1 snapshot | """With no prior visit, load a pypi project ends up with 1 snapshot | ||||
| """ | """ | ||||
| url = "https://pypi.org/project/nexter" | url = "https://pypi.org/project/nexter" | ||||
| loader = PyPILoader(swh_storage, url) | loader = PyPILoader(swh_storage, url) | ||||
| actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
| expected_snapshot_id = hash_to_bytes("a27e638a4dad6fbfa273c6ebec1c4bf320fb84c6") | expected_snapshot_id = hash_to_bytes("a136ee226316276c347d7be3da07df5828605927") | ||||
| 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="pypi", snapshot=expected_snapshot_id | swh_storage, url, status="full", type="pypi", snapshot=expected_snapshot_id | ||||
| ) | ) | ||||
| expected_snapshot = Snapshot( | expected_snapshot = Snapshot( | ||||
| id=expected_snapshot_id, | id=expected_snapshot_id, | ||||
| branches={ | branches={ | ||||
| b"releases/1.1.0/nexter-1.1.0.zip": SnapshotBranch( | b"releases/1.1.0/nexter-1.1.0.zip": SnapshotBranch( | ||||
| target=hash_to_bytes("4c99891f93b81450385777235a37b5e966dd1571"), | target=hash_to_bytes("9478c9981887fdf5ada3f1fcb20c81069cdf4c44"), | ||||
| target_type=TargetType.REVISION, | target_type=TargetType.RELEASE, | ||||
| ), | ), | ||||
| b"releases/1.1.0/nexter-1.1.0.tar.gz": SnapshotBranch( | b"releases/1.1.0/nexter-1.1.0.tar.gz": SnapshotBranch( | ||||
| target=hash_to_bytes("0bf88f5760cca7665d0af4d6575d9301134fe11a"), | target=hash_to_bytes("b3391cb4007fb6872c4dfab476a7cfe7443a1bb4"), | ||||
| target_type=TargetType.REVISION, | target_type=TargetType.RELEASE, | ||||
| ), | ), | ||||
| }, | }, | ||||
| ) | ) | ||||
| check_snapshot(expected_snapshot, swh_storage) | check_snapshot(expected_snapshot, swh_storage) | ||||
| def test_pypi_artifact_with_no_intrinsic_metadata(swh_storage, requests_mock_datadir): | def test_pypi_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 | ||||
| Show All 25 Lines | def test_pypi_origin_not_found(swh_storage, requests_mock_datadir): | ||||
| assert loader.load() == {"status": "failed"} | assert loader.load() == {"status": "failed"} | ||||
| assert_last_visit_matches( | assert_last_visit_matches( | ||||
| swh_storage, url, status="not_found", type="pypi", snapshot=None | swh_storage, url, status="not_found", type="pypi", snapshot=None | ||||
| ) | ) | ||||
| def test_pypi_build_revision_missing_version_in_pkg_info(swh_storage, tmp_path): | def test_pypi_build_release_missing_version_in_pkg_info(swh_storage, tmp_path): | ||||
| """Simulate revision build when Version field is missing in PKG-INFO file.""" | """Simulate release build when Version field is missing in PKG-INFO file.""" | ||||
| url = "https://pypi.org/project/GermlineFilter" | url = "https://pypi.org/project/GermlineFilter" | ||||
| # create package info | # create package info | ||||
| p_info = PyPIPackageInfo( | p_info = PyPIPackageInfo( | ||||
| url=url, | url=url, | ||||
| filename="GermlineFilter-1.2.tar.gz", | filename="GermlineFilter-1.2.tar.gz", | ||||
| directory_extrinsic_metadata=[], | directory_extrinsic_metadata=[], | ||||
| raw_info={}, | raw_info={}, | ||||
| comment_text="", | comment_text="", | ||||
| Show All 12 Lines | |||||
| Author: Cristian Caloian (OICR) | Author: Cristian Caloian (OICR) | ||||
| Author-email: cristian.caloian@oicr.on.ca | Author-email: cristian.caloian@oicr.on.ca | ||||
| License: UNKNOWN | License: UNKNOWN | ||||
| Description: UNKNOWN | Description: UNKNOWN | ||||
| Platform: UNKNOWN""" | Platform: UNKNOWN""" | ||||
| ) | ) | ||||
| directory = hash_to_bytes("8b864d66f356afe35033d58f8e03b7c23a66751f") | directory = hash_to_bytes("8b864d66f356afe35033d58f8e03b7c23a66751f") | ||||
| # attempt to build revision | # attempt to build release | ||||
| loader = PyPILoader(swh_storage, url) | loader = PyPILoader(swh_storage, url) | ||||
| revision = loader.build_revision(p_info, str(tmp_path), directory) | release = loader.build_release("1.2", p_info, str(tmp_path), directory) | ||||
| # without comment_text and version in PKG-INFO, message should be empty | # without comment_text and version in PKG-INFO, message should be empty | ||||
| assert revision.message == b"" | assert release.message == b"" | ||||
| def test_filter_out_invalid_sdists(swh_storage, requests_mock): | def test_filter_out_invalid_sdists(swh_storage, requests_mock): | ||||
| project_name = "swh-test-sdist-filtering" | project_name = "swh-test-sdist-filtering" | ||||
| version = "1.0.0" | version = "1.0.0" | ||||
| url = f"https://pypi.org/project/{project_name}" | url = f"https://pypi.org/project/{project_name}" | ||||
| json_url = f"https://pypi.org/pypi/{project_name}/json" | json_url = f"https://pypi.org/pypi/{project_name}/json" | ||||
| Show All 29 Lines | |||||
this looks weird.
Should be one of the other branch name instead, ain't it?