self = <swh.storage.tests.test_cassandra.TestCassandraStorage object at 0x7fd6682d6b38> swh_storage = <swh.storage.cassandra.storage.CassandraStorage object at 0x7fd66f195da0> mocker = <pytest_mock.plugin.MockFixture object at 0x7fd668209278> sample_data = <swh.storage.tests.storage_data.StorageData object at 0x7fd668209518> def test_content_get_metadata_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_cgtfsh(algo, hash_): nonlocal called called += 1 assert algo in ("sha1", "sha1_git") return [123456] mocker.patch.object( swh_storage._cql_runner, "content_get_tokens_from_single_hash", mock_cgtfsh, ) # For all tokens, always return cont and cont2 cols = list(set(cont.to_dict()) - {"data"}) Row = namedtuple("Row", cols) def mock_cgft(token): nonlocal called called += 1 return [ Row(**{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 ) actual_result = swh_storage.content_get_metadata([cont.sha1]) assert called == 2 # dropping extra column not returned expected_cont = attr.evolve(cont, data=None, ctime=None).to_dict() > del expected_cont["ctime"] # forced to pop it as to_dict does not E KeyError: 'ctime' .tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_cassandra.py:276: KeyError