Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
Show First 20 Lines • Show All 3,962 Lines • ▼ Show 20 Lines | def test_generate_content_get_range_limit(self, swh_storage, swh_contents): | ||||
# retrieve next part | # retrieve next part | ||||
actual_results2 = swh_storage.content_get_range(start=end, end=end) | actual_results2 = swh_storage.content_get_range(start=end, end=end) | ||||
assert actual_results2["next"] is None | assert actual_results2["next"] is None | ||||
actual_contents2 = actual_results2["contents"] | actual_contents2 = actual_results2["contents"] | ||||
assert len(actual_contents2) == 1 | assert len(actual_contents2) == 1 | ||||
assert_contents_ok([contents_map[get_sha1s[-1]]], actual_contents2, ["sha1"]) | assert_contents_ok([contents_map[get_sha1s[-1]]], actual_contents2, ["sha1"]) | ||||
def test_origin_get_range_from_zero(self, swh_storage, swh_origins): | |||||
actual_origins = list( | |||||
swh_storage.origin_get_range(origin_from=0, origin_count=0) | |||||
) | |||||
assert len(actual_origins) == 0 | |||||
actual_origins = list( | |||||
swh_storage.origin_get_range(origin_from=0, origin_count=1) | |||||
) | |||||
assert len(actual_origins) == 1 | |||||
assert actual_origins[0]["id"] == 1 | |||||
assert actual_origins[0]["url"] == swh_origins[0].url | |||||
@pytest.mark.parametrize( | |||||
"origin_from,origin_count", | |||||
[(1, 1), (1, 10), (1, 20), (1, 101), (11, 0), (11, 10), (91, 11)], | |||||
) | |||||
def test_origin_get_range( | |||||
self, swh_storage, swh_origins, origin_from, origin_count | |||||
): | |||||
actual_origins = list( | |||||
swh_storage.origin_get_range( | |||||
origin_from=origin_from, origin_count=origin_count | |||||
) | |||||
) | |||||
origins_with_id = list(enumerate(swh_origins, start=1)) | |||||
expected_origins = [ | |||||
{"url": origin.url, "id": origin_id,} | |||||
for (origin_id, origin) in origins_with_id[ | |||||
origin_from - 1 : origin_from + origin_count - 1 | |||||
] | |||||
] | |||||
assert actual_origins == expected_origins | |||||
@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: | ||||
actual_page = swh_storage.origin_list(page_token=page_token, limit=limit) | actual_page = swh_storage.origin_list(page_token=page_token, limit=limit) | ||||
▲ Show 20 Lines • Show All 344 Lines • Show Last 20 Lines |