self = <swh.storage.tests.test_cassandra.TestCassandraStorage object at 0x7f1c743665f8>
swh_storage = <swh.storage.validate.ValidatingProxyStorage object at 0x7f1c74332ef0>
def test_stat_counters(self, swh_storage):
expected_keys = ["content", "directory", "origin", "revision"]
# Initially, all counters are 0
swh_storage.refresh_stat_counters()
counters = swh_storage.stat_counters()
assert set(expected_keys) <= set(counters)
for key in expected_keys:
assert counters[key] == 0
# Add a content. Only the content counter should increase.
swh_storage.content_add([data.cont])
swh_storage.refresh_stat_counters()
counters = swh_storage.stat_counters()
assert set(expected_keys) <= set(counters)
for key in expected_keys:
if key != "content":
assert counters[key] == 0
assert counters["content"] == 1
# Add other objects. Check their counter increased as well.
origin_url = 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"],
)
]
)
swh_storage.directory_add([data.dir])
> swh_storage.revision_add([data.revision])
.tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_storage.py:2955:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/py3/lib/python3.7/site-packages/swh/storage/validate.py:104: in revision_add
[dict_converter(Revision, r) for r in revisions]
.tox/py3/lib/python3.7/site-packages/swh/storage/cassandra/storage.py:433: in revision_add
revobject = revision_to_db(revision)
.tox/py3/lib/python3.7/site-packages/swh/storage/cassandra/converters.py:38: in revision_to_db
db_revision["metadata"] = json.dumps(db_revision["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 0x7f1bcae16cc0>
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