self = <swh.storage.tests.test_cassandra.TestCassandraStorage object at 0x7f1944574ef0>
swh_storage = <swh.storage.cassandra.storage.CassandraStorage object at 0x7f194457ce48>
mocker = <pytest_mock.plugin.MockFixture object at 0x7f19444a27b8>
sample_data_model = {'authority': [MetadataAuthority(type=<MetadataAuthorityType.DEPOSIT_CLIENT: 'deposit_client'>, url='http://hal.inria....11', length=7, status='visible', data=b'424242\n', ctime=datetime.datetime(2019, 12, 1, 0, 0, tzinfo=tzlocal()))], ...}
def test_content_find_murmur3_collision(
self, swh_storage, mocker, sample_data_model
):
"""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_model["content"][:2]
]
# always return a token
def mock_cgtfsh(algo, hash_):
nonlocal called
called += 1
assert algo in ("sha1", "sha1_git")
return [123456]
mocker.patch.object(
> swh_storage.storage._cql_runner,
"content_get_tokens_from_single_hash",
mock_cgtfsh,
)
E AttributeError: 'CassandraStorage' object has no attribute 'storage'
.tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_cassandra.py:309: AttributeError
TEST RESULT
TEST RESULT
- Run At
- Jul 21 2020, 9:55 AM