swh_storage = <swh.storage.retry.RetryingProxyStorage object at 0x7f77d22988d0>
deposit_client = <swh.loader.package.deposit.loader.ApiClient object at 0x7f77d2303160>
requests_mock_datadir = <requests_mock.mocker.Mocker object at 0x7f77d2303588>
def test_deposit_loading_ok(swh_storage, deposit_client, requests_mock_datadir):
url = "https://hal-test.archives-ouvertes.fr/some-external-id"
deposit_id = 666
loader = DepositLoader(swh_storage, url, deposit_id, deposit_client)
actual_load_status = loader.load()
expected_snapshot_id = "b2b327b33dc85818bd23c3ccda8b7e675a66ecbd"
assert actual_load_status == {
"status": "eventful",
"snapshot_id": expected_snapshot_id,
}
assert_last_visit_matches(loader.storage, url, status="full", type="deposit")
stats = get_stats(loader.storage)
assert {
"content": 303,
"directory": 12,
"origin": 1,
"origin_visit": 1,
"release": 0,
"revision": 1,
"skipped_content": 0,
"snapshot": 1,
} == stats
revision_id_hex = "637318680351f5d78856d13264faebbd91efe9bb"
revision_id = hash_to_bytes(revision_id_hex)
expected_snapshot = Snapshot(
id=hash_to_bytes(expected_snapshot_id),
branches={
b"HEAD": SnapshotBranch(
target=revision_id, target_type=TargetType.REVISION,
),
},
)
check_snapshot(expected_snapshot, storage=loader.storage)
revision = loader.storage.revision_get([revision_id])[0]
assert revision is not None
# check metadata
fetcher = MetadataFetcher(name="swh-deposit", version="0.0.1",)
authority = MetadataAuthority(
type=MetadataAuthorityType.DEPOSIT_CLIENT,
url="https://hal-test.archives-ouvertes.fr/",
)
# Check origin metadata
orig_meta = loader.storage.raw_extrinsic_metadata_get(
Origin(url).swhid(), authority
)
assert orig_meta.next_page_token is None
raw_meta = loader.client.metadata_get(deposit_id)
all_metadata_raw: List[str] = raw_meta["metadata_raw"]
# 2 raw metadata xml + 1 json dict
assert len(orig_meta.results) == len(all_metadata_raw) + 1
orig_meta0 = orig_meta.results[0]
assert orig_meta0.authority == authority
assert orig_meta0.fetcher == fetcher
# Check directory metadata
directory_id = hash_to_hex(revision.directory)
directory_swhid = CoreSWHID(
> object_type=ObjectType.DIRECTORY, object_id=directory_id
)
.tox/py3/lib/python3.7/site-packages/swh/loader/package/deposit/tests/test_deposit.py:255:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
<attrs generated init swh.model.identifiers.CoreSWHID>:11: in __init__
__attr_validator_object_id(self, __attr_object_id, self.object_id)
.tox/py3/lib/python3.7/site-packages/attr/_make.py:2723: in __call__
v(inst, attr, value)
.tox/py3/lib/python3.7/site-packages/attrs_strict/_type_validation.py:105: in _validator
_validate_elements(attribute, field, attribute.type)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
attribute = Attribute(name='object_id', default=NOTHING, validator=_AndValidator(_validators=(<function type_validator.<locals>._v...t=True, metadata=mappingproxy({}), type=<class 'bytes'>, converter=None, kw_only=True, inherited=True, on_setattr=None)
value = 'fd2df12dc5534c1da1e9180b915102fb6f601d19'
expected_type = <class 'bytes'>
def _validate_elements(attribute, value, expected_type):
if expected_type is None:
return
base_type = _get_base_type(expected_type)
if base_type == typing.Any:
return
if isinstance(base_type, (str, ForwardRef)):
# These base_types happen when you have string annotations and cannot
# be used in isinstance.
raise _StringAnnotationError()
elif base_type == Literal or base_type == type(Literal): # type: ignore
_handle_literal(attribute, value, expected_type)
elif base_type == typing.Union: # type: ignore
_handle_union(attribute, value, expected_type)
elif not isinstance(value, base_type):
> raise AttributeTypeError(value, attribute)
E attrs_strict._error.AttributeTypeError: object_id must be <class 'bytes'> (got fd2df12dc5534c1da1e9180b915102fb6f601d19 that is a <class 'str'>)
.tox/py3/lib/python3.7/site-packages/attrs_strict/_type_validation.py:131: AttributeTypeError
TEST RESULT
TEST RESULT
- Run At
- Mar 2 2021, 10:24 AM