feat: add own-branch lock adoption recovery to gitea_lock_issue (Closes #442) #461
+44
-4
@@ -14,6 +14,8 @@ this module additionally records whether they passed for proof purposes.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
ADOPT = "adopt_existing_branch"
|
||||
BLOCK_COMPETING = "block_competing_branch"
|
||||
NO_MATCH = "no_matching_branch"
|
||||
@@ -33,25 +35,58 @@ def _branch_sha(entry) -> str | None:
|
||||
return None
|
||||
|
||||
|
||||
def _branch_carries_issue_marker(branch_name: str, issue_number: int) -> bool:
|
||||
"""Return True when *branch_name* references issue *issue_number* exactly.
|
||||
|
||||
Uses a numeric word-boundary so ``issue-42`` does not match inside
|
||||
``issue-420`` (AC6 / #440).
|
||||
"""
|
||||
name = (branch_name or "").strip()
|
||||
if not name:
|
||||
return False
|
||||
pattern = rf"(?:^|/)issue-{int(issue_number)}(?![0-9])"
|
||||
return re.search(pattern, name) is not None
|
||||
|
||||
|
||||
def assess_own_branch_adoption(
|
||||
*,
|
||||
issue_number: int,
|
||||
requested_branch: str,
|
||||
existing_branches,
|
||||
) -> dict:
|
||||
"""Decide whether an existing matching branch is adoptable."""
|
||||
marker = f"issue-{issue_number}"
|
||||
"""Decide whether an existing matching branch is adoptable.
|
||||
|
||||
Args:
|
||||
issue_number: The tracking issue number being locked.
|
||||
requested_branch: The exact branch the caller wants to lock.
|
||||
existing_branches: Iterable of remote branch entries — either names or
|
||||
dicts with ``name`` and optional ``commit_sha``.
|
||||
|
||||
Returns:
|
||||
dict with:
|
||||
* ``outcome`` — one of ADOPT / BLOCK_COMPETING / NO_MATCH
|
||||
* ``adopt`` (bool), ``block`` (bool)
|
||||
* ``reason`` (str)
|
||||
* ``matched_branch`` (str | None), ``matched_head_sha`` (str | None)
|
||||
* ``competing_branches`` (list[str])
|
||||
|
||||
ADOPT: the issue's exact branch exists and no other same-issue branch does.
|
||||
BLOCK_COMPETING: at least one same-issue branch is not the requested branch.
|
||||
NO_MATCH: no branch carries the issue marker — normal lock path applies.
|
||||
"""
|
||||
requested = (requested_branch or "").strip()
|
||||
|
||||
matches: list[tuple[str, str | None]] = []
|
||||
for entry in existing_branches or []:
|
||||
name = _branch_name(entry).strip()
|
||||
if marker in name:
|
||||
if _branch_carries_issue_marker(name, issue_number):
|
||||
matches.append((name, _branch_sha(entry)))
|
||||
|
||||
competing = sorted({name for name, _ in matches if name != requested})
|
||||
exact = [(name, sha) for name, sha in matches if name == requested]
|
||||
|
||||
# Fail closed whenever any non-requested same-issue branch exists, even if
|
||||
# the requested branch is also present: ownership is then ambiguous.
|
||||
if competing:
|
||||
return {
|
||||
"outcome": BLOCK_COMPETING,
|
||||
@@ -103,7 +138,12 @@ def build_adoption_proof(
|
||||
lock_file_path: str,
|
||||
lock_file_status: str,
|
||||
) -> dict:
|
||||
"""Assemble the proof block returned by ``gitea_lock_issue`` on adoption."""
|
||||
"""Assemble the proof block returned by ``gitea_lock_issue`` on adoption.
|
||||
|
||||
Requirement #4: adoption results must carry issue number, branch name,
|
||||
branch head commit, adoption reason, no-existing-PR proof, no-competing-
|
||||
live-lock proof, and lock file path/status.
|
||||
"""
|
||||
return {
|
||||
"issue_number": issue_number,
|
||||
"branch_name": branch_name,
|
||||
|
||||
@@ -25,6 +25,16 @@ class TestAssessOwnBranchAdoption(unittest.TestCase):
|
||||
)
|
||||
self.assertEqual(result["outcome"], ADOPT)
|
||||
self.assertTrue(result["adopt"])
|
||||
self.assertFalse(result["block"])
|
||||
self.assertEqual(result["matched_branch"], REQ)
|
||||
self.assertEqual(result["matched_head_sha"], "934688a")
|
||||
|
||||
def test_exact_own_branch_adopted_when_sha_missing(self):
|
||||
result = assess_own_branch_adoption(
|
||||
issue_number=420, requested_branch=REQ, existing_branches=[REQ]
|
||||
)
|
||||
self.assertEqual(result["outcome"], ADOPT)
|
||||
self.assertIsNone(result["matched_head_sha"])
|
||||
|
||||
def test_different_branch_same_issue_blocks(self):
|
||||
result = assess_own_branch_adoption(
|
||||
@@ -34,6 +44,20 @@ class TestAssessOwnBranchAdoption(unittest.TestCase):
|
||||
)
|
||||
self.assertEqual(result["outcome"], BLOCK_COMPETING)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertFalse(result["adopt"])
|
||||
self.assertIn("feat/issue-420-other-work", result["competing_branches"])
|
||||
self.assertIn("fail closed", result["reason"])
|
||||
|
||||
def test_own_branch_plus_competing_branch_blocks(self):
|
||||
# Ambiguous ownership: fail closed even though the exact branch exists.
|
||||
result = assess_own_branch_adoption(
|
||||
issue_number=420,
|
||||
requested_branch=REQ,
|
||||
existing_branches=[{"name": REQ}, {"name": "feat/issue-420-rogue"}],
|
||||
)
|
||||
self.assertEqual(result["outcome"], BLOCK_COMPETING)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertEqual(result["competing_branches"], ["feat/issue-420-rogue"])
|
||||
|
||||
def test_no_matching_branch_is_normal_path(self):
|
||||
result = assess_own_branch_adoption(
|
||||
@@ -42,10 +66,44 @@ class TestAssessOwnBranchAdoption(unittest.TestCase):
|
||||
existing_branches=[{"name": "feat/issue-999-unrelated"}],
|
||||
)
|
||||
self.assertEqual(result["outcome"], NO_MATCH)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertFalse(result["adopt"])
|
||||
|
||||
def test_empty_branch_list_is_normal_path(self):
|
||||
result = assess_own_branch_adoption(
|
||||
issue_number=420, requested_branch=REQ, existing_branches=[]
|
||||
)
|
||||
self.assertEqual(result["outcome"], NO_MATCH)
|
||||
|
||||
def test_higher_issue_number_branch_does_not_block_lower_issue_adoption(self):
|
||||
# issue-420 must not be treated as competing work for issue #42.
|
||||
own_branch = "feat/issue-42-widget"
|
||||
result = assess_own_branch_adoption(
|
||||
issue_number=42,
|
||||
requested_branch=own_branch,
|
||||
existing_branches=[
|
||||
{"name": own_branch, "commit_sha": "abc1234"},
|
||||
{"name": "feat/issue-420-server-code-parity"},
|
||||
],
|
||||
)
|
||||
self.assertEqual(result["outcome"], ADOPT)
|
||||
self.assertTrue(result["adopt"])
|
||||
self.assertFalse(result["block"])
|
||||
self.assertEqual(result["matched_branch"], own_branch)
|
||||
|
||||
def test_unrelated_higher_number_branch_is_ignored_without_own_branch(self):
|
||||
result = assess_own_branch_adoption(
|
||||
issue_number=42,
|
||||
requested_branch="feat/issue-42-thing",
|
||||
existing_branches=[{"name": "feat/issue-420-server-code-parity"}],
|
||||
)
|
||||
self.assertEqual(result["outcome"], NO_MATCH)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertFalse(result["adopt"])
|
||||
|
||||
|
||||
class TestBuildAdoptionProof(unittest.TestCase):
|
||||
def test_proof_has_required_fields(self):
|
||||
def test_proof_has_all_required_fields(self):
|
||||
assessment = assess_own_branch_adoption(
|
||||
issue_number=420,
|
||||
requested_branch=REQ,
|
||||
@@ -60,8 +118,20 @@ class TestBuildAdoptionProof(unittest.TestCase):
|
||||
lock_file_path="/tmp/example-lock.json",
|
||||
lock_file_status="written",
|
||||
)
|
||||
for key in (
|
||||
"issue_number",
|
||||
"branch_name",
|
||||
"branch_head_commit",
|
||||
"adoption_reason",
|
||||
"no_existing_pr_proof",
|
||||
"no_competing_live_lock_proof",
|
||||
"lock_file_path",
|
||||
"lock_file_status",
|
||||
):
|
||||
self.assertIn(key, proof)
|
||||
self.assertEqual(proof["branch_head_commit"], "934688a")
|
||||
self.assertTrue(proof["no_existing_pr_proof"])
|
||||
self.assertTrue(proof["no_competing_live_lock_proof"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user