fix(author): refresh durable issue-lock head after branch sync; recover merge-sync-drifted dead-session locks (#871)
`gitea_update_pr_branch_by_merge` advanced a PR's remote head but never advanced the linked durable issue lock's recorded head. After the owning session died the drifted lock became unrecoverable and no further synchronization was possible (PR #866 / issue #855). Write-side (prevents future drift): - issue_lock_store.assess/apply_durable_lock_head_refresh: on a successful sync, CAS-refresh the durable lock's recorded head from the exact expected PR head to the resulting head, re-verifying repo/issue/branch/worktree/ identity/profile/live-session ownership, with read-after-write verification. - gitea_update_pr_branch_by_merge now refreshes the lock after the remote advance and reports a PARTIAL LIFECYCLE FAILURE (success=False) when the refresh fails, instead of falsely reporting a full synchronization. Read-side (recovers already-drifted locks): - issue_lock_worktree.read_merge_sync_provenance: server-side git observation proving a remote head is a sanctioned base-into-branch merge that preserved the branch mainline back to the recorded head. - issue_lock_recovery: new HEAD_RELATION_REMOTE_MERGE_SYNCED accepts a dead-session lock whose recorded head is a strict merge-sync ancestor of the live PR head — and only that. Rewrites, rebases, force-pushes, non-ancestor heads, dirty worktrees, live/competing owners, and wrong repo/issue/branch/ identity/profile all stay protected. No existing exact-head, branch-protection, parity, workspace, identity, role, or mutation-safety gate is weakened. All provenance is server-derived; nothing is reachable from an MCP caller. Tests: tests/test_issue_871_durable_lock_head_refresh.py (32 cases) covering first/second sync, CAS, ownership, partial-failure, merge-sync recovery happy-path and every fail-closed branch. Full suite: 4789 passed, 13 pre- existing baseline failures unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01FZPyVh2DGczQrDxtqwGH5p
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,630 @@
|
||||
"""Durable linked-issue lock head refresh + merge-sync dead-session recovery (#871).
|
||||
|
||||
``gitea_update_pr_branch_by_merge`` advances a PR's *remote* head but historically
|
||||
never advanced the linked durable issue lock's recorded head. After the owning
|
||||
session died the drifted lock became unrecoverable and no further synchronization
|
||||
was possible (PR #866 / issue #855).
|
||||
|
||||
Two halves are covered:
|
||||
|
||||
* the write-side refresh (``issue_lock_store.assess/apply_durable_lock_head_refresh``)
|
||||
that records the new synced head under compare-and-swap with read-after-write; and
|
||||
* the read-side recovery relation (``issue_lock_recovery`` +
|
||||
``issue_lock_worktree.read_merge_sync_provenance``) that lets a dead-session lock
|
||||
whose recorded head is a merge-sync *ancestor* of the live PR head be recovered —
|
||||
and nothing else.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import issue_lock_recovery # noqa: E402
|
||||
import issue_lock_store # noqa: E402
|
||||
import issue_lock_worktree # noqa: E402
|
||||
|
||||
ISSUE = 8710
|
||||
PR_NUMBER = 8711
|
||||
BRANCH = f"fix/issue-{ISSUE}-durable-lock-head-refresh"
|
||||
IDENTITY = "example-user"
|
||||
PROFILE = "example-author"
|
||||
OLD = "a" * 40
|
||||
NEW1 = "b" * 40
|
||||
NEW2 = "c" * 40
|
||||
BASE = "d" * 40
|
||||
REMOTE = "prgs"
|
||||
ORG = "ExampleOrg"
|
||||
REPO = "ExampleRepo"
|
||||
|
||||
|
||||
def dead_pid() -> int:
|
||||
proc = subprocess.Popen([sys.executable, "-c", "pass"])
|
||||
proc.wait()
|
||||
return proc.pid
|
||||
|
||||
|
||||
def future_ts(hours: int = 4) -> str:
|
||||
return (
|
||||
(datetime.now(timezone.utc) + timedelta(hours=hours))
|
||||
.isoformat()
|
||||
.replace("+00:00", "Z")
|
||||
)
|
||||
|
||||
|
||||
def _git(cwd, *args):
|
||||
return subprocess.run(
|
||||
["git", "-C", cwd, *args],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
|
||||
|
||||
def _rev(cwd, ref="HEAD") -> str:
|
||||
return _git(cwd, "rev-parse", ref).stdout.strip()
|
||||
|
||||
|
||||
def build_merge_sync_repo(tmp: str) -> dict:
|
||||
"""Build a repo where a feature branch was synced by merging master in.
|
||||
|
||||
Returns a dict with the prior (branch) head, the synced merge-commit head,
|
||||
the master tip, plus a rebase-style linear descendant and an unrelated head.
|
||||
"""
|
||||
_git(tmp, "init", "-q", "-b", "master")
|
||||
_git(tmp, "config", "user.email", "[email protected]")
|
||||
_git(tmp, "config", "user.name", "T")
|
||||
Path(tmp, "base.txt").write_text("base\n")
|
||||
_git(tmp, "add", "-A")
|
||||
_git(tmp, "commit", "-q", "-m", "root")
|
||||
|
||||
# Feature branch cut from root, one commit — this is the PRIOR/recorded head.
|
||||
_git(tmp, "checkout", "-q", "-b", BRANCH)
|
||||
Path(tmp, "feature.txt").write_text("feature\n")
|
||||
_git(tmp, "add", "-A")
|
||||
_git(tmp, "commit", "-q", "-m", "feature work")
|
||||
prior = _rev(tmp)
|
||||
|
||||
# Master advances (the base the sync will merge in).
|
||||
_git(tmp, "checkout", "-q", "master")
|
||||
Path(tmp, "base.txt").write_text("base\nmore\n")
|
||||
_git(tmp, "add", "-A")
|
||||
_git(tmp, "commit", "-q", "-m", "master advance")
|
||||
master_tip = _rev(tmp)
|
||||
|
||||
# Sync: merge master INTO the feature branch → merge commit, first parent = prior.
|
||||
_git(tmp, "checkout", "-q", BRANCH)
|
||||
_git(tmp, "merge", "-q", "--no-ff", "-m", "Merge master into feature", "master")
|
||||
synced = _rev(tmp)
|
||||
|
||||
# A plain linear descendant of prior (NOT a merge) — a rebase/extra-commit shape.
|
||||
_git(tmp, "checkout", "-q", "-b", "linear-branch", prior)
|
||||
Path(tmp, "extra.txt").write_text("extra\n")
|
||||
_git(tmp, "add", "-A")
|
||||
_git(tmp, "commit", "-q", "-m", "extra linear commit")
|
||||
linear = _rev(tmp)
|
||||
|
||||
# An unrelated root (force-push / rewritten history shape).
|
||||
unrelated_dir = tempfile.mkdtemp()
|
||||
_git(unrelated_dir, "init", "-q", "-b", "x")
|
||||
_git(unrelated_dir, "config", "user.email", "[email protected]")
|
||||
_git(unrelated_dir, "config", "user.name", "T")
|
||||
Path(unrelated_dir, "z.txt").write_text("z\n")
|
||||
_git(unrelated_dir, "add", "-A")
|
||||
_git(unrelated_dir, "commit", "-q", "-m", "unrelated")
|
||||
unrelated = _rev(unrelated_dir)
|
||||
|
||||
# Leave the worktree checked out on the feature branch at the PRIOR head, as
|
||||
# a dead author session that never advanced would have left it.
|
||||
_git(tmp, "checkout", "-q", BRANCH)
|
||||
_git(tmp, "reset", "-q", "--hard", prior)
|
||||
|
||||
return {
|
||||
"prior": prior,
|
||||
"master_tip": master_tip,
|
||||
"synced": synced,
|
||||
"linear": linear,
|
||||
"unrelated": unrelated,
|
||||
}
|
||||
|
||||
|
||||
# ─────────────────────────── write-side refresh ───────────────────────────
|
||||
|
||||
|
||||
class TestDurableLockHeadRefresh(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.lock_dir = tempfile.mkdtemp()
|
||||
self.wt = tempfile.mkdtemp()
|
||||
lock_data = {
|
||||
"issue_number": ISSUE,
|
||||
"branch_name": BRANCH,
|
||||
"worktree_path": self.wt,
|
||||
"remote": REMOTE,
|
||||
"org": ORG,
|
||||
"repo": REPO,
|
||||
"claimant": {"username": IDENTITY, "profile": PROFILE},
|
||||
"work_lease": {
|
||||
"operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
|
||||
"issue_number": ISSUE,
|
||||
"branch": BRANCH,
|
||||
"worktree_path": self.wt,
|
||||
"claimant": {"username": IDENTITY, "profile": PROFILE},
|
||||
"expires_at": future_ts(),
|
||||
},
|
||||
}
|
||||
issue_lock_store.bind_session_lock(lock_data, lock_dir=self.lock_dir)
|
||||
|
||||
def _apply(self, **over):
|
||||
kw = dict(
|
||||
remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE,
|
||||
branch_name=BRANCH, worktree_path=self.wt, pr_number=PR_NUMBER,
|
||||
identity=IDENTITY, profile=PROFILE, current_pid=os.getpid(),
|
||||
expected_old_head=OLD, new_head=NEW1, synced_at=future_ts(0),
|
||||
base_head=BASE, lock_dir=self.lock_dir,
|
||||
)
|
||||
kw.update(over)
|
||||
return issue_lock_store.apply_durable_lock_head_refresh(**kw)
|
||||
|
||||
def _load(self):
|
||||
return issue_lock_store.load_issue_lock(
|
||||
remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE,
|
||||
lock_dir=self.lock_dir,
|
||||
)
|
||||
|
||||
def test_first_sync_updates_recorded_head(self):
|
||||
"""AC1: first base sync writes the resulting head to the durable lock."""
|
||||
res = self._apply()
|
||||
self.assertTrue(res["refreshed"], res["reasons"])
|
||||
self.assertTrue(res["read_after_write_ok"])
|
||||
self.assertEqual(self._load().get("synced_pr_head"), NEW1)
|
||||
|
||||
def test_second_sync_after_master_advance(self):
|
||||
"""AC2: a later master advance permits a second sanctioned sync."""
|
||||
self.assertTrue(self._apply()["refreshed"])
|
||||
res2 = self._apply(expected_old_head=NEW1, new_head=NEW2)
|
||||
self.assertTrue(res2["refreshed"], res2["reasons"])
|
||||
self.assertEqual(self._load().get("synced_pr_head"), NEW2)
|
||||
history = self._load().get("branch_sync_history")
|
||||
self.assertEqual(len(history), 2)
|
||||
self.assertEqual(history[0]["last_synced_pr_head"], NEW1)
|
||||
self.assertEqual(history[1]["prior_pr_head"], NEW1)
|
||||
|
||||
def test_cas_detects_concurrent_head_change(self):
|
||||
"""AC6: CAS refuses when the recorded synced head is not the old head."""
|
||||
self.assertTrue(self._apply()["refreshed"]) # recorded head now NEW1
|
||||
# A second sync claiming the old head is still OLD must fail closed.
|
||||
res = self._apply(expected_old_head=OLD, new_head=NEW2)
|
||||
self.assertFalse(res["refreshed"])
|
||||
self.assertTrue(any("CAS" in r or "concurrent" in r for r in res["reasons"]))
|
||||
self.assertEqual(self._load().get("synced_pr_head"), NEW1)
|
||||
|
||||
def test_wrong_issue_fails_closed(self):
|
||||
res = self._apply(issue_number=999999)
|
||||
self.assertFalse(res["refreshed"])
|
||||
|
||||
def test_wrong_branch_fails_closed(self):
|
||||
res = self._apply(branch_name="fix/issue-8710-wrong")
|
||||
self.assertFalse(res["refreshed"])
|
||||
|
||||
def test_wrong_repo_fails_closed(self):
|
||||
res = self._apply(repo="OtherRepo")
|
||||
self.assertFalse(res["refreshed"])
|
||||
|
||||
def test_wrong_identity_fails_closed(self):
|
||||
res = self._apply(identity="intruder")
|
||||
self.assertFalse(res["refreshed"])
|
||||
|
||||
def test_wrong_profile_fails_closed(self):
|
||||
res = self._apply(profile="prgs-reviewer")
|
||||
self.assertFalse(res["refreshed"])
|
||||
|
||||
def test_foreign_session_fails_closed(self):
|
||||
"""A refresh is not a recovery: the current process must own the lock."""
|
||||
path = issue_lock_store.lock_file_path(
|
||||
remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE,
|
||||
lock_dir=self.lock_dir,
|
||||
)
|
||||
rec = issue_lock_store.read_lock_file(path)
|
||||
rec["session_pid"] = dead_pid()
|
||||
rec["pid"] = rec["session_pid"]
|
||||
issue_lock_store.save_lock_file(path, rec)
|
||||
res = self._apply()
|
||||
self.assertFalse(res["refreshed"])
|
||||
self.assertTrue(any("current session" in r or "live owner" in r for r in res["reasons"]))
|
||||
|
||||
def test_new_equals_old_fails_closed(self):
|
||||
res = self._apply(expected_old_head=OLD, new_head=OLD)
|
||||
self.assertFalse(res["refreshed"])
|
||||
|
||||
def test_non_full_sha_fails_closed(self):
|
||||
self.assertFalse(self._apply(new_head="deadbeef")["refreshed"])
|
||||
self.assertFalse(self._apply(expected_old_head="xyz")["refreshed"])
|
||||
|
||||
def test_no_lock_fails_closed(self):
|
||||
assessment = issue_lock_store.assess_durable_lock_head_refresh(
|
||||
None, remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE,
|
||||
branch_name=BRANCH, worktree_path=self.wt, pr_number=PR_NUMBER,
|
||||
identity=IDENTITY, profile=PROFILE, current_pid=os.getpid(),
|
||||
expected_old_head=OLD, new_head=NEW1,
|
||||
)
|
||||
self.assertFalse(assessment["allowed"])
|
||||
|
||||
|
||||
# ─────────────────────── merge-sync provenance (real git) ───────────────────
|
||||
|
||||
|
||||
class TestMergeSyncProvenanceObservation(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.mkdtemp()
|
||||
self.shas = build_merge_sync_repo(self.tmp)
|
||||
|
||||
def test_merge_sync_is_recognized(self):
|
||||
obs = issue_lock_worktree.read_merge_sync_provenance(
|
||||
self.tmp, prior_head_sha=self.shas["prior"],
|
||||
synced_head_sha=self.shas["synced"],
|
||||
)
|
||||
self.assertTrue(obs["is_merge_sync"], obs["reasons"])
|
||||
self.assertTrue(obs["prior_is_ancestor"])
|
||||
self.assertTrue(obs["synced_is_merge"])
|
||||
self.assertTrue(obs["first_parent_reaches_prior"])
|
||||
|
||||
def test_linear_descendant_is_not_a_merge_sync(self):
|
||||
"""A plain non-merge descendant (rebase/extra commit) is not a sync."""
|
||||
obs = issue_lock_worktree.read_merge_sync_provenance(
|
||||
self.tmp, prior_head_sha=self.shas["prior"],
|
||||
synced_head_sha=self.shas["linear"],
|
||||
)
|
||||
self.assertTrue(obs["probe_ok"])
|
||||
self.assertFalse(obs["is_merge_sync"])
|
||||
self.assertFalse(obs["synced_is_merge"])
|
||||
|
||||
def test_unrelated_history_fails_closed(self):
|
||||
"""A rewritten/force-pushed head where prior is unreachable fails closed."""
|
||||
obs = issue_lock_worktree.read_merge_sync_provenance(
|
||||
self.tmp, prior_head_sha=self.shas["prior"],
|
||||
synced_head_sha=self.shas["unrelated"],
|
||||
)
|
||||
self.assertFalse(obs["is_merge_sync"])
|
||||
|
||||
def test_missing_args_fail_closed(self):
|
||||
obs = issue_lock_worktree.read_merge_sync_provenance(
|
||||
self.tmp, prior_head_sha=None, synced_head_sha=self.shas["synced"],
|
||||
)
|
||||
self.assertFalse(obs["is_merge_sync"])
|
||||
|
||||
|
||||
# ──────────────────── merge-sync dead-session recovery ──────────────────────
|
||||
|
||||
|
||||
def make_dead_lock(worktree, **over):
|
||||
pid = dead_pid()
|
||||
lock = {
|
||||
"issue_number": ISSUE,
|
||||
"branch_name": BRANCH,
|
||||
"worktree_path": worktree,
|
||||
"remote": REMOTE,
|
||||
"org": ORG,
|
||||
"repo": REPO,
|
||||
"session_pid": pid,
|
||||
"pid": pid,
|
||||
"claimant": {"username": IDENTITY, "profile": PROFILE},
|
||||
"work_lease": {
|
||||
"operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
|
||||
"issue_number": ISSUE,
|
||||
"branch": BRANCH,
|
||||
"worktree_path": worktree,
|
||||
"claimant": {"username": IDENTITY, "profile": PROFILE},
|
||||
"expires_at": future_ts(),
|
||||
},
|
||||
}
|
||||
lock.update(over)
|
||||
return lock
|
||||
|
||||
|
||||
def sync_prov(prior, synced, **over):
|
||||
d = {
|
||||
"prior_head_sha": prior,
|
||||
"synced_head_sha": synced,
|
||||
"probe_ok": True,
|
||||
"prior_present": True,
|
||||
"synced_present": True,
|
||||
"prior_is_ancestor": True,
|
||||
"synced_is_merge": True,
|
||||
"first_parent_reaches_prior": True,
|
||||
"is_merge_sync": True,
|
||||
"first_parent_sha": prior,
|
||||
"parent_count": 2,
|
||||
"proof": f"{synced} merged base into branch above {prior}",
|
||||
"reasons": [],
|
||||
}
|
||||
d.update(over)
|
||||
return d
|
||||
|
||||
|
||||
class TestMergeSyncRecovery(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.mkdtemp()
|
||||
self.shas = build_merge_sync_repo(self.tmp)
|
||||
self.prior = self.shas["prior"]
|
||||
self.synced = self.shas["synced"]
|
||||
|
||||
def _assess(self, **over):
|
||||
lock = over.pop("_lock", None) or make_dead_lock(self.tmp)
|
||||
kw = dict(
|
||||
issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.tmp,
|
||||
remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE,
|
||||
current_branch=BRANCH, porcelain_status="",
|
||||
head_sha=self.prior, remote_head_sha=self.synced,
|
||||
pr_head_sha=self.synced, pr_number=PR_NUMBER,
|
||||
competing_live_locks=[], candidate_branches=[BRANCH],
|
||||
current_pid=os.getpid(),
|
||||
remote_branch_exists=True,
|
||||
sync_provenance=sync_prov(self.prior, self.synced),
|
||||
)
|
||||
kw.update(over)
|
||||
return issue_lock_recovery.assess_dead_session_lock_recovery(lock, **kw)
|
||||
|
||||
def test_merge_sync_drift_is_recoverable(self):
|
||||
"""AC3/AC4: dead session, recorded head is a merge-sync ancestor of PR head."""
|
||||
res = self._assess()
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED, res["reasons"])
|
||||
self.assertEqual(
|
||||
res["evidence"]["head_relation"],
|
||||
issue_lock_recovery.HEAD_RELATION_REMOTE_MERGE_SYNCED,
|
||||
)
|
||||
self.assertEqual(res["evidence"]["accepted_head"], self.synced)
|
||||
|
||||
def test_missing_provenance_fails_closed(self):
|
||||
"""No server-derived provenance → cannot accept a remote ahead of local."""
|
||||
res = self._assess(sync_provenance=None)
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_non_ancestor_recorded_head_fails_closed(self):
|
||||
"""AC7: provenance that does not prove ancestry is rejected."""
|
||||
res = self._assess(
|
||||
sync_provenance=sync_prov(
|
||||
self.prior, self.synced, prior_is_ancestor=False, is_merge_sync=False,
|
||||
reasons=["prior head is not an ancestor"],
|
||||
)
|
||||
)
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_force_pushed_history_fails_closed(self):
|
||||
"""AC8: a rewritten head (not a merge sync) stays protected."""
|
||||
res = self._assess(
|
||||
sync_provenance=sync_prov(
|
||||
self.prior, self.synced, is_merge_sync=False, synced_is_merge=False,
|
||||
reasons=["not a merge-based sync"],
|
||||
)
|
||||
)
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_provenance_for_other_commits_fails_closed(self):
|
||||
"""Provenance whose endpoints differ from the heads under assessment is rejected."""
|
||||
res = self._assess(
|
||||
sync_provenance=sync_prov("f" * 40, self.synced),
|
||||
)
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_dirty_worktree_fails_closed(self):
|
||||
"""AC11: dirty worktrees remain protected."""
|
||||
res = self._assess(porcelain_status=" M feature.txt\n")
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_live_owner_fails_closed(self):
|
||||
"""AC10: a live recorded owner is not a dead-session recovery."""
|
||||
lock = make_dead_lock(self.tmp, session_pid=os.getpid(), pid=os.getpid())
|
||||
res = self._assess(_lock=lock)
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_competing_claimant_fails_closed(self):
|
||||
"""AC13: a competing live lock blocks recovery."""
|
||||
res = self._assess(
|
||||
competing_live_locks=[{
|
||||
"issue_number": ISSUE, "branch_name": BRANCH,
|
||||
"worktree_path": "/some/other/wt", "pid": os.getpid(),
|
||||
}]
|
||||
)
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_wrong_branch_fails_closed(self):
|
||||
"""AC9: worktree on a different branch fails closed."""
|
||||
res = self._assess(current_branch="fix/issue-8710-other")
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_wrong_identity_fails_closed(self):
|
||||
res = self._assess(identity="intruder")
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_pr_head_mismatch_fails_closed(self):
|
||||
"""The open PR must sit at the synced remote head."""
|
||||
res = self._assess(pr_head_sha="e" * 40)
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED)
|
||||
|
||||
def test_owning_pr_evidence_for_merge_sync(self):
|
||||
res = self._assess()
|
||||
ev = issue_lock_recovery.owning_pr_recovery_evidence(res)
|
||||
self.assertIsNotNone(ev)
|
||||
self.assertEqual(ev["pr_number"], PR_NUMBER)
|
||||
self.assertEqual(ev["head_sha"], self.synced)
|
||||
self.assertEqual(
|
||||
ev["head_relation"],
|
||||
issue_lock_recovery.HEAD_RELATION_REMOTE_MERGE_SYNCED,
|
||||
)
|
||||
|
||||
def test_recovered_owning_pr_from_persisted_record(self):
|
||||
res = self._assess()
|
||||
record = issue_lock_recovery.build_recovery_record(res, recovered_at=future_ts(0))
|
||||
lock = {"issue_number": ISSUE, "branch_name": BRANCH,
|
||||
"dead_session_recovery": record}
|
||||
rebuilt = issue_lock_recovery.recovered_owning_pr_from_lock(lock)
|
||||
self.assertIsNotNone(rebuilt)
|
||||
self.assertEqual(rebuilt["head_sha"], self.synced)
|
||||
self.assertEqual(
|
||||
rebuilt["head_relation"],
|
||||
issue_lock_recovery.HEAD_RELATION_REMOTE_MERGE_SYNCED,
|
||||
)
|
||||
|
||||
|
||||
class TestExistingRelationsUnchanged(unittest.TestCase):
|
||||
"""AC14/AC15: equal-head recovery still works; merge-sync did not weaken it."""
|
||||
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.mkdtemp()
|
||||
self.shas = build_merge_sync_repo(self.tmp)
|
||||
|
||||
def test_equal_head_recovery_still_sanctioned(self):
|
||||
# Worktree at prior head; remote also at prior head → the #753 equal case.
|
||||
prior = self.shas["prior"]
|
||||
lock = make_dead_lock(self.tmp)
|
||||
res = issue_lock_recovery.assess_dead_session_lock_recovery(
|
||||
lock, issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.tmp,
|
||||
remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE,
|
||||
current_branch=BRANCH, porcelain_status="",
|
||||
head_sha=prior, remote_head_sha=prior,
|
||||
pr_head_sha=prior, pr_number=PR_NUMBER,
|
||||
competing_live_locks=[], candidate_branches=[BRANCH],
|
||||
current_pid=os.getpid(), remote_branch_exists=True,
|
||||
)
|
||||
self.assertEqual(res["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED, res["reasons"])
|
||||
self.assertEqual(
|
||||
res["evidence"]["head_relation"], issue_lock_recovery.HEAD_RELATION_EQUAL,
|
||||
)
|
||||
|
||||
|
||||
class TestUpdatePrWrapperPartialFailure(unittest.TestCase):
|
||||
"""AC5/AC16: the tool advances the remote head then refreshes the durable lock.
|
||||
|
||||
When the durable refresh fails after the remote advance, the tool must report a
|
||||
partial lifecycle failure and NOT a fully successful synchronization. Exact PR-
|
||||
head / base-head pinning is preserved (delegated to the real preflight, stubbed
|
||||
here only to isolate the post-update lifecycle branch).
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
import gitea_mcp_server as gms # noqa: E402
|
||||
self.gms = gms
|
||||
self._orig = {}
|
||||
|
||||
def _patch(name, value):
|
||||
self._orig[name] = getattr(gms, name)
|
||||
setattr(gms, name, value)
|
||||
|
||||
_patch("get_profile", lambda *a, **k: {
|
||||
"allowed_operations": ["gitea.branch.push"],
|
||||
"forbidden_operations": [],
|
||||
"profile_name": "prgs-author",
|
||||
})
|
||||
_patch("_role_kind", lambda *a, **k: "author")
|
||||
_patch("_profile_operation_gate", lambda *a, **k: None)
|
||||
_patch("_permission_block_report", lambda *a, **k: {})
|
||||
_patch("_resolve", lambda *a, **k: ("gitea.prgs.cc", ORG, REPO))
|
||||
_patch("_verify_role_mutation_workspace", lambda *a, **k: None)
|
||||
_patch("_get_workspace_porcelain", lambda *a, **k: "")
|
||||
_patch("_canonical_local_git_root", lambda *a, **k: "/x")
|
||||
_patch("_master_parity_block", lambda *a, **k: None)
|
||||
_patch("_auth", lambda *a, **k: {"token": "x"})
|
||||
_patch("repo_api_url", lambda *a, **k: "http://api")
|
||||
_patch("_redact", lambda s: s)
|
||||
_patch("_work_lease_claimant", lambda *a, **k: {
|
||||
"username": IDENTITY, "profile": PROFILE,
|
||||
})
|
||||
_patch("_prove_author_ownership_for_pr", lambda *a, **k: {
|
||||
"has_author_lock": True, "matched_issue": ISSUE,
|
||||
"matched_via": "branch", "linked_issues": [ISSUE],
|
||||
"recovered_owning_pr": None, "reasons": [],
|
||||
})
|
||||
|
||||
# Real preflight is unit-tested elsewhere; stub it to isolate the
|
||||
# post-update durable-lock lifecycle branch under test.
|
||||
orig_pf = gms.pr_sync_status.assess_update_pr_branch_preflight
|
||||
self._orig_pf = orig_pf
|
||||
gms.pr_sync_status.assess_update_pr_branch_preflight = (
|
||||
lambda *a, **k: {"mutation_allowed": True, "reasons": [], "performed": False}
|
||||
)
|
||||
|
||||
# Sequence the two GET /pulls calls: OLD before update, NEW after.
|
||||
self._pull_calls = {"n": 0}
|
||||
|
||||
def fake_api_request(method, url, auth, *a, **k):
|
||||
m = method.upper()
|
||||
if m == "GET" and url.endswith(f"/pulls/{PR_NUMBER}"):
|
||||
self._pull_calls["n"] += 1
|
||||
head = OLD if self._pull_calls["n"] == 1 else NEW1
|
||||
return {
|
||||
"state": "open",
|
||||
"head": {"sha": head, "ref": BRANCH},
|
||||
"base": {"sha": BASE, "ref": "master"},
|
||||
"mergeable": True, "title": "t", "body": "b",
|
||||
}
|
||||
if m == "GET" and "/branches/" in url:
|
||||
return {"commit": {"id": BASE}}
|
||||
if m == "POST" and "/update" in url:
|
||||
return {}
|
||||
return {}
|
||||
|
||||
_patch("api_request", fake_api_request)
|
||||
|
||||
def tearDown(self):
|
||||
for name, value in self._orig.items():
|
||||
setattr(self.gms, name, value)
|
||||
self.gms.pr_sync_status.assess_update_pr_branch_preflight = self._orig_pf
|
||||
|
||||
def _run(self):
|
||||
return self.gms.gitea_update_pr_branch_by_merge(
|
||||
pr_number=PR_NUMBER,
|
||||
expected_pr_head_sha=OLD,
|
||||
expected_base_head_sha=BASE,
|
||||
remote=REMOTE,
|
||||
worktree_path="/tmp/branches/wt-871",
|
||||
)
|
||||
|
||||
def test_partial_failure_when_refresh_fails(self):
|
||||
self._orig["apply_durable_lock_head_refresh"] = (
|
||||
self.gms.issue_lock_store.apply_durable_lock_head_refresh
|
||||
)
|
||||
self.gms.issue_lock_store.apply_durable_lock_head_refresh = (
|
||||
lambda **k: {"refreshed": False, "reasons": ["forced refresh failure"]}
|
||||
)
|
||||
try:
|
||||
res = self._run()
|
||||
finally:
|
||||
self.gms.issue_lock_store.apply_durable_lock_head_refresh = (
|
||||
self._orig["apply_durable_lock_head_refresh"]
|
||||
)
|
||||
self.assertTrue(res["performed"])
|
||||
self.assertEqual(res["new_pr_head_sha"], NEW1)
|
||||
self.assertFalse(res["success"])
|
||||
self.assertTrue(res["partial_lifecycle_failure"])
|
||||
self.assertFalse(res["durable_lock_refreshed"])
|
||||
|
||||
def test_full_success_when_refresh_succeeds(self):
|
||||
self._orig["apply_durable_lock_head_refresh"] = (
|
||||
self.gms.issue_lock_store.apply_durable_lock_head_refresh
|
||||
)
|
||||
self.gms.issue_lock_store.apply_durable_lock_head_refresh = (
|
||||
lambda **k: {"refreshed": True, "read_after_write_ok": True,
|
||||
"new_head": NEW1, "reasons": ["ok"]}
|
||||
)
|
||||
try:
|
||||
res = self._run()
|
||||
finally:
|
||||
self.gms.issue_lock_store.apply_durable_lock_head_refresh = (
|
||||
self._orig["apply_durable_lock_head_refresh"]
|
||||
)
|
||||
self.assertTrue(res["success"])
|
||||
self.assertTrue(res["performed"])
|
||||
self.assertTrue(res["durable_lock_refreshed"])
|
||||
self.assertTrue(res["fully_synchronized"])
|
||||
self.assertEqual(res["new_pr_head_sha"], NEW1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user