diff --git a/swh/storage/tests/conftest.py b/swh/storage/tests/conftest.py --- a/swh/storage/tests/conftest.py +++ b/swh/storage/tests/conftest.py @@ -13,8 +13,12 @@ except ImportError: pytest_cov = None +from typing import Iterable + +from swh.model.model import BaseContent from swh.model.tests.generate_testdata import gen_contents, gen_origins from swh.storage import get_storage +from swh.storage.interface import StorageInterface # define tests profile. Full documentation is at: # https://hypothesis.readthedocs.io/en/latest/settings.html#settings-profiles @@ -51,10 +55,10 @@ @pytest.fixture -def swh_contents(swh_storage): - contents = gen_contents(n=20) - swh_storage.content_add([c for c in contents if c["status"] != "absent"]) - swh_storage.skipped_content_add([c for c in contents if c["status"] == "absent"]) +def swh_contents(swh_storage: StorageInterface) -> Iterable[BaseContent]: + contents = [BaseContent.from_dict(c) for c in gen_contents(n=20)] + swh_storage.content_add([c for c in contents if c.status != "absent"]) + swh_storage.skipped_content_add([c for c in contents if c.status == "absent"]) return contents diff --git a/swh/storage/tests/test_storage.py b/swh/storage/tests/test_storage.py --- a/swh/storage/tests/test_storage.py +++ b/swh/storage/tests/test_storage.py @@ -589,7 +589,7 @@ def test_content_get_partition(self, swh_storage, swh_contents): """content_get_partition paginates results if limit exceeded""" - expected_contents = [c for c in swh_contents if c["status"] != "absent"] + expected_contents = [c.to_dict() for c in swh_contents if c.status != "absent"] actual_contents = [] for i in range(16): @@ -602,7 +602,7 @@ def test_content_get_partition_full(self, swh_storage, swh_contents): """content_get_partition for a single partition returns all available contents""" - expected_contents = [c for c in swh_contents if c["status"] != "absent"] + expected_contents = [c.to_dict() for c in swh_contents if c.status != "absent"] actual_result = swh_storage.content_get_partition(0, 1) assert actual_result["next_page_token"] is None @@ -614,7 +614,7 @@ """content_get_partition when at least one of the partitions is empty""" 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 # the partitions is empty @@ -644,7 +644,7 @@ def test_generate_content_get_partition_pagination(self, swh_storage, swh_contents): """content_get_partition returns contents within range provided""" - expected_contents = [c for c in swh_contents if c["status"] != "absent"] + expected_contents = [c.to_dict() for c in swh_contents if c.status != "absent"] # retrieve contents actual_contents = [] @@ -3724,7 +3724,7 @@ class TestStorageGeneratedData: def test_generate_content_get(self, swh_storage, swh_contents): - contents_with_data = [c for c in swh_contents if c["status"] != "absent"] + contents_with_data = [c.to_dict() for c in swh_contents if c.status != "absent"] # input the list of sha1s we want from storage get_sha1s = [c["sha1"] for c in contents_with_data] @@ -3735,7 +3735,7 @@ def test_generate_content_get_metadata(self, swh_storage, swh_contents): # input the list of sha1s we want from storage - expected_contents = [c for c in swh_contents if c["status"] != "absent"] + expected_contents = [c.to_dict() for c in swh_contents if c.status != "absent"] get_sha1s = [c["sha1"] for c in expected_contents] # retrieve contents @@ -3755,9 +3755,9 @@ def test_generate_content_get_range(self, swh_storage, swh_contents): """content_get_range returns complete range""" - present_contents = [c for c in swh_contents if c["status"] != "absent"] + present_contents = [c.to_dict() for c in swh_contents if c.status != "absent"] - get_sha1s = sorted([c["sha1"] for c in swh_contents if c["status"] != "absent"]) + get_sha1s = sorted([c.sha1 for c in swh_contents if c.status != "absent"]) start = get_sha1s[2] end = get_sha1s[-2] actual_result = swh_storage.content_get_range(start, end) @@ -3773,7 +3773,7 @@ def test_generate_content_get_range_full(self, swh_storage, swh_contents): """content_get_range for a full range returns all available contents""" - present_contents = [c for c in swh_contents if c["status"] != "absent"] + present_contents = [c.to_dict() for c in swh_contents if c.status != "absent"] start = b"0" * 40 end = b"f" * 40 @@ -3805,7 +3805,7 @@ def test_generate_content_get_range_no_limit(self, swh_storage, swh_contents): """content_get_range returns contents within range provided""" # input the list of sha1s we want from storage - get_sha1s = sorted([c["sha1"] for c in swh_contents if c["status"] != "absent"]) + get_sha1s = sorted([c.sha1 for c in swh_contents if c.status != "absent"]) start = get_sha1s[0] end = get_sha1s[-1] @@ -3816,15 +3816,15 @@ assert actual_result["next"] is None assert len(actual_contents) == len(get_sha1s) - expected_contents = [c for c in swh_contents if c["status"] != "absent"] + expected_contents = [c.to_dict() for c in swh_contents if c.status != "absent"] assert_contents_ok(expected_contents, actual_contents, ["sha1"]) def test_generate_content_get_range_limit(self, swh_storage, swh_contents): """content_get_range paginates results if limit exceeded""" - contents_map = {c["sha1"]: c for c in swh_contents} + contents_map = {c.sha1: c.to_dict() for c in swh_contents} # input the list of sha1s we want from storage - get_sha1s = sorted([c["sha1"] for c in swh_contents if c["status"] != "absent"]) + get_sha1s = sorted([c.sha1 for c in swh_contents if c.status != "absent"]) start = get_sha1s[0] end = get_sha1s[-1]