self = <swh.storage.tests.test_cassandra.TestCassandraStorage object at 0x7f54984ca588>
swh_storage = <swh.storage.validate.ValidatingProxyStorage object at 0x7f54981a37f0>
def test_origin_visit_status_get_latest(self, swh_storage):
origin1 = Origin.from_dict(data.origin)
swh_storage.origin_add_one(data.origin)
# to have some reference visits
ov1, ov2 = swh_storage.origin_visit_add(
[
OriginVisit(
origin=origin1.url, date=data.date_visit1, type=data.type_visit1,
),
OriginVisit(
origin=origin1.url, date=data.date_visit2, type=data.type_visit2,
),
]
)
snapshot = Snapshot.from_dict(data.complete_snapshot)
swh_storage.snapshot_add([snapshot])
date_now = now()
date_now = round_to_milliseconds(date_now)
assert data.date_visit1 < data.date_visit2
assert data.date_visit2 < date_now
ovs1 = OriginVisitStatus(
origin=origin1.url,
visit=ov1.visit,
date=data.date_visit1,
status="partial",
snapshot=None,
)
ovs2 = OriginVisitStatus(
origin=origin1.url,
visit=ov1.visit,
date=data.date_visit2,
status="ongoing",
snapshot=None,
)
ovs3 = OriginVisitStatus(
origin=origin1.url,
visit=ov2.visit,
date=data.date_visit2 + datetime.timedelta(minutes=1), # to not be ignored
status="ongoing",
snapshot=None,
)
ovs4 = OriginVisitStatus(
origin=origin1.url,
visit=ov2.visit,
date=date_now,
status="full",
snapshot=snapshot.id,
metadata={"something": "wicked"},
)
> swh_storage.origin_visit_status_add([ovs1, ovs2, ovs3, ovs4])
.tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_storage.py:2325:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.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 0x7f5583d10cc0>
o = <swh.model.collections.ImmutableDict object at 0x7f5498191cf8>
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:30 AM