Skip to content
Snippets Groups Projects
Commit 6b1e2a0b authored by Jayesh's avatar Jayesh 🐈
Browse files

Add a function to resolve any snapshot branch to its final target

Returns the final branch target along with the resolve chain.
Resolve chain will be an empty list for every branch other than
an alias type. Supports breaking the chain if it gets too long.
parent 7246a37e
No related branches found
Tags v1.7.3
No related merge requests found
......@@ -168,6 +168,49 @@ def visits_and_snapshots_get_from_revision(
yield (visit, visit_status, snapshot)
def snapshot_resolve_branch_target(
storage: StorageInterface,
snapshot_id: Sha1Git,
branch_obj: Optional[SnapshotBranch],
max_length: int = 100,
) -> Tuple[Optional[SnapshotBranch], List[bytes]]:
"""
Return the final target of a snapshot branch along with its
resolve chain. Return 'None' as the target in case the chain is
longer than the given max_length or has cycles
Args:
storage: Storage instance
snapshot_id: snapshot identifier
branch_obj: snapshot branch object to resolve
max_length: maximum chain length before breaking
Returns:
The first branch that isn't an alias and the resolve chain
"""
resolve_chain: List[bytes] = []
while branch_obj and branch_obj.target_type == TargetType.ALIAS:
target_branch = branch_obj.target # target branch name
resolve_chain.append(target_branch)
# query for the new branch
branches = storage.snapshot_get_branches(
snapshot_id, branches_from=target_branch, branches_count=1
)
if (
not branches
# The target branch is missing
or branches["branches"].get(target_branch) is None
# Circular reference
or resolve_chain.count(target_branch) > 1
# Too many re-directs
or len(resolve_chain) >= max_length
):
# chain broke without a target, this will still return the resolve_chain list
branch_obj = None
break
branch_obj = branches["branches"][target_branch]
return branch_obj, resolve_chain
def snapshot_resolve_alias(
storage: StorageInterface, snapshot_id: Sha1Git, alias_name: bytes
) -> Optional[SnapshotBranch]:
......@@ -196,18 +239,7 @@ def snapshot_resolve_alias(
return None
last_branch = snapshot["branches"][alias_name]
seen_aliases = {alias_name}
while last_branch is not None and last_branch.target_type == TargetType.ALIAS:
if last_branch.target in seen_aliases:
return None
alias_target = last_branch.target
snapshot = storage.snapshot_get_branches(
snapshot_id, branches_from=alias_target, branches_count=1
)
assert snapshot is not None
last_branch = snapshot["branches"].get(alias_target)
seen_aliases.add(alias_target)
return last_branch
target_branch, _ = snapshot_resolve_branch_target(
storage, snapshot_id=snapshot_id, branch_obj=last_branch
)
return target_branch
......@@ -19,6 +19,7 @@ from swh.storage.algos.snapshot import (
snapshot_get_latest,
snapshot_id_get_from_revision,
snapshot_resolve_alias,
snapshot_resolve_branch_target,
visits_and_snapshots_get_from_revision,
)
from swh.storage.tests.conftest import function_scoped_fixture_check
......@@ -431,3 +432,100 @@ def test_snapshot_resolve_alias_cycle_found(swh_storage):
swh_storage.snapshot_add([snapshot])
assert snapshot_resolve_alias(swh_storage, snapshot.id, alias1_name) is None
def test_snapshot_resolve_snapshot_branch_chain(swh_storage, sample_data):
alias1_name = b"alias_1"
alias2_name = b"alias_2"
alias3_name = b"alias_3"
alias4_name = b"alias_4"
alias1_branch_info = SnapshotBranch(
target=alias2_name, target_type=TargetType.ALIAS
)
alias2_branch_info = SnapshotBranch(
target=alias3_name, target_type=TargetType.ALIAS
)
alias3_branch_info = SnapshotBranch(
target=alias4_name, target_type=TargetType.ALIAS
)
alias4_branch_info = SnapshotBranch(
target=sample_data.revisions[0].id, target_type=TargetType.REVISION
)
snapshot = Snapshot(
branches={
alias1_name: alias1_branch_info,
alias2_name: alias2_branch_info,
alias3_name: alias3_branch_info,
alias4_name: alias4_branch_info,
}
)
swh_storage.snapshot_add([snapshot])
target, chain = snapshot_resolve_branch_target(
swh_storage, snapshot_id=snapshot.id, branch_obj=alias1_branch_info
)
assert target == alias4_branch_info
assert chain == [b"alias_2", b"alias_3", b"alias_4"]
def test_snapshot_resolve_snapshot_branch_too_long_chain(swh_storage, sample_data):
alias1_name = b"alias_1"
alias2_name = b"alias_2"
alias3_name = b"alias_3"
alias4_name = b"alias_4"
alias1_branch_info = SnapshotBranch(
target=alias2_name, target_type=TargetType.ALIAS
)
alias2_branch_info = SnapshotBranch(
target=alias3_name, target_type=TargetType.ALIAS
)
alias3_branch_info = SnapshotBranch(
target=alias4_name, target_type=TargetType.ALIAS
)
alias4_branch_info = SnapshotBranch(
target=sample_data.revisions[0].id, target_type=TargetType.REVISION
)
snapshot = Snapshot(
branches={
alias1_name: alias1_branch_info,
alias2_name: alias2_branch_info,
alias3_name: alias3_branch_info,
alias4_name: alias4_branch_info,
}
)
swh_storage.snapshot_add([snapshot])
target, chain = snapshot_resolve_branch_target(
swh_storage,
snapshot_id=snapshot.id,
branch_obj=alias1_branch_info,
max_length=2,
)
assert target is None
assert chain == [b"alias_2", b"alias_3"]
def test_snapshot_resolve_snapshot_branch_cyclic_chain(swh_storage):
alias1_name = b"alias_1"
alias2_name = b"alias_2"
alias1_branch_info = SnapshotBranch(
target=alias2_name, target_type=TargetType.ALIAS
)
alias2_branch_info = SnapshotBranch(
target=alias1_name, target_type=TargetType.ALIAS
)
snapshot = Snapshot(
branches={
alias1_name: alias1_branch_info,
alias2_name: alias2_branch_info,
}
)
swh_storage.snapshot_add([snapshot])
target, chain = snapshot_resolve_branch_target(
swh_storage,
snapshot_id=snapshot.id,
branch_obj=alias1_branch_info,
)
assert target is None
assert chain == [b"alias_2", b"alias_1", b"alias_2"]
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment