Gitea-Tools/tests/test_issue_871_durable_lock_head_refresh.py

"""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()