Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 2,894 Lines • ▼ Show 20 Lines | def test_snapshot_get_latest__missing_snapshot(self): | ||||
| visit1_id = origin_visit1['visit'] | visit1_id = origin_visit1['visit'] | ||||
| origin_visit2 = self.storage.origin_visit_add(origin_id, | origin_visit2 = self.storage.origin_visit_add(origin_id, | ||||
| self.date_visit2) | self.date_visit2) | ||||
| visit2_id = origin_visit2['visit'] | visit2_id = origin_visit2['visit'] | ||||
| # Two visits, both with no snapshot: latest snapshot is None | # Two visits, both with no snapshot: latest snapshot is None | ||||
| self.assertIsNone(self.storage.snapshot_get_latest(origin_id)) | self.assertIsNone(self.storage.snapshot_get_latest(origin_id)) | ||||
| # Add unknown snapshot to visit1, latest snapshot = None | # Add unknown snapshot to visit1, check that the inconsistency is | ||||
| # detected | |||||
| self.storage.origin_visit_update( | self.storage.origin_visit_update( | ||||
| origin_id, visit1_id, snapshot=self.complete_snapshot['id']) | origin_id, visit1_id, snapshot=self.complete_snapshot['id']) | ||||
| self.assertIsNone(self.storage.snapshot_get_latest(origin_id)) | with self.assertRaises(ValueError): | ||||
| self.storage.snapshot_get_latest(origin_id) | |||||
| # Status filter: both visits are status=ongoing, so no snapshot | # Status filter: both visits are status=ongoing, so no snapshot | ||||
| # returned | # returned | ||||
| self.assertIsNone( | self.assertIsNone( | ||||
| self.storage.snapshot_get_latest(origin_id, | self.storage.snapshot_get_latest(origin_id, | ||||
| allowed_statuses=['full']) | allowed_statuses=['full']) | ||||
| ) | ) | ||||
| # Mark the first visit as completed and check status filter again | # Mark the first visit as completed and check status filter again | ||||
| self.storage.origin_visit_update(origin_id, visit1_id, status='full') | self.storage.origin_visit_update(origin_id, visit1_id, status='full') | ||||
| self.assertIsNone( | with self.assertRaises(ValueError): | ||||
| self.storage.snapshot_get_latest(origin_id, | self.storage.snapshot_get_latest(origin_id, | ||||
| allowed_statuses=['full']), | allowed_statuses=['full']), | ||||
| ) | |||||
| # Actually add the snapshot and check status filter again | # Actually add the snapshot and check status filter again | ||||
| self.storage.snapshot_add([self.complete_snapshot]) | self.storage.snapshot_add([self.complete_snapshot]) | ||||
| self.assertEqual( | self.assertEqual( | ||||
| self.complete_snapshot, | self.complete_snapshot, | ||||
| self.storage.snapshot_get_latest(origin_id) | self.storage.snapshot_get_latest(origin_id) | ||||
| ) | ) | ||||
| # Add unknown snapshot to visit2 and check that the old snapshot | # Add unknown snapshot to visit2 and check that the inconsistency | ||||
| # is still returned | # is detected | ||||
| self.storage.origin_visit_update( | self.storage.origin_visit_update( | ||||
| origin_id, visit2_id, snapshot=self.empty_snapshot['id']) | origin_id, visit2_id, snapshot=self.snapshot['id']) | ||||
| self.assertEqual( | with self.assertRaises(ValueError): | ||||
| self.complete_snapshot, | self.storage.snapshot_get_latest(origin_id) | ||||
| self.storage.snapshot_get_latest(origin_id)) | |||||
| # Actually add that snapshot and check that the new one is returned | # Actually add that snapshot and check that the new one is returned | ||||
| self.storage.snapshot_add([self.empty_snapshot]) | self.storage.snapshot_add([self.snapshot]) | ||||
| self.assertEqual( | self.assertEqual( | ||||
| self.empty_snapshot, | self.snapshot, | ||||
| self.storage.snapshot_get_latest(origin_id) | self.storage.snapshot_get_latest(origin_id) | ||||
| ) | ) | ||||
| def test_snapshot_get_latest__legacy_add(self): | def test_snapshot_get_latest__legacy_add(self): | ||||
| origin_id = self.storage.origin_add_one(self.origin) | origin_id = self.storage.origin_add_one(self.origin) | ||||
| origin_visit1 = self.storage.origin_visit_add(origin_id, | origin_visit1 = self.storage.origin_visit_add(origin_id, | ||||
| self.date_visit1) | self.date_visit1) | ||||
| visit1_id = origin_visit1['visit'] | visit1_id = origin_visit1['visit'] | ||||
| ▲ Show 20 Lines • Show All 1,035 Lines • Show Last 20 Lines | |||||