Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 1,869 Lines • ▼ Show 20 Lines | def test_origin_visit_get_by(self, swh_storage, sample_data): | ||||
| date=now(), | date=now(), | ||||
| status="full", | status="full", | ||||
| snapshot=snapshot.id, | snapshot=snapshot.id, | ||||
| metadata=visit1_metadata, | metadata=visit1_metadata, | ||||
| ) | ) | ||||
| ] | ] | ||||
| ) | ) | ||||
| expected_origin_visit = origin_visit1.to_dict() | actual_visit = swh_storage.origin_visit_get_by(origin_url, origin_visit1.visit) | ||||
| expected_origin_visit.update( | assert actual_visit == origin_visit1 | ||||
| { | |||||
| "origin": origin_url, | |||||
| "visit": origin_visit1.visit, | |||||
| "date": sample_data.date_visit2, | |||||
| "type": sample_data.type_visit2, | |||||
| "metadata": visit1_metadata, | |||||
| "status": "full", | |||||
| "snapshot": snapshot.id, | |||||
| } | |||||
| ) | |||||
| # when | def test_origin_visit_get_by__no_result(self, swh_storage, sample_data): | ||||
| actual_origin_visit1 = swh_storage.origin_visit_get_by( | actual_visit = swh_storage.origin_visit_get_by("unknown", 10) # unknown origin | ||||
| origin_url, origin_visit1.visit | assert actual_visit is None | ||||
| ) | |||||
| # then | |||||
| assert actual_origin_visit1 == expected_origin_visit | |||||
| def test_origin_visit_get_by__unknown_origin(self, swh_storage): | |||||
| assert swh_storage.origin_visit_get_by("foo", 10) is None | |||||
| def test_origin_visit_get_by_no_result(self, swh_storage, sample_data): | |||||
| origin = sample_data.origin | origin = sample_data.origin | ||||
| swh_storage.origin_add([origin]) | swh_storage.origin_add([origin]) | ||||
| actual_origin_visit = swh_storage.origin_visit_get_by(origin.url, 999) | actual_visit = swh_storage.origin_visit_get_by(origin.url, 999) # unknown visit | ||||
| assert actual_origin_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 result | ||||
| assert swh_storage.origin_visit_get_latest("unknown-origin") is None | assert swh_storage.origin_visit_get_latest("unknown-origin") is None | ||||
| ▲ Show 20 Lines • Show All 742 Lines • ▼ Show 20 Lines | def test_snapshot_add_get(self, swh_storage, sample_data): | ||||
| origin = sample_data.origin | origin = sample_data.origin | ||||
| swh_storage.origin_add([origin]) | swh_storage.origin_add([origin]) | ||||
| visit = OriginVisit( | visit = 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, | ||||
| ) | ) | ||||
| origin_visit1 = swh_storage.origin_visit_add([visit])[0] | ov1 = swh_storage.origin_visit_add([visit])[0] | ||||
| visit_id = origin_visit1.visit | |||||
| swh_storage.snapshot_add([snapshot]) | swh_storage.snapshot_add([snapshot]) | ||||
| swh_storage.origin_visit_status_add( | swh_storage.origin_visit_status_add( | ||||
| [ | [ | ||||
| OriginVisitStatus( | OriginVisitStatus( | ||||
| origin=origin.url, | origin=origin.url, | ||||
| visit=origin_visit1.visit, | visit=ov1.visit, | ||||
| date=now(), | date=now(), | ||||
| status="ongoing", | status="ongoing", | ||||
| snapshot=snapshot.id, | snapshot=snapshot.id, | ||||
| ) | ) | ||||
| ] | ] | ||||
| ) | ) | ||||
| expected_snapshot = {**snapshot.to_dict(), "next_branch": None} | expected_snapshot = {**snapshot.to_dict(), "next_branch": None} | ||||
| by_id = swh_storage.snapshot_get(snapshot.id) | by_id = swh_storage.snapshot_get(snapshot.id) | ||||
| assert by_id == expected_snapshot | assert by_id == expected_snapshot | ||||
| by_ov = swh_storage.snapshot_get_by_origin_visit(origin.url, visit_id) | by_ov = swh_storage.snapshot_get_by_origin_visit(origin.url, ov1.visit) | ||||
| assert by_ov == expected_snapshot | assert by_ov == expected_snapshot | ||||
| origin_visit_info = swh_storage.origin_visit_get_by(origin.url, visit_id) | actual_visit = swh_storage.origin_visit_get_by(origin.url, ov1.visit) | ||||
| assert origin_visit_info["snapshot"] == snapshot.id | assert actual_visit == ov1 | ||||
| visit_status = swh_storage.origin_visit_status_get_latest( | |||||
| origin.url, ov1.visit, require_snapshot=True | |||||
| ) | |||||
| assert visit_status.snapshot == snapshot.id | |||||
| def test_snapshot_add_twice__by_origin_visit(self, swh_storage, sample_data): | def test_snapshot_add_twice__by_origin_visit(self, swh_storage, sample_data): | ||||
| snapshot = sample_data.snapshot | snapshot = sample_data.snapshot | ||||
| origin = sample_data.origin | origin = sample_data.origin | ||||
| swh_storage.origin_add([origin]) | swh_storage.origin_add([origin]) | ||||
| ov1 = swh_storage.origin_visit_add( | ov1 = swh_storage.origin_visit_add( | ||||
| [ | [ | ||||
| ▲ Show 20 Lines • Show All 1,476 Lines • Show Last 20 Lines | |||||