Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/storage_tests.py
| Show All 13 Lines | |||||
| from unittest.mock import MagicMock | from unittest.mock import MagicMock | ||||
| import attr | import attr | ||||
| from hypothesis import HealthCheck, given, settings, strategies | from hypothesis import HealthCheck, given, settings, strategies | ||||
| import pytest | import pytest | ||||
| from swh.core.api.classes import stream_results | from swh.core.api.classes import stream_results | ||||
| from swh.model import from_disk | from swh.model import from_disk | ||||
| from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import DEFAULT_ALGORITHMS, hash_to_bytes | ||||
| from swh.model.hypothesis_strategies import objects | from swh.model.hypothesis_strategies import objects | ||||
| from swh.model.identifiers import CoreSWHID, ObjectType | from swh.model.identifiers import CoreSWHID, ObjectType | ||||
| from swh.model.model import ( | from swh.model.model import ( | ||||
| Content, | Content, | ||||
| Directory, | Directory, | ||||
| ExtID, | ExtID, | ||||
| Origin, | Origin, | ||||
| OriginVisit, | OriginVisit, | ||||
| ▲ Show 20 Lines • Show All 599 Lines • ▼ Show 20 Lines | def test_content_get_partition_pagination_generate(self, swh_storage, swh_contents): | ||||
| if page_token is None: | if page_token is None: | ||||
| break | break | ||||
| assert len(actual_contents) == len(expected_contents) | assert len(actual_contents) == len(expected_contents) | ||||
| for content in actual_contents: | for content in actual_contents: | ||||
| assert content in expected_contents | assert content in expected_contents | ||||
| def test_content_get(self, swh_storage, sample_data): | @pytest.mark.parametrize("algo", DEFAULT_ALGORITHMS) | ||||
| def test_content_get(self, swh_storage, sample_data, algo): | |||||
| cont1, cont2 = sample_data.contents[:2] | cont1, cont2 = sample_data.contents[:2] | ||||
| swh_storage.content_add([cont1, cont2]) | swh_storage.content_add([cont1, cont2]) | ||||
| actual_contents = swh_storage.content_get([cont1.sha1, cont2.sha1]) | actual_contents = swh_storage.content_get( | ||||
| [getattr(cont1, algo), getattr(cont2, algo)], algo | |||||
| ) | |||||
| # we only retrieve the metadata so no data nor ctime within | # we only retrieve the metadata so no data nor ctime within | ||||
| expected_contents = [attr.evolve(c, data=None) for c in [cont1, cont2]] | expected_contents = [attr.evolve(c, data=None) for c in [cont1, cont2]] | ||||
| assert actual_contents == expected_contents | assert actual_contents == expected_contents | ||||
| for content in actual_contents: | for content in actual_contents: | ||||
| assert content.ctime is None | assert content.ctime is None | ||||
| def test_content_get_missing_sha1(self, swh_storage, sample_data): | @pytest.mark.parametrize("algo", DEFAULT_ALGORITHMS) | ||||
| def test_content_get_missing(self, swh_storage, sample_data, algo): | |||||
| cont1, cont2 = sample_data.contents[:2] | cont1, cont2 = sample_data.contents[:2] | ||||
| assert cont1.sha1 != cont2.sha1 | assert cont1.sha1 != cont2.sha1 | ||||
| missing_cont = sample_data.skipped_content | missing_cont = sample_data.skipped_content | ||||
| swh_storage.content_add([cont1, cont2]) | swh_storage.content_add([cont1, cont2]) | ||||
| actual_contents = swh_storage.content_get( | actual_contents = swh_storage.content_get( | ||||
| [cont1.sha1, cont2.sha1, missing_cont.sha1] | [getattr(cont1, algo), getattr(cont2, algo), getattr(missing_cont, algo)], | ||||
| algo, | |||||
| ) | ) | ||||
| expected_contents = [ | expected_contents = [ | ||||
| attr.evolve(c, data=None) if c else None for c in [cont1, cont2, None] | attr.evolve(c, data=None) if c else None for c in [cont1, cont2, None] | ||||
| ] | ] | ||||
| assert actual_contents == expected_contents | assert actual_contents == expected_contents | ||||
| def test_content_get_random(self, swh_storage, sample_data): | def test_content_get_random(self, swh_storage, sample_data): | ||||
| cont, cont2, cont3 = sample_data.contents[:3] | cont, cont2, cont3 = sample_data.contents[:3] | ||||
| swh_storage.content_add([cont, cont2, cont3]) | swh_storage.content_add([cont, cont2, cont3]) | ||||
| assert swh_storage.content_get_random() in { | assert swh_storage.content_get_random() in { | ||||
| cont.sha1_git, | cont.sha1_git, | ||||
| cont2.sha1_git, | cont2.sha1_git, | ||||
| cont3.sha1_git, | cont3.sha1_git, | ||||
| } | } | ||||
| def test_directory_add(self, swh_storage, sample_data): | def test_directory_add(self, swh_storage, sample_data): | ||||
| content = sample_data.content | content = sample_data.content | ||||
| directory = sample_data.directories[1] | directory = sample_data.directories[1] | ||||
| assert directory.entries[0].target == content.sha1_git | assert directory.entries[0].target == content.sha1_git | ||||
| swh_storage.content_add([content]) | swh_storage.content_add([content]) | ||||
olasd: This could probably be `pytest.mark.parametrize`d | |||||
| init_missing = list(swh_storage.directory_missing([directory.id])) | init_missing = list(swh_storage.directory_missing([directory.id])) | ||||
| assert [directory.id] == init_missing | assert [directory.id] == init_missing | ||||
| actual_result = swh_storage.directory_add([directory]) | actual_result = swh_storage.directory_add([directory]) | ||||
| assert actual_result == {"directory:add": 1} | assert actual_result == {"directory:add": 1} | ||||
| assert ("directory", directory) in list( | assert ("directory", directory) in list( | ||||
| ▲ Show 20 Lines • Show All 3,709 Lines • Show Last 20 Lines | |||||
This could probably be pytest.mark.parametrized