self = <swh.storage.tests.test_cassandra.TestCassandraStorage object at 0x7fda9042aeb8>
swh_storage = <swh.storage.validate.ValidatingProxyStorage object at 0x7fda903c99b0>
def test_content_add_from_lazy_content(self, swh_storage):
lazy_content = LazyContent.from_dict({**data.cont, "data": b"nope",})
insertion_start_time = datetime.datetime.now(tz=datetime.timezone.utc)
# bypass the validation proxy for now, to directly put a dict
actual_result = swh_storage.storage.content_add([lazy_content])
insertion_end_time = datetime.datetime.now(tz=datetime.timezone.utc)
assert actual_result == {
"content:add": 1,
"content:add:bytes": data.cont["length"],
}
# the fact that we retrieve the content object from the storage with
# the correct 'data' field ensures it has been 'called'
assert list(swh_storage.content_get([data.cont["sha1"]])) == [
{"sha1": data.cont["sha1"], "data": data.cont["data"]}
]
expected_cont = data.cont
del expected_cont["data"]
contents = [
obj
for (obj_type, obj) in swh_storage.journal_writer.journal.objects
if obj_type == "content"
]
assert len(contents) == 1
for obj in contents:
assert insertion_start_time <= obj.ctime
assert obj.ctime <= insertion_end_time
obj_d = obj.to_dict()
del obj_d["ctime"]
assert obj_d == expected_cont
swh_storage.refresh_stat_counters()
> assert swh_storage.stat_counters()["content"] == 1
E assert 4 == 1
E +4
E -1
.tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_storage.py:240: AssertionError
TEST RESULT
TEST RESULT
- Run At
- Apr 21 2020, 12:34 PM