swh_storage = <swh.storage.retry.RetryingProxyStorage object at 0x7f0c6dd6ad68>
deposit_client = <swh.loader.package.deposit.loader.ApiClient object at 0x7f0c6dd69470>
requests_mock_datadir = <requests_mock.mocker.Mocker object at 0x7f0c6dd69518>
def test_deposit_loading_ok_2(swh_storage, deposit_client, requests_mock_datadir):
"""Field dates should be se appropriately
"""
external_id = "some-external-id"
url = f"https://hal-test.archives-ouvertes.fr/{external_id}"
deposit_id = 777
loader = DepositLoader(
swh_storage, url, deposit_id, deposit_client, default_filename="archive.zip"
)
actual_load_status = loader.load()
expected_snapshot_id = "3e68440fdd7c81d283f8f3aebb6f0c8657864192"
assert actual_load_status == {
"status": "eventful",
"snapshot_id": expected_snapshot_id,
}
assert_last_visit_matches(loader.storage, url, status="full", type="deposit")
revision_id = "564d18943d71be80d0d73b43a77cfb205bcde96c"
expected_snapshot = Snapshot(
id=hash_to_bytes(expected_snapshot_id),
branches={
b"HEAD": SnapshotBranch(
target=hash_to_bytes(revision_id), target_type=TargetType.REVISION
)
},
)
check_snapshot(expected_snapshot, storage=loader.storage)
raw_meta = loader.client.metadata_get(deposit_id)
# Ensure the date fields are set appropriately in the revision
# Retrieve the revision
revision = loader.storage.revision_get([hash_to_bytes(revision_id)])[0]
assert revision
assert revision.date.to_dict() == raw_meta["deposit"]["author_date"]
assert revision.committer_date.to_dict() == raw_meta["deposit"]["committer_date"]
read_api = f"{DEPOSIT_URL}/{deposit_id}/meta/"
provider = {
"provider_name": "hal",
"provider_type": "deposit_client",
"provider_url": "https://hal-test.archives-ouvertes.fr/",
"metadata": None,
}
tool = {
"name": "swh-deposit",
"version": "0.0.1",
"configuration": {"sword_version": "2"},
}
assert revision.metadata == {
"extrinsic": {
"provider": read_api,
"raw": {
"origin": {"type": "deposit", "url": url,},
"origin_metadata": {
"metadata": raw_meta["metadata_dict"],
"provider": provider,
"tool": tool,
},
},
"when": revision.metadata["extrinsic"]["when"], # dynamic
},
"original_artifact": [
{
"checksums": {
"sha1": "f8c63d7c890a7453498e6cf9fef215d85ec6801d",
"sha256": "474bf646aeeff6d945eb752b1a9f8a40f3d81a88909ee7bd2d08cc822aa361e6", # noqa
},
"filename": "archive.zip",
"length": 956830,
"url": "https://deposit.softwareheritage.org/1/private/777/raw/",
}
],
}
fetcher = MetadataFetcher(name="swh-deposit", version="0.0.1",)
authority = MetadataAuthority(
type=MetadataAuthorityType.DEPOSIT_CLIENT,
url="https://hal-test.archives-ouvertes.fr/",
)
# Check the origin metadata swh side
origin_extrinsic_metadata = loader.storage.raw_extrinsic_metadata_get(
Origin(url).swhid(), authority
)
assert origin_extrinsic_metadata.next_page_token is None
all_metadata_raw: List[str] = raw_meta["metadata_raw"]
# 1 raw metadata xml + 1 json dict
assert len(origin_extrinsic_metadata.results) == len(all_metadata_raw) + 1
origin_swhid = Origin(url).swhid()
expected_metadata = []
for idx, raw_meta in enumerate(all_metadata_raw):
origin_meta = origin_extrinsic_metadata.results[idx]
expected_metadata.append(
RawExtrinsicMetadata(
target=origin_swhid,
discovery_date=origin_meta.discovery_date,
metadata=raw_meta.encode(),
format="sword-v2-atom-codemeta-v2",
authority=authority,
fetcher=fetcher,
)
)
origin_metadata = {
"metadata": all_metadata_raw,
"provider": provider,
"tool": tool,
}
expected_metadata.append(
RawExtrinsicMetadata(
target=origin_swhid,
discovery_date=origin_extrinsic_metadata.results[-1].discovery_date,
metadata=json.dumps(origin_metadata).encode(),
format="original-artifacts-json",
authority=authority,
fetcher=fetcher,
)
)
assert len(origin_extrinsic_metadata.results) == len(expected_metadata)
for orig_meta in origin_extrinsic_metadata.results:
assert orig_meta in expected_metadata
# Check the revision metadata swh side
directory_swhid = ExtendedSWHID(
object_type=ExtendedObjectType.DIRECTORY, object_id=revision.directory
)
actual_directory_metadata = loader.storage.raw_extrinsic_metadata_get(
directory_swhid, authority
)
assert actual_directory_metadata.next_page_token is None
assert len(actual_directory_metadata.results) == len(all_metadata_raw)
revision_swhid = CoreSWHID(
object_type=ObjectType.REVISION, object_id=hash_to_bytes(revision_id)
)
dir_metadata_template = RawExtrinsicMetadata(
target=directory_swhid,
format="sword-v2-atom-codemeta-v2",
authority=authority,
fetcher=fetcher,
origin=url,
revision=revision_swhid,
# to satisfy the constructor
discovery_date=now(),
metadata=b"",
)
expected_directory_metadata = []
for idx, raw_meta in enumerate(all_metadata_raw):
dir_metadata = actual_directory_metadata.results[idx]
expected_directory_metadata.append(
RawExtrinsicMetadata.from_dict(
{
**{
k: v
for (k, v) in dir_metadata_template.to_dict().items()
if k != "id"
},
"discovery_date": dir_metadata.discovery_date,
"metadata": raw_meta.encode(),
}
)
)
> assert actual_directory_metadata.results == expected_directory_metadata
E AssertionError: assert [RawExtrinsic...8;\x94q\x87')] == [RawExtrinsic...07\xe3O\xcc')]
E At index 0 diff: RawExtrinsicMetadata(target=ExtendedSWHID(namespace='swh', scheme_version=1, object_id=b'\xfd-\xf1-\xc5SL\x1d\xa1\xe9\x18\x0b\x91Q\x02\xfbo`\x1d\x19', object_type=<ExtendedObjectType.DIRECTORY: 'dir'>), discovery_date=datetime.datetime(2021, 3, 23, 10, 10, 1, tzinfo=datetime.timezone.utc), authority=MetadataAuthority(type=<MetadataAuthorityType.DEPOSIT_CLIENT: 'deposit_client'>, url='https://hal-test.archives-ouvertes.fr/', metadata=None), fetcher=MetadataFetcher(name='swh-deposit', version='0.0.1', metadata=None), format='sword-v2-atom-codemeta-v2', me...
E
E ...Full output truncated (7 lines hidden), use '-vv' to show
.tox/py3/lib/python3.7/site-packages/swh/loader/package/deposit/tests/test_deposit.py:474: AssertionError
TEST RESULT
TEST RESULT
- Run At
- Mar 23 2021, 11:10 AM