Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 1,676 Lines • ▼ Show 20 Lines | def test_origin_visit_add(self, swh_storage): | ||||
| {**ovs2.to_dict(), "type": ov2.type}, | {**ovs2.to_dict(), "type": ov2.type}, | ||||
| ] | ] | ||||
| assert len(expected_visits) == len(actual_origin_visits) | assert len(expected_visits) == len(actual_origin_visits) | ||||
| for visit in expected_visits: | for visit in expected_visits: | ||||
| assert visit in actual_origin_visits | assert visit in actual_origin_visits | ||||
| actual_objects = set(swh_storage.journal_writer.journal.objects) | actual_objects = list(swh_storage.journal_writer.journal.objects) | ||||
| # we write to the journal as many times as we call the endpoint | expected_objects = list( | ||||
| assert actual_objects == set( | |||||
| [("origin", origin1)] | [("origin", origin1)] | ||||
| + [("origin_visit", visit) for visit in [ov1, ov2]] * 2 | + [("origin_visit", visit) for visit in [ov1, ov2]] * 2 | ||||
| + [("origin_visit_status", ovs) for ovs in [ovs1, ovs2]] | + [("origin_visit_status", ovs) for ovs in [ovs1, ovs2]] | ||||
| ) | ) | ||||
| for obj in expected_objects: | |||||
| assert obj in actual_objects | |||||
| def test_origin_visit_add_validation(self, swh_storage): | def test_origin_visit_add_validation(self, swh_storage): | ||||
| """Unknown origin when adding visits should raise""" | """Unknown origin when adding visits should raise""" | ||||
| visit = OriginVisit( | visit = OriginVisit( | ||||
| origin="something-unknown", date=now(), type=data.type_visit1, | origin="something-unknown", date=now(), type=data.type_visit1, | ||||
| ) | ) | ||||
| with pytest.raises(StorageArgumentException, match="Unknown origin"): | with pytest.raises(StorageArgumentException, match="Unknown origin"): | ||||
| swh_storage.origin_visit_add([visit]) | swh_storage.origin_visit_add([visit]) | ||||
| ▲ Show 20 Lines • Show All 1,115 Lines • ▼ Show 20 Lines | def test_snapshot_add_twice__by_origin_visit(self, swh_storage): | ||||
| ("origin_visit", ov1), | ("origin_visit", ov1), | ||||
| ("origin_visit_status", ovs1), | ("origin_visit_status", ovs1), | ||||
| ("snapshot", Snapshot.from_dict(data.snapshot)), | ("snapshot", Snapshot.from_dict(data.snapshot)), | ||||
| ("origin_visit_status", ovs2), | ("origin_visit_status", ovs2), | ||||
| ("origin_visit", ov2), | ("origin_visit", ov2), | ||||
| ("origin_visit_status", ovs3), | ("origin_visit_status", ovs3), | ||||
| ("origin_visit_status", ovs4), | ("origin_visit_status", ovs4), | ||||
| ] | ] | ||||
| for obj in expected_objects: | for obj in expected_objects: | ||||
| assert obj in actual_objects | assert obj in actual_objects | ||||
| def test_snapshot_get_random(self, swh_storage): | def test_snapshot_get_random(self, swh_storage): | ||||
| swh_storage.snapshot_add( | swh_storage.snapshot_add( | ||||
| [data.snapshot, data.empty_snapshot, data.complete_snapshot] | [data.snapshot, data.empty_snapshot, data.complete_snapshot] | ||||
| ) | ) | ||||
| ▲ Show 20 Lines • Show All 1,081 Lines • Show Last 20 Lines | |||||