sample_data = <swh.storage.tests.storage_data.StorageData object at 0x7fb7f97122e8>
def test_buffering_proxy_storage_release_threshold_hit(sample_data) -> None:
releases = sample_data.releases
threshold = 2
assert len(releases) > threshold
storage = get_storage_with_buffer_config(
min_batch_size={"release": threshold,} # configuration set
)
> s = storage.release_add(releases)
.tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_buffer.py:371:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.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 0x7fb7f97127f0>
releases = [Release(name=b'v0.0.1', message=b'synthetic release', target=b'\x01\xa7\x11O6\xfd\xdd^\xf2Q\x1b,\xad\xda#zh\xad\xbb\x...econds=0), offset=-120, negative_utc=False), metadata=None, id=b'i\x02\xbdL\x82\xb7\xd1\x9aB\x1d"J\xed\xab+t\x19~B\r')]
def release_add(self, releases: List[Release]) -> Dict:
> 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:635: AttributeError
TEST RESULT
TEST RESULT
- Run At
- Apr 22 2021, 8:33 PM