Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/algos/test_snapshot.py
| Show First 20 Lines • Show All 47 Lines • ▼ Show 20 Lines | def test_snapshot_large(swh_storage, branch_name, branch_target): # noqa | ||||
| assert snapshot == returned_snapshot | assert snapshot == returned_snapshot | ||||
| def test_snapshot_get_latest_none(swh_storage): | def test_snapshot_get_latest_none(swh_storage): | ||||
| """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 | |||||
| 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 then nothing is found | # no snapshot on origin visit so None | ||||
| origin = Origin.from_dict(data.origin) | origin = Origin.from_dict(data.origin) | ||||
| swh_storage.origin_add_one(origin) | swh_storage.origin_add_one(origin) | ||||
| swh_storage.origin_visit_add( | swh_storage.origin_visit_add( | ||||
| [ | [ | ||||
| OriginVisit( | 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, | ||||
| ) | ) | ||||
| ] | ] | ||||
| ) | ) | ||||
| 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) | |||||
| assert ov1 is not None | |||||
| visit_id = ov1["visit"] | |||||
| # visit references a snapshot but the snapshot does not exist in backend for some | |||||
| # reason | |||||
| complete_snapshot = Snapshot.from_dict(data.complete_snapshot) | |||||
| swh_storage.origin_visit_status_add( | |||||
| [ | |||||
| OriginVisitStatus( | |||||
| origin=origin.url, | |||||
| visit=visit_id, | |||||
| date=data.date_visit2, | |||||
| status="partial", | |||||
| snapshot=complete_snapshot.id, | |||||
| ) | |||||
| ] | |||||
| ) | |||||
| # so we do not find it | |||||
| assert snapshot_get_latest(swh_storage, origin.url) is None | |||||
| assert snapshot_get_latest(swh_storage, origin.url, branches_count=1) is None | |||||
| def test_snapshot_get_latest(swh_storage): | def test_snapshot_get_latest(swh_storage): | ||||
| origin = Origin.from_dict(data.origin) | origin = Origin.from_dict(data.origin) | ||||
| swh_storage.origin_add_one(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, | ||||
| ▲ Show 20 Lines • Show All 70 Lines • Show Last 20 Lines | |||||