From f04cf449752e91472aa7f7a1eeb25fe661bd1c05 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 1 Jul 2026 15:03:49 -0400 Subject: [PATCH] feat: gate gitea_merge_pr behind identity/profile/eligibility + confirmation (#16) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the ungated gitea_merge_pr with a gated merge workflow. This is now the only merge path the MCP server exposes; the merge API is called only after every safety gate passes. Gates (fail-closed at each step): 1. Merge method is merge | squash | rebase. 2. Explicit confirmation: confirmation must equal "MERGE PR " (without it, zero API calls are made). 3. Reuse gitea_check_pr_eligibility (#14) with action 'merge': proves the authenticated identity, the active profile (and that it allows merge), the PR author, blocks self-merge, requires the PR open, and fails closed when the PR is not mergeable or mergeability is unknown. 4. Optional expected_head_sha: refuse if the PR head moved. 5. Optional expected_changed_files: refuse if the PR's changed file set differs. 6. Redundant self-merge block (auth user == PR author). The force/ignore-checks option was removed — Gitea's own mergeable signal (which reflects branch-protection required reviews/checks) must be positive, so required approval/check state is honoured, never bypassed. Output reports performed, authenticated user, profile name, PR author, PR number, head SHA checked, merge method, gates passed/blocked, and merge result / merge commit — never a token, auth header, or credential. Error text is scrubbed via _redact. Surface audit: no ungated merge path remains. The /merge endpoint appears only inside gitea_merge_pr; gitea_review_pr fails closed on merge=True before any API call; gitea_submit_pr_review has no merge parameter and 'merge' is not a reviewable action. Tests assert all three. Tests cover: merge succeeds only when all gates pass; self-author blocked; unknown identity/profile blocked; profile without merge permission blocked; missing/wrong confirmation blocked (no API call); head-SHA mismatch blocked; changed-files mismatch blocked; closed PR blocked; non-mergeable blocked; unknown mergeability fail-closed; no merge call when gates fail; invalid merge method rejected; output and error redaction; and the no-ungated-merge-path audit. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 2 +- mcp_server.py | 182 ++++++++++++++++++++--- tests/test_mcp_server.py | 305 ++++++++++++++++++++++++++++++++++++++- 3 files changed, 462 insertions(+), 27 deletions(-) diff --git a/README.md b/README.md index c124ee8..352a07c 100644 --- a/README.md +++ b/README.md @@ -44,7 +44,7 @@ Any MCP-compatible agent (Antigravity, Claude Code, etc.) can call these tools n | `gitea_edit_pr` | Edit details of an existing pull request | | `gitea_list_prs` | List pull requests with state/remote | | `gitea_view_pr` | Get full details of a single pull request | -| `gitea_merge_pr` | Merge a pull request (merge, squash, or rebase) | +| `gitea_merge_pr` | Gated merge: merge/squash/rebase only after identity+profile+eligibility gates pass, explicit `confirmation="MERGE PR "`, optional head-SHA and changed-files pinning (no self-merge, no force) | | `gitea_review_pr` | Legacy wrapper for `gitea_submit_pr_review` (merging disabled) | | `gitea_delete_branch` | Delete a remote branch | | `gitea_close_issue` | Close an issue by number | diff --git a/mcp_server.py b/mcp_server.py index 6ba1a81..3810c3f 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -671,47 +671,193 @@ def gitea_commit_files( } +# Merge methods supported by the Gitea merge API. +_MERGE_METHODS = ("merge", "squash", "rebase") + + @mcp.tool() def gitea_merge_pr( pr_number: int, + confirmation: str = "", + expected_head_sha: str | None = None, + expected_changed_files: list[str] | None = None, do: str = "merge", title: str | None = None, message: str | None = None, - force: bool = False, remote: str = "dadeschools", host: str | None = None, org: str | None = None, repo: str | None = None, ) -> dict: - """Merge a Gitea pull request. + """Gated merge of a Gitea pull request (#16). + + This is the ONLY merge path this server exposes, and it mutates only after + every safety gate passes. No ungated merge tool remains: legacy + ``gitea_review_pr`` fails closed on ``merge=True`` and + ``gitea_submit_pr_review`` never merges. + + Gate order (fail-closed at each step; the merge API is called only if all + gates pass): + + 1. Merge method (``do``) is 'merge', 'squash', or 'rebase'. + 2. Explicit confirmation: ``confirmation`` must equal ``"MERGE PR "``. + Without it, the tool makes no API calls at all. + 3. Reuse ``gitea_check_pr_eligibility`` (#14) with action 'merge': this + proves the authenticated identity, the active profile (and that it + allows merge), the PR author, blocks self-merge, requires the PR to be + open, and fails closed when the PR is not mergeable or mergeability is + unknown. + 4. If ``expected_head_sha`` is given and the PR head moved → refuse. + 5. If ``expected_changed_files`` is given and the PR's changed file set + differs → refuse. + 6. Redundant self-merge block (authenticated user == PR author). + + No force / ignore-checks option is exposed. Gitea's own ``mergeable`` signal + (which reflects branch-protection required reviews and status checks) must + be positive, so required approval/check state is honoured, never bypassed. + + Never returns the token, Authorization header, or any credential material. Args: pr_number: The PR number to merge. + confirmation: Must be exactly ``"MERGE PR "`` or merge is refused. + expected_head_sha: Strongly recommended. If set and the PR head differs, refuse. + expected_changed_files: Optional. If set and the PR's changed file set + differs, refuse. do: Merge style — 'merge', 'squash', or 'rebase'. - title: Optional merge title. - message: Optional merge message. - force: Force merge, ignoring status checks. + title: Optional merge commit title. + message: Optional merge commit message. remote: Known instance — 'dadeschools' or 'prgs'. host: Override the Gitea host. org: Override the owner/organization. repo: Override the repository name. Returns: - dict with 'success' and 'message'. + dict describing the attempt: performed, authenticated user, profile + name, PR author, PR number, head SHA checked, merge method, + reasons/gates passed or blocked, and merge result / merge commit if + available. Never secrets. """ - h, o, r = _resolve(remote, host, org, repo) - auth = _auth(h) - url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}/merge" - payload = { - "Do": do, - "force_merge": force, + do = (do or "").strip().lower() + result = { + "performed": False, + "authenticated_user": None, + "profile_name": get_profile()["profile_name"], + "pr_author": None, + "pr_number": pr_number, + "head_sha": None, + "expected_head_sha": expected_head_sha, + "merge_method": do, + "mergeable": None, + "remote": remote if remote in REMOTES else None, + "merge_result": None, + "merge_commit": None, + "reasons": [], } - if title: - payload["MergeTitleField"] = title - if message: - payload["MergeMessageField"] = message - api_request("POST", url, auth, payload) - return {"success": True, "message": f"PR #{pr_number} merged via '{do}'."} + reasons = result["reasons"] + + # Gate 1 — valid merge method (no API call on a bad method). + if do not in _MERGE_METHODS: + reasons.append( + f"unknown merge method '{do}'; expected one of {list(_MERGE_METHODS)}" + ) + return result + + # Gate 2 — explicit confirmation (fail fast; zero API calls without it). + expected_confirmation = f"MERGE PR {pr_number}" + if (confirmation or "").strip() != expected_confirmation: + reasons.append( + f"explicit confirmation required: pass confirmation='{expected_confirmation}'" + ) + return result + + # Gate 3 — reuse #14 eligibility (identity + profile + merge-allowed + + # author + self-merge block + open + mergeable/unknown fail-closed). + # Read-only GETs only. + elig = gitea_check_pr_eligibility( + pr_number=pr_number, action="merge", remote=remote, + host=host, org=org, repo=repo, + ) + result["authenticated_user"] = elig.get("authenticated_user") + result["profile_name"] = elig.get("profile_name", result["profile_name"]) + result["pr_author"] = elig.get("pr_author") + result["head_sha"] = elig.get("head_sha") + result["mergeable"] = elig.get("mergeable") + if not elig.get("eligible"): + reasons.append("eligibility check for 'merge' failed (fail closed)") + reasons.extend(elig.get("reasons", [])) + return result + + # Gate 4 — head SHA must match if the caller pinned a reviewed SHA. + actual_sha = result["head_sha"] + if expected_head_sha and actual_sha and expected_head_sha != actual_sha: + reasons.append( + "expected head SHA does not match current PR head (fail closed)" + ) + return result + if not actual_sha: + # Unreachable — eligibility fails closed without a head SHA — but never + # merge a PR whose head commit we could not read. + reasons.append("PR head SHA unavailable (fail closed)") + return result + + h, o, r = _resolve(remote, host, org, repo) + + # Gate 5 — changed files must match the reviewed set, if provided. + if expected_changed_files is not None: + try: + auth = _auth(h) + files = api_request( + "GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}/files", auth + ) + actual_files = sorted( + (f or {}).get("filename", "") for f in (files or []) + ) + except Exception as exc: # noqa: BLE001 — redact before surfacing + reasons.append( + f"could not verify changed files (fail closed): {_redact(str(exc))}" + ) + return result + result["changed_files"] = actual_files + if actual_files != sorted(expected_changed_files): + reasons.append( + "PR changed files do not match expected_changed_files (fail closed)" + ) + return result + + # Gate 6 — redundant self-merge block (belt-and-suspenders over #14). + auth_user = result["authenticated_user"] + pr_author = result["pr_author"] + if auth_user and pr_author and auth_user == pr_author: + reasons.append("self-merge blocked (authenticated user is PR author)") + return result + + # All gates passed — perform the single merge mutation. + try: + auth = _auth(h) + merge_url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}/merge" + payload = {"Do": do} + if title: + payload["MergeTitleField"] = title + if message: + payload["MergeMessageField"] = message + api_request("POST", merge_url, auth, payload) + # Best-effort read-back of the merge commit SHA (redacted on error). + try: + merged = api_request( + "GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth + ) + result["merge_commit"] = (merged or {}).get("merged_commit_sha") + except Exception: + result["merge_commit"] = None + except Exception as exc: # noqa: BLE001 — redact before surfacing + reasons.append(f"merge failed: {_redact(str(exc))}") + return result + + result["performed"] = True + result["merge_result"] = f"PR #{pr_number} merged via '{do}'." + reasons.append(f"all gates passed; merged PR #{pr_number} via '{do}'") + return result @mcp.tool() diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 3f86942..c7440c8 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -298,20 +298,309 @@ class TestViewPR(unittest.TestCase): # Merge PR # --------------------------------------------------------------------------- class TestMergePR(unittest.TestCase): + """Gated merge workflow (#16). gitea_merge_pr is the only merge path.""" + + def _pr(self, author, state="open", sha="abc123", mergeable=True): + return { + "user": {"login": author}, + "state": state, + "head": {"sha": sha}, + "mergeable": mergeable, + } + + def _confirm(self, n): + return f"MERGE PR {n}" + + def _assert_no_merge_call(self, mock_api): + for c in mock_api.call_args_list: + method, url = c.args[0], c.args[1] + self.assertFalse( + method == "POST" and url.endswith("/merge"), + f"unexpected merge mutation: {method} {url}", + ) + + # -- success -------------------------------------------------------------- @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_merge_pr(self, _auth, mock_api): - mock_api.return_value = {} - result = gitea_merge_pr(pr_number=1, do="squash", title="T", message="M", force=True) - self.assertTrue(result["success"]) - self.assertIn("merged", result["message"]) - # Check payload - payload = mock_api.call_args[0][3] + def test_merge_succeeds_when_all_gates_pass(self, _auth, mock_api): + mock_api.side_effect = [ + {"login": "merger-bot"}, self._pr("author-bot"), + {}, # merge POST + {"merged_commit_sha": "mergecommit99"}, # read-back + ] + env = {"GITEA_PROFILE_NAME": "gitea-merger", + "GITEA_ALLOWED_OPERATIONS": "read,merge"} + with patch.dict(os.environ, env, clear=True): + r = gitea_merge_pr( + pr_number=8, confirmation=self._confirm(8), + expected_head_sha="abc123", do="squash", + title="T", message="M", remote="prgs") + self.assertTrue(r["performed"]) + self.assertEqual(r["authenticated_user"], "merger-bot") + self.assertEqual(r["pr_author"], "author-bot") + self.assertEqual(r["head_sha"], "abc123") + self.assertEqual(r["merge_method"], "squash") + self.assertEqual(r["merge_commit"], "mergecommit99") + # 3rd call is the merge POST with the requested method/title/message. + merge_call = mock_api.call_args_list[2] + self.assertEqual(merge_call.args[0], "POST") + self.assertTrue(merge_call.args[1].endswith("/pulls/8/merge")) + payload = merge_call.args[3] self.assertEqual(payload["Do"], "squash") self.assertEqual(payload["MergeTitleField"], "T") self.assertEqual(payload["MergeMessageField"], "M") - self.assertEqual(payload["force_merge"], True) + self.assertNotIn("force_merge", payload) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_expected_changed_files_match_allows(self, _auth, mock_api): + mock_api.side_effect = [ + {"login": "merger-bot"}, self._pr("author-bot"), + [{"filename": "a.py"}, {"filename": "b.py"}], # files + {}, # merge POST + {"merged_commit_sha": "c1"}, # read-back + ] + env = {"GITEA_PROFILE_NAME": "gitea-merger", + "GITEA_ALLOWED_OPERATIONS": "read,merge"} + with patch.dict(os.environ, env, clear=True): + r = gitea_merge_pr( + pr_number=8, confirmation=self._confirm(8), + expected_changed_files=["b.py", "a.py"], remote="prgs") + self.assertTrue(r["performed"]) + + # -- confirmation --------------------------------------------------------- + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_missing_confirmation_blocks_with_no_api_call(self, _auth, mock_api): + env = {"GITEA_PROFILE_NAME": "gitea-merger", + "GITEA_ALLOWED_OPERATIONS": "read,merge"} + with patch.dict(os.environ, env, clear=True): + r = gitea_merge_pr(pr_number=8, confirmation="", remote="prgs") + self.assertFalse(r["performed"]) + self.assertTrue(any("explicit confirmation required" in x for x in r["reasons"])) + mock_api.assert_not_called() + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_wrong_confirmation_blocks(self, _auth, mock_api): + env = {"GITEA_PROFILE_NAME": "gitea-merger", + "GITEA_ALLOWED_OPERATIONS": "read,merge"} + with patch.dict(os.environ, env, clear=True): + r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 9", remote="prgs") + self.assertFalse(r["performed"]) + mock_api.assert_not_called() + + # -- identity / profile / eligibility fail-closed ------------------------- + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_self_author_cannot_merge(self, _auth, mock_api): + mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")] + env = {"GITEA_PROFILE_NAME": "gitea-merger", + "GITEA_ALLOWED_OPERATIONS": "read,merge"} + with patch.dict(os.environ, env, clear=True): + r = gitea_merge_pr( + pr_number=8, confirmation=self._confirm(8), remote="prgs") + self.assertFalse(r["performed"]) + self.assertIn("authenticated user is PR author", r["reasons"]) + self._assert_no_merge_call(mock_api) + + @patch("mcp_server.get_auth_header", return_value=None) + def test_unknown_identity_blocks(self, _auth): + env = {"GITEA_PROFILE_NAME": "gitea-merger", + "GITEA_ALLOWED_OPERATIONS": "read,merge"} + with patch.dict(os.environ, env, clear=True): + r = gitea_merge_pr( + pr_number=8, confirmation=self._confirm(8), remote="prgs") + self.assertFalse(r["performed"]) + self.assertIsNone(r["authenticated_user"]) + self.assertIn("authenticated identity could not be determined", r["reasons"]) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_unknown_profile_blocks(self, _auth, mock_api): + mock_api.side_effect = [{"login": "merger-bot"}, self._pr("author-bot")] + env = {} # no GITEA_ALLOWED_OPERATIONS → empty allowed ops + with patch.dict(os.environ, env, clear=True): + r = gitea_merge_pr( + pr_number=8, confirmation=self._confirm(8), remote="prgs") + self.assertFalse(r["performed"]) + self.assertIn( + "profile has no configured allowed operations (fail closed)", + r["reasons"]) + self._assert_no_merge_call(mock_api) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_profile_without_merge_permission_blocks(self, _auth, mock_api): + mock_api.side_effect = [{"login": "merger-bot"}, self._pr("author-bot")] + env = {"GITEA_PROFILE_NAME": "gitea-reviewer", + "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} + with patch.dict(os.environ, env, clear=True): + r = gitea_merge_pr( + pr_number=8, confirmation=self._confirm(8), remote="prgs") + self.assertFalse(r["performed"]) + self.assertIn("profile is not allowed to merge", r["reasons"]) + self._assert_no_merge_call(mock_api) + + # -- PR state / mergeability ---------------------------------------------- + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_closed_pr_blocks(self, _auth, mock_api): + mock_api.side_effect = [ + {"login": "merger-bot"}, self._pr("author-bot", state="closed")] + env = {"GITEA_PROFILE_NAME": "gitea-merger", + "GITEA_ALLOWED_OPERATIONS": "read,merge"} + with patch.dict(os.environ, env, clear=True): + r = gitea_merge_pr( + pr_number=8, confirmation=self._confirm(8), remote="prgs") + self.assertFalse(r["performed"]) + self.assertIn("PR is not open (state=closed)", r["reasons"]) + self._assert_no_merge_call(mock_api) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_non_mergeable_pr_blocks(self, _auth, mock_api): + mock_api.side_effect = [ + {"login": "merger-bot"}, self._pr("author-bot", mergeable=False)] + env = {"GITEA_PROFILE_NAME": "gitea-merger", + "GITEA_ALLOWED_OPERATIONS": "read,merge"} + with patch.dict(os.environ, env, clear=True): + r = gitea_merge_pr( + pr_number=8, confirmation=self._confirm(8), remote="prgs") + self.assertFalse(r["performed"]) + self.assertIn("PR is not mergeable", r["reasons"]) + self._assert_no_merge_call(mock_api) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_unknown_mergeability_fails_closed(self, _auth, mock_api): + mock_api.side_effect = [ + {"login": "merger-bot"}, self._pr("author-bot", mergeable=None)] + env = {"GITEA_PROFILE_NAME": "gitea-merger", + "GITEA_ALLOWED_OPERATIONS": "read,merge"} + with patch.dict(os.environ, env, clear=True): + r = gitea_merge_pr( + pr_number=8, confirmation=self._confirm(8), remote="prgs") + self.assertFalse(r["performed"]) + self.assertIn("PR mergeability unknown", r["reasons"]) + self._assert_no_merge_call(mock_api) + + # -- head SHA / changed files --------------------------------------------- + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_head_sha_mismatch_blocks(self, _auth, mock_api): + mock_api.side_effect = [ + {"login": "merger-bot"}, self._pr("author-bot", sha="abc123")] + env = {"GITEA_PROFILE_NAME": "gitea-merger", + "GITEA_ALLOWED_OPERATIONS": "read,merge"} + with patch.dict(os.environ, env, clear=True): + r = gitea_merge_pr( + pr_number=8, confirmation=self._confirm(8), + expected_head_sha="deadbeef", remote="prgs") + self.assertFalse(r["performed"]) + self.assertIn( + "expected head SHA does not match current PR head (fail closed)", + r["reasons"]) + self._assert_no_merge_call(mock_api) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_changed_files_mismatch_blocks(self, _auth, mock_api): + mock_api.side_effect = [ + {"login": "merger-bot"}, self._pr("author-bot"), + [{"filename": "a.py"}, {"filename": "c.py"}], # actual files + ] + env = {"GITEA_PROFILE_NAME": "gitea-merger", + "GITEA_ALLOWED_OPERATIONS": "read,merge"} + with patch.dict(os.environ, env, clear=True): + r = gitea_merge_pr( + pr_number=8, confirmation=self._confirm(8), + expected_changed_files=["a.py", "b.py"], remote="prgs") + self.assertFalse(r["performed"]) + self.assertIn( + "PR changed files do not match expected_changed_files (fail closed)", + r["reasons"]) + self._assert_no_merge_call(mock_api) + + # -- misc ----------------------------------------------------------------- + + @patch("mcp_server.api_request") + def test_invalid_merge_method_rejected(self, mock_api): + with patch.dict(os.environ, {}, clear=True): + r = gitea_merge_pr( + pr_number=8, confirmation="MERGE PR 8", do="octopus", remote="prgs") + self.assertFalse(r["performed"]) + self.assertTrue(any("unknown merge method" in x for x in r["reasons"])) + mock_api.assert_not_called() + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_output_redacts_secrets(self, _auth, mock_api): + mock_api.side_effect = [ + {"login": "merger-bot"}, self._pr("author-bot"), + {}, {"merged_commit_sha": "c1"}, + ] + env = {"GITEA_PROFILE_NAME": "gitea-merger", + "GITEA_ALLOWED_OPERATIONS": "read,merge", + "GITEA_TOKEN": "super-secret-token"} + with patch.dict(os.environ, env, clear=True): + r = gitea_merge_pr( + pr_number=8, confirmation=self._confirm(8), remote="prgs") + blob = repr(r).lower() + for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()): + self.assertNotIn(secret, blob) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_merge_error_message_redacts_credential(self, _auth, mock_api): + mock_api.side_effect = [ + {"login": "merger-bot"}, self._pr("author-bot"), + RuntimeError("HTTP 500: token abc-secret-xyz rejected"), + ] + env = {"GITEA_PROFILE_NAME": "gitea-merger", + "GITEA_ALLOWED_OPERATIONS": "read,merge"} + with patch.dict(os.environ, env, clear=True): + r = gitea_merge_pr( + pr_number=8, confirmation=self._confirm(8), remote="prgs") + self.assertFalse(r["performed"]) + blob = repr(r) + self.assertIn("[REDACTED]", blob) + self.assertNotIn("abc-secret-xyz", blob) + + +class TestNoUngatedMergePath(unittest.TestCase): + """Prove no other exposed tool can merge (#16 surface audit).""" + + def test_submit_pr_review_has_no_merge(self): + import inspect + import mcp_server + # No merge parameter on the review-mutation tool. + params = inspect.signature(mcp_server.gitea_submit_pr_review).parameters + self.assertNotIn("merge", params) + # And 'merge' is not a reviewable action. + self.assertNotIn("merge", mcp_server._REVIEW_ACTIONS) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_review_pr_merge_true_makes_no_api_call(self, _auth, mock_api): + r = gitea_review_pr(pr_number=1, event="APPROVE", body="x", merge=True) + self.assertFalse(r["success"]) + self.assertEqual(mock_api.call_count, 0) + + def test_only_merge_endpoint_is_in_gated_tool(self): + # Source-level audit: the merge endpoint appears only inside + # gitea_merge_pr, which is the gated path. + import inspect + import mcp_server + src = inspect.getsource(mcp_server) + self.assertEqual(src.count("/merge\""), 1) + merge_src = inspect.getsource(mcp_server.gitea_merge_pr) + self.assertIn("/merge\"", merge_src) # ---------------------------------------------------------------------------