self = <swh.storage.tests.test_cassandra.TestCassandraStorage object at 0x7f007abe8518>
swh_storage = <swh.storage.cassandra.storage.CassandraStorage object at 0x7f008a59b780>
mocker = <pytest_mock.plugin.MockerFixture object at 0x7f008a5a1048>
sample_data = <swh.storage.tests.storage_data.StorageData object at 0x7f008a5a1ef0>
def test_content_find_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 collisions.
"""
called = 0
cont, cont2 = [attr.evolve(c, ctime=now()) for c in 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 and cont2
cols = list(set(cont.to_dict()) - {"data"})
def mock_cgft(token):
nonlocal called
called += 1
return [
ContentRow(**{col: getattr(cont, col) for col in cols})
for cont in [cont, cont2]
]
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:394:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.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 0x7f008a5a5940>
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 0x7f008a59b4a8> 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