self = <swh.storage.tests.test_cassandra.TestCassandraStorage object at 0x7f1c944767b8>
swh_storage = <swh.storage.validate.ValidatingProxyStorage object at 0x7f1cd4664cc0>
def test_origin_visit_get_by(self, swh_storage):
origin_url = swh_storage.origin_add_one(data.origin)
origin_url2 = swh_storage.origin_add_one(data.origin2)
visit = OriginVisit(
origin=origin_url, date=data.date_visit2, type=data.type_visit2,
)
origin_visit1 = swh_storage.origin_visit_add([visit])[0]
swh_storage.snapshot_add([data.snapshot])
swh_storage.origin_visit_status_add(
[
OriginVisitStatus(
origin=origin_url,
visit=origin_visit1.visit,
date=now(),
status="ongoing",
snapshot=data.snapshot["id"],
)
]
)
# Add some other {origin, visit} entries
visit2 = OriginVisit(
origin=origin_url, date=data.date_visit3, type=data.type_visit3,
)
visit3 = OriginVisit(
origin=origin_url2, date=data.date_visit3, type=data.type_visit3,
)
swh_storage.origin_visit_add([visit2, visit3])
# when
visit1_metadata = {
"contents": 42,
"directories": 22,
}
swh_storage.origin_visit_status_add(
[
OriginVisitStatus(
origin=origin_url,
visit=origin_visit1.visit,
date=now(),
status="full",
snapshot=data.snapshot["id"],
> metadata=visit1_metadata,
)
]
)
.tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_storage.py:2031:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/py3/lib/python3.7/site-packages/swh/storage/cassandra/storage.py:844: in origin_visit_status_add
self._origin_visit_status_add(visit_status)
.tox/py3/lib/python3.7/site-packages/swh/storage/cassandra/storage.py:832: in _origin_visit_status_add
self._cql_runner.origin_visit_status_add_one(visit_status)
.tox/py3/lib/python3.7/site-packages/swh/storage/cassandra/cql.py:106: in newf
self, *args, **kwargs, statement=self._prepared_statements[f.__name__]
.tox/py3/lib/python3.7/site-packages/swh/storage/cassandra/cql.py:782: in origin_visit_status_add_one
metadata = json.dumps(visit_update.metadata)
/usr/lib/python3.7/json/__init__.py:231: in dumps
return _default_encoder.encode(obj)
/usr/lib/python3.7/json/encoder.py:199: in encode
chunks = self.iterencode(o, _one_shot=True)
/usr/lib/python3.7/json/encoder.py:257: in iterencode
return _iterencode(o, 0)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <json.encoder.JSONEncoder object at 0x7f1d7fd98cc0>
o = <swh.model.collections.ImmutableDict object at 0x7f1d7f1a8d30>
def default(self, o):
"""Implement this method in a subclass such that it returns
a serializable object for ``o``, or calls the base implementation
(to raise a ``TypeError``).
For example, to support arbitrary iterators, you could
implement default like this::
def default(self, o):
try:
iterable = iter(o)
except TypeError:
pass
else:
return list(iterable)
# Let the base class default method raise the TypeError
return JSONEncoder.default(self, o)
"""
> raise TypeError(f'Object of type {o.__class__.__name__} '
f'is not JSON serializable')
E TypeError: Object of type ImmutableDict is not JSON serializable
/usr/lib/python3.7/json/encoder.py:179: TypeError
TEST RESULT
TEST RESULT
- Run At
- Jul 9 2020, 10:49 AM