Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 1,875 Lines • ▼ Show 20 Lines | def test_snapshot_add_validation(self, swh_storage): | ||||
| snap = copy.deepcopy(data.snapshot) | snap = copy.deepcopy(data.snapshot) | ||||
| snap['branches'][b'foo'] = {'target': b'\x42'*20} | snap['branches'][b'foo'] = {'target': b'\x42'*20} | ||||
| with pytest.raises(KeyError, match='target_type'): | with pytest.raises(KeyError, match='target_type'): | ||||
| swh_storage.snapshot_add([snap]) | swh_storage.snapshot_add([snap]) | ||||
| def test_snapshot_add_count_branches(self, swh_storage): | def test_snapshot_add_count_branches(self, swh_storage): | ||||
| origin_id = swh_storage.origin_add_one(data.origin) | |||||
| origin_visit1 = swh_storage.origin_visit_add( | |||||
| origin_id, data.date_visit1) | |||||
| visit_id = origin_visit1['visit'] | |||||
| actual_result = swh_storage.snapshot_add([data.complete_snapshot]) | actual_result = swh_storage.snapshot_add([data.complete_snapshot]) | ||||
| swh_storage.origin_visit_update( | |||||
| origin_id, visit_id, snapshot=data.complete_snapshot['id']) | |||||
| assert actual_result == {'snapshot:add': 1} | assert actual_result == {'snapshot:add': 1} | ||||
| snp_id = data.complete_snapshot['id'] | snp_id = data.complete_snapshot['id'] | ||||
| snp_size = swh_storage.snapshot_count_branches(snp_id) | snp_size = swh_storage.snapshot_count_branches(snp_id) | ||||
| expected_snp_size = { | expected_snp_size = { | ||||
| 'alias': 1, | 'alias': 1, | ||||
| 'content': 1, | 'content': 1, | ||||
| 'directory': 2, | 'directory': 2, | ||||
| 'release': 1, | 'release': 1, | ||||
| 'revision': 1, | 'revision': 1, | ||||
| 'snapshot': 1, | 'snapshot': 1, | ||||
| None: 1 | None: 1 | ||||
| } | } | ||||
| assert snp_size == expected_snp_size | assert snp_size == expected_snp_size | ||||
| def test_snapshot_add_get_paginated(self, swh_storage): | def test_snapshot_add_get_paginated(self, swh_storage): | ||||
| origin_id = swh_storage.origin_add_one(data.origin) | |||||
| origin_visit1 = swh_storage.origin_visit_add( | |||||
| origin_id, data.date_visit1) | |||||
| visit_id = origin_visit1['visit'] | |||||
| swh_storage.snapshot_add([data.complete_snapshot]) | swh_storage.snapshot_add([data.complete_snapshot]) | ||||
| swh_storage.origin_visit_update( | |||||
| origin_id, visit_id, | |||||
| snapshot=data.complete_snapshot['id']) | |||||
| snp_id = data.complete_snapshot['id'] | snp_id = data.complete_snapshot['id'] | ||||
| branches = data.complete_snapshot['branches'] | branches = data.complete_snapshot['branches'] | ||||
| branch_names = list(sorted(branches)) | branch_names = list(sorted(branches)) | ||||
| # Test branch_from | # Test branch_from | ||||
| snapshot = swh_storage.snapshot_get_branches( | snapshot = swh_storage.snapshot_get_branches( | ||||
| snp_id, branches_from=b'release') | snp_id, branches_from=b'release') | ||||
| ▲ Show 20 Lines • Show All 79 Lines • ▼ Show 20 Lines | def test_snapshot_add_get_filtered(self, swh_storage): | ||||
| if tgt and tgt['target_type'] == 'alias' | if tgt and tgt['target_type'] == 'alias' | ||||
| }, | }, | ||||
| 'next_branch': None, | 'next_branch': None, | ||||
| } | } | ||||
| assert snapshot == expected_snapshot | assert snapshot == expected_snapshot | ||||
| def test_snapshot_add_get_filtered_and_paginated(self, swh_storage): | def test_snapshot_add_get_filtered_and_paginated(self, swh_storage): | ||||
| origin_id = swh_storage.origin_add_one(data.origin) | |||||
| origin_visit1 = swh_storage.origin_visit_add( | |||||
| origin_id, data.date_visit1) | |||||
| visit_id = origin_visit1['visit'] | |||||
| swh_storage.snapshot_add([data.complete_snapshot]) | swh_storage.snapshot_add([data.complete_snapshot]) | ||||
| swh_storage.origin_visit_update( | |||||
| origin_id, visit_id, snapshot=data.complete_snapshot['id']) | |||||
| snp_id = data.complete_snapshot['id'] | snp_id = data.complete_snapshot['id'] | ||||
| branches = data.complete_snapshot['branches'] | branches = data.complete_snapshot['branches'] | ||||
| branch_names = list(sorted(branches)) | branch_names = list(sorted(branches)) | ||||
| # Test branch_from | # Test branch_from | ||||
| snapshot = swh_storage.snapshot_get_branches( | snapshot = swh_storage.snapshot_get_branches( | ||||
| ▲ Show 20 Lines • Show All 1,441 Lines • Show Last 20 Lines | |||||