Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/algos/test_snapshot.py
| Show First 20 Lines • Show All 273 Lines • ▼ Show 20 Lines | |||||
| def test_snapshot_resolve_aliases_unknown_snapshot(swh_storage): | def test_snapshot_resolve_aliases_unknown_snapshot(swh_storage): | ||||
| assert snapshot_resolve_alias(swh_storage, b"foo", b"HEAD") is None | assert snapshot_resolve_alias(swh_storage, b"foo", b"HEAD") is None | ||||
| def test_snapshot_resolve_aliases_no_aliases(swh_storage): | def test_snapshot_resolve_aliases_no_aliases(swh_storage): | ||||
| snapshot = Snapshot(branches={}) | snapshot = Snapshot(branches={}) | ||||
| swh_storage.snapshot_add([snapshot]) | swh_storage.snapshot_add([snapshot]) | ||||
| assert snapshot_resolve_alias(swh_storage, snapshot.id, b"HEAD") == ([], None) | assert snapshot_resolve_alias(swh_storage, snapshot.id, b"HEAD") is None | ||||
| def test_snapshot_resolve_alias(swh_storage, sample_data): | def test_snapshot_resolve_alias(swh_storage, sample_data): | ||||
| rev_branch_name = b"revision_branch" | rev_branch_name = b"revision_branch" | ||||
| rel_branch_name = b"release_branch" | rel_branch_name = b"release_branch" | ||||
| rev_alias1_name = b"rev_alias1" | rev_alias1_name = b"rev_alias1" | ||||
| rev_alias2_name = b"rev_alias2" | rev_alias2_name = b"rev_alias2" | ||||
| rev_alias3_name = b"rev_alias3" | rev_alias3_name = b"rev_alias3" | ||||
| Show All 25 Lines | snapshot = Snapshot( | ||||
| rev_alias1_name: rev_alias1_branch_info, | rev_alias1_name: rev_alias1_branch_info, | ||||
| rev_alias2_name: rev_alias2_branch_info, | rev_alias2_name: rev_alias2_branch_info, | ||||
| rev_alias3_name: rev_alias3_branch_info, | rev_alias3_name: rev_alias3_branch_info, | ||||
| rel_alias_name: rel_alias_branch_info, | rel_alias_name: rel_alias_branch_info, | ||||
| } | } | ||||
| ) | ) | ||||
| swh_storage.snapshot_add([snapshot]) | swh_storage.snapshot_add([snapshot]) | ||||
| for alias_name, expected_branches in ( | for alias_name, expected_branch in ( | ||||
| (rev_alias1_name, ([rev_alias1_branch_info], rev_branch_info)), | (rev_alias1_name, rev_branch_info), | ||||
| ( | (rev_alias2_name, rev_branch_info,), | ||||
| rev_alias2_name, | (rev_alias3_name, rev_branch_info,), | ||||
| ([rev_alias2_branch_info, rev_alias1_branch_info], rev_branch_info), | (rel_alias_name, rel_branch_info), | ||||
| ), | |||||
| ( | |||||
| rev_alias3_name, | |||||
| ( | |||||
| [ | |||||
| rev_alias3_branch_info, | |||||
| rev_alias2_branch_info, | |||||
| rev_alias1_branch_info, | |||||
| ], | |||||
| rev_branch_info, | |||||
| ), | |||||
| ), | |||||
| (rel_alias_name, ([rel_alias_branch_info], rel_branch_info)), | |||||
| ): | ): | ||||
| branches = snapshot_resolve_alias(swh_storage, snapshot.id, alias_name) | assert ( | ||||
| assert branches == expected_branches | snapshot_resolve_alias(swh_storage, snapshot.id, alias_name) | ||||
| == expected_branch | |||||
| ) | |||||
| def test_snapshot_resolve_alias_dangling_branch(swh_storage): | def test_snapshot_resolve_alias_dangling_branch(swh_storage): | ||||
| dangling_branch_name = b"dangling_branch" | dangling_branch_name = b"dangling_branch" | ||||
| alias_name = b"rev_alias" | alias_name = b"rev_alias" | ||||
| alias_branch = SnapshotBranch( | alias_branch = SnapshotBranch( | ||||
| target=dangling_branch_name, target_type=TargetType.ALIAS | target=dangling_branch_name, target_type=TargetType.ALIAS | ||||
| ) | ) | ||||
| snapshot = Snapshot( | snapshot = Snapshot( | ||||
| branches={dangling_branch_name: None, alias_name: alias_branch,} | branches={dangling_branch_name: None, alias_name: alias_branch,} | ||||
| ) | ) | ||||
| swh_storage.snapshot_add([snapshot]) | swh_storage.snapshot_add([snapshot]) | ||||
| branches = snapshot_resolve_alias(swh_storage, snapshot.id, alias_name) | assert snapshot_resolve_alias(swh_storage, snapshot.id, alias_name) is None | ||||
| assert branches == ([alias_branch], None) | |||||
| def test_snapshot_resolve_alias_missing_branch(swh_storage): | def test_snapshot_resolve_alias_missing_branch(swh_storage): | ||||
| missing_branch_name = b"missing_branch" | missing_branch_name = b"missing_branch" | ||||
| alias_name = b"rev_alias" | alias_name = b"rev_alias" | ||||
| alias_branch = SnapshotBranch( | alias_branch = SnapshotBranch( | ||||
| target=missing_branch_name, target_type=TargetType.ALIAS | target=missing_branch_name, target_type=TargetType.ALIAS | ||||
| ) | ) | ||||
| snapshot = Snapshot(id=b"42" * 10, branches={alias_name: alias_branch,}) | snapshot = Snapshot(id=b"42" * 10, branches={alias_name: alias_branch,}) | ||||
| swh_storage.snapshot_add([snapshot]) | swh_storage.snapshot_add([snapshot]) | ||||
| branches = snapshot_resolve_alias(swh_storage, snapshot.id, alias_name) | assert snapshot_resolve_alias(swh_storage, snapshot.id, alias_name) is None | ||||
| assert branches == ([alias_branch], None) | |||||
| def test_snapshot_resolve_alias_cycle_found(swh_storage): | def test_snapshot_resolve_alias_cycle_found(swh_storage): | ||||
| alias1_name = b"alias_1" | alias1_name = b"alias_1" | ||||
| alias2_name = b"alias_2" | alias2_name = b"alias_2" | ||||
| alias3_name = b"alias_3" | alias3_name = b"alias_3" | ||||
| alias4_name = b"alias_4" | alias4_name = b"alias_4" | ||||
| Show All 15 Lines | snapshot = Snapshot( | ||||
| alias1_name: alias1_branch_info, | alias1_name: alias1_branch_info, | ||||
| alias2_name: alias2_branch_info, | alias2_name: alias2_branch_info, | ||||
| alias3_name: alias3_branch_info, | alias3_name: alias3_branch_info, | ||||
| alias4_name: alias4_branch_info, | alias4_name: alias4_branch_info, | ||||
| } | } | ||||
| ) | ) | ||||
| swh_storage.snapshot_add([snapshot]) | swh_storage.snapshot_add([snapshot]) | ||||
| branches = snapshot_resolve_alias(swh_storage, snapshot.id, alias1_name) | assert snapshot_resolve_alias(swh_storage, snapshot.id, alias1_name) is None | ||||
| assert branches == ( | |||||
| [alias1_branch_info, alias2_branch_info, alias3_branch_info], | |||||
| alias4_branch_info, | |||||
| ) | |||||