sample_data = <swh.storage.tests.storage_data.StorageData object at 0x7f3a3c8f19b0>
def test_buffering_proxy_storage_revision_deduplicate(sample_data) -> None:
revisions = sample_data.revisions[:2]
storage = get_storage_with_buffer_config(min_batch_size={"revision": 2,})
s = storage.revision_add([revisions[0], revisions[0]])
assert s == {}
s = storage.revision_add([revisions[0]])
assert s == {}
> s = storage.revision_add([revisions[1]])
.tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_buffer.py:326:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/py3/lib/python3.7/site-packages/swh/storage/buffer.py:138: in object_add
return self.flush()
.tox/py3/lib/python3.7/site-packages/swh/storage/buffer.py:156: in flush
stats = add_fn(list(batch))
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <swh.storage.in_memory.InMemoryStorage object at 0x7f3a3c8f14a8>
revisions = [Revision(message=b'hello', author=Person(fullname=b'Nicolas Dandrimont <nicolas@example.com> ', name=b'Nicolas Dandri...1b,\xad\xda#zh\xad\xbb\x12',), id=b'\xa6F\xdd\x94\xc9\x12\x82\x96Y\xb2*\x1e~\x14=/\xa5\xeb\xde\x1b', extra_headers=())]
def revision_add(self, revisions: List[Revision]) -> Dict:
# Filter-out revisions already in the database
> if not self._allow_overwrite:
E AttributeError: 'InMemoryStorage' object has no attribute '_allow_overwrite'
.tox/py3/lib/python3.7/site-packages/swh/storage/cassandra/storage.py:526: AttributeError
TEST RESULT
TEST RESULT
- Run At
- Apr 29 2021, 10:23 AM