Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/deposit/tests/test_deposit.py
Show First 20 Lines • Show All 200 Lines • ▼ Show 20 Lines | def test_deposit_loading_ok(swh_config, requests_mock_datadir): | ||||
fetcher = MetadataFetcher(name="swh-deposit", version="0.0.1",) | fetcher = MetadataFetcher(name="swh-deposit", version="0.0.1",) | ||||
authority = MetadataAuthority( | authority = MetadataAuthority( | ||||
type=MetadataAuthorityType.DEPOSIT_CLIENT, | type=MetadataAuthorityType.DEPOSIT_CLIENT, | ||||
url="https://hal-test.archives-ouvertes.fr/", | url="https://hal-test.archives-ouvertes.fr/", | ||||
) | ) | ||||
# Check origin metadata | # Check origin metadata | ||||
orig_meta = loader.storage.object_metadata_get( | orig_meta = loader.storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.ORIGIN, url, authority | MetadataTargetType.ORIGIN, url, authority | ||||
) | ) | ||||
assert orig_meta is not None | assert orig_meta is not None | ||||
assert isinstance(orig_meta, dict) | assert isinstance(orig_meta, dict) | ||||
assert len(orig_meta["results"]) == 1 | assert len(orig_meta["results"]) == 1 | ||||
assert orig_meta["next_page_token"] is None | assert orig_meta["next_page_token"] is None | ||||
orig_meta0 = orig_meta["results"][0] | orig_meta0 = orig_meta["results"][0] | ||||
assert orig_meta0.authority == authority | assert orig_meta0.authority == authority | ||||
assert orig_meta0.fetcher == fetcher | assert orig_meta0.fetcher == fetcher | ||||
# Check revision metadata | # Check revision metadata | ||||
revision_swhid = SWHID(object_type="revision", object_id=revision_id) | revision_swhid = SWHID(object_type="revision", object_id=revision_id) | ||||
rev_meta = loader.storage.object_metadata_get( | rev_meta = loader.storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.REVISION, revision_swhid, authority | MetadataTargetType.REVISION, revision_swhid, authority | ||||
) | ) | ||||
assert rev_meta is not None | assert rev_meta is not None | ||||
assert isinstance(rev_meta, dict) | assert isinstance(rev_meta, dict) | ||||
assert len(rev_meta["results"]) == 1 | assert len(rev_meta["results"]) == 1 | ||||
assert rev_meta["next_page_token"] is None | assert rev_meta["next_page_token"] is None | ||||
rev_meta0 = rev_meta["results"][0] | rev_meta0 = rev_meta["results"][0] | ||||
assert rev_meta0.authority == authority | assert rev_meta0.authority == authority | ||||
▲ Show 20 Lines • Show All 106 Lines • ▼ Show 20 Lines | def test_deposit_loading_ok_2(swh_config, requests_mock_datadir): | ||||
fetcher = MetadataFetcher(name="swh-deposit", version="0.0.1",) | fetcher = MetadataFetcher(name="swh-deposit", version="0.0.1",) | ||||
authority = MetadataAuthority( | authority = MetadataAuthority( | ||||
type=MetadataAuthorityType.DEPOSIT_CLIENT, | type=MetadataAuthorityType.DEPOSIT_CLIENT, | ||||
url="https://hal-test.archives-ouvertes.fr/", | url="https://hal-test.archives-ouvertes.fr/", | ||||
) | ) | ||||
# Check the origin metadata swh side | # Check the origin metadata swh side | ||||
orig_meta = loader.storage.object_metadata_get( | orig_meta = loader.storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.ORIGIN, url, authority | MetadataTargetType.ORIGIN, url, authority | ||||
) | ) | ||||
assert orig_meta is not None | assert orig_meta is not None | ||||
assert isinstance(orig_meta, dict) | assert isinstance(orig_meta, dict) | ||||
assert len(orig_meta["results"]) == 1 | assert len(orig_meta["results"]) == 1 | ||||
assert orig_meta["next_page_token"] is None | assert orig_meta["next_page_token"] is None | ||||
assert len(orig_meta["results"]) == 1 | assert len(orig_meta["results"]) == 1 | ||||
Show All 18 Lines | expected_metadata = RawExtrinsicMetadata( | ||||
authority=authority, | authority=authority, | ||||
fetcher=fetcher, | fetcher=fetcher, | ||||
) | ) | ||||
assert orig_meta0 == expected_metadata | assert orig_meta0 == expected_metadata | ||||
# Check the revision metadata swh side | # Check the revision metadata swh side | ||||
revision_swhid = SWHID(object_type="revision", object_id=revision_id) | revision_swhid = SWHID(object_type="revision", object_id=revision_id) | ||||
rev_meta = loader.storage.object_metadata_get( | rev_meta = loader.storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.REVISION, revision_swhid, authority | MetadataTargetType.REVISION, revision_swhid, authority | ||||
) | ) | ||||
assert rev_meta is not None | assert rev_meta is not None | ||||
assert isinstance(rev_meta, dict) | assert isinstance(rev_meta, dict) | ||||
assert len(rev_meta["results"]) == 1 | assert len(rev_meta["results"]) == 1 | ||||
assert rev_meta["next_page_token"] is None | assert rev_meta["next_page_token"] is None | ||||
assert len(rev_meta["results"]) == 1 | assert len(rev_meta["results"]) == 1 | ||||
Show All 30 Lines |