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 @@ -3152,6 +3152,60 @@ origin_visits = list(self.storage.origin_visit_get(1)) self.assertEqual(origin_visits, []) + @given(gen_contents(min_size=4, max_size=4)) + def test_generate_content_get_range_start(self, contents): + """content_get_range paginates results if limit exceeded""" + self.reset_storage_tables() + contents_map = {c['sha1']: c for c in contents} + + # add contents to storage + self.storage.content_add(contents) + + # input the list of sha1s we want from storage + get_sha1s = sorted([c['sha1'] for c in contents]) + start = get_sha1s[1] + end = get_sha1s[-1] + + actual_result = self.storage.content_get_range(start, end) + + actual_contents = actual_result['contents'] + actual_next = actual_result['next'] + + self.assertEqual(3, len(actual_contents)) + self.assertIsNone(actual_next) + + expected_contents = [contents_map[sha1] for sha1 in get_sha1s[1:]] + keys_to_check = set(contents[0].keys()) - {'data'} + self.assert_contents_ok(expected_contents, actual_contents, + keys_to_check) + + @given(gen_contents(min_size=4, max_size=4)) + def test_generate_content_get_range_end(self, contents): + """content_get_range paginates results if limit exceeded""" + self.reset_storage_tables() + contents_map = {c['sha1']: c for c in contents} + + # add contents to storage + self.storage.content_add(contents) + + # input the list of sha1s we want from storage + get_sha1s = sorted([c['sha1'] for c in contents]) + start = get_sha1s[0] + end = get_sha1s[-2] + + actual_result = self.storage.content_get_range(start, end) + + actual_contents = actual_result['contents'] + actual_next = actual_result['next'] + + self.assertEqual(3, len(actual_contents)) + self.assertIsNone(actual_next) + + expected_contents = [contents_map[sha1] for sha1 in get_sha1s[:-1]] + keys_to_check = set(contents[0].keys()) - {'data'} + self.assert_contents_ok(expected_contents, actual_contents, + keys_to_check) + @given(strategies.sets(origins().map(lambda x: tuple(x.to_dict().items())), min_size=6, max_size=15)) def test_origin_get_range(self, new_origins):