Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
Show First 20 Lines • Show All 2,146 Lines • ▼ Show 20 Lines | def test_origin_visit_upsert_existing(self, swh_storage): | ||||
"snapshot": None, | "snapshot": None, | ||||
} | } | ||||
] | ] | ||||
data1 = { | data1 = { | ||||
"origin": origin_url, | "origin": origin_url, | ||||
"date": data.date_visit2, | "date": data.date_visit2, | ||||
"visit": origin_visit1.visit, | "visit": origin_visit1.visit, | ||||
"type": data.type_visit1, | |||||
"status": "ongoing", | "status": "ongoing", | ||||
"metadata": None, | "metadata": None, | ||||
"snapshot": None, | "snapshot": None, | ||||
} | } | ||||
data2 = { | data2 = { | ||||
"origin": origin_url, | "origin": origin_url, | ||||
"date": data.date_visit2, | "date": data.date_visit2, | ||||
"visit": origin_visit1.visit, | "visit": origin_visit1.visit, | ||||
"type": data.type_visit1, | "type": data.type_visit1, | ||||
"status": "full", | "status": "full", | ||||
"metadata": None, | "metadata": None, | ||||
"snapshot": None, | "snapshot": None, | ||||
} | } | ||||
assert list(swh_storage.journal_writer.journal.objects) == [ | actual_written_objects = list(swh_storage.journal_writer.journal.objects) | ||||
assert actual_written_objects == [ | |||||
("origin", Origin.from_dict(data.origin2)), | ("origin", Origin.from_dict(data.origin2)), | ||||
("origin_visit", OriginVisit.from_dict(data1)), | ( | ||||
"origin_visit", | |||||
OriginVisit.from_dict({**data1, "type": data.type_visit1,}), | |||||
), | |||||
("origin_visit_status", OriginVisitStatus.from_dict(data1)), | |||||
("origin_visit", OriginVisit.from_dict(data2)), | ("origin_visit", OriginVisit.from_dict(data2)), | ||||
] | ] | ||||
def test_origin_visit_upsert_missing_visit_id(self, swh_storage): | def test_origin_visit_upsert_missing_visit_id(self, swh_storage): | ||||
# given | # given | ||||
origin_url = swh_storage.origin_add_one(data.origin2) | origin_url = swh_storage.origin_add_one(data.origin2) | ||||
# then | # then | ||||
▲ Show 20 Lines • Show All 158 Lines • ▼ Show 20 Lines | def test_snapshot_add_get_empty(self, swh_storage): | ||||
actual_result = swh_storage.snapshot_add([data.empty_snapshot]) | actual_result = swh_storage.snapshot_add([data.empty_snapshot]) | ||||
assert actual_result == {"snapshot:add": 1} | assert actual_result == {"snapshot:add": 1} | ||||
swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
origin_url, | origin_url, | ||||
origin_visit1.visit, | origin_visit1.visit, | ||||
status="ongoing", | status="ongoing", | ||||
snapshot=data.empty_snapshot["id"], | snapshot=data.empty_snapshot["id"], | ||||
# date=data.date_visit2 | |||||
) | ) | ||||
by_id = swh_storage.snapshot_get(data.empty_snapshot["id"]) | by_id = swh_storage.snapshot_get(data.empty_snapshot["id"]) | ||||
assert by_id == {**data.empty_snapshot, "next_branch": None} | assert by_id == {**data.empty_snapshot, "next_branch": None} | ||||
by_ov = swh_storage.snapshot_get_by_origin_visit(origin_url, visit_id) | by_ov = swh_storage.snapshot_get_by_origin_visit(origin_url, visit_id) | ||||
assert by_ov == {**data.empty_snapshot, "next_branch": None} | assert by_ov == {**data.empty_snapshot, "next_branch": None} | ||||
data1 = { | data1 = { | ||||
"origin": origin_url, | "origin": origin_url, | ||||
"date": data.date_visit1, | "date": data.date_visit1, | ||||
"visit": origin_visit1.visit, | "visit": origin_visit1.visit, | ||||
"type": data.type_visit1, | |||||
"status": "ongoing", | "status": "ongoing", | ||||
"metadata": None, | "metadata": None, | ||||
"snapshot": None, | "snapshot": None, | ||||
} | } | ||||
data2 = { | data2 = { | ||||
"origin": origin_url, | "origin": origin_url, | ||||
"date": data.date_visit1, | "date": data.date_visit1, | ||||
"visit": origin_visit1.visit, | "visit": origin_visit1.visit, | ||||
"type": data.type_visit1, | |||||
"status": "ongoing", | "status": "ongoing", | ||||
"metadata": None, | "metadata": None, | ||||
"snapshot": data.empty_snapshot["id"], | "snapshot": data.empty_snapshot["id"], | ||||
} | } | ||||
assert list(swh_storage.journal_writer.journal.objects) == [ | actual_objects = list(swh_storage.journal_writer.journal.objects) | ||||
assert actual_objects == [ | |||||
("origin", Origin.from_dict(data.origin)), | ("origin", Origin.from_dict(data.origin)), | ||||
("origin_visit", OriginVisit.from_dict(data1)), | ( | ||||
"origin_visit", | |||||
OriginVisit.from_dict({**data1, "type": data.type_visit1},), | |||||
), | |||||
("origin_visit_status", OriginVisitStatus.from_dict(data1)), | |||||
("snapshot", Snapshot.from_dict(data.empty_snapshot)), | ("snapshot", Snapshot.from_dict(data.empty_snapshot)), | ||||
("origin_visit", OriginVisit.from_dict(data2)), | ( | ||||
"origin_visit", | |||||
OriginVisit.from_dict({**data2, "type": data.type_visit1,}), | |||||
), | |||||
# ("origin_visit_status", OriginVisitStatus.from_dict(data2)), | |||||
] | ] | ||||
def test_snapshot_add_get_complete(self, swh_storage): | def test_snapshot_add_get_complete(self, swh_storage): | ||||
origin_url = data.origin["url"] | origin_url = data.origin["url"] | ||||
origin_url = swh_storage.origin_add_one(data.origin) | origin_url = swh_storage.origin_add_one(data.origin) | ||||
origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
origin_url, date=data.date_visit1, type=data.type_visit1 | origin_url, date=data.date_visit1, type=data.type_visit1 | ||||
) | ) | ||||
▲ Show 20 Lines • Show All 341 Lines • ▼ Show 20 Lines | def test_snapshot_add_twice__by_origin_visit(self, swh_storage): | ||||
by_ov2 = swh_storage.snapshot_get_by_origin_visit(origin_url, visit2_id) | by_ov2 = swh_storage.snapshot_get_by_origin_visit(origin_url, visit2_id) | ||||
assert by_ov2 == {**data.snapshot, "next_branch": None} | assert by_ov2 == {**data.snapshot, "next_branch": None} | ||||
data1 = { | data1 = { | ||||
"origin": origin_url, | "origin": origin_url, | ||||
"date": data.date_visit1, | "date": data.date_visit1, | ||||
"visit": origin_visit1.visit, | "visit": origin_visit1.visit, | ||||
"type": data.type_visit1, | # "type": data.type_visit1, | ||||
"status": "ongoing", | "status": "ongoing", | ||||
"metadata": None, | "metadata": None, | ||||
"snapshot": None, | "snapshot": None, | ||||
} | } | ||||
data2 = { | data2 = { | ||||
"origin": origin_url, | "origin": origin_url, | ||||
"date": data.date_visit1, | "date": data.date_visit1, | ||||
"visit": origin_visit1.visit, | "visit": origin_visit1.visit, | ||||
"type": data.type_visit1, | "type": data.type_visit1, | ||||
"status": "ongoing", | "status": "ongoing", | ||||
"metadata": None, | "metadata": None, | ||||
"snapshot": data.snapshot["id"], | "snapshot": data.snapshot["id"], | ||||
} | } | ||||
data3 = { | data3 = { | ||||
"origin": origin_url, | "origin": origin_url, | ||||
"date": data.date_visit2, | "date": data.date_visit2, | ||||
"visit": origin_visit2.visit, | "visit": origin_visit2.visit, | ||||
"type": data.type_visit2, | # "type": data.type_visit2, | ||||
"status": "ongoing", | "status": "ongoing", | ||||
"metadata": None, | "metadata": None, | ||||
"snapshot": None, | "snapshot": None, | ||||
} | } | ||||
data4 = { | data4 = { | ||||
"origin": origin_url, | "origin": origin_url, | ||||
"date": data.date_visit2, | "date": data.date_visit2, | ||||
"visit": origin_visit2.visit, | "visit": origin_visit2.visit, | ||||
"type": data.type_visit2, | "type": data.type_visit2, | ||||
"status": "ongoing", | "status": "ongoing", | ||||
"metadata": None, | "metadata": None, | ||||
"snapshot": data.snapshot["id"], | "snapshot": data.snapshot["id"], | ||||
} | } | ||||
assert list(swh_storage.journal_writer.journal.objects) == [ | assert list(swh_storage.journal_writer.journal.objects) == [ | ||||
("origin", Origin.from_dict(data.origin)), | ("origin", Origin.from_dict(data.origin)), | ||||
("origin_visit", OriginVisit.from_dict(data1)), | ( | ||||
"origin_visit", | |||||
OriginVisit.from_dict({**data1, "type": data.type_visit1}), | |||||
), | |||||
("origin_visit_status", OriginVisitStatus.from_dict(data1)), | |||||
("snapshot", Snapshot.from_dict(data.snapshot)), | ("snapshot", Snapshot.from_dict(data.snapshot)), | ||||
("origin_visit", OriginVisit.from_dict(data2)), | ("origin_visit", OriginVisit.from_dict(data2)), | ||||
("origin_visit", OriginVisit.from_dict(data3)), | ( | ||||
"origin_visit", | |||||
OriginVisit.from_dict({**data3, "type": data.type_visit2}), | |||||
), | |||||
("origin_visit_status", OriginVisitStatus.from_dict(data3)), | |||||
("origin_visit", OriginVisit.from_dict(data4)), | ("origin_visit", OriginVisit.from_dict(data4)), | ||||
] | ] | ||||
def test_snapshot_get_latest(self, swh_storage): | def test_snapshot_get_latest(self, swh_storage): | ||||
origin_url = swh_storage.origin_add_one(data.origin) | origin_url = swh_storage.origin_add_one(data.origin) | ||||
origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
origin_url, date=data.date_visit1, type=data.type_visit1 | origin_url, date=data.date_visit1, type=data.type_visit1 | ||||
) | ) | ||||
▲ Show 20 Lines • Show All 1,202 Lines • Show Last 20 Lines |