sample_data = <swh.storage.tests.storage_data.StorageData object at 0x7fab01e32b70>
def test_buffering_proxy_storage_directory_deduplicate(sample_data) -> None:
directories = sample_data.directories[:2]
storage = get_storage_with_buffer_config(min_batch_size={"directory": 2,})
s = storage.directory_add([directories[0], directories[0]])
assert s == {}
s = storage.directory_add([directories[0]])
assert s == {}
> s = storage.directory_add([directories[1]])
.tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_buffer.py:271:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.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 0x7fab01e32588>
directories = [Directory(entries=(DirectoryEntry(name=b'oof', type='file', target=b'6\xfa\xdew\x19<\xb6\xd2\xbd\x82aa\xa0\x97\x9dd\x...x04', perms=<DentryPerms.directory: 16384>)), id=b'RV\xe8V\xa0\xa0\x89\x89f\xd6\xba\x14\xfe\xb48\x8b\x8b\x82\xd3\x02')]
def directory_add(self, directories: List[Directory]) -> Dict:
to_add = {d.id: d for d in directories}.values()
> if self._check_missing:
E AttributeError: 'InMemoryStorage' object has no attribute '_check_missing'
.tox/py3/lib/python3.7/site-packages/swh/storage/cassandra/storage.py:397: AttributeError
TEST RESULT
TEST RESULT
- Run At
- Apr 22 2021, 8:40 PM