Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
Show First 20 Lines • Show All 1,247 Lines • ▼ Show 20 Lines | def _generate_random_visits(self, nb_visits=100, start=0, end=7): | ||||
days = random.randint(0, 28) | days = random.randint(0, 28) | ||||
weeks = random.randint(start, end) | weeks = random.randint(start, end) | ||||
date_visit = today - timedelta( | date_visit = today - timedelta( | ||||
weeks=weeks, hours=hours, minutes=minutes, seconds=seconds, days=days | weeks=weeks, hours=hours, minutes=minutes, seconds=seconds, days=days | ||||
) | ) | ||||
visits.append(date_visit) | visits.append(date_visit) | ||||
return visits | return visits | ||||
def test_origin_visit_get__unknown_origin(self, swh_storage): | |||||
actual_visit = swh_storage.origin_visit_get("foo") | |||||
assert actual_visit == {} | |||||
def test_origin_visit_get__validation_failure(self, swh_storage, sample_data): | |||||
origin = sample_data.origin | |||||
swh_storage.origin_add([origin]) | |||||
with pytest.raises( | |||||
StorageArgumentException, match="page_token must be a string" | |||||
): | |||||
swh_storage.origin_visit_get(origin.url, page_token=10) # not a string | |||||
with pytest.raises( | |||||
StorageArgumentException, match="order must be one of asc, desc" | |||||
): | |||||
swh_storage.origin_visit_get(origin.url, order="foobar") # wrong order | |||||
def test_origin_visit_get_all(self, swh_storage, sample_data): | def test_origin_visit_get_all(self, swh_storage, sample_data): | ||||
origin = sample_data.origin | origin = sample_data.origin | ||||
swh_storage.origin_add([origin]) | swh_storage.origin_add([origin]) | ||||
visits = swh_storage.origin_visit_add( | ov1, ov2, ov3 = swh_storage.origin_visit_add( | ||||
[ | [ | ||||
OriginVisit( | OriginVisit( | ||||
origin=origin.url, | origin=origin.url, | ||||
date=sample_data.date_visit1, | date=sample_data.date_visit1, | ||||
type=sample_data.type_visit1, | type=sample_data.type_visit1, | ||||
), | ), | ||||
OriginVisit( | OriginVisit( | ||||
origin=origin.url, | origin=origin.url, | ||||
date=sample_data.date_visit2, | date=sample_data.date_visit2, | ||||
type=sample_data.type_visit2, | type=sample_data.type_visit2, | ||||
), | ), | ||||
OriginVisit( | OriginVisit( | ||||
origin=origin.url, | origin=origin.url, | ||||
date=sample_data.date_visit2, | date=sample_data.date_visit2, | ||||
type=sample_data.type_visit2, | type=sample_data.type_visit2, | ||||
), | ), | ||||
] | ] | ||||
) | ) | ||||
ov1, ov2, ov3 = [ | |||||
{**v.to_dict(), "status": "created", "snapshot": None, "metadata": None,} | |||||
for v in visits | |||||
] | |||||
# order asc, no pagination, no limit | # order asc, no pagination, no limit | ||||
all_visits = list(swh_storage.origin_visit_get(origin.url)) | actual_result = swh_storage.origin_visit_get(origin.url) | ||||
assert all_visits == [ov1, ov2, ov3] | assert actual_result.get("next_page_token") is None | ||||
assert actual_result["visits"] == [ov1, ov2, ov3] | |||||
# order asc, no pagination, limit | # order asc, no pagination, limit | ||||
all_visits2 = list(swh_storage.origin_visit_get(origin.url, limit=2)) | actual_result = swh_storage.origin_visit_get(origin.url, limit=2) | ||||
vlorentz: No need to test `next_page_token`, it's opaque. Just check it's not None. (same comment for… | |||||
assert all_visits2 == [ov1, ov2] | assert actual_result["next_page_token"] == str(ov3.visit) | ||||
assert actual_result["visits"] == [ov1, ov2] | |||||
# order asc, pagination, no limit | # order asc, pagination, no limit | ||||
all_visits3 = list( | actual_result = swh_storage.origin_visit_get( | ||||
Not Done Inline Actionsreuse next_page_token from the previous call here. (same comment for other occurences below) vlorentz: reuse `next_page_token` from the previous call here. (same comment for other occurences below) | |||||
Done Inline ActionsWell, i'd like to keep the existing tests consistent with what we have already. ardumont: Well, i'd like to keep the existing tests consistent with what we have already.
completely make… | |||||
swh_storage.origin_visit_get(origin.url, last_visit=ov1["visit"]) | origin.url, page_token=str(ov1.visit) | ||||
) | ) | ||||
assert all_visits3 == [ov2, ov3] | assert actual_result.get("next_page_token") is None | ||||
assert actual_result["visits"] == [ov2, ov3] | |||||
Not Done Inline Actionsnope (should be opaque) vlorentz: nope (should be opaque) | |||||
Done Inline Actionswell, yes but i also said that i wanted to keep the existing tests comparable with the existing ones... If i'm starting to use the opaque identifier here, it will become unreadable... ardumont: well, yes but i also said that i wanted to keep the existing tests comparable with the existing… | |||||
Not Done Inline Actionsmeh. vlorentz: meh. | |||||
# order asc, pagination, limit | # order asc, pagination, limit | ||||
all_visits4 = list( | actual_result = swh_storage.origin_visit_get( | ||||
swh_storage.origin_visit_get(origin.url, last_visit=ov2["visit"], limit=1) | origin.url, page_token=str(ov2.visit), limit=1 | ||||
) | ) | ||||
assert all_visits4 == [ov3] | assert actual_result.get("next_page_token") is None | ||||
assert actual_result["visits"] == [ov3] | |||||
Not Done Inline Actionssame vlorentz: same | |||||
# order desc, no pagination, no limit | # order desc, no pagination, no limit | ||||
all_visits5 = list(swh_storage.origin_visit_get(origin.url, order="desc")) | actual_result = swh_storage.origin_visit_get(origin.url, order="desc") | ||||
assert all_visits5 == [ov3, ov2, ov1] | assert actual_result.get("next_page_token") is None | ||||
assert actual_result["visits"] == [ov3, ov2, ov1] | |||||
# order desc, no pagination, limit | # order desc, no pagination, limit | ||||
all_visits6 = list( | actual_result = swh_storage.origin_visit_get(origin.url, limit=2, order="desc") | ||||
swh_storage.origin_visit_get(origin.url, limit=2, order="desc") | assert actual_result["next_page_token"] == str(ov1.visit) | ||||
) | assert actual_result["visits"] == [ov3, ov2] | ||||
assert all_visits6 == [ov3, ov2] | |||||
Not Done Inline Actionsget the next page of results and test it vlorentz: get the next page of results and test it | |||||
# order desc, pagination, no limit | # order desc, pagination, no limit | ||||
all_visits7 = list( | actual_result = swh_storage.origin_visit_get( | ||||
swh_storage.origin_visit_get( | origin.url, page_token=str(ov3.visit), order="desc" | ||||
Not Done Inline Actionssame vlorentz: same | |||||
origin.url, last_visit=ov3["visit"], order="desc" | |||||
) | ) | ||||
) | assert actual_result.get("next_page_token") is None | ||||
assert all_visits7 == [ov2, ov1] | assert actual_result["visits"] == [ov2, ov1] | ||||
Not Done Inline Actionsassert actual_page.next_page_token is None vlorentz: assert actual_page.next_page_token is None | |||||
Done Inline ActionsI thought it enough already. ardumont: I thought it enough already. | |||||
# order desc, pagination, limit | # order desc, pagination, limit | ||||
all_visits8 = list( | actual_result = swh_storage.origin_visit_get( | ||||
swh_storage.origin_visit_get( | origin.url, page_token=str(ov3.visit), order="desc", limit=1 | ||||
origin.url, last_visit=ov3["visit"], order="desc", limit=1 | |||||
) | |||||
) | ) | ||||
assert all_visits8 == [ov2] | assert actual_result["next_page_token"] == str(ov1.visit) | ||||
assert actual_result["visits"] == [ov2] | |||||
def test_origin_visit_get__unknown_origin(self, swh_storage): | |||||
assert [] == list(swh_storage.origin_visit_get("foo")) | |||||
Done Inline Actionsyou should get the next page of results and test it too vlorentz: you should get the next page of results and test it too | |||||
def test_origin_visit_status_get_random(self, swh_storage, sample_data): | def test_origin_visit_status_get_random(self, swh_storage, sample_data): | ||||
origins = sample_data.origins[:2] | origins = sample_data.origins[:2] | ||||
swh_storage.origin_add(origins) | swh_storage.origin_add(origins) | ||||
# Add some random visits within the selection range | # Add some random visits within the selection range | ||||
visits = self._generate_random_visits() | visits = self._generate_random_visits() | ||||
visit_type = "git" | visit_type = "git" | ||||
▲ Show 20 Lines • Show All 224 Lines • ▼ Show 20 Lines | def test_origin_visit_add(self, swh_storage, sample_data): | ||||
actual_origin_visits = list(swh_storage.origin_visit_get(origin1.url)) | actual_origin_visits = list(swh_storage.origin_visit_get(origin1.url)) | ||||
expected_visits = [ | expected_visits = [ | ||||
{**ovs1.to_dict(), "type": ov1.type}, | {**ovs1.to_dict(), "type": ov1.type}, | ||||
{**ovs2.to_dict(), "type": ov2.type}, | {**ovs2.to_dict(), "type": ov2.type}, | ||||
] | ] | ||||
assert len(expected_visits) == len(actual_origin_visits) | assert len(expected_visits) == len(actual_origin_visits) | ||||
for visit in expected_visits: | for visit in expected_visits: | ||||
assert visit in actual_origin_visits | assert visit in actual_origin_visits | ||||
Not Done Inline Actionsequivalent to: assert set(actual_visits) == {ov1, ov2} vlorentz: equivalent to:
```
assert set(actual_visits) == {ov1, ov2}
``` | |||||
Not Done Inline Actionsyou forgot this comment vlorentz: you forgot this comment | |||||
actual_objects = list(swh_storage.journal_writer.journal.objects) | actual_objects = list(swh_storage.journal_writer.journal.objects) | ||||
expected_objects = list( | expected_objects = list( | ||||
[("origin", origin1)] | [("origin", origin1)] | ||||
+ [("origin_visit", visit) for visit in [ov1, ov2]] * 2 | + [("origin_visit", visit) for visit in [ov1, ov2]] * 2 | ||||
+ [("origin_visit_status", ovs) for ovs in [ovs1, ovs2]] | + [("origin_visit_status", ovs) for ovs in [ovs1, ovs2]] | ||||
) | ) | ||||
for obj in expected_objects: | for obj in expected_objects: | ||||
▲ Show 20 Lines • Show All 302 Lines • ▼ Show 20 Lines | def test_origin_visit_get_by__no_result(self, swh_storage, sample_data): | ||||
swh_storage.origin_add([origin]) | swh_storage.origin_add([origin]) | ||||
actual_visit = swh_storage.origin_visit_get_by(origin.url, 999) # unknown visit | actual_visit = swh_storage.origin_visit_get_by(origin.url, 999) # unknown visit | ||||
assert actual_visit is None | assert actual_visit is None | ||||
def test_origin_visit_get_latest_none(self, swh_storage, sample_data): | def test_origin_visit_get_latest_none(self, swh_storage, sample_data): | ||||
"""Origin visit get latest on unknown objects should return nothing | """Origin visit get latest on unknown objects should return nothing | ||||
""" | """ | ||||
# unknown origin so no result | # unknown origin so no actual_result | ||||
assert swh_storage.origin_visit_get_latest("unknown-origin") is None | assert swh_storage.origin_visit_get_latest("unknown-origin") is None | ||||
# unknown type | # unknown type | ||||
origin = sample_data.origin | origin = sample_data.origin | ||||
swh_storage.origin_add([origin]) | swh_storage.origin_add([origin]) | ||||
assert swh_storage.origin_visit_get_latest(origin.url, type="unknown") is None | assert swh_storage.origin_visit_get_latest(origin.url, type="unknown") is None | ||||
def test_origin_visit_get_latest_filter_type(self, swh_storage, sample_data): | def test_origin_visit_get_latest_filter_type(self, swh_storage, sample_data): | ||||
▲ Show 20 Lines • Show All 258 Lines • ▼ Show 20 Lines | def test_origin_visit_status_get_latest(self, swh_storage, sample_data): | ||||
date=date_now, | date=date_now, | ||||
status="full", | status="full", | ||||
snapshot=snapshot.id, | snapshot=snapshot.id, | ||||
metadata={"something": "wicked"}, | metadata={"something": "wicked"}, | ||||
) | ) | ||||
swh_storage.origin_visit_status_add([ovs1, ovs2, ovs3, ovs4]) | swh_storage.origin_visit_status_add([ovs1, ovs2, ovs3, ovs4]) | ||||
# unknown origin so no result | # unknown origin so no actual_result | ||||
actual_origin_visit = swh_storage.origin_visit_status_get_latest( | actual_origin_visit = swh_storage.origin_visit_status_get_latest( | ||||
"unknown-origin", ov1.visit | "unknown-origin", ov1.visit | ||||
) | ) | ||||
assert actual_origin_visit is None | assert actual_origin_visit is None | ||||
# unknown visit so no result | # unknown visit so no actual_result | ||||
actual_origin_visit = swh_storage.origin_visit_status_get_latest( | actual_origin_visit = swh_storage.origin_visit_status_get_latest( | ||||
ov1.origin, ov1.visit + 10 | ov1.origin, ov1.visit + 10 | ||||
) | ) | ||||
assert actual_origin_visit is None | assert actual_origin_visit is None | ||||
# Two visits, both with no snapshot, take the most recent | # Two visits, both with no snapshot, take the most recent | ||||
actual_origin_visit2 = swh_storage.origin_visit_status_get_latest( | actual_origin_visit2 = swh_storage.origin_visit_status_get_latest( | ||||
origin1.url, ov1.visit | origin1.url, ov1.visit | ||||
▲ Show 20 Lines • Show All 785 Lines • ▼ Show 20 Lines | def test_content_find_with_duplicate_input(self, swh_storage, sample_data): | ||||
expected_content, | expected_content, | ||||
expected_duplicated_content, | expected_duplicated_content, | ||||
actual_result[0], | actual_result[0], | ||||
actual_result[1], | actual_result[1], | ||||
]: | ]: | ||||
dict_.pop(key, None) | dict_.pop(key, None) | ||||
expected_result = [expected_content, expected_duplicated_content] | expected_result = [expected_content, expected_duplicated_content] | ||||
for result in expected_result: | for actual_result in expected_result: | ||||
assert result in actual_result | assert actual_result in actual_result | ||||
def test_content_find_with_duplicate_sha256(self, swh_storage, sample_data): | def test_content_find_with_duplicate_sha256(self, swh_storage, sample_data): | ||||
content = sample_data.content | content = sample_data.content | ||||
hashes = {} | hashes = {} | ||||
# Create fake data with colliding sha256 | # Create fake data with colliding sha256 | ||||
for hashalgo in ("sha1", "sha1_git", "blake2s256"): | for hashalgo in ("sha1", "sha1_git", "blake2s256"): | ||||
value = bytearray(getattr(content, hashalgo)) | value = bytearray(getattr(content, hashalgo)) | ||||
▲ Show 20 Lines • Show All 78 Lines • ▼ Show 20 Lines | def test_content_find_with_duplicate_blake2s256(self, swh_storage, sample_data): | ||||
expected_content, | expected_content, | ||||
expected_duplicated_content, | expected_duplicated_content, | ||||
actual_result[0], | actual_result[0], | ||||
actual_result[1], | actual_result[1], | ||||
]: | ]: | ||||
dict_.pop(key, None) | dict_.pop(key, None) | ||||
expected_result = [expected_content, expected_duplicated_content] | expected_result = [expected_content, expected_duplicated_content] | ||||
for result in expected_result: | for actual_result in expected_result: | ||||
assert result in actual_result | assert actual_result in actual_result | ||||
# Find with both sha256 and blake2s256 | # Find with both sha256 and blake2s256 | ||||
actual_result = list( | actual_result = list( | ||||
swh_storage.content_find( | swh_storage.content_find( | ||||
{ | { | ||||
"sha256": duplicated_content.sha256, | "sha256": duplicated_content.sha256, | ||||
"blake2s256": duplicated_content.blake2s256, | "blake2s256": duplicated_content.blake2s256, | ||||
} | } | ||||
▲ Show 20 Lines • Show All 96 Lines • ▼ Show 20 Lines | def test_content_metadata_add(self, swh_storage, sample_data): | ||||
object_type="content", object_id=hash_to_bytes(content.sha1_git) | object_type="content", object_id=hash_to_bytes(content.sha1_git) | ||||
) | ) | ||||
swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
swh_storage.raw_extrinsic_metadata_add(content_metadata) | swh_storage.raw_extrinsic_metadata_add(content_metadata) | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.CONTENT, content_swhid, authority | MetadataTargetType.CONTENT, content_swhid, authority | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert list(sorted(result["results"], key=lambda x: x.discovery_date,)) == list( | assert list( | ||||
content_metadata | sorted(actual_result["results"], key=lambda x: x.discovery_date,) | ||||
) | ) == list(content_metadata) | ||||
def test_content_metadata_add_duplicate(self, swh_storage, sample_data): | def test_content_metadata_add_duplicate(self, swh_storage, sample_data): | ||||
"""Duplicates should be silently updated.""" | """Duplicates should be silently updated.""" | ||||
content = sample_data.content | content = sample_data.content | ||||
fetcher = sample_data.metadata_fetcher | fetcher = sample_data.metadata_fetcher | ||||
authority = sample_data.metadata_authority | authority = sample_data.metadata_authority | ||||
content_metadata, content_metadata2 = sample_data.content_metadata[:2] | content_metadata, content_metadata2 = sample_data.content_metadata[:2] | ||||
content_swhid = SWHID( | content_swhid = SWHID( | ||||
object_type="content", object_id=hash_to_bytes(content.sha1_git) | object_type="content", object_id=hash_to_bytes(content.sha1_git) | ||||
) | ) | ||||
new_content_metadata2 = attr.evolve( | new_content_metadata2 = attr.evolve( | ||||
content_metadata2, format="new-format", metadata=b"new-metadata", | content_metadata2, format="new-format", metadata=b"new-metadata", | ||||
) | ) | ||||
swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
swh_storage.raw_extrinsic_metadata_add([content_metadata, content_metadata2]) | swh_storage.raw_extrinsic_metadata_add([content_metadata, content_metadata2]) | ||||
swh_storage.raw_extrinsic_metadata_add([new_content_metadata2]) | swh_storage.raw_extrinsic_metadata_add([new_content_metadata2]) | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.CONTENT, content_swhid, authority | MetadataTargetType.CONTENT, content_swhid, authority | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
expected_results1 = (content_metadata, new_content_metadata2) | expected_results1 = (content_metadata, new_content_metadata2) | ||||
expected_results2 = (content_metadata, content_metadata2) | expected_results2 = (content_metadata, content_metadata2) | ||||
assert tuple(sorted(result["results"], key=lambda x: x.discovery_date,)) in ( | assert tuple( | ||||
sorted(actual_result["results"], key=lambda x: x.discovery_date,) | |||||
) in ( | |||||
expected_results1, # cassandra | expected_results1, # cassandra | ||||
expected_results2, # postgresql | expected_results2, # postgresql | ||||
) | ) | ||||
def test_content_metadata_get(self, swh_storage, sample_data): | def test_content_metadata_get(self, swh_storage, sample_data): | ||||
content, content2 = sample_data.contents[:2] | content, content2 = sample_data.contents[:2] | ||||
fetcher, fetcher2 = sample_data.fetchers[:2] | fetcher, fetcher2 = sample_data.fetchers[:2] | ||||
authority, authority2 = sample_data.authorities[:2] | authority, authority2 = sample_data.authorities[:2] | ||||
Show All 14 Lines | def test_content_metadata_get(self, swh_storage, sample_data): | ||||
[ | [ | ||||
content1_metadata1, | content1_metadata1, | ||||
content1_metadata2, | content1_metadata2, | ||||
content1_metadata3, | content1_metadata3, | ||||
content2_metadata, | content2_metadata, | ||||
] | ] | ||||
) | ) | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.CONTENT, content1_swhid, authority | MetadataTargetType.CONTENT, content1_swhid, authority | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert [content1_metadata1, content1_metadata2] == list( | assert [content1_metadata1, content1_metadata2] == list( | ||||
sorted(result["results"], key=lambda x: x.discovery_date,) | sorted(actual_result["results"], key=lambda x: x.discovery_date,) | ||||
) | ) | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.CONTENT, content1_swhid, authority2 | MetadataTargetType.CONTENT, content1_swhid, authority2 | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert [content1_metadata3] == list( | assert [content1_metadata3] == list( | ||||
sorted(result["results"], key=lambda x: x.discovery_date,) | sorted(actual_result["results"], key=lambda x: x.discovery_date,) | ||||
) | ) | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.CONTENT, content2_swhid, authority | MetadataTargetType.CONTENT, content2_swhid, authority | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert [content2_metadata] == list(result["results"],) | assert [content2_metadata] == list(actual_result["results"],) | ||||
def test_content_metadata_get_after(self, swh_storage, sample_data): | def test_content_metadata_get_after(self, swh_storage, sample_data): | ||||
content = sample_data.content | content = sample_data.content | ||||
fetcher = sample_data.metadata_fetcher | fetcher = sample_data.metadata_fetcher | ||||
authority = sample_data.metadata_authority | authority = sample_data.metadata_authority | ||||
content_metadata, content_metadata2 = sample_data.content_metadata[:2] | content_metadata, content_metadata2 = sample_data.content_metadata[:2] | ||||
content_swhid = SWHID(object_type="content", object_id=content.sha1_git) | content_swhid = SWHID(object_type="content", object_id=content.sha1_git) | ||||
swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
swh_storage.raw_extrinsic_metadata_add([content_metadata, content_metadata2]) | swh_storage.raw_extrinsic_metadata_add([content_metadata, content_metadata2]) | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.CONTENT, | MetadataTargetType.CONTENT, | ||||
content_swhid, | content_swhid, | ||||
authority, | authority, | ||||
after=content_metadata.discovery_date - timedelta(seconds=1), | after=content_metadata.discovery_date - timedelta(seconds=1), | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert [content_metadata, content_metadata2] == list( | assert [content_metadata, content_metadata2] == list( | ||||
sorted(result["results"], key=lambda x: x.discovery_date,) | sorted(actual_result["results"], key=lambda x: x.discovery_date,) | ||||
) | ) | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.CONTENT, | MetadataTargetType.CONTENT, | ||||
content_swhid, | content_swhid, | ||||
authority, | authority, | ||||
after=content_metadata.discovery_date, | after=content_metadata.discovery_date, | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert result["results"] == [content_metadata2] | assert actual_result["results"] == [content_metadata2] | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.CONTENT, | MetadataTargetType.CONTENT, | ||||
content_swhid, | content_swhid, | ||||
authority, | authority, | ||||
after=content_metadata2.discovery_date, | after=content_metadata2.discovery_date, | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert result["results"] == [] | assert actual_result["results"] == [] | ||||
def test_content_metadata_get_paginate(self, swh_storage, sample_data): | def test_content_metadata_get_paginate(self, swh_storage, sample_data): | ||||
content = sample_data.content | content = sample_data.content | ||||
fetcher = sample_data.metadata_fetcher | fetcher = sample_data.metadata_fetcher | ||||
authority = sample_data.metadata_authority | authority = sample_data.metadata_authority | ||||
content_metadata, content_metadata2 = sample_data.content_metadata[:2] | content_metadata, content_metadata2 = sample_data.content_metadata[:2] | ||||
content_swhid = SWHID(object_type="content", object_id=content.sha1_git) | content_swhid = SWHID(object_type="content", object_id=content.sha1_git) | ||||
swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
swh_storage.raw_extrinsic_metadata_add([content_metadata, content_metadata2]) | swh_storage.raw_extrinsic_metadata_add([content_metadata, content_metadata2]) | ||||
swh_storage.raw_extrinsic_metadata_get( | swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.CONTENT, content_swhid, authority | MetadataTargetType.CONTENT, content_swhid, authority | ||||
) | ) | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.CONTENT, content_swhid, authority, limit=1 | MetadataTargetType.CONTENT, content_swhid, authority, limit=1 | ||||
) | ) | ||||
assert result["next_page_token"] is not None | assert actual_result["next_page_token"] is not None | ||||
assert result["results"] == [content_metadata] | assert actual_result["results"] == [content_metadata] | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.CONTENT, | MetadataTargetType.CONTENT, | ||||
content_swhid, | content_swhid, | ||||
authority, | authority, | ||||
limit=1, | limit=1, | ||||
page_token=result["next_page_token"], | page_token=actual_result["next_page_token"], | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert result["results"] == [content_metadata2] | assert actual_result["results"] == [content_metadata2] | ||||
def test_content_metadata_get_paginate_same_date(self, swh_storage, sample_data): | def test_content_metadata_get_paginate_same_date(self, swh_storage, sample_data): | ||||
content = sample_data.content | content = sample_data.content | ||||
fetcher1, fetcher2 = sample_data.fetchers[:2] | fetcher1, fetcher2 = sample_data.fetchers[:2] | ||||
authority = sample_data.metadata_authority | authority = sample_data.metadata_authority | ||||
content_metadata, content_metadata2 = sample_data.content_metadata[:2] | content_metadata, content_metadata2 = sample_data.content_metadata[:2] | ||||
content_swhid = SWHID(object_type="content", object_id=content.sha1_git) | content_swhid = SWHID(object_type="content", object_id=content.sha1_git) | ||||
swh_storage.metadata_fetcher_add([fetcher1, fetcher2]) | swh_storage.metadata_fetcher_add([fetcher1, fetcher2]) | ||||
swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
new_content_metadata2 = attr.evolve( | new_content_metadata2 = attr.evolve( | ||||
content_metadata2, | content_metadata2, | ||||
discovery_date=content_metadata2.discovery_date, | discovery_date=content_metadata2.discovery_date, | ||||
fetcher=attr.evolve(fetcher2, metadata=None), | fetcher=attr.evolve(fetcher2, metadata=None), | ||||
) | ) | ||||
swh_storage.raw_extrinsic_metadata_add( | swh_storage.raw_extrinsic_metadata_add( | ||||
[content_metadata, new_content_metadata2] | [content_metadata, new_content_metadata2] | ||||
) | ) | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.CONTENT, content_swhid, authority, limit=1 | MetadataTargetType.CONTENT, content_swhid, authority, limit=1 | ||||
) | ) | ||||
assert result["next_page_token"] is not None | assert actual_result["next_page_token"] is not None | ||||
assert result["results"] == [content_metadata] | assert actual_result["results"] == [content_metadata] | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.CONTENT, | MetadataTargetType.CONTENT, | ||||
content_swhid, | content_swhid, | ||||
authority, | authority, | ||||
limit=1, | limit=1, | ||||
page_token=result["next_page_token"], | page_token=actual_result["next_page_token"], | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert result["results"] == [new_content_metadata2] | assert actual_result["results"] == [new_content_metadata2] | ||||
def test_content_metadata_get__invalid_id(self, swh_storage, sample_data): | def test_content_metadata_get__invalid_id(self, swh_storage, sample_data): | ||||
origin = sample_data.origin | origin = sample_data.origin | ||||
fetcher = sample_data.metadata_fetcher | fetcher = sample_data.metadata_fetcher | ||||
authority = sample_data.metadata_authority | authority = sample_data.metadata_authority | ||||
content_metadata, content_metadata2 = sample_data.content_metadata[:2] | content_metadata, content_metadata2 = sample_data.content_metadata[:2] | ||||
swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
Show All 13 Lines | def test_origin_metadata_add(self, swh_storage, sample_data): | ||||
assert swh_storage.origin_add([origin]) == {"origin:add": 1} | assert swh_storage.origin_add([origin]) == {"origin:add": 1} | ||||
swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
swh_storage.raw_extrinsic_metadata_add([origin_metadata, origin_metadata2]) | swh_storage.raw_extrinsic_metadata_add([origin_metadata, origin_metadata2]) | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.ORIGIN, origin.url, authority | MetadataTargetType.ORIGIN, origin.url, authority | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert list(sorted(result["results"], key=lambda x: x.discovery_date)) == [ | assert list( | ||||
origin_metadata, | sorted(actual_result["results"], key=lambda x: x.discovery_date) | ||||
origin_metadata2, | ) == [origin_metadata, origin_metadata2,] | ||||
] | |||||
def test_origin_metadata_add_duplicate(self, swh_storage, sample_data): | def test_origin_metadata_add_duplicate(self, swh_storage, sample_data): | ||||
"""Duplicates should be silently updated.""" | """Duplicates should be silently updated.""" | ||||
origin = sample_data.origin | origin = sample_data.origin | ||||
fetcher = sample_data.metadata_fetcher | fetcher = sample_data.metadata_fetcher | ||||
authority = sample_data.metadata_authority | authority = sample_data.metadata_authority | ||||
origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | ||||
assert swh_storage.origin_add([origin]) == {"origin:add": 1} | assert swh_storage.origin_add([origin]) == {"origin:add": 1} | ||||
new_origin_metadata2 = attr.evolve( | new_origin_metadata2 = attr.evolve( | ||||
origin_metadata2, format="new-format", metadata=b"new-metadata", | origin_metadata2, format="new-format", metadata=b"new-metadata", | ||||
) | ) | ||||
swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
swh_storage.raw_extrinsic_metadata_add([origin_metadata, origin_metadata2]) | swh_storage.raw_extrinsic_metadata_add([origin_metadata, origin_metadata2]) | ||||
swh_storage.raw_extrinsic_metadata_add([new_origin_metadata2]) | swh_storage.raw_extrinsic_metadata_add([new_origin_metadata2]) | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.ORIGIN, origin.url, authority | MetadataTargetType.ORIGIN, origin.url, authority | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
# which of the two behavior happens is backend-specific. | # which of the two behavior happens is backend-specific. | ||||
expected_results1 = (origin_metadata, new_origin_metadata2) | expected_results1 = (origin_metadata, new_origin_metadata2) | ||||
expected_results2 = (origin_metadata, origin_metadata2) | expected_results2 = (origin_metadata, origin_metadata2) | ||||
assert tuple(sorted(result["results"], key=lambda x: x.discovery_date,)) in ( | assert tuple( | ||||
sorted(actual_result["results"], key=lambda x: x.discovery_date,) | |||||
) in ( | |||||
expected_results1, # cassandra | expected_results1, # cassandra | ||||
expected_results2, # postgresql | expected_results2, # postgresql | ||||
) | ) | ||||
def test_origin_metadata_get(self, swh_storage, sample_data): | def test_origin_metadata_get(self, swh_storage, sample_data): | ||||
origin, origin2 = sample_data.origins[:2] | origin, origin2 = sample_data.origins[:2] | ||||
fetcher, fetcher2 = sample_data.fetchers[:2] | fetcher, fetcher2 = sample_data.fetchers[:2] | ||||
authority, authority2 = sample_data.authorities[:2] | authority, authority2 = sample_data.authorities[:2] | ||||
Show All 9 Lines | def test_origin_metadata_get(self, swh_storage, sample_data): | ||||
swh_storage.metadata_authority_add([authority, authority2]) | swh_storage.metadata_authority_add([authority, authority2]) | ||||
swh_storage.metadata_fetcher_add([fetcher, fetcher2]) | swh_storage.metadata_fetcher_add([fetcher, fetcher2]) | ||||
swh_storage.raw_extrinsic_metadata_add( | swh_storage.raw_extrinsic_metadata_add( | ||||
[origin1_metadata1, origin1_metadata2, origin1_metadata3, origin2_metadata] | [origin1_metadata1, origin1_metadata2, origin1_metadata3, origin2_metadata] | ||||
) | ) | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.ORIGIN, origin.url, authority | MetadataTargetType.ORIGIN, origin.url, authority | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert [origin1_metadata1, origin1_metadata2] == list( | assert [origin1_metadata1, origin1_metadata2] == list( | ||||
sorted(result["results"], key=lambda x: x.discovery_date,) | sorted(actual_result["results"], key=lambda x: x.discovery_date,) | ||||
) | ) | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.ORIGIN, origin.url, authority2 | MetadataTargetType.ORIGIN, origin.url, authority2 | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert [origin1_metadata3] == list( | assert [origin1_metadata3] == list( | ||||
sorted(result["results"], key=lambda x: x.discovery_date,) | sorted(actual_result["results"], key=lambda x: x.discovery_date,) | ||||
) | ) | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.ORIGIN, origin2.url, authority | MetadataTargetType.ORIGIN, origin2.url, authority | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert [origin2_metadata] == list(result["results"],) | assert [origin2_metadata] == list(actual_result["results"],) | ||||
def test_origin_metadata_get_after(self, swh_storage, sample_data): | def test_origin_metadata_get_after(self, swh_storage, sample_data): | ||||
origin = sample_data.origin | origin = sample_data.origin | ||||
fetcher = sample_data.metadata_fetcher | fetcher = sample_data.metadata_fetcher | ||||
authority = sample_data.metadata_authority | authority = sample_data.metadata_authority | ||||
origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | ||||
assert swh_storage.origin_add([origin]) == {"origin:add": 1} | assert swh_storage.origin_add([origin]) == {"origin:add": 1} | ||||
swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
swh_storage.raw_extrinsic_metadata_add([origin_metadata, origin_metadata2]) | swh_storage.raw_extrinsic_metadata_add([origin_metadata, origin_metadata2]) | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.ORIGIN, | MetadataTargetType.ORIGIN, | ||||
origin.url, | origin.url, | ||||
authority, | authority, | ||||
after=origin_metadata.discovery_date - timedelta(seconds=1), | after=origin_metadata.discovery_date - timedelta(seconds=1), | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert list(sorted(result["results"], key=lambda x: x.discovery_date,)) == [ | assert list( | ||||
origin_metadata, | sorted(actual_result["results"], key=lambda x: x.discovery_date,) | ||||
origin_metadata2, | ) == [origin_metadata, origin_metadata2,] | ||||
] | |||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.ORIGIN, | MetadataTargetType.ORIGIN, | ||||
origin.url, | origin.url, | ||||
authority, | authority, | ||||
after=origin_metadata.discovery_date, | after=origin_metadata.discovery_date, | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert result["results"] == [origin_metadata2] | assert actual_result["results"] == [origin_metadata2] | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.ORIGIN, | MetadataTargetType.ORIGIN, | ||||
origin.url, | origin.url, | ||||
authority, | authority, | ||||
after=origin_metadata2.discovery_date, | after=origin_metadata2.discovery_date, | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert result["results"] == [] | assert actual_result["results"] == [] | ||||
def test_origin_metadata_get_paginate(self, swh_storage, sample_data): | def test_origin_metadata_get_paginate(self, swh_storage, sample_data): | ||||
origin = sample_data.origin | origin = sample_data.origin | ||||
fetcher = sample_data.metadata_fetcher | fetcher = sample_data.metadata_fetcher | ||||
authority = sample_data.metadata_authority | authority = sample_data.metadata_authority | ||||
origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | ||||
assert swh_storage.origin_add([origin]) == {"origin:add": 1} | assert swh_storage.origin_add([origin]) == {"origin:add": 1} | ||||
swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
swh_storage.raw_extrinsic_metadata_add([origin_metadata, origin_metadata2]) | swh_storage.raw_extrinsic_metadata_add([origin_metadata, origin_metadata2]) | ||||
swh_storage.raw_extrinsic_metadata_get( | swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.ORIGIN, origin.url, authority | MetadataTargetType.ORIGIN, origin.url, authority | ||||
) | ) | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.ORIGIN, origin.url, authority, limit=1 | MetadataTargetType.ORIGIN, origin.url, authority, limit=1 | ||||
) | ) | ||||
assert result["next_page_token"] is not None | assert actual_result["next_page_token"] is not None | ||||
assert result["results"] == [origin_metadata] | assert actual_result["results"] == [origin_metadata] | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.ORIGIN, | MetadataTargetType.ORIGIN, | ||||
origin.url, | origin.url, | ||||
authority, | authority, | ||||
limit=1, | limit=1, | ||||
page_token=result["next_page_token"], | page_token=actual_result["next_page_token"], | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert result["results"] == [origin_metadata2] | assert actual_result["results"] == [origin_metadata2] | ||||
def test_origin_metadata_get_paginate_same_date(self, swh_storage, sample_data): | def test_origin_metadata_get_paginate_same_date(self, swh_storage, sample_data): | ||||
origin = sample_data.origin | origin = sample_data.origin | ||||
fetcher1, fetcher2 = sample_data.fetchers[:2] | fetcher1, fetcher2 = sample_data.fetchers[:2] | ||||
authority = sample_data.metadata_authority | authority = sample_data.metadata_authority | ||||
origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | ||||
assert swh_storage.origin_add([origin]) == {"origin:add": 1} | assert swh_storage.origin_add([origin]) == {"origin:add": 1} | ||||
swh_storage.metadata_fetcher_add([fetcher1, fetcher2]) | swh_storage.metadata_fetcher_add([fetcher1, fetcher2]) | ||||
swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
new_origin_metadata2 = attr.evolve( | new_origin_metadata2 = attr.evolve( | ||||
origin_metadata2, | origin_metadata2, | ||||
discovery_date=origin_metadata2.discovery_date, | discovery_date=origin_metadata2.discovery_date, | ||||
fetcher=attr.evolve(fetcher2, metadata=None), | fetcher=attr.evolve(fetcher2, metadata=None), | ||||
) | ) | ||||
swh_storage.raw_extrinsic_metadata_add([origin_metadata, new_origin_metadata2]) | swh_storage.raw_extrinsic_metadata_add([origin_metadata, new_origin_metadata2]) | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.ORIGIN, origin.url, authority, limit=1 | MetadataTargetType.ORIGIN, origin.url, authority, limit=1 | ||||
) | ) | ||||
assert result["next_page_token"] is not None | assert actual_result["next_page_token"] is not None | ||||
assert result["results"] == [origin_metadata] | assert actual_result["results"] == [origin_metadata] | ||||
result = swh_storage.raw_extrinsic_metadata_get( | actual_result = swh_storage.raw_extrinsic_metadata_get( | ||||
MetadataTargetType.ORIGIN, | MetadataTargetType.ORIGIN, | ||||
origin.url, | origin.url, | ||||
authority, | authority, | ||||
limit=1, | limit=1, | ||||
page_token=result["next_page_token"], | page_token=actual_result["next_page_token"], | ||||
) | ) | ||||
assert result["next_page_token"] is None | assert actual_result["next_page_token"] is None | ||||
assert result["results"] == [new_origin_metadata2] | assert actual_result["results"] == [new_origin_metadata2] | ||||
def test_origin_metadata_add_missing_authority(self, swh_storage, sample_data): | def test_origin_metadata_add_missing_authority(self, swh_storage, sample_data): | ||||
origin = sample_data.origin | origin = sample_data.origin | ||||
fetcher = sample_data.metadata_fetcher | fetcher = sample_data.metadata_fetcher | ||||
origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | ||||
assert swh_storage.origin_add([origin]) == {"origin:add": 1} | assert swh_storage.origin_add([origin]) == {"origin:add": 1} | ||||
swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
▲ Show 20 Lines • Show All 194 Lines • ▼ Show 20 Lines | class TestStorageGeneratedData: | ||||
@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: | ||||
result = swh_storage.origin_list(page_token=page_token, limit=limit) | actual_result = swh_storage.origin_list(page_token=page_token, limit=limit) | ||||
assert len(result["origins"]) <= limit | assert len(actual_result["origins"]) <= limit | ||||
returned_origins.extend(origin["url"] for origin in result["origins"]) | returned_origins.extend( | ||||
origin["url"] for origin in actual_result["origins"] | |||||
) | |||||
i += 1 | i += 1 | ||||
page_token = result.get("next_page_token") | page_token = actual_result.get("next_page_token") | ||||
if page_token is None: | if page_token is None: | ||||
assert i * limit >= len(swh_origins) | assert i * limit >= len(swh_origins) | ||||
break | break | ||||
else: | else: | ||||
assert len(result["origins"]) == limit | assert len(actual_result["origins"]) == limit | ||||
expected_origins = [origin.url for origin in swh_origins] | expected_origins = [origin.url for origin in swh_origins] | ||||
assert sorted(returned_origins) == sorted(expected_origins) | assert sorted(returned_origins) == sorted(expected_origins) | ||||
def test_origin_count(self, swh_storage, sample_data): | def test_origin_count(self, swh_storage, sample_data): | ||||
swh_storage.origin_add(sample_data.origins) | swh_storage.origin_add(sample_data.origins) | ||||
assert swh_storage.origin_count("github") == 3 | assert swh_storage.origin_count("github") == 3 | ||||
▲ Show 20 Lines • Show All 325 Lines • Show Last 20 Lines |
No need to test next_page_token, it's opaque. Just check it's not None. (same comment for other occurences below)