Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show All 18 Lines | |||||
| import psycopg2 | import psycopg2 | ||||
| import pytest | import pytest | ||||
| from hypothesis import given, strategies, settings, HealthCheck | from hypothesis import given, strategies, settings, HealthCheck | ||||
| from typing import ClassVar, Optional | from typing import ClassVar, Optional | ||||
| from swh.model import from_disk, identifiers | from swh.model import from_disk, identifiers | ||||
| from swh.model.model import SHA1_SIZE | |||||
| from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
| from swh.model.hypothesis_strategies import objects | from swh.model.hypothesis_strategies import objects | ||||
| from swh.storage import HashCollision | from swh.storage import HashCollision | ||||
| from swh.storage.converters import origin_url_to_sha1 as sha1 | from swh.storage.converters import origin_url_to_sha1 as sha1 | ||||
| from .storage_data import data | from .storage_data import data | ||||
| ▲ Show 20 Lines • Show All 355 Lines • ▼ Show 20 Lines | def test_content_get_partition_full(self, swh_storage, swh_contents): | ||||
| actual_result = swh_storage.content_get_partition(0, 1) | actual_result = swh_storage.content_get_partition(0, 1) | ||||
| assert actual_result['next_page_token'] is None | assert actual_result['next_page_token'] is None | ||||
| actual_contents = actual_result['contents'] | actual_contents = actual_result['contents'] | ||||
| assert_contents_ok( | assert_contents_ok( | ||||
| expected_contents, actual_contents, ['sha1']) | expected_contents, actual_contents, ['sha1']) | ||||
| def test_content_get_partition_empty(self, swh_storage, swh_contents): | def test_content_get_partition_empty(self, swh_storage, swh_contents): | ||||
| """content_get_partition for an empty partition returns nothing""" | """content_get_partition when at least one of the partitions is | ||||
| first_sha1 = min(content['sha1'] for content in swh_contents) | empty""" | ||||
| first_sha1 = int.from_bytes(first_sha1, 'big') | expected_contents = {cont['sha1'] for cont in swh_contents | ||||
| # nb_partitions = smallest power of 2 such that first_sha1 is not in | if cont['status'] != 'absent'} | ||||
| # the first partition | # nb_partitions = smallest power of 2 such that at least one of | ||||
| nb_partitions = \ | # the partitions is empty | ||||
| 1 << (SHA1_SIZE*8 - math.floor(math.log2(first_sha1)) + 1) | nb_partitions = 1 << math.floor(math.log2(len(swh_contents)) + 1) | ||||
| actual_result = swh_storage.content_get_partition(0, nb_partitions) | seen_sha1s = [] | ||||
| for i in range(nb_partitions): | |||||
| actual_result = swh_storage.content_get_partition( | |||||
| i, nb_partitions, limit=len(swh_contents)+1) | |||||
| for cont in actual_result['contents']: | |||||
| seen_sha1s.append(cont['sha1']) | |||||
| # Limit is higher than the max number of results | |||||
| assert actual_result['next_page_token'] is None | assert actual_result['next_page_token'] is None | ||||
| assert len(actual_result['contents']) == 0 | |||||
| assert set(seen_sha1s) == expected_contents | |||||
| def test_content_get_partition_limit_none(self, swh_storage): | def test_content_get_partition_limit_none(self, swh_storage): | ||||
| """content_get_partition call with wrong limit input should fail""" | """content_get_partition call with wrong limit input should fail""" | ||||
| with pytest.raises(ValueError) as e: | with pytest.raises(ValueError) as e: | ||||
| swh_storage.content_get_partition(1, 16, limit=None) | swh_storage.content_get_partition(1, 16, limit=None) | ||||
| assert e.value.args == ('Development error: limit should not be None',) | assert e.value.args == ('Development error: limit should not be None',) | ||||
| ▲ Show 20 Lines • Show All 3,313 Lines • Show Last 20 Lines | |||||