Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/algos/test_snapshot.py
| Show All 38 Lines | def test_snapshot_large(swh_storage, branch_name, branch_target): # noqa | ||||
| ) | ) | ||||
| swh_storage.snapshot_add([snapshot]) | swh_storage.snapshot_add([snapshot]) | ||||
| returned_snapshot = snapshot_get_all_branches(swh_storage, snapshot.id) | returned_snapshot = snapshot_get_all_branches(swh_storage, snapshot.id) | ||||
| assert snapshot.to_dict() == returned_snapshot | assert snapshot.to_dict() == returned_snapshot | ||||
| def test_snapshot_get_latest_none(swh_storage, sample_data_model): | def test_snapshot_get_latest_none(swh_storage, sample_data): | ||||
| """Retrieve latest snapshot on unknown origin or origin without snapshot should | """Retrieve latest snapshot on unknown origin or origin without snapshot should | ||||
| yield no result | yield no result | ||||
| """ | """ | ||||
| # unknown origin so None | # unknown origin so None | ||||
| assert snapshot_get_latest(swh_storage, "unknown-origin") is None | assert snapshot_get_latest(swh_storage, "unknown-origin") is None | ||||
| # no snapshot on origin visit so None | # no snapshot on origin visit so None | ||||
| origin = sample_data_model["origin"][0] | origin = sample_data["origin"][0] | ||||
| swh_storage.origin_add([origin]) | swh_storage.origin_add([origin]) | ||||
| origin_visit, origin_visit2 = sample_data_model["origin_visit"][:2] | origin_visit, origin_visit2 = sample_data["origin_visit"][:2] | ||||
| assert origin_visit.origin == origin.url | assert origin_visit.origin == origin.url | ||||
| swh_storage.origin_visit_add([origin_visit]) | swh_storage.origin_visit_add([origin_visit]) | ||||
| assert snapshot_get_latest(swh_storage, origin.url) is None | assert snapshot_get_latest(swh_storage, origin.url) is None | ||||
| ov1 = swh_storage.origin_visit_get_latest(origin.url) | ov1 = swh_storage.origin_visit_get_latest(origin.url) | ||||
| assert ov1 is not None | assert ov1 is not None | ||||
| visit_id = ov1["visit"] | visit_id = ov1["visit"] | ||||
| # visit references a snapshot but the snapshot does not exist in backend for some | # visit references a snapshot but the snapshot does not exist in backend for some | ||||
| # reason | # reason | ||||
| complete_snapshot = sample_data_model["snapshot"][2] | complete_snapshot = sample_data["snapshot"][2] | ||||
| swh_storage.origin_visit_status_add( | swh_storage.origin_visit_status_add( | ||||
| [ | [ | ||||
| OriginVisitStatus( | OriginVisitStatus( | ||||
| origin=origin.url, | origin=origin.url, | ||||
| visit=visit_id, | visit=visit_id, | ||||
| date=origin_visit2.date, | date=origin_visit2.date, | ||||
| status="partial", | status="partial", | ||||
| snapshot=complete_snapshot.id, | snapshot=complete_snapshot.id, | ||||
| ) | ) | ||||
| ] | ] | ||||
| ) | ) | ||||
| # so we do not find it | # so we do not find it | ||||
| assert snapshot_get_latest(swh_storage, origin.url) is None | assert snapshot_get_latest(swh_storage, origin.url) is None | ||||
| assert snapshot_get_latest(swh_storage, origin.url, branches_count=1) is None | assert snapshot_get_latest(swh_storage, origin.url, branches_count=1) is None | ||||
| def test_snapshot_get_latest(swh_storage, sample_data_model): | def test_snapshot_get_latest(swh_storage, sample_data): | ||||
| origin = sample_data_model["origin"][0] | origin = sample_data["origin"][0] | ||||
| swh_storage.origin_add([origin]) | swh_storage.origin_add([origin]) | ||||
| visit1, visit2 = sample_data_model["origin_visit"][:2] | visit1, visit2 = sample_data["origin_visit"][:2] | ||||
| assert visit1.origin == origin.url | assert visit1.origin == origin.url | ||||
| swh_storage.origin_visit_add([visit1]) | swh_storage.origin_visit_add([visit1]) | ||||
| ov1 = swh_storage.origin_visit_get_latest(origin.url) | ov1 = swh_storage.origin_visit_get_latest(origin.url) | ||||
| visit_id = ov1["visit"] | visit_id = ov1["visit"] | ||||
| # Add snapshot to visit1, latest snapshot = visit 1 snapshot | # Add snapshot to visit1, latest snapshot = visit 1 snapshot | ||||
| complete_snapshot = sample_data_model["snapshot"][2] | complete_snapshot = sample_data["snapshot"][2] | ||||
| swh_storage.snapshot_add([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=visit_id, | visit=visit_id, | ||||
| date=visit2.date, | date=visit2.date, | ||||
| ▲ Show 20 Lines • Show All 42 Lines • Show Last 20 Lines | |||||