diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index f4fb0aa..810e546 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -1282,12 +1282,18 @@ def _verify_role_mutation_workspace( worktree_path: str | None = None, worktree: str | None = None, task: str | None = None, + org: str | None = None, + repo: str | None = None, ) -> str: """Bind reviewer/merger mutations to the active namespace workspace (#510). #683: must NOT early-return solely because pytest/unittest is loaded. Production workspace binding always runs; test isolation uses explicit env fixtures / force-on flags, never a production short-circuit here. + + org/repo are forwarded into anti-stomp (#604) so callers that pass explicit + repository targets (e.g. prgs + Gitea-Tools when REMOTES defaults to + Timesheet) do not fail closed with wrong_repo after a successful resolve. """ # Check running runtimes to prevent stale mutations @@ -1350,7 +1356,13 @@ def _verify_role_mutation_workspace( ) ) resolved = assessment["mutation_workspace"] - verify_preflight_purity(remote, worktree_path=resolved, task=task) + verify_preflight_purity( + remote, + worktree_path=resolved, + task=task, + org=org, + repo=repo, + ) return resolved @@ -4521,7 +4533,11 @@ def _evaluate_pr_review_submission( ) -> dict: """Shared gate chain for live submit and dry-run review tools.""" _verify_role_mutation_workspace( - remote, worktree_path=worktree_path, task="review_pr" + remote, + worktree_path=worktree_path, + task="review_pr", + org=org, + repo=repo, ) action = (action or "").strip().lower() workflow_blockers = _review_workflow_load_gate_reasons() if live else [] @@ -7619,7 +7635,11 @@ def gitea_merge_pr( available. Never secrets. """ _verify_role_mutation_workspace( - remote, worktree_path=worktree_path, task="merge_pr" + remote, + worktree_path=worktree_path, + task="merge_pr", + org=org, + repo=repo, ) allowed = get_profile()["allowed_operations"] forbidden = get_profile()["forbidden_operations"] @@ -10652,8 +10672,14 @@ def gitea_acquire_reviewer_pr_lease( # task=acquire_reviewer_pr_lease so verify_preflight_purity runs shared #604 # anti-stomp for the declared lease-acquire mutation inventory entry. + # Forward explicit org/repo so anti-stomp does not default bare remote=prgs + # to Timesheet when the local git remote is Gitea-Tools (#604 wrong_repo). _verify_role_mutation_workspace( - remote, worktree=worktree, task="acquire_reviewer_pr_lease" + remote, + worktree=worktree, + task="acquire_reviewer_pr_lease", + org=org, + repo=repo, ) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) @@ -10736,6 +10762,200 @@ def gitea_acquire_reviewer_pr_lease( } +@mcp.tool() +def gitea_acquire_merger_pr_lease( + pr_number: int, + worktree: str, + candidate_head: str | None = None, + target_branch: str = "master", + target_branch_sha: str | None = None, + issue_number: int | None = None, + session_id: str | None = None, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Acquire a per-PR lease for a merger session when no reviewer lease exists (#718). + + Merger sessions should normally adopt an active reviewer lease using + gitea_adopt_merger_pr_lease. However, if no active reviewer lease exists + or it has expired, a merger can acquire one directly using this tool. + """ + read_block = _profile_operation_gate("gitea.read") + if read_block: + return { + "success": False, + "acquired": False, + "reasons": read_block, + "permission_report": _permission_block_report("gitea.read"), + } + comment_block = _profile_operation_gate("gitea.pr.comment") + if comment_block: + return { + "success": False, + "acquired": False, + "reasons": comment_block, + "permission_report": _permission_block_report("gitea.pr.comment"), + } + merge_block = _profile_operation_gate("gitea.pr.merge") + if merge_block: + return { + "success": False, + "acquired": False, + "reasons": merge_block, + "permission_report": _permission_block_report("gitea.pr.merge"), + } + + head_pin = (candidate_head or "").strip() + if not head_pin: + return { + "success": False, + "acquired": False, + "reasons": [ + "candidate_head is required for merger lease acquisition " + "(exact-head scoping; fail closed)" + ], + } + + try: + _verify_role_mutation_workspace( + remote, + worktree=worktree, + task="acquire_merger_pr_lease", + org=org, + repo=repo, + ) + except RuntimeError as e: + return { + "success": False, + "acquired": False, + "reasons": [str(e)], + } + + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + profile = get_profile() + identity = _authenticated_username(h) or profile.get("username") or "" + sid = (session_id or "").strip() or reviewer_pr_lease.new_session_id() + repo_label = f"{o}/{r}" + + comments = _fetch_pr_comments( + pr_number, remote=remote, host=host, org=org, repo=repo) + + pr_live = api_request( + "GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {} + live_head = ( + (pr_live.get("head") or {}).get("sha") + or pr_live.get("head_sha") + or "" + ) + live_head = str(live_head).strip() + # Fail closed when live PR head cannot be resolved — never proceed with + # an unpinned/unknown head even if candidate_head was supplied (#718). + if not live_head: + return { + "success": False, + "acquired": False, + "reasons": [ + "live PR head could not be resolved from GET /pulls response " + "(exact-head scoping; fail closed)" + ], + "live_head": None, + "candidate_head": head_pin, + } + if live_head != head_pin: + return { + "success": False, + "acquired": False, + "reasons": [ + f"candidate_head {head_pin[:12]}… does not match live PR head " + f"{live_head[:12]}… (exact-head scoping; fail closed)" + ], + "live_head": live_head, + "candidate_head": head_pin, + } + + pr_merged_or_closed = bool( + pr_live.get("merged") or pr_live.get("merged_at") + ) or (str(pr_live.get("state") or "").strip().lower() == "closed") + + assessment = reviewer_pr_lease.assess_acquire_lease( + comments, + pr_number=pr_number, + reviewer_identity=identity, + profile=profile.get("profile_name") or "unknown", + session_id=sid, + repo=repo_label, + issue_number=issue_number, + worktree=worktree, + candidate_head=head_pin, + target_branch=target_branch, + target_branch_sha=target_branch_sha, + pr_merged_or_closed=pr_merged_or_closed, + ) + if not assessment.get("acquire_allowed"): + return { + "success": False, + "acquired": False, + "reasons": assessment.get("reasons") or [], + "existing_lease": assessment.get("existing_lease"), + "post_merge_moot": assessment.get("post_merge_moot", False), + } + + body = assessment["lease_body"] + comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments" + with _audited( + "comment_pr", + host=h, + remote=remote, + org=o, + repo=r, + pr_number=pr_number, + request_metadata={"source": "acquire_merger_pr_lease"}, + ): + posted = api_request("POST", comment_url, auth, {"body": body}) + + if not isinstance(posted, dict) or not posted.get("id"): + return { + "success": False, + "acquired": False, + "reasons": [ + "lease comment POST failed or returned no comment id " + "(no partial session lease recorded)" + ], + } + + session_lease = reviewer_pr_lease.record_session_lease({ + "pr_number": pr_number, + "issue_number": issue_number, + "session_id": sid, + "reviewer_identity": identity, + "profile": profile.get("profile_name"), + "worktree": worktree, + "phase": "claimed", + "candidate_head": head_pin, + "target_branch": target_branch, + "target_branch_sha": target_branch_sha, + "repo": repo_label, + "comment_id": posted.get("id"), + }, lease_provenance=merger_lease_adoption.build_lease_provenance( + source=merger_lease_adoption.SOURCE_ACQUIRE_MERGER, + comment_id=posted.get("id"), + )) + return { + "success": True, + "acquired": True, + "pr_number": pr_number, + "session_id": sid, + "comment_id": posted.get("id"), + "session_lease": session_lease, + "candidate_head": head_pin, + "repo": repo_label, + "reasons": [], + } + + @mcp.tool() def gitea_adopt_merger_pr_lease( pr_number: int, @@ -10797,7 +11017,11 @@ def gitea_adopt_merger_pr_lease( # task=adopt_merger_pr_lease so verify_preflight_purity runs shared #604 anti-stomp. _verify_role_mutation_workspace( - remote, worktree=worktree, task="adopt_merger_pr_lease" + remote, + worktree=worktree, + task="adopt_merger_pr_lease", + org=org, + repo=repo, ) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) @@ -11519,7 +11743,11 @@ def gitea_release_reviewer_pr_lease( dict reporting release status. """ _verify_role_mutation_workspace( - remote, worktree=worktree, task="review_pr" + remote, + worktree=worktree, + task="review_pr", + org=org, + repo=repo, ) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) @@ -14631,11 +14859,56 @@ def gitea_resolve_task_capability( TASK_MAP = task_capability_map.TASK_CAPABILITY_MAP if task not in TASK_MAP: - raise ValueError(f"Unknown task/action: '{task}' (fail closed)") + # #723: structured fail-closed unknown_task (never raise into internal_error). + profile = get_profile() + h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None) + username = _authenticated_username(h) if h else None + active_role_kind = _profile_role_kind(profile) + switching = gitea_config.is_runtime_switching_enabled() + result = { + "requested_task": task, + "required_operation_permission": "unknown", + "required_role_kind": "unknown", + "active_profile": profile.get("profile_name", "unknown"), + "active_identity": username, + "active_role_kind": active_role_kind, + "active_profile_allowed_operations": profile.get("allowed_operations") or [], + "active_profile_permission_allowed": False, + "allowed_in_current_session": False, + "available_in_session": False, + "configured": False, + "restart_required": False, + "stop_required": True, + "task_role_guidance": [ + f"STOP: Unknown task/action: '{task}' (fail closed)" + ], + "matching_configured_profile": [], + "runtime_switching_supported": switching, + "different_mcp_namespace_required": False, + "exact_safe_next_action": ( + f"None; task '{task}' is unrecognized by the capability map." + ), + "reason": f"Unknown task/action: '{task}' (fail closed)", + "reason_code": "unknown_task", + "mutation_performed": False, + } + role_session_router.sync_route_from_capability(result) + was_terminal = capability_stop_terminal.is_active() + terminal = capability_stop_terminal.sync_from_capability_result(result) + if terminal: + result["terminal_mode"] = True + result["terminal_report"] = ( + capability_stop_terminal.build_terminal_report(result) + ) + elif was_terminal: + result["cleared_stale_denial"] = True + return result required_permission = task_capability_map.required_permission(task) required_role = task_capability_map.required_role(task) role_exclusive_tasks = { + "acquire_reviewer_pr_lease", + "gitea_acquire_reviewer_pr_lease", "review_pr", "approve_pr", "request_changes_pr", @@ -14643,6 +14916,10 @@ def gitea_resolve_task_capability( "pr_queue_cleanup", "pr-queue-cleanup", "merge_pr", + "acquire_merger_pr_lease", + "gitea_acquire_merger_pr_lease", + "adopt_merger_pr_lease", + "gitea_adopt_merger_pr_lease", "create_branch", "push_branch", "create_pr", diff --git a/merger_lease_adoption.py b/merger_lease_adoption.py index 821779a..46ae926 100644 --- a/merger_lease_adoption.py +++ b/merger_lease_adoption.py @@ -18,11 +18,13 @@ DEFAULT_ADOPTION_REASON = "merger-handoff-approved-head" SOURCE_ADOPT = "gitea_adopt_merger_pr_lease" SOURCE_ACQUIRE = "gitea_acquire_reviewer_pr_lease" +SOURCE_ACQUIRE_MERGER = "gitea_acquire_merger_pr_lease" SOURCE_HEARTBEAT = "gitea_heartbeat_reviewer_pr_lease" SANCTIONED_PROVENANCE_SOURCES = frozenset({ SOURCE_ADOPT, SOURCE_ACQUIRE, + SOURCE_ACQUIRE_MERGER, SOURCE_HEARTBEAT, }) @@ -209,8 +211,10 @@ _SANCTIONED_LEASE_EVIDENCE_RE = re.compile( r"(?is)\b(" r"gitea_adopt_merger_pr_lease|" r"gitea_acquire_reviewer_pr_lease|" + r"gitea_acquire_merger_pr_lease|" r"lease_proof_source\s*[:=]\s*gitea_adopt_merger_pr_lease|" r"lease_proof_source\s*[:=]\s*gitea_acquire_reviewer_pr_lease|" + r"lease_proof_source\s*[:=]\s*gitea_acquire_merger_pr_lease|" r"lease_proof_kind\s*[:=]\s*sanctioned_adoption|" r"lease_proof_kind\s*[:=]\s*sanctioned_acquire|" r"adoption_comment_id\s*[:=]|" diff --git a/task_capability_map.py b/task_capability_map.py index 6fc01ab..4e4197c 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -73,6 +73,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.branch.push", "role": "author", }, + "acquire_reviewer_pr_lease": { + "permission": "gitea.pr.comment", + "role": "reviewer", + }, + "gitea_acquire_reviewer_pr_lease": { + "permission": "gitea.pr.comment", + "role": "reviewer", + }, "review_pr": { "permission": "gitea.pr.review", "role": "reviewer", @@ -107,6 +115,18 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.pr.comment", "role": "merger", }, + "gitea_adopt_merger_pr_lease": { + "permission": "gitea.pr.comment", + "role": "merger", + }, + "acquire_merger_pr_lease": { + "permission": "gitea.pr.comment", + "role": "merger", + }, + "gitea_acquire_merger_pr_lease": { + "permission": "gitea.pr.comment", + "role": "merger", + }, # #691: guarded non-owner cleanup of obsolete comment-backed reviewer leases. # Apply path posts lease release + audit comments (gitea.pr.comment). "cleanup_obsolete_reviewer_comment_lease": { diff --git a/tests/test_merger_lease_acquire.py b/tests/test_merger_lease_acquire.py new file mode 100644 index 0000000..77bc118 --- /dev/null +++ b/tests/test_merger_lease_acquire.py @@ -0,0 +1,466 @@ +"""Merger lease acquisition + capability resolution tests (#718, #723).""" + +from __future__ import annotations + +import os +import sys +import unittest +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import gitea_mcp_server as mcp_server +import reviewer_pr_lease as leases +import task_capability_map as tcm + + +HEAD = "a" * 40 +LIVE_HEAD = HEAD + + +class TestMergerLeaseAcquireTool(unittest.TestCase): + def setUp(self): + mcp_server._preflight_whoami_called = True + mcp_server._preflight_capability_called = True + mcp_server._preflight_resolved_role = "merger" + mcp_server._preflight_resolved_task = "acquire_merger_pr_lease" + leases.clear_session_lease() + + def tearDown(self): + leases.clear_session_lease() + + def _merger_profile(self, **extra): + p = { + "username": "merger-user", + "profile_name": "prgs-merger", + "role": "merger", + "allowed_operations": [ + "gitea.read", + "gitea.pr.comment", + "gitea.pr.merge", + ], + "forbidden_operations": [ + "gitea.pr.approve", + "gitea.pr.review", + "gitea.pr.request_changes", + ], + } + p.update(extra) + return p + + @patch("gitea_mcp_server.api_request") + @patch("gitea_mcp_server._auth", return_value="token pass") + @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")) + @patch("gitea_mcp_server.get_profile") + @patch("gitea_mcp_server._fetch_pr_comments", return_value=[]) + @patch("gitea_mcp_server._verify_role_mutation_workspace") + def test_acquire_merger_lease_success( + self, _verify, _comments, mock_profile, _resolve, _auth, mock_api + ): + mock_profile.return_value = self._merger_profile() + mock_api.side_effect = [ + {"state": "open", "head": {"sha": LIVE_HEAD}}, + {"id": 456}, + ] + with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}): + res = mcp_server.gitea_acquire_merger_pr_lease( + pr_number=718, + worktree="branches/issue-718", + session_id="merger-session", + candidate_head=HEAD, + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + ) + self.assertTrue(res["success"], res) + self.assertTrue(res["acquired"]) + self.assertEqual(res["comment_id"], 456) + self.assertEqual(res["session_id"], "merger-session") + self.assertEqual(res.get("repo"), "Scaled-Tech-Consulting/Gitea-Tools") + body = mock_api.call_args_list[-1][0][3]["body"] + self.assertIn("reviewer_identity: merger-user", body) + self.assertIn("profile: prgs-merger", body) + session_lease = leases._SESSION_LEASE + self.assertIsNotNone(session_lease) + provenance = session_lease.get("lease_provenance") or {} + self.assertEqual(provenance.get("source"), "gitea_acquire_merger_pr_lease") + + @patch("gitea_mcp_server.api_request") + @patch("gitea_mcp_server._auth", return_value="token pass") + @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")) + @patch("gitea_mcp_server.get_profile") + @patch("gitea_mcp_server._fetch_pr_comments") + @patch("gitea_mcp_server._verify_role_mutation_workspace") + def test_acquire_merger_lease_fails_if_active_exists( + self, _verify, _comments, mock_profile, _resolve, _auth, mock_api + ): + mock_profile.return_value = self._merger_profile() + active_body = leases.format_lease_body( + repo="Scaled-Tech-Consulting/Gitea-Tools", + pr_number=718, + issue_number=None, + reviewer_identity="other-user", + profile="prgs-reviewer", + session_id="other-session", + worktree="branches/review-1", + phase="claimed", + candidate_head=HEAD, + target_branch="master", + target_branch_sha="b" * 40, + last_activity=datetime.now(timezone.utc), + ) + _comments.return_value = [ + {"id": 1, "body": active_body, "user": {"login": "other-user"}} + ] + mock_api.return_value = {"state": "open", "head": {"sha": LIVE_HEAD}} + with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}): + res = mcp_server.gitea_acquire_merger_pr_lease( + pr_number=718, + worktree="branches/issue-718", + session_id="merger-session", + candidate_head=HEAD, + ) + self.assertFalse(res["success"]) + self.assertFalse(res["acquired"]) + self.assertIn("already has active reviewer lease", " ".join(res["reasons"])) + self.assertEqual( + [c for c in mock_api.call_args_list if c[0][0] == "POST"], [] + ) + self.assertIsNone(leases._SESSION_LEASE) + + @patch("gitea_mcp_server.get_profile") + @patch("gitea_mcp_server._verify_role_mutation_workspace") + def test_acquire_merger_lease_fails_workspace_binding(self, _verify, mock_profile): + mock_profile.return_value = self._merger_profile() + _verify.side_effect = RuntimeError("Branches-only mutation guard") + with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}): + res = mcp_server.gitea_acquire_merger_pr_lease( + pr_number=718, + worktree="branches/issue-718", + session_id="merger-session", + candidate_head=HEAD, + ) + self.assertFalse(res["success"]) + self.assertIn("Branches-only mutation guard", res["reasons"][0]) + self.assertIsNone(leases._SESSION_LEASE) + + @patch("gitea_mcp_server.api_request") + @patch("gitea_mcp_server._auth", return_value="token pass") + @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")) + @patch("gitea_mcp_server.get_profile") + @patch("gitea_mcp_server._fetch_pr_comments", return_value=[]) + @patch("gitea_mcp_server._verify_role_mutation_workspace") + def test_acquire_merger_forwards_explicit_org_repo_to_workspace_verify( + self, mock_verify, _comments, mock_profile, _resolve, _auth, mock_api + ): + """Explicit org/repo must reach anti-stomp (prgs bare default is Timesheet).""" + mock_profile.return_value = self._merger_profile() + mock_api.side_effect = [ + {"state": "open", "head": {"sha": LIVE_HEAD}}, + {"id": 789}, + ] + with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}): + res = mcp_server.gitea_acquire_merger_pr_lease( + pr_number=718, + worktree="branches/issue-718", + session_id="merger-session", + candidate_head=HEAD, + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + ) + self.assertTrue(res["success"], res) + kwargs = mock_verify.call_args.kwargs + self.assertEqual(kwargs.get("org"), "Scaled-Tech-Consulting") + self.assertEqual(kwargs.get("repo"), "Gitea-Tools") + self.assertEqual(kwargs.get("task"), "acquire_merger_pr_lease") + + @patch("gitea_mcp_server.get_profile") + def test_acquire_merger_requires_candidate_head(self, mock_profile): + mock_profile.return_value = self._merger_profile() + res = mcp_server.gitea_acquire_merger_pr_lease( + pr_number=718, + worktree="branches/issue-718", + candidate_head=None, + ) + self.assertFalse(res["success"]) + self.assertTrue(any("candidate_head is required" in r for r in res["reasons"])) + + @patch("gitea_mcp_server.api_request") + @patch("gitea_mcp_server._auth", return_value="token pass") + @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")) + @patch("gitea_mcp_server.get_profile") + @patch("gitea_mcp_server._fetch_pr_comments", return_value=[]) + @patch("gitea_mcp_server._verify_role_mutation_workspace") + def test_acquire_merger_head_mismatch( + self, _verify, _comments, mock_profile, _resolve, _auth, mock_api + ): + mock_profile.return_value = self._merger_profile() + mock_api.return_value = {"state": "open", "head": {"sha": "b" * 40}} + res = mcp_server.gitea_acquire_merger_pr_lease( + pr_number=718, + worktree="branches/issue-718", + candidate_head=HEAD, + ) + self.assertFalse(res["success"]) + self.assertTrue(any("does not match live PR head" in r for r in res["reasons"])) + self.assertIsNone(leases._SESSION_LEASE) + + @patch("gitea_mcp_server.api_request") + @patch("gitea_mcp_server._auth", return_value="token pass") + @patch("gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")) + @patch("gitea_mcp_server.get_profile") + @patch("gitea_mcp_server._fetch_pr_comments", return_value=[]) + @patch("gitea_mcp_server._verify_role_mutation_workspace") + def test_acquire_merger_fails_closed_when_live_head_unresolvable( + self, _verify, _comments, mock_profile, _resolve, _auth, mock_api + ): + """Empty/missing live head must not silently proceed past exact-head gate.""" + mock_profile.return_value = self._merger_profile() + # PR payload present but head.sha missing / empty / non-dict. + mock_api.return_value = {"state": "open", "head": {}} + res = mcp_server.gitea_acquire_merger_pr_lease( + pr_number=718, + worktree="branches/issue-718", + session_id="merger-session", + candidate_head=HEAD, + ) + self.assertFalse(res["success"], res) + self.assertFalse(res.get("acquired"), res) + self.assertTrue( + any("live PR head could not be resolved" in r for r in res["reasons"]), + res, + ) + self.assertIsNone(res.get("live_head")) + self.assertEqual(res.get("candidate_head"), HEAD) + self.assertEqual( + [c for c in mock_api.call_args_list if c[0][0] == "POST"], [] + ) + self.assertIsNone(leases._SESSION_LEASE) + + # Completely empty GET /pulls body also fails closed. + mock_api.return_value = {} + res2 = mcp_server.gitea_acquire_merger_pr_lease( + pr_number=718, + worktree="branches/issue-718", + candidate_head=HEAD, + ) + self.assertFalse(res2["success"], res2) + self.assertTrue( + any("live PR head could not be resolved" in r for r in res2["reasons"]), + res2, + ) + self.assertIsNone(leases._SESSION_LEASE) + + @patch("gitea_mcp_server._profile_operation_gate") + def test_author_denied_merge_permission(self, mock_gate): + """Author profile lacks gitea.pr.merge → fail closed before mutation.""" + def gate(op): + if op == "gitea.pr.merge": + return [f"profile lacks {op}"] + return None + mock_gate.side_effect = gate + res = mcp_server.gitea_acquire_merger_pr_lease( + pr_number=718, + worktree="branches/issue-718", + candidate_head=HEAD, + ) + self.assertFalse(res["success"]) + self.assertTrue(any("gitea.pr.merge" in r for r in res["reasons"])) + + @patch("gitea_mcp_server._profile_operation_gate") + def test_reviewer_denied_merge_permission(self, mock_gate): + def gate(op): + if op == "gitea.pr.merge": + return [f"profile lacks {op}"] + return None + mock_gate.side_effect = gate + res = mcp_server.gitea_acquire_merger_pr_lease( + pr_number=718, + worktree="branches/issue-718", + candidate_head=HEAD, + ) + self.assertFalse(res["success"]) + self.assertTrue(any("gitea.pr.merge" in r for r in res["reasons"])) + + +class TestCapabilityMapLeaseTasks(unittest.TestCase): + def test_merger_acquire_map_entries(self): + for task in ("acquire_merger_pr_lease", "gitea_acquire_merger_pr_lease"): + self.assertEqual(tcm.required_permission(task), "gitea.pr.comment") + self.assertEqual(tcm.required_role(task), "merger") + + def test_reviewer_acquire_map_entries(self): + for task in ("acquire_reviewer_pr_lease", "gitea_acquire_reviewer_pr_lease"): + self.assertEqual(tcm.required_permission(task), "gitea.pr.comment") + self.assertEqual(tcm.required_role(task), "reviewer") + + def test_adopt_merger_aliases(self): + for task in ("adopt_merger_pr_lease", "gitea_adopt_merger_pr_lease"): + self.assertEqual(tcm.required_role(task), "merger") + + +class TestResolveTaskCapabilityLeaseAndUnknown(unittest.TestCase): + def setUp(self): + mcp_server._process_boot_head_sha = None + if hasattr(mcp_server, "capability_stop_terminal"): + mcp_server.capability_stop_terminal.clear() + + @patch.object(mcp_server, "record_mutation_authority", return_value=None) + @patch.object(mcp_server, "init_review_decision_lock", return_value=None) + @patch.object(mcp_server, "record_preflight_check", return_value=None) + @patch.object(mcp_server, "_ensure_matching_profile", return_value=None) + @patch.object(mcp_server, "_authenticated_username", return_value="sysadmin") + @patch.object(mcp_server, "get_profile") + @patch.object(mcp_server.gitea_config, "load_config") + def test_resolve_acquire_reviewer_authorized( + self, mock_cfg, mock_profile, *_mocks + ): + mock_profile.return_value = { + "profile_name": "prgs-reviewer", + "role": "reviewer", + "allowed_operations": ["gitea.pr.comment", "gitea.read", "gitea.pr.review"], + "forbidden_operations": [], + } + mock_cfg.return_value = { + "profiles": { + "prgs-reviewer": { + "role": "reviewer", + "allowed_operations": [ + "gitea.pr.comment", + "gitea.read", + "gitea.pr.review", + ], + "forbidden_operations": [], + } + } + } + with patch.dict(os.environ, {"GITEA_MCP_PROFILE": "prgs-reviewer"}, clear=False): + for task in ("acquire_reviewer_pr_lease", "gitea_acquire_reviewer_pr_lease"): + res = mcp_server.gitea_resolve_task_capability(task=task, remote="prgs") + self.assertEqual(res["required_role_kind"], "reviewer", res) + self.assertEqual( + res["required_operation_permission"], "gitea.pr.comment", res + ) + self.assertTrue(res.get("allowed_in_current_session"), res) + + @patch.object(mcp_server, "record_mutation_authority", return_value=None) + @patch.object(mcp_server, "init_review_decision_lock", return_value=None) + @patch.object(mcp_server, "record_preflight_check", return_value=None) + @patch.object(mcp_server, "_ensure_matching_profile", return_value=None) + @patch.object(mcp_server, "_authenticated_username", return_value="jcwalker3") + @patch.object(mcp_server, "get_profile") + @patch.object(mcp_server.gitea_config, "load_config") + def test_resolve_acquire_reviewer_author_denied( + self, mock_cfg, mock_profile, *_mocks + ): + mock_profile.return_value = { + "profile_name": "prgs-author", + "role": "author", + "allowed_operations": ["gitea.pr.comment", "gitea.branch.push", "gitea.read"], + "forbidden_operations": [], + } + mock_cfg.return_value = { + "profiles": { + "prgs-author": { + "role": "author", + "allowed_operations": [ + "gitea.pr.comment", + "gitea.branch.push", + "gitea.read", + ], + "forbidden_operations": [], + }, + "prgs-reviewer": { + "role": "reviewer", + "allowed_operations": ["gitea.pr.comment", "gitea.pr.review"], + "forbidden_operations": [], + }, + } + } + with patch.dict(os.environ, {"GITEA_MCP_PROFILE": "prgs-author"}, clear=False): + res = mcp_server.gitea_resolve_task_capability( + task="acquire_reviewer_pr_lease", remote="prgs" + ) + self.assertFalse(res.get("allowed_in_current_session"), res) + self.assertEqual(res["required_role_kind"], "reviewer") + guidance = " ".join(res.get("task_role_guidance") or []) + self.assertTrue( + "cannot perform reviewer" in guidance.lower() + or "reviewer" in guidance.lower() + or res.get("stop_required"), + res, + ) + + @patch.object(mcp_server, "record_mutation_authority", return_value=None) + @patch.object(mcp_server, "init_review_decision_lock", return_value=None) + @patch.object(mcp_server, "record_preflight_check", return_value=None) + @patch.object(mcp_server, "_ensure_matching_profile", return_value=None) + @patch.object(mcp_server, "_authenticated_username", return_value="sysadmin") + @patch.object(mcp_server, "get_profile") + def test_resolve_unknown_task_structured_no_raise(self, mock_profile, *_mocks): + mock_profile.return_value = { + "profile_name": "prgs-reviewer", + "role": "reviewer", + "allowed_operations": ["gitea.pr.comment"], + "forbidden_operations": [], + } + # Must not raise ValueError into the tool boundary. + res = mcp_server.gitea_resolve_task_capability( + task="some_made_up_task", remote="prgs" + ) + self.assertIsInstance(res, dict) + self.assertEqual(res.get("reason_code"), "unknown_task", res) + self.assertFalse(res.get("allowed_in_current_session")) + self.assertTrue(res.get("stop_required")) + self.assertIs(res.get("mutation_performed"), False) + self.assertNotIn("token", str(res).lower()) + self.assertNotIn("password", str(res).lower()) + guidance = " ".join(res.get("task_role_guidance") or []) + self.assertIn("Unknown task/action", guidance) + + @patch.object(mcp_server, "record_mutation_authority", return_value=None) + @patch.object(mcp_server, "init_review_decision_lock", return_value=None) + @patch.object(mcp_server, "record_preflight_check", return_value=None) + @patch.object(mcp_server, "_ensure_matching_profile", return_value=None) + @patch.object(mcp_server, "_authenticated_username", return_value="sysadmin") + @patch.object(mcp_server, "get_profile") + @patch.object(mcp_server.gitea_config, "load_config") + def test_resolve_merger_acquire_aliases( + self, mock_cfg, mock_profile, *_mocks + ): + mock_profile.return_value = { + "profile_name": "prgs-merger", + "role": "merger", + "allowed_operations": [ + "gitea.read", + "gitea.pr.comment", + "gitea.pr.merge", + ], + "forbidden_operations": [], + } + mock_cfg.return_value = { + "profiles": { + "prgs-merger": { + "role": "merger", + "allowed_operations": [ + "gitea.read", + "gitea.pr.comment", + "gitea.pr.merge", + ], + "forbidden_operations": [], + } + } + } + with patch.dict(os.environ, {"GITEA_MCP_PROFILE": "prgs-merger"}, clear=False): + for task in ("acquire_merger_pr_lease", "gitea_acquire_merger_pr_lease"): + res = mcp_server.gitea_resolve_task_capability(task=task, remote="prgs") + self.assertEqual(res["required_role_kind"], "merger", res) + self.assertEqual( + res["required_operation_permission"], "gitea.pr.comment", res + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_resolve_task_capability.py b/tests/test_resolve_task_capability.py index 23f05bc..07e5769 100644 --- a/tests/test_resolve_task_capability.py +++ b/tests/test_resolve_task_capability.py @@ -231,9 +231,16 @@ class TestResolveTaskCapability(unittest.TestCase): self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"]) def test_resolve_unknown_task_fails_closed(self): + # #723: unknown tasks return structured unknown_task (no ValueError escape). with patch.dict(os.environ, self._env("author-profile")): - with self.assertRaises(ValueError): - mcp_server.gitea_resolve_task_capability(task="invalid_task_xyz", remote="prgs") + res = mcp_server.gitea_resolve_task_capability( + task="invalid_task_xyz", remote="prgs" + ) + self.assertIsInstance(res, dict) + self.assertEqual(res.get("reason_code"), "unknown_task", res) + self.assertFalse(res.get("allowed_in_current_session")) + self.assertTrue(res.get("stop_required")) + self.assertIs(res.get("mutation_performed"), False) # Additional regression tests per #145 for permission boundaries and structured guidance def test_issue_comment_does_not_imply_close(self): @@ -439,8 +446,11 @@ class TestResolveTaskCapability(unittest.TestCase): res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs") self.assertEqual(res["required_operation_permission"], "gitea.pr.close") for unknown in ("close_pull_request", "close", "pr_close"): - with self.assertRaises(ValueError): - mcp_server.gitea_resolve_task_capability(task=unknown, remote="prgs") + unk = mcp_server.gitea_resolve_task_capability( + task=unknown, remote="prgs" + ) + self.assertEqual(unk.get("reason_code"), "unknown_task", unk) + self.assertFalse(unk.get("allowed_in_current_session")) if __name__ == "__main__": unittest.main()