diff --git a/swh/storage/algos/snapshot.py b/swh/storage/algos/snapshot.py --- a/swh/storage/algos/snapshot.py +++ b/swh/storage/algos/snapshot.py @@ -3,7 +3,7 @@ # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information -from typing import Iterator, List, Optional, Tuple, cast +from typing import Iterator, List, Optional, Tuple from swh.model.hashutil import hash_to_hex from swh.model.model import ( @@ -196,25 +196,23 @@ if alias_name not in snapshot["branches"]: return ([], None) - branch_info = snapshot["branches"][alias_name] - branches = [branch_info] + last_branch = snapshot["branches"][alias_name] + branches = [] seen_aliases = {alias_name} while ( - branch_info is not None - and branch_info.target_type == TargetType.ALIAS - and branch_info.target not in seen_aliases + last_branch is not None + and last_branch.target_type == TargetType.ALIAS + and last_branch.target not in seen_aliases ): - alias_target = branch_info.target + branches.append(last_branch) + alias_target = last_branch.target snapshot = storage.snapshot_get_branches( snapshot_id, branches_from=alias_target, branches_count=1 ) assert snapshot is not None - if alias_target not in snapshot["branches"]: - break + last_branch = snapshot["branches"].get(alias_target) seen_aliases.add(alias_target) - branch_info = snapshot["branches"][alias_target] - branches.append(branch_info) - return (cast(List[SnapshotBranch], branches[:-1]), branches[-1]) + return (branches, last_branch) diff --git a/swh/storage/tests/algos/test_snapshot.py b/swh/storage/tests/algos/test_snapshot.py --- a/swh/storage/tests/algos/test_snapshot.py +++ b/swh/storage/tests/algos/test_snapshot.py @@ -348,17 +348,32 @@ dangling_branch_name = b"dangling_branch" alias_name = b"rev_alias" - alias_branch_info = SnapshotBranch( + alias_branch = SnapshotBranch( target=dangling_branch_name, target_type=TargetType.ALIAS ) snapshot = Snapshot( - branches={dangling_branch_name: None, alias_name: alias_branch_info,} + branches={dangling_branch_name: None, alias_name: alias_branch,} ) swh_storage.snapshot_add([snapshot]) branches = snapshot_resolve_alias(swh_storage, snapshot.id, alias_name) - assert branches == ([alias_branch_info], None) + assert branches == ([alias_branch], None) + + +def test_snapshot_resolve_alias_missing_branch(swh_storage): + missing_branch_name = b"missing_branch" + alias_name = b"rev_alias" + + alias_branch = SnapshotBranch( + target=missing_branch_name, target_type=TargetType.ALIAS + ) + + snapshot = Snapshot(id=b"42" * 10, branches={alias_name: alias_branch,}) + swh_storage.snapshot_add([snapshot]) + + branches = snapshot_resolve_alias(swh_storage, snapshot.id, alias_name) + assert branches == ([alias_branch], None) def test_snapshot_resolve_alias_cycle_found(swh_storage):