self = <swh.storage.tests.test_cassandra.TestCassandraStorage object at 0x7f008f06b470>
swh_storage = <swh.storage.cassandra.storage.CassandraStorage object at 0x7f0079b274a8>
mocker = <pytest_mock.plugin.MockerFixture object at 0x7f0079230cc0>
sample_data = <swh.storage.tests.storage_data.StorageData object at 0x7f00792307f0>
def test_content_add_murmur3_collision(self, swh_storage, mocker, sample_data):
"""The Murmur3 token is used as link from index tables to the main
table; and non-matching contents with colliding murmur3-hash
are filtered-out when reading the main table.
This test checks the content methods do filter out these collision.
"""
called = 0
cont, cont2 = sample_data.contents[:2]
# always return a token
def mock_cgtfsa(algo, hashes):
nonlocal called
called += 1
assert algo in ("sha1", "sha1_git")
return [123456]
mocker.patch.object(
swh_storage._cql_runner, "content_get_tokens_from_single_algo", mock_cgtfsa,
)
# For all tokens, always return cont
def mock_cgft(token):
nonlocal called
called += 1
return [
ContentRow(
length=10,
ctime=datetime.datetime.now(),
status="present",
**{algo: getattr(cont, algo) for algo in HASH_ALGORITHMS},
)
]
mocker.patch.object(
> swh_storage._cql_runner, "content_get_from_token", mock_cgft
)
.tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_cassandra.py:303:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/py3/lib/python3.7/site-packages/pytest_mock/plugin.py:230: in object
**kwargs
.tox/py3/lib/python3.7/site-packages/pytest_mock/plugin.py:183: in _start_patch
mocked = p.start() # type: unittest.mock.MagicMock
/usr/lib/python3.7/unittest/mock.py:1399: in start
result = self.__enter__()
/usr/lib/python3.7/unittest/mock.py:1268: in __enter__
original, local = self.get_original()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <unittest.mock._patch object at 0x7f00792304a8>
def get_original(self):
target = self.getter()
name = self.attribute
original = DEFAULT
local = False
try:
original = target.__dict__[name]
except (AttributeError, KeyError):
original = getattr(target, name, DEFAULT)
else:
local = True
if name in _builtins and isinstance(target, ModuleType):
self.create = True
if not self.create and original is DEFAULT:
raise AttributeError(
> "%s does not have the attribute %r" % (target, name)
)
E AttributeError: <swh.storage.cassandra.cql.CqlRunner object at 0x7f0079b27dd8> does not have the attribute 'content_get_from_token'
/usr/lib/python3.7/unittest/mock.py:1242: AttributeError
TEST RESULT
TEST RESULT
- Run At
- Sep 15 2021, 3:20 PM