Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 324 Lines • ▼ Show 20 Lines | def test_content_update(self, swh_storage, sample_data): | ||||
| # alter the sha1_git for example | # alter the sha1_git for example | ||||
| cont1b = attr.evolve( | cont1b = attr.evolve( | ||||
| cont1, sha1_git=hash_to_bytes("3a60a5275d0333bf13468e8b3dcab90f4046e654") | cont1, sha1_git=hash_to_bytes("3a60a5275d0333bf13468e8b3dcab90f4046e654") | ||||
| ) | ) | ||||
| swh_storage.content_update([cont1b.to_dict()], keys=["sha1_git"]) | swh_storage.content_update([cont1b.to_dict()], keys=["sha1_git"]) | ||||
| results = swh_storage.content_get_metadata([cont1.sha1]) | actual_contents = swh_storage.content_get([cont1.sha1]) | ||||
| expected_content = attr.evolve(cont1b, data=None) | |||||
| expected_content = attr.evolve(cont1b, data=None, ctime=None).to_dict() | assert actual_contents == [expected_content] | ||||
| assert tuple(results[cont1.sha1]) == (expected_content,) | |||||
| def test_content_add_metadata(self, swh_storage, sample_data): | def test_content_add_metadata(self, swh_storage, sample_data): | ||||
| cont = attr.evolve(sample_data.content, data=None, ctime=now()) | cont = attr.evolve(sample_data.content, data=None, ctime=now()) | ||||
| actual_result = swh_storage.content_add_metadata([cont]) | actual_result = swh_storage.content_add_metadata([cont]) | ||||
| assert actual_result == { | assert actual_result == { | ||||
| "content:add": 1, | "content:add": 1, | ||||
| } | } | ||||
| expected_cont = cont.to_dict() | expected_cont = cont | ||||
| expected_cont.pop("ctime", None) | assert swh_storage.content_get([cont.sha1]) == [expected_cont] | ||||
| assert tuple(swh_storage.content_get_metadata([cont.sha1])[cont.sha1]) == ( | |||||
| expected_cont, | |||||
| ) | |||||
| contents = [ | contents = [ | ||||
| obj | obj | ||||
| for (obj_type, obj) in swh_storage.journal_writer.journal.objects | for (obj_type, obj) in swh_storage.journal_writer.journal.objects | ||||
| if obj_type == "content" | if obj_type == "content" | ||||
| ] | ] | ||||
| assert len(contents) == 1 | assert len(contents) == 1 | ||||
| for obj in contents: | for obj in contents: | ||||
| obj = attr.evolve(obj, ctime=None) | obj = attr.evolve(obj, ctime=None) | ||||
| ▲ Show 20 Lines • Show All 252 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_metadata(self, swh_storage, sample_data): | def test_content_get(self, swh_storage, sample_data): | ||||
| 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_md = swh_storage.content_get_metadata([cont1.sha1, cont2.sha1]) | actual_contents = swh_storage.content_get([cont1.sha1, cont2.sha1]) | ||||
| # we only retrieve the metadata so no data nor ctime within | # we only retrieve the metadata so no data nor ctime within | ||||
| expected_cont1, expected_cont2 = [ | expected_contents = [attr.evolve(c, data=None) for c in [cont1, cont2]] | ||||
| attr.evolve(c, data=None, ctime=None).to_dict() for c in [cont1, cont2] | |||||
| ] | |||||
| assert tuple(actual_md[cont1.sha1]) == (expected_cont1,) | assert actual_contents == expected_contents | ||||
| assert tuple(actual_md[cont2.sha1]) == (expected_cont2,) | |||||
| assert len(actual_md.keys()) == 2 | |||||
| def test_content_get_metadata_missing_sha1(self, swh_storage, sample_data): | def test_content_get_missing_sha1(self, swh_storage, sample_data): | ||||
| cont1, cont2 = sample_data.contents[:2] | cont1, cont2 = sample_data.contents[:2] | ||||
| 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_metadata([missing_cont.sha1]) | actual_contents = swh_storage.content_get( | ||||
| [cont1.sha1, cont2.sha1, missing_cont.sha1] | |||||
| ) | |||||
| assert len(actual_contents) == 1 | expected_contents = [ | ||||
| assert tuple(actual_contents[missing_cont.sha1]) == () | attr.evolve(c, data=None) if c else None for c in [cont1, cont2, None] | ||||
| ] | |||||
| 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, | ||||
| ▲ Show 20 Lines • Show All 3,170 Lines • ▼ Show 20 Lines | def test_generate_content_get_data(self, swh_storage, swh_contents): | ||||
| contents_with_data = [c for c in swh_contents if c.status != "absent"] | contents_with_data = [c for c in swh_contents if c.status != "absent"] | ||||
| # retrieve contents | # retrieve contents | ||||
| for content in contents_with_data: | for content in contents_with_data: | ||||
| actual_content_data = swh_storage.content_get_data(content.sha1) | actual_content_data = swh_storage.content_get_data(content.sha1) | ||||
| assert actual_content_data is not None | assert actual_content_data is not None | ||||
| assert actual_content_data == content.data | assert actual_content_data == content.data | ||||
| def test_generate_content_get_metadata(self, swh_storage, swh_contents): | def test_generate_content_get(self, swh_storage, swh_contents): | ||||
| # input the list of sha1s we want from storage | expected_contents = [ | ||||
| expected_contents = [c.to_dict() for c in swh_contents if c.status != "absent"] | attr.evolve(c, data=None) for c in swh_contents if c.status != "absent" | ||||
| get_sha1s = [c["sha1"] for c in expected_contents] | ] | ||||
| # retrieve contents | |||||
| meta_contents = swh_storage.content_get_metadata(get_sha1s) | |||||
| assert len(list(meta_contents)) == len(get_sha1s) | |||||
| actual_contents = [] | |||||
| for contents in meta_contents.values(): | |||||
| actual_contents.extend(contents) | |||||
| keys_to_check = {"length", "status", "sha1", "sha1_git", "sha256", "blake2s256"} | actual_contents = swh_storage.content_get([c.sha1 for c in expected_contents]) | ||||
| assert_contents_ok( | assert len(actual_contents) == len(expected_contents) | ||||
| expected_contents, actual_contents, keys_to_check=keys_to_check | assert actual_contents == expected_contents | ||||
| ) | |||||
| @pytest.mark.parametrize("limit", [1, 7, 10, 100, 1000]) | @pytest.mark.parametrize("limit", [1, 7, 10, 100, 1000]) | ||||
| def test_origin_list(self, swh_storage, swh_origins, limit): | def test_origin_list(self, swh_storage, swh_origins, limit): | ||||
| returned_origins = [] | returned_origins = [] | ||||
| page_token = None | page_token = None | ||||
| i = 0 | i = 0 | ||||
| while True: | while True: | ||||
| ▲ Show 20 Lines • Show All 345 Lines • Show Last 20 Lines | |||||