Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 538 Lines • ▼ Show 20 Lines | def test_content_missing_per_sha1_git(self, swh_storage, sample_data): | ||||
| contents = [cont.sha1_git, cont2.sha1_git, missing_cont.sha1_git] | contents = [cont.sha1_git, cont2.sha1_git, missing_cont.sha1_git] | ||||
| missing_contents = swh_storage.content_missing_per_sha1_git(contents) | missing_contents = swh_storage.content_missing_per_sha1_git(contents) | ||||
| assert list(missing_contents) == [missing_cont.sha1_git] | assert list(missing_contents) == [missing_cont.sha1_git] | ||||
| def test_content_get_partition(self, swh_storage, swh_contents): | def test_content_get_partition(self, swh_storage, swh_contents): | ||||
| """content_get_partition paginates results if limit exceeded""" | """content_get_partition paginates results if limit exceeded""" | ||||
| expected_contents = [c.to_dict() for c in swh_contents if c.status != "absent"] | expected_contents = [ | ||||
ardumont: I remove the data as otherwise the comparison below will explode.
Content returned do not have… | |||||
| attr.evolve(c, data=None) for c in swh_contents if c.status != "absent" | |||||
| ] | |||||
| actual_contents = [] | actual_contents = [] | ||||
| for i in range(16): | for i in range(16): | ||||
| actual_result = swh_storage.content_get_partition(i, 16) | actual_result = swh_storage.content_get_partition(i, 16) | ||||
| assert actual_result["next_page_token"] is None | assert actual_result.next_page_token is None | ||||
| actual_contents.extend(actual_result["contents"]) | actual_contents.extend(actual_result.results) | ||||
| assert_contents_ok(expected_contents, actual_contents, ["sha1"]) | assert len(actual_contents) == len(expected_contents) | ||||
| for content in actual_contents: | |||||
| assert content in expected_contents | |||||
Done Inline Actionsi'm missing an assertion on the length here. ardumont: i'm missing an assertion on the length here. | |||||
| def test_content_get_partition_full(self, swh_storage, swh_contents): | def test_content_get_partition_full(self, swh_storage, swh_contents): | ||||
| """content_get_partition for a single partition returns all available | """content_get_partition for a single partition returns all available contents | ||||
| contents""" | |||||
| expected_contents = [c.to_dict() for c in swh_contents if c.status != "absent"] | """ | ||||
| expected_contents = [ | |||||
| attr.evolve(c, data=None) for c in swh_contents if c.status != "absent" | |||||
| ] | |||||
| 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.results | ||||
| assert_contents_ok(expected_contents, actual_contents, ["sha1"]) | assert len(actual_contents) == len(expected_contents) | ||||
| for content in actual_contents: | |||||
| assert content in expected_contents | |||||
| 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 when at least one of the partitions is | """content_get_partition when at least one of the partitions is empty""" | ||||
| empty""" | |||||
| expected_contents = { | expected_contents = { | ||||
| cont.sha1 for cont in swh_contents if cont.status != "absent" | cont.sha1 for cont in swh_contents if cont.status != "absent" | ||||
| } | } | ||||
| # nb_partitions = smallest power of 2 such that at least one of | # nb_partitions = smallest power of 2 such that at least one of | ||||
| # the partitions is empty | # the partitions is empty | ||||
| nb_partitions = 1 << math.floor(math.log2(len(swh_contents)) + 1) | nb_partitions = 1 << math.floor(math.log2(len(swh_contents)) + 1) | ||||
| seen_sha1s = [] | seen_sha1s = [] | ||||
| for i in range(nb_partitions): | for i in range(nb_partitions): | ||||
| actual_result = swh_storage.content_get_partition( | actual_result = swh_storage.content_get_partition( | ||||
| i, nb_partitions, limit=len(swh_contents) + 1 | i, nb_partitions, limit=len(swh_contents) + 1 | ||||
| ) | ) | ||||
| for cont in actual_result["contents"]: | for content in actual_result.results: | ||||
| seen_sha1s.append(cont["sha1"]) | seen_sha1s.append(content.sha1) | ||||
| # Limit is higher than the max number of results | # 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 set(seen_sha1s) == expected_contents | 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(StorageArgumentException) as e: | with pytest.raises(StorageArgumentException, match="limit should not be None"): | ||||
| swh_storage.content_get_partition(1, 16, limit=None) | swh_storage.content_get_partition(1, 16, limit=None) | ||||
| assert e.value.args == ("limit should not be None",) | def test_content_get_partition_pagination_generate(self, swh_storage, swh_contents): | ||||
| def test_generate_content_get_partition_pagination(self, swh_storage, swh_contents): | |||||
| """content_get_partition returns contents within range provided""" | """content_get_partition returns contents within range provided""" | ||||
| expected_contents = [c.to_dict() for c in swh_contents if c.status != "absent"] | expected_contents = [ | ||||
| attr.evolve(c, data=None) for c in swh_contents if c.status != "absent" | |||||
| ] | |||||
| # retrieve contents | # retrieve contents | ||||
| actual_contents = [] | actual_contents = [] | ||||
| for i in range(4): | for i in range(4): | ||||
| page_token = None | page_token = None | ||||
| while True: | while True: | ||||
| actual_result = swh_storage.content_get_partition( | actual_result = swh_storage.content_get_partition( | ||||
| i, 4, limit=3, page_token=page_token | i, 4, limit=3, page_token=page_token | ||||
| ) | ) | ||||
| actual_contents.extend(actual_result["contents"]) | actual_contents.extend(actual_result.results) | ||||
| page_token = actual_result["next_page_token"] | page_token = actual_result.next_page_token | ||||
| if page_token is None: | if page_token is None: | ||||
| break | break | ||||
| assert_contents_ok(expected_contents, actual_contents, ["sha1"]) | assert len(actual_contents) == len(expected_contents) | ||||
| for content in actual_contents: | |||||
| assert content in expected_contents | |||||
| def test_content_get_metadata(self, swh_storage, sample_data): | def test_content_get_metadata(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_md = swh_storage.content_get_metadata([cont1.sha1, cont2.sha1]) | ||||
| ▲ Show 20 Lines • Show All 3,580 Lines • Show Last 20 Lines | |||||
I remove the data as otherwise the comparison below will explode.
Content returned do not have their data field filled.