Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/algos/test_snapshot.py
| Show All 20 Lines | |||||
| @pytest.mark.db | @pytest.mark.db | ||||
| @pytest.mark.property_based | @pytest.mark.property_based | ||||
| class TestSnapshotAllBranches(StorageTestFixture, unittest.TestCase): | class TestSnapshotAllBranches(StorageTestFixture, unittest.TestCase): | ||||
| @given(origins().map(lambda x: x.to_dict()), | @given(origins().map(lambda x: x.to_dict()), | ||||
| datetimes(), | datetimes(), | ||||
| snapshots(min_size=0, max_size=10, only_objects=False)) | snapshots(min_size=0, max_size=10, only_objects=False)) | ||||
| def test_snapshot_small(self, origin, ts, snapshot): | def test_snapshot_small(self, origin, ts, snapshot): | ||||
| snapshot = snapshot.to_dict() | snapshot = snapshot.to_dict() | ||||
| origin_id = self.storage.origin_add_one(origin) | self.storage.snapshot_add([snapshot]) | ||||
| visit = self.storage.origin_visit_add(origin_id, ts) | |||||
| self.storage.snapshot_add(origin_id, visit['visit'], snapshot) | |||||
| returned_snapshot = snapshot_get_all_branches(self.storage, | returned_snapshot = snapshot_get_all_branches(self.storage, | ||||
| snapshot['id']) | snapshot['id']) | ||||
| self.assertEqual(snapshot, returned_snapshot) | self.assertEqual(snapshot, returned_snapshot) | ||||
| @given(origins().map(lambda x: x.to_dict()), | @given(origins().map(lambda x: x.to_dict()), | ||||
| datetimes(), | datetimes(), | ||||
| branch_names(), branch_targets(only_objects=True)) | branch_names(), branch_targets(only_objects=True)) | ||||
| def test_snapshot_large(self, origin, ts, branch_name, branch_target): | def test_snapshot_large(self, origin, ts, branch_name, branch_target): | ||||
| branch_target = branch_target.to_dict() | branch_target = branch_target.to_dict() | ||||
| origin_id = self.storage.origin_add_one(origin) | |||||
| visit = self.storage.origin_visit_add(origin_id, ts) | |||||
| snapshot = { | snapshot = { | ||||
| 'branches': { | 'branches': { | ||||
| b'%s%05d' % (branch_name, i): branch_target | b'%s%05d' % (branch_name, i): branch_target | ||||
| for i in range(10000) | for i in range(10000) | ||||
| } | } | ||||
| } | } | ||||
| snapshot['id'] = identifier_to_bytes(snapshot_identifier(snapshot)) | snapshot['id'] = identifier_to_bytes(snapshot_identifier(snapshot)) | ||||
| self.storage.snapshot_add(origin_id, visit['visit'], snapshot) | self.storage.snapshot_add([snapshot]) | ||||
| returned_snapshot = snapshot_get_all_branches(self.storage, | returned_snapshot = snapshot_get_all_branches(self.storage, | ||||
| snapshot['id']) | snapshot['id']) | ||||
| self.assertEqual(snapshot, returned_snapshot) | self.assertEqual(snapshot, returned_snapshot) | ||||