Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_cassandra.py
| Show All 11 Lines | |||||
| from typing import Dict | from typing import Dict | ||||
| import attr | import attr | ||||
| import pytest | import pytest | ||||
| from swh.core.api.classes import stream_results | from swh.core.api.classes import stream_results | ||||
| from swh.storage import get_storage | from swh.storage import get_storage | ||||
| from swh.storage.cassandra import create_keyspace | from swh.storage.cassandra import create_keyspace | ||||
| from swh.storage.cassandra.model import ContentRow | from swh.storage.cassandra.model import ContentRow, ExtIDRow | ||||
| from swh.storage.cassandra.schema import HASH_ALGORITHMS, TABLES | from swh.storage.cassandra.schema import HASH_ALGORITHMS, TABLES | ||||
| from swh.storage.tests.storage_tests import ( | from swh.storage.tests.storage_tests import ( | ||||
| TestStorageGeneratedData as _TestStorageGeneratedData, | TestStorageGeneratedData as _TestStorageGeneratedData, | ||||
| ) | ) | ||||
| from swh.storage.tests.storage_tests import TestStorage as _TestStorage | from swh.storage.tests.storage_tests import TestStorage as _TestStorage | ||||
| from swh.storage.utils import now | from swh.storage.utils import now | ||||
| CONFIG_TEMPLATE = """ | CONFIG_TEMPLATE = """ | ||||
| ▲ Show 20 Lines • Show All 347 Lines • ▼ Show 20 Lines | ): | ||||
| # as we duplicated the returned results, dropping duplicate should yield | # as we duplicated the returned results, dropping duplicate should yield | ||||
| # the original length | # the original length | ||||
| assert len(set(actual_results)) == len(sample_data.contents) | assert len(set(actual_results)) == len(sample_data.contents) | ||||
| @pytest.mark.skip("content_update is not yet implemented for Cassandra") | @pytest.mark.skip("content_update is not yet implemented for Cassandra") | ||||
| def test_content_update(self): | def test_content_update(self): | ||||
| pass | pass | ||||
| def test_extid_murmur3_collision(self, swh_storage, mocker, sample_data): | |||||
| """The Murmur3 token is used as link from index table to the main | |||||
| table; and non-matching extid with colliding murmur3-hash | |||||
| are filtered-out when reading the main table. | |||||
| This test checks the extid methods do filter out these collision. | |||||
| """ | |||||
| swh_storage.extid_add(sample_data.extids) | |||||
| # For any token, always return all extids, i.e. make as if all tokens | |||||
| # for all extid entries collide | |||||
| def mock_egft(token): | |||||
| return [ | |||||
| ExtIDRow( | |||||
| extid_type=extid.extid_type, | |||||
| extid=extid.extid, | |||||
| target_type=extid.target.object_type.value, | |||||
| target=extid.target.object_id, | |||||
| ) | |||||
| for extid in sample_data.extids | |||||
| ] | |||||
| mocker.patch.object( | |||||
| swh_storage._cql_runner, "extid_get_from_token", mock_egft, | |||||
| ) | |||||
| for extid in sample_data.extids: | |||||
| extids = swh_storage.extid_get_from_target( | |||||
| target_type=extid.target.object_type, ids=[extid.target.object_id] | |||||
| ) | |||||
| assert extids == [extid] | |||||
| @pytest.mark.skip( | @pytest.mark.skip( | ||||
| 'The "person" table of the pgsql is a legacy thing, and not ' | 'The "person" table of the pgsql is a legacy thing, and not ' | ||||
| "supported by the cassandra backend." | "supported by the cassandra backend." | ||||
| ) | ) | ||||
| def test_person_fullname_unicity(self): | def test_person_fullname_unicity(self): | ||||
| pass | pass | ||||
| @pytest.mark.skip( | @pytest.mark.skip( | ||||
| Show All 28 Lines | |||||