fix: align test mocks with #399 lease gates and head SHA requirements

PR #416 introduced pr_work_lease reviewer blocks and mandatory full 40-hex expected_head_sha on mark/merge paths. Update audit, review, permission-report, terminal hard-stop, inventory, and schema tests to patch lease comment fetches, seed ready review decisions, and use consistent HEAD_SHA constants so the full suite passes again.
2026-07-08 01:26:36 -04:00
parent 55b3658d93
commit d4dc9aa854
7 changed files with 159 additions and 59 deletions
@@ -284,21 +284,38 @@ class TestSimpleToolAudit(_AuditWiringBase):
        self.assertEqual(result["number"], 9)


+_NO_PR_WORK_LEASE_BLOCK = {"block": False, "reasons": [], "mutation_allowed": True}
+
+
 class TestGatedToolAudit(_AuditWiringBase):

    def setUp(self):
        super().setUp()
+        from mcp_server import init_review_decision_lock
        from tests.test_mcp_server import _install_owned_reviewer_lease
        import reviewer_pr_lease

+        # init_review_decision_lock clears any prior session lease (#407).
+        init_review_decision_lock("prgs", "review_pr")
        self._lease_patch = _install_owned_reviewer_lease(8)
        self._lease_patch.start()
        self._auth_identity_patch = patch(
            "mcp_server._authenticated_username", return_value="reviewer-bot"
        )
        self._auth_identity_patch.start()
+        self._pr_lease_comments_patch = patch(
+            "mcp_server._list_pr_lease_comments", return_value=[]
+        )
+        self._pr_lease_comments_patch.start()
+        self._pr_work_lease_patch = patch(
+            "mcp_server._pr_work_lease_reviewer_block",
+            return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
+        )
+        self._pr_work_lease_patch.start()
        self.addCleanup(self._auth_identity_patch.stop)
        self.addCleanup(self._lease_patch.stop)
+        self.addCleanup(self._pr_lease_comments_patch.stop)
+        self.addCleanup(self._pr_work_lease_patch.stop)
        self.addCleanup(reviewer_pr_lease.clear_session_lease)

    def _pr(self, author, state="open", sha="abc123", mergeable=True):
@@ -349,7 +366,9 @@ class TestGatedToolAudit(_AuditWiringBase):
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_submit_review_success_audited(self, _auth, mock_api):
+        # mark_final_review_decision and submit each run eligibility (user + PR).
        mock_api.side_effect = [
+            {"login": "reviewer-bot"}, self._pr("author-bot"),
            {"login": "reviewer-bot"}, self._pr("author-bot"),
            {"id": 7, "state": "APPROVED"},
            [{"id": 7, "user": {"login": "reviewer-bot"}, "state": "APPROVED",
@@ -358,23 +377,16 @@ class TestGatedToolAudit(_AuditWiringBase):
        env = self._env(GITEA_PROFILE_NAME="gitea-reviewer",
                        GITEA_ALLOWED_OPERATIONS="read,review,approve")
        with patch.dict(os.environ, env, clear=True):
-            from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
-            from tests.test_mcp_server import _install_owned_reviewer_lease
-            import reviewer_pr_lease
+            from mcp_server import gitea_mark_final_review_decision

-            init_review_decision_lock("prgs", "review_pr")
-            gitea_mark_final_review_decision(8, "approve", remote="prgs")
-            lease_patch = _install_owned_reviewer_lease(8)
-            lease_patch.start()
-            try:
+            gitea_mark_final_review_decision(
+                8, "approve", expected_head_sha="abc123", remote="prgs",
+            )
            r = gitea_submit_pr_review(
                pr_number=8, action="approve",
                body="LGTM", remote="prgs",
                final_review_decision_ready=True,
            )
-            finally:
-                lease_patch.stop()
-                reviewer_pr_lease.clear_session_lease()
        self.assertTrue(r["performed"])
        recs = self._records()
        self.assertEqual(len(recs), 1)
@@ -122,15 +122,19 @@ class TestShaCannotBypassSelfReview(unittest.TestCase):
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_review_tool_refuses_self_approval_despite_sha(self, _auth, mock_api, mock_get_all):
-        mock_get_all.return_value = [{"number": 9, "title": "PR 9", "state": "open", "head": {"ref": "branch9", "sha": "abc1234"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}]
+        from mcp_server import init_review_decision_lock
+        from tests.test_mcp_server import FULL_HEAD_SHA, _seed_ready_review_decision
+
+        head_sha = FULL_HEAD_SHA
+        mock_get_all.return_value = [{"number": 9, "title": "PR 9", "state": "open", "head": {"ref": "branch9", "sha": head_sha}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}]
        mock_api.side_effect = [
            {"login": "jcwalker3"}, # /user (inventory)
            {"login": "jcwalker3"}, # /user (submit eligibility)
-            {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/9
+            {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": head_sha}, "mergeable": True}, # /pulls/9
        ]
-        from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
        init_review_decision_lock("prgs", "review_pr")
-        gitea_mark_final_review_decision(9, "approve", remote="prgs")
+        with patch("mcp_server._list_pr_lease_comments", return_value=[]):
+            _seed_ready_review_decision(9, "approve", sha=head_sha, remote="prgs")
        env = self._env(SHA_WOULD_BE_REVIEWER, "reviewer")
        with patch.dict(os.environ, env, clear=True):
            r = gitea_review_pr(
@@ -47,6 +47,7 @@ import gitea_config  # noqa: E402
 import mcp_server

 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
+FULL_HEAD_SHA = "a" * 40

 _NO_BLOCKER_FEEDBACK = {
    "success": True,
@@ -139,6 +140,45 @@ def _install_owned_reviewer_lease(
    )


+def _seed_ready_review_decision(
+    pr_number,
+    action,
+    *,
+    sha=FULL_HEAD_SHA,
+    remote="prgs",
+    org=None,
+    repo=None,
+):
+    """Mark the review-decision lock ready without mark_final API calls (#399)."""
+    import mcp_server as _m
+
+    resolved_org, resolved_repo = org, repo
+    if remote in _m.REMOTES:
+        _, resolved_org, resolved_repo = _m._resolve(remote, None, org, repo)
+    profile_name = (_m.get_profile().get("profile_name") or "").strip()
+    session_lock = (
+        (os.environ.get(_m.SESSION_PROFILE_LOCK_ENV) or "").strip()
+        or profile_name
+    )
+    _m._save_review_decision_lock({
+        "task": "review_pr",
+        "remote": remote,
+        "session_pid": os.getpid(),
+        "session_profile": profile_name,
+        "session_profile_lock": session_lock,
+        "final_review_decision_ready": True,
+        "ready_pr_number": pr_number,
+        "ready_action": action,
+        "ready_expected_head_sha": sha,
+        "ready_remote": remote,
+        "ready_org": resolved_org,
+        "ready_repo": resolved_repo,
+        "live_mutations": [],
+        "correction_authorized": False,
+        "correction_reason": None,
+    })
+
+
 # Issue-write tools are profile-gated (#69).
 ISSUE_WRITE_ENV = {
    "GITEA_ALLOWED_OPERATIONS": (
@@ -1104,16 +1144,18 @@ class TestReviewPR(unittest.TestCase):
            "forbidden_operations": [],
            "base_url": None,
        }
-        mock_get_all.return_value = [{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1234"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}]
+        head_sha = FULL_HEAD_SHA
+        mock_get_all.return_value = [{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": head_sha}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}]
        # mock_api responses: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility)
        mock_api.side_effect = [
            {"login": "jcwalker3"}, # /api/v1/user (inventory)
            {"login": "jcwalker3"}, # /api/v1/user (submit eligibility)
-            {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/1
+            {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": head_sha}, "mergeable": True}, # /pulls/1
        ]
        from mcp_server import init_review_decision_lock
        init_review_decision_lock("prgs", "review_pr")
-        gitea_mark_final_review_decision(1, "approve", remote="prgs", expected_head_sha="abc123")
+        with patch("mcp_server._list_pr_lease_comments", return_value=[]):
+            _seed_ready_review_decision(1, "approve", sha=head_sha, remote="prgs")
        result = gitea_review_pr(
            pr_number=1,
            event="APPROVE",
@@ -1769,7 +1811,7 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
    """Block incidental live review mutations during validation."""

    PR = 203
-    SHA = "abc123"
+    SHA = FULL_HEAD_SHA

    def _pr(self, author, sha=SHA):
        return {
@@ -1798,7 +1840,13 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
            "mcp_server._list_pr_lease_comments", return_value=[]
        )
        self._pr_lease_comments_patch.start()
+        self._pr_work_lease_patch = patch(
+            "mcp_server._pr_work_lease_reviewer_block",
+            return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
+        )
+        self._pr_work_lease_patch.start()
        self.addCleanup(self._pr_lease_comments_patch.stop)
+        self.addCleanup(self._pr_work_lease_patch.stop)


    def _env(self):
@@ -2406,6 +2454,11 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
        self.mock_audit = patch("gitea_audit.write_event").start()
        # gitea.pr.close: closing a PR via gitea_edit_pr is capability-gated (#216).
        patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["read", "merge", "edit", "close", "gitea.pr.close", "gitea.issue.close"], "audit_label": "test", "forbidden_operations": []}).start()
+        patch("mcp_server._list_pr_lease_comments", return_value=[]).start()
+        patch(
+            "mcp_server._pr_work_lease_reviewer_block",
+            return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
+        ).start()

    def tearDown(self):
        patch.stopall()
@@ -2452,7 +2505,8 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
    def test_merge_pr_with_closes_removes_label(self):
        import reviewer_pr_lease

-        lease_patch = _install_owned_reviewer_lease(1, head_sha="sha123")
+        head_sha = FULL_HEAD_SHA
+        lease_patch = _install_owned_reviewer_lease(1, head_sha=head_sha)
        lease_patch.start()
        self.addCleanup(lease_patch.stop)
        self.addCleanup(reviewer_pr_lease.clear_session_lease)
@@ -2461,12 +2515,12 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
            if method == "GET" and "/user" in url:
                return {"login": "merger"}
            if method == "GET" and url.endswith("/reviews"):
-                return [_formal_review("reviewer", "APPROVED", sha="sha123")]
+                return [_formal_review("reviewer", "APPROVED", sha=head_sha)]
            if method == "GET" and "pulls/1" in url and "/files" not in url:
                return {
                    "user": {"login": "author"},
                    "state": "open",
-                    "head": {"sha": "sha123", "ref": "feat/my-branch"},
+                    "head": {"sha": head_sha, "ref": "feat/my-branch"},
                    "base": {"ref": "main"},
                    "mergeable": True,
                    "merged_commit_sha": "merge123",
@@ -2486,14 +2540,18 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
            return {}
        self.mock_api.side_effect = api_side_effect

-        res = gitea_merge_pr(pr_number=1, confirmation="MERGE PR 1", do="merge")
+        res = gitea_merge_pr(
+            pr_number=1, confirmation="MERGE PR 1", do="merge",
+            expected_head_sha=head_sha,
+        )
        self.assertTrue(res["performed"])
        self.assertEqual(res["cleanup_status"].get(123), "released")

    def test_merge_pr_with_branch_name_removes_label(self):
        import reviewer_pr_lease

-        lease_patch = _install_owned_reviewer_lease(1, head_sha="sha123")
+        head_sha = FULL_HEAD_SHA
+        lease_patch = _install_owned_reviewer_lease(1, head_sha=head_sha)
        lease_patch.start()
        self.addCleanup(lease_patch.stop)
        self.addCleanup(reviewer_pr_lease.clear_session_lease)
@@ -2502,12 +2560,12 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
            if method == "GET" and "/user" in url:
                return {"login": "merger"}
            if method == "GET" and url.endswith("/reviews"):
-                return [_formal_review("reviewer", "APPROVED", sha="sha123")]
+                return [_formal_review("reviewer", "APPROVED", sha=head_sha)]
            if method == "GET" and "pulls/1" in url and "/files" not in url:
                return {
                    "user": {"login": "author"},
                    "state": "open",
-                    "head": {"sha": "sha123", "ref": "fix/issue-123-slug"},
+                    "head": {"sha": head_sha, "ref": "fix/issue-123-slug"},
                    "base": {"ref": "main"},
                    "mergeable": True,
                    "merged_commit_sha": "merge123",
@@ -2527,7 +2585,10 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
            return {}
        self.mock_api.side_effect = api_side_effect

-        res = gitea_merge_pr(pr_number=1, confirmation="MERGE PR 1", do="merge")
+        res = gitea_merge_pr(
+            pr_number=1, confirmation="MERGE PR 1", do="merge",
+            expected_head_sha=head_sha,
+        )
        self.assertTrue(res["performed"])
        self.assertEqual(res["cleanup_status"].get(123), "released")

@@ -230,7 +230,9 @@ class TestEligibilityDenialReport(PermissionReportBase):
            return PR_PAYLOAD
        mock_api.side_effect = fake_api
        mcp_server.init_review_decision_lock("prgs", "review_pr")
-        mcp_server.gitea_mark_final_review_decision(42, "approve", remote="prgs")
+        from tests.test_mcp_server import _seed_ready_review_decision
+
+        _seed_ready_review_decision(42, "approve", remote="prgs")
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_submit_pr_review(
                pr_number=42, action="approve", body="lgtm", remote="prgs",
@@ -272,7 +274,9 @@ class TestReviewCommentPathUsesCanonicalOp(PermissionReportBase):
            return PR_PAYLOAD
        mock_api.side_effect = fake_api
        mcp_server.init_review_decision_lock("prgs", "review_pr")
-        mcp_server.gitea_mark_final_review_decision(42, "comment", remote="prgs")
+        from tests.test_mcp_server import _seed_ready_review_decision
+
+        _seed_ready_review_decision(42, "comment", remote="prgs")
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_submit_pr_review(
                pr_number=42, action="comment", body="finding", remote="prgs",
@@ -128,18 +128,29 @@ class TestPRQueueInventory(unittest.TestCase):
            "forbidden_operations": [],
            "base_url": None,
        }
+
+        from mcp_server import init_review_decision_lock
+        from tests.test_mcp_server import (
+            FULL_HEAD_SHA,
+            _install_owned_reviewer_lease,
+            _seed_ready_review_decision,
+        )
+        import reviewer_pr_lease
+
+        head_sha = FULL_HEAD_SHA
        mock_fetch.return_value = _final_page_fetch([
-            {"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "other_user"}}
+            {"number": 1, "title": "PR 1", "state": "open",
+             "head": {"ref": "branch1", "sha": head_sha},
+             "base": {"ref": "master"}, "mergeable": True,
+             "user": {"login": "other_user"}}
        ])
-        # mock_api: inventory whoami, eligibility whoami, eligibility PR,
-        # POST review (#244: state + visible-verdict GET reviews).
        mock_api.side_effect = [
            {"login": "reviewer1"},
            {"login": "reviewer1"},
            {
                "user": {"login": "other_user"},
                "state": "open",
-                "head": {"sha": "abc1"},
+                "head": {"sha": head_sha},
                "mergeable": True,
            },
            {"id": 100, "state": "APPROVED"},
@@ -148,22 +159,19 @@ class TestPRQueueInventory(unittest.TestCase):
                    "id": 100,
                    "user": {"login": "reviewer1"},
                    "state": "APPROVED",
-                    "commit_id": "abc1",
+                    "commit_id": head_sha,
                    "submitted_at": "2026-07-06T10:00:00Z",
                    "dismissed": False,
                }
            ],
        ]

-        from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
-        from tests.test_mcp_server import _install_owned_reviewer_lease
-        import reviewer_pr_lease
-
-        with patch("mcp_server._authenticated_username", return_value="reviewer1"):
+        with patch("mcp_server._authenticated_username", return_value="reviewer1"), \
+                patch("mcp_server._list_pr_lease_comments", return_value=[]):
            init_review_decision_lock("prgs", "review_pr")
-            gitea_mark_final_review_decision(1, "approve", remote="prgs")
+            _seed_ready_review_decision(1, "approve", sha=head_sha, remote="prgs")
            lease_patch = _install_owned_reviewer_lease(
-                1, head_sha="abc1", session_id="inventory-review-lease",
+                1, head_sha=head_sha, session_id="inventory-review-lease",
            )
            lease_patch.start()
            self.addCleanup(lease_patch.stop)
@@ -19,7 +19,11 @@ def _minimal_review_report(**overrides):
        "- Issue/PR: #182 / PR #203",
        "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
        "- Files changed: review_proofs.py",
-        "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
+        "- Validation: pass: pytest tests/test_review_proofs.py -q in branches/review-203",
+        "- Reviewed head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Final live head SHA before approval: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Final live head SHA before merge: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Push occurred during validation: no",
        "- Mutations: review only",
        "- File edits by reviewer: none",
        "- Worktree/index mutations: none",
@@ -80,7 +84,7 @@ class TestReviewFinalReportSchema(unittest.TestCase):

    def test_reviewed_head_without_validation_blocks(self):
        report = _minimal_review_report().replace(
-            "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
+            "- Validation: pass: pytest tests/test_review_proofs.py -q in branches/review-203",
            "- Validation: not run",
        )
        report += "\n- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
@@ -12,6 +12,8 @@ from unittest.mock import patch

 import mcp_server

+HEAD_SHA = "a" * 40
+

 def _lock(mutations=None, correction=False):
    return {
@@ -127,10 +129,20 @@ def _feedback(blocking, stale=False, success=True):
        "success": success,
        "has_blocking_change_requests": blocking,
        "review_feedback_stale": stale,
-        "current_head_sha": "abc123",
+        "current_head_sha": HEAD_SHA,
    }


+def _mark(action, pr_number=6, **kwargs):
+    kwargs.setdefault("expected_head_sha", HEAD_SHA)
+    no_lease_block = {"block": False, "reasons": [], "mutation_allowed": True}
+    with patch("mcp_server._list_pr_lease_comments", return_value=[]), \
+            patch("mcp_server._pr_work_lease_reviewer_block", return_value=no_lease_block):
+        return mcp_server.gitea_mark_final_review_decision(
+            pr_number=pr_number, action=action, remote="prgs", **kwargs
+        )
+
+
 class TestDuplicateRequestChangesSuppression(unittest.TestCase):
    def tearDown(self):
        mcp_server._save_review_decision_lock(None)
@@ -139,8 +151,7 @@ class TestDuplicateRequestChangesSuppression(unittest.TestCase):
        _seed()
        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
                          return_value=_feedback(blocking=True, stale=False)):
-            result = mcp_server.gitea_mark_final_review_decision(
-                pr_number=6, action="request_changes", remote="prgs")
+            result = _mark("request_changes")
        self.assertFalse(result["marked_ready"])
        self.assertTrue(
            any("duplicate" in r for r in result["reasons"]),
@@ -150,16 +161,14 @@ class TestDuplicateRequestChangesSuppression(unittest.TestCase):
        _seed()
        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
                          return_value=_feedback(blocking=True, stale=True)):
-            result = mcp_server.gitea_mark_final_review_decision(
-                pr_number=6, action="request_changes", remote="prgs")
+            result = _mark("request_changes")
        self.assertTrue(result["marked_ready"], result.get("reasons"))

    def test_request_changes_allowed_when_no_blocker(self):
        _seed()
        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
                          return_value=_feedback(blocking=False)):
-            result = mcp_server.gitea_mark_final_review_decision(
-                pr_number=6, action="request_changes", remote="prgs")
+            result = _mark("request_changes")
        self.assertTrue(result["marked_ready"], result.get("reasons"))

    def test_request_changes_fails_closed_when_feedback_unavailable(self):
@@ -167,8 +176,7 @@ class TestDuplicateRequestChangesSuppression(unittest.TestCase):
        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
                          return_value=_feedback(blocking=False,
                                                 success=False)):
-            result = mcp_server.gitea_mark_final_review_decision(
-                pr_number=6, action="request_changes", remote="prgs")
+            result = _mark("request_changes")
        self.assertFalse(result["marked_ready"])
        self.assertTrue(
            any("could not verify" in r for r in result["reasons"]),
@@ -178,8 +186,7 @@ class TestDuplicateRequestChangesSuppression(unittest.TestCase):
        _seed()
        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
                          side_effect=AssertionError("must not be called")):
-            result = mcp_server.gitea_mark_final_review_decision(
-                pr_number=6, action="approve", remote="prgs")
+            result = _mark("approve")
        self.assertTrue(result["marked_ready"], result.get("reasons"))