Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show All 30 Lines | from swh.model.model import ( | ||||
| Directory, | Directory, | ||||
| MetadataTargetType, | MetadataTargetType, | ||||
| Origin, | Origin, | ||||
| OriginVisit, | OriginVisit, | ||||
| OriginVisitStatus, | OriginVisitStatus, | ||||
| Person, | Person, | ||||
| Release, | Release, | ||||
| Revision, | Revision, | ||||
| SkippedContent, | |||||
| Snapshot, | Snapshot, | ||||
| ) | ) | ||||
| from swh.model.hypothesis_strategies import objects | from swh.model.hypothesis_strategies import objects | ||||
| from swh.storage import get_storage | from swh.storage import get_storage | ||||
| from swh.storage.converters import origin_url_to_sha1 as sha1 | from swh.storage.converters import origin_url_to_sha1 as sha1 | ||||
| from swh.storage.exc import HashCollision, StorageArgumentException | from swh.storage.exc import HashCollision, StorageArgumentException | ||||
| from swh.storage.interface import StorageInterface | from swh.storage.interface import StorageInterface | ||||
| from swh.storage.utils import content_hex_hashes, now | from swh.storage.utils import content_hex_hashes, now | ||||
| ▲ Show 20 Lines • Show All 54 Lines • ▼ Show 20 Lines | def test_round_to_milliseconds(): | ||||
| for (ms, expected_ms) in [(0, 0), (1000, 1000), (555555, 555000), (999500, 999000)]: | for (ms, expected_ms) in [(0, 0), (1000, 1000), (555555, 555000), (999500, 999000)]: | ||||
| date = date.replace(microsecond=ms) | date = date.replace(microsecond=ms) | ||||
| actual_date = round_to_milliseconds(date) | actual_date = round_to_milliseconds(date) | ||||
| assert actual_date.microsecond == expected_ms | assert actual_date.microsecond == expected_ms | ||||
| class LazyContent(Content): | class LazyContent(Content): | ||||
| def with_data(self): | def with_data(self): | ||||
| return Content.from_dict({**self.to_dict(), "data": data.cont["data"]}) | raw_data = data.content.data | ||||
| return Content.from_dict({**self.to_dict(), "data": raw_data}) | |||||
| class TestStorage: | class TestStorage: | ||||
| """Main class for Storage testing. | """Main class for Storage testing. | ||||
| This class is used as-is to test local storage (see TestLocalStorage | This class is used as-is to test local storage (see TestLocalStorage | ||||
| below) and remote storage (see TestRemoteStorage in | below) and remote storage (see TestRemoteStorage in | ||||
| test_remote_storage.py. | test_remote_storage.py. | ||||
| ▲ Show 20 Lines • Show All 346 Lines • ▼ Show 20 Lines | class TestStorage: | ||||
| @given( | @given( | ||||
| strategies.sets( | strategies.sets( | ||||
| elements=strategies.sampled_from(["sha256", "sha1_git", "blake2s256"]), | elements=strategies.sampled_from(["sha256", "sha1_git", "blake2s256"]), | ||||
| min_size=0, | min_size=0, | ||||
| ) | ) | ||||
| ) | ) | ||||
| def test_content_missing(self, swh_storage, algos): | def test_content_missing(self, swh_storage, algos): | ||||
| algos |= {"sha1"} | algos |= {"sha1"} | ||||
| cont = Content.from_dict(data.cont2) | content, missing_content = [data.content2, data.missing_content] | ||||
| missing_cont = SkippedContent.from_dict(data.missing_cont) | swh_storage.content_add([content]) | ||||
| swh_storage.content_add([cont]) | |||||
| test_contents = [cont.to_dict()] | test_contents = [content.to_dict()] | ||||
| missing_per_hash = defaultdict(list) | missing_per_hash = defaultdict(list) | ||||
| for i in range(256): | for i in range(256): | ||||
| test_content = missing_cont.to_dict() | test_content = missing_content.to_dict() | ||||
| for hash in algos: | for hash in algos: | ||||
| test_content[hash] = bytes([i]) + test_content[hash][1:] | test_content[hash] = bytes([i]) + test_content[hash][1:] | ||||
| missing_per_hash[hash].append(test_content[hash]) | missing_per_hash[hash].append(test_content[hash]) | ||||
| test_contents.append(test_content) | test_contents.append(test_content) | ||||
| assert set(swh_storage.content_missing(test_contents)) == set( | assert set(swh_storage.content_missing(test_contents)) == set( | ||||
| missing_per_hash["sha1"] | missing_per_hash["sha1"] | ||||
| ) | ) | ||||
| for hash in algos: | for hash in algos: | ||||
| assert set( | assert set( | ||||
| swh_storage.content_missing(test_contents, key_hash=hash) | swh_storage.content_missing(test_contents, key_hash=hash) | ||||
| ) == set(missing_per_hash[hash]) | ) == set(missing_per_hash[hash]) | ||||
| @pytest.mark.property_based | @pytest.mark.property_based | ||||
| @given( | @given( | ||||
| strategies.sets( | strategies.sets( | ||||
| elements=strategies.sampled_from(["sha256", "sha1_git", "blake2s256"]), | elements=strategies.sampled_from(["sha256", "sha1_git", "blake2s256"]), | ||||
| min_size=0, | min_size=0, | ||||
| ) | ) | ||||
| ) | ) | ||||
| def test_content_missing_unknown_algo(self, swh_storage, algos): | def test_content_missing_unknown_algo(self, swh_storage, algos): | ||||
| algos |= {"sha1"} | algos |= {"sha1"} | ||||
| cont = Content.from_dict(data.cont2) | content, missing_content = [data.content2, data.missing_content] | ||||
| missing_cont = SkippedContent.from_dict(data.missing_cont) | swh_storage.content_add([content]) | ||||
| swh_storage.content_add([cont]) | |||||
| test_contents = [cont.to_dict()] | test_contents = [content.to_dict()] | ||||
| missing_per_hash = defaultdict(list) | missing_per_hash = defaultdict(list) | ||||
| for i in range(16): | for i in range(16): | ||||
| test_content = missing_cont.to_dict() | test_content = missing_content.to_dict() | ||||
| for hash in algos: | for hash in algos: | ||||
| test_content[hash] = bytes([i]) + test_content[hash][1:] | test_content[hash] = bytes([i]) + test_content[hash][1:] | ||||
| missing_per_hash[hash].append(test_content[hash]) | missing_per_hash[hash].append(test_content[hash]) | ||||
| test_content["nonexisting_algo"] = b"\x00" | test_content["nonexisting_algo"] = b"\x00" | ||||
| test_contents.append(test_content) | test_contents.append(test_content) | ||||
| assert set(swh_storage.content_missing(test_contents)) == set( | assert set(swh_storage.content_missing(test_contents)) == set( | ||||
| missing_per_hash["sha1"] | missing_per_hash["sha1"] | ||||
| ▲ Show 20 Lines • Show All 126 Lines • ▼ Show 20 Lines | def test_content_get_metadata_missing_sha1(self, swh_storage, sample_data_model): | ||||
| swh_storage.content_add([cont1, cont2]) | swh_storage.content_add([cont1, cont2]) | ||||
| actual_contents = swh_storage.content_get_metadata([missing_cont.sha1]) | actual_contents = swh_storage.content_get_metadata([missing_cont.sha1]) | ||||
| assert len(actual_contents) == 1 | assert len(actual_contents) == 1 | ||||
| assert tuple(actual_contents[missing_cont.sha1]) == () | assert tuple(actual_contents[missing_cont.sha1]) == () | ||||
| def test_content_get_random(self, swh_storage, sample_data_model): | def test_content_get_random(self, swh_storage, sample_data_model): | ||||
| cont, cont2 = sample_data_model["content"][:2] | cont, cont2, cont3 = sample_data_model["content"][:3] | ||||
| cont3 = sample_data_model["content_no_data"][0] | |||||
| 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, | ||||
| } | } | ||||
| ▲ Show 20 Lines • Show All 2,246 Lines • ▼ Show 20 Lines | def test_content_find_with_present_content(self, swh_storage, sample_data_model): | ||||
| actually_present = swh_storage.content_find(content.hashes()) | actually_present = swh_storage.content_find(content.hashes()) | ||||
| assert 1 == len(actually_present) | assert 1 == len(actually_present) | ||||
| actually_present[0].pop("ctime") | actually_present[0].pop("ctime") | ||||
| assert actually_present[0] == expected_content | assert actually_present[0] == expected_content | ||||
| def test_content_find_with_non_present_content( | def test_content_find_with_non_present_content( | ||||
| self, swh_storage, sample_data_model | self, swh_storage, sample_data_model | ||||
| ): | ): | ||||
| missing_content = sample_data_model["content_no_data"][0] | missing_content = sample_data_model["skipped_content"][0] | ||||
| # 1. with something that does not exist | # 1. with something that does not exist | ||||
| actually_present = swh_storage.content_find({"sha1": missing_content.sha1}) | actually_present = swh_storage.content_find({"sha1": missing_content.sha1}) | ||||
| assert actually_present == [] | assert actually_present == [] | ||||
| # 2. with something that does not exist | # 2. with something that does not exist | ||||
| actually_present = swh_storage.content_find( | actually_present = swh_storage.content_find( | ||||
| {"sha1_git": missing_content.sha1_git} | {"sha1_git": missing_content.sha1_git} | ||||
| ▲ Show 20 Lines • Show All 1,221 Lines • Show Last 20 Lines | |||||