Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
Show First 20 Lines • Show All 1,788 Lines • ▼ Show 20 Lines | def test_origin_visit_status_add(self, swh_storage): | ||||
assert origin_visit2["metadata"] == {"intrinsic": "something"} | assert origin_visit2["metadata"] == {"intrinsic": "something"} | ||||
actual_objects = list(swh_storage.journal_writer.journal.objects) | actual_objects = list(swh_storage.journal_writer.journal.objects) | ||||
expected_origins = [origin1, origin2] | expected_origins = [origin1, origin2] | ||||
expected_visits = [ov1, ov2] | expected_visits = [ov1, ov2] | ||||
expected_visit_statuses = [] | expected_visit_statuses = [] | ||||
for visit in expected_visits: # out of origin-visit-add calls | for visit in expected_visits: # out of origin-visit-add calls | ||||
visit_status = visit.to_dict() | visit_status = visit.to_dict() | ||||
ardumont: mmmph, wrong location, that should go in the test around origin-visit-get-latest (we are in the… | |||||
visit_status.pop("type") | visit_status.pop("type") | ||||
expected_visit_statuses.append(OriginVisitStatus.from_dict(visit_status)) | expected_visit_statuses.append(OriginVisitStatus.from_dict(visit_status)) | ||||
# out of origin-visit-status add calls | # out of origin-visit-status add calls | ||||
expected_visit_statuses += [visit_status1, visit_status2] | expected_visit_statuses += [visit_status1, visit_status2] | ||||
expected_objects = ( | expected_objects = ( | ||||
[("origin", o) for o in expected_origins] | [("origin", o) for o in expected_origins] | ||||
+ [("origin_visit", v) for v in expected_visits] | + [("origin_visit", v) for v in expected_visits] | ||||
▲ Show 20 Lines • Show All 189 Lines • ▼ Show 20 Lines | |||||
def test_origin_visit_get_by__unknown_origin(self, swh_storage): | def test_origin_visit_get_by__unknown_origin(self, swh_storage): | ||||
assert swh_storage.origin_visit_get_by("foo", 10) is None | assert swh_storage.origin_visit_get_by("foo", 10) is None | ||||
def test_origin_visit_get_by_no_result(self, swh_storage): | def test_origin_visit_get_by_no_result(self, swh_storage): | ||||
swh_storage.origin_add([data.origin]) | swh_storage.origin_add([data.origin]) | ||||
actual_origin_visit = swh_storage.origin_visit_get_by(data.origin["url"], 999) | actual_origin_visit = swh_storage.origin_visit_get_by(data.origin["url"], 999) | ||||
assert actual_origin_visit is None | assert actual_origin_visit is None | ||||
def test_origin_visit_get_latest(self, swh_storage): | def test_origin_visit_get_latest(self, swh_storage): | ||||
olasdUnsubmitted Not Done Inline ActionsThis test is now 180 lines long. Could we consider splitting it into more readable chunks? I think comparing the results for various filters should be split into separate tests, instead of done sequentially in a single huge test (that I have a hard time following). olasd: This test is now 180 lines long.
Could we consider splitting it into more readable chunks? I… | |||||
ardumontAuthorUnsubmitted Done Inline Actionsgranted! I have a hard time myself. ardumont: granted!
I have a hard time myself. | |||||
ardumontAuthorUnsubmitted Done Inline ActionsI'll start by getting the type filtering test in another test for that diff as that makes sense here. ardumont: I'll start by getting the type filtering test in another test for that diff as that makes sense… | |||||
origin_url = swh_storage.origin_add_one(data.origin) | origin = Origin.from_dict(data.origin) | ||||
swh_storage.origin_add_one(origin) | |||||
visit1 = OriginVisit( | visit1 = OriginVisit( | ||||
origin=origin_url, | origin=origin.url, | ||||
date=data.date_visit1, | date=data.date_visit1, | ||||
type=data.type_visit1, | type=data.type_visit1, | ||||
status="ongoing", | status="ongoing", | ||||
snapshot=None, | snapshot=None, | ||||
) | ) | ||||
visit2 = OriginVisit( | visit2 = OriginVisit( | ||||
origin=origin_url, | origin=origin.url, | ||||
date=data.date_visit2, | date=data.date_visit2, | ||||
type=data.type_visit2, | type=data.type_visit2, | ||||
status="ongoing", | status="ongoing", | ||||
snapshot=None, | snapshot=None, | ||||
) | ) | ||||
# Add a visit with the same date as the previous one | # Add a visit with the same date as the previous one | ||||
visit3 = OriginVisit( | visit3 = OriginVisit( | ||||
origin=origin_url, | origin=origin.url, | ||||
date=data.date_visit2, | date=data.date_visit2, | ||||
type=data.type_visit2, | type=data.type_visit2, | ||||
status="ongoing", | status="ongoing", | ||||
snapshot=None, | snapshot=None, | ||||
) | ) | ||||
assert data.type_visit1 != data.type_visit2 | |||||
assert data.date_visit1 < data.date_visit2 | |||||
ov1, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3]) | ov1, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3]) | ||||
origin_visit1 = swh_storage.origin_visit_get_by(origin.url, ov1.visit) | |||||
origin_visit2 = swh_storage.origin_visit_get_by(origin.url, ov2.visit) | |||||
origin_visit3 = swh_storage.origin_visit_get_by(origin.url, ov3.visit) | |||||
assert data.type_visit1 != data.type_visit2 | |||||
# Check type filter is ok | |||||
actual_ov1 = swh_storage.origin_visit_get_latest( | |||||
origin.url, type=data.type_visit1, | |||||
) | |||||
assert actual_ov1 == origin_visit1 | |||||
actual_ov3 = swh_storage.origin_visit_get_latest( | |||||
origin.url, type=data.type_visit2, | |||||
) | |||||
assert actual_ov3 == origin_visit3 | |||||
new_type = "npm" | |||||
assert new_type not in [data.type_visit1, data.type_visit2] | |||||
assert ( | |||||
swh_storage.origin_visit_get_latest( | |||||
origin.url, type=new_type, # no visit matching that type | |||||
) | |||||
is None | |||||
) | |||||
origin_visit1 = swh_storage.origin_visit_get_by(origin_url, ov1.visit) | |||||
origin_visit2 = swh_storage.origin_visit_get_by(origin_url, ov2.visit) | |||||
origin_visit3 = swh_storage.origin_visit_get_by(origin_url, ov3.visit) | |||||
# Two visits, both with no snapshot | # Two visits, both with no snapshot | ||||
assert origin_visit3 == swh_storage.origin_visit_get_latest(origin_url) | assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url) | ||||
assert ( | assert ( | ||||
swh_storage.origin_visit_get_latest(origin_url, require_snapshot=True) | swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True) | ||||
is None | is None | ||||
) | ) | ||||
# Add snapshot to visit1; require_snapshot=True makes it return | # Add snapshot to visit1; require_snapshot=True makes it return | ||||
# visit1 and require_snapshot=False still returns visit2 | # visit1 and require_snapshot=False still returns visit2 | ||||
swh_storage.snapshot_add([data.complete_snapshot]) | complete_snapshot = Snapshot.from_dict(data.complete_snapshot) | ||||
swh_storage.snapshot_add([complete_snapshot]) | |||||
swh_storage.origin_visit_status_add( | swh_storage.origin_visit_status_add( | ||||
[ | [ | ||||
OriginVisitStatus( | OriginVisitStatus( | ||||
origin=origin_url, | origin=origin.url, | ||||
visit=ov1.visit, | visit=ov1.visit, | ||||
date=now(), | date=now(), | ||||
status="ongoing", | status="ongoing", | ||||
snapshot=data.complete_snapshot["id"], | snapshot=complete_snapshot.id, | ||||
) | ) | ||||
] | ] | ||||
) | ) | ||||
assert { | assert { | ||||
**origin_visit1, | **origin_visit1, | ||||
"snapshot": data.complete_snapshot["id"], | "snapshot": complete_snapshot.id, | ||||
} == swh_storage.origin_visit_get_latest(origin_url, require_snapshot=True) | } == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True) | ||||
assert origin_visit3 == swh_storage.origin_visit_get_latest(origin_url) | assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url) | ||||
# Status filter: all three visits are status=ongoing, so no visit | # Status filter: all three visits are status=ongoing, so no visit | ||||
# returned | # returned | ||||
assert ( | assert ( | ||||
swh_storage.origin_visit_get_latest(origin_url, allowed_statuses=["full"]) | swh_storage.origin_visit_get_latest(origin.url, allowed_statuses=["full"]) | ||||
is None | is None | ||||
) | ) | ||||
# Mark the first visit as completed and check status filter again | # Mark the first visit as completed and check status filter again | ||||
swh_storage.origin_visit_status_add( | swh_storage.origin_visit_status_add( | ||||
[ | [ | ||||
OriginVisitStatus( | OriginVisitStatus( | ||||
origin=origin_url, | origin=origin.url, | ||||
visit=ov1.visit, | visit=ov1.visit, | ||||
date=now(), | date=now(), | ||||
status="full", | status="full", | ||||
snapshot=data.complete_snapshot["id"], | snapshot=complete_snapshot.id, | ||||
) | ) | ||||
] | ] | ||||
) | ) | ||||
assert { | assert { | ||||
**origin_visit1, | **origin_visit1, | ||||
"snapshot": data.complete_snapshot["id"], | "snapshot": complete_snapshot.id, | ||||
"status": "full", | "status": "full", | ||||
} == swh_storage.origin_visit_get_latest(origin_url, allowed_statuses=["full"]) | } == swh_storage.origin_visit_get_latest(origin.url, allowed_statuses=["full"]) | ||||
assert origin_visit3 == swh_storage.origin_visit_get_latest(origin_url) | assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url) | ||||
# Add snapshot to visit2 and check that the new snapshot is returned | # Add snapshot to visit2 and check that the new snapshot is returned | ||||
swh_storage.snapshot_add([data.empty_snapshot]) | empty_snapshot = Snapshot.from_dict(data.empty_snapshot) | ||||
swh_storage.snapshot_add([empty_snapshot]) | |||||
swh_storage.origin_visit_status_add( | swh_storage.origin_visit_status_add( | ||||
[ | [ | ||||
OriginVisitStatus( | OriginVisitStatus( | ||||
origin=origin_url, | origin=origin.url, | ||||
visit=ov2.visit, | visit=ov2.visit, | ||||
date=now(), | date=now(), | ||||
status="ongoing", | status="ongoing", | ||||
snapshot=data.empty_snapshot["id"], | snapshot=empty_snapshot.id, | ||||
) | ) | ||||
] | ] | ||||
) | ) | ||||
assert { | assert { | ||||
**origin_visit2, | **origin_visit2, | ||||
"snapshot": data.empty_snapshot["id"], | "snapshot": empty_snapshot.id, | ||||
} == swh_storage.origin_visit_get_latest(origin_url, require_snapshot=True) | } == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True) | ||||
assert origin_visit3 == swh_storage.origin_visit_get_latest(origin_url) | assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url) | ||||
# Check that the status filter is still working | # Check that the status filter is still working | ||||
assert { | assert { | ||||
**origin_visit1, | **origin_visit1, | ||||
"snapshot": data.complete_snapshot["id"], | "snapshot": complete_snapshot.id, | ||||
"status": "full", | "status": "full", | ||||
} == swh_storage.origin_visit_get_latest(origin_url, allowed_statuses=["full"]) | } == swh_storage.origin_visit_get_latest(origin.url, allowed_statuses=["full"]) | ||||
# Add snapshot to visit3 (same date as visit2) | # Add snapshot to visit3 (same date as visit2) | ||||
swh_storage.snapshot_add([data.complete_snapshot]) | |||||
swh_storage.origin_visit_status_add( | swh_storage.origin_visit_status_add( | ||||
[ | [ | ||||
OriginVisitStatus( | OriginVisitStatus( | ||||
origin=origin_url, | origin=origin.url, | ||||
visit=ov3.visit, | visit=ov3.visit, | ||||
date=now(), | date=now(), | ||||
status="ongoing", | status="ongoing", | ||||
snapshot=data.complete_snapshot["id"], | snapshot=complete_snapshot.id, | ||||
) | ) | ||||
] | ] | ||||
) | ) | ||||
assert { | assert { | ||||
**origin_visit1, | **origin_visit1, | ||||
"snapshot": data.complete_snapshot["id"], | "snapshot": complete_snapshot.id, | ||||
"status": "full", | "status": "full", | ||||
} == swh_storage.origin_visit_get_latest(origin_url, allowed_statuses=["full"]) | } == swh_storage.origin_visit_get_latest(origin.url, allowed_statuses=["full"]) | ||||
assert { | assert { | ||||
**origin_visit1, | **origin_visit1, | ||||
"snapshot": data.complete_snapshot["id"], | "snapshot": complete_snapshot.id, | ||||
"status": "full", | "status": "full", | ||||
} == swh_storage.origin_visit_get_latest( | } == swh_storage.origin_visit_get_latest( | ||||
origin_url, allowed_statuses=["full"], require_snapshot=True | origin.url, allowed_statuses=["full"], require_snapshot=True | ||||
) | ) | ||||
assert { | assert { | ||||
**origin_visit3, | **origin_visit3, | ||||
"snapshot": data.complete_snapshot["id"], | "snapshot": complete_snapshot.id, | ||||
} == swh_storage.origin_visit_get_latest(origin_url) | } == swh_storage.origin_visit_get_latest(origin.url) | ||||
assert { | assert { | ||||
**origin_visit3, | **origin_visit3, | ||||
"snapshot": data.complete_snapshot["id"], | "snapshot": complete_snapshot.id, | ||||
} == swh_storage.origin_visit_get_latest(origin_url, require_snapshot=True) | } == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True) | ||||
def test_origin_visit_status_get_latest(self, swh_storage): | def test_origin_visit_status_get_latest(self, swh_storage): | ||||
origin1 = Origin.from_dict(data.origin) | origin1 = Origin.from_dict(data.origin) | ||||
swh_storage.origin_add_one(data.origin) | swh_storage.origin_add_one(data.origin) | ||||
# to have some reference visits | # to have some reference visits | ||||
ov1, ov2 = swh_storage.origin_visit_add( | ov1, ov2 = swh_storage.origin_visit_add( | ||||
▲ Show 20 Lines • Show All 1,937 Lines • Show Last 20 Lines |
mmmph, wrong location, that should go in the test around origin-visit-get-latest (we are in the origin-visit-status-add test here ;)