provenance_storage = <RemoteProvenanceStorage url=mock://example.com/>
repo = 'cmdbts2'
@pytest.mark.parametrize(
"repo",
("cmdbts2",),
)
def test_provenance_storage_revision(
provenance_storage: ProvenanceStorageInterface,
repo: str,
) -> None:
"""Tests revision methods for every `ProvenanceStorageInterface` implementation."""
# Read data/README.md for more details on how these datasets are generated.
data = load_repo_data(repo)
# Test revision methods.
# Add all revisions present in the current repo to the storage, assigning their
# dates and an arbitrary origin to each one. Then check that the returned results
# when querying are the same.
origin = next(iter(data["origin"]))
org_sha1 = hash_to_bytes(origin_identifier(origin))
# Origin must be inserted in advance.
assert provenance_storage.origin_set_url({org_sha1: origin["url"]})
dates = {rev["id"]: ts2dt(rev["date"]) for rev in data["revision"]}
orgs = {rev["id"]: org_sha1 for rev in data["revision"]}
assert set(dates.keys()) == set(orgs.keys())
revs = {
rev: RevisionData(date, org)
for sha1, date in dates.items()
for rev, org in orgs.items()
if rev == sha1
}
assert dates
assert orgs
assert provenance_storage.revision_set_date(dates)
assert provenance_storage.revision_set_origin(orgs)
assert provenance_storage.revision_get(set(revs.keys())) == revs
> assert provenance_storage.entity_get_all(EntityType.REVISION) == set(revs.keys())
.tox/py3/lib/python3.7/site-packages/swh/provenance/tests/test_provenance_storage.py:227:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/py3/lib/python3.7/site-packages/swh/core/api/__init__.py:181: in meth_
return self.post(meth._endpoint_path, post_data)
.tox/py3/lib/python3.7/site-packages/swh/core/api/__init__.py:278: in post
return self._decode_response(response)
.tox/py3/lib/python3.7/site-packages/swh/core/api/__init__.py:354: in _decode_response
self.raise_for_status(response)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <RemoteProvenanceStorage url=mock://example.com/>
response = <Response [500]>
def raise_for_status(self, response) -> None:
"""check response HTTP status code and raise an exception if it denotes an
error; do nothing otherwise
"""
status_code = response.status_code
status_class = response.status_code // 100
if status_code == 404:
raise RemoteException(payload="404 not found", response=response)
exception = None
# TODO: only old servers send pickled error; stop trying to unpickle
# after they are all upgraded
try:
if status_class == 4:
data = self._decode_response(response, check_status=False)
if isinstance(data, dict):
# TODO: remove "exception" key check once all servers
# are using new schema
exc_data = data["exception"] if "exception" in data else data
for exc_type in self.reraise_exceptions:
if exc_type.__name__ == exc_data["type"]:
exception = exc_type(*exc_data["args"])
break
else:
exception = RemoteException(payload=exc_data, response=response)
else:
exception = pickle.loads(data)
elif status_class == 5:
data = self._decode_response(response, check_status=False)
if "exception_pickled" in data:
exception = pickle.loads(data["exception_pickled"])
else:
# TODO: remove "exception" key check once all servers
# are using new schema
exc_data = data["exception"] if "exception" in data else data
exception = RemoteException(payload=exc_data, response=response)
except (TypeError, pickle.UnpicklingError):
raise RemoteException(payload=data, response=response)
if exception:
> raise exception from None
E swh.core.api.RemoteException: <!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 3.2 Final//EN">
E <title>500 Internal Server Error</title>
E <h1>Internal Server Error</h1>
E <p>The server encountered an internal error and was unable to complete your request. Either the server is overloaded or there is an error in the application.</p>
.tox/py3/lib/python3.7/site-packages/swh/core/api/__init__.py:344: RemoteException
TEST RESULT
TEST RESULT
- Run At
- Sep 9 2021, 4:48 PM