From 4f894009f6c49b55d706f0673212e07c1375f7cc Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Fri, 17 Jul 2026 10:09:36 -0400 Subject: [PATCH] fix(mcp): complete merger lease acquire + structured capability resolve (#718, #723) Harden gitea_acquire_merger_pr_lease with required candidate_head, live-head match, and fail-closed POST handling. Add capability-map aliases for reviewer and merger lease acquire/adopt. Make unknown tasks return structured unknown_task (no ValueError escape). Expand tests for roles, head scoping, aliases, and unknown_task. Closes #718. Addresses #723 resolver unknown_task and lease task mapping. --- gitea_mcp_server.py | 104 ++++++- task_capability_map.py | 16 ++ tests/test_merger_lease_acquire.py | 387 +++++++++++++++++++++----- tests/test_resolve_task_capability.py | 18 +- 4 files changed, 451 insertions(+), 74 deletions(-) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 2141b11..74982f3 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -10004,8 +10004,21 @@ def gitea_acquire_merger_pr_lease( "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") + _verify_role_mutation_workspace( + remote, worktree=worktree, task="acquire_merger_pr_lease" + ) except RuntimeError as e: return { "success": False, @@ -10022,13 +10035,31 @@ def gitea_acquire_merger_pr_lease( 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() + if live_head and 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, @@ -10038,7 +10069,7 @@ def gitea_acquire_merger_pr_lease( repo=repo_label, issue_number=issue_number, worktree=worktree, - candidate_head=candidate_head, + candidate_head=head_pin, target_branch=target_branch, target_branch_sha=target_branch_sha, pr_merged_or_closed=pr_merged_or_closed, @@ -10065,6 +10096,16 @@ def gitea_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, @@ -10073,7 +10114,7 @@ def gitea_acquire_merger_pr_lease( "profile": profile.get("profile_name"), "worktree": worktree, "phase": "claimed", - "candidate_head": candidate_head, + "candidate_head": head_pin, "target_branch": target_branch, "target_branch_sha": target_branch_sha, "repo": repo_label, @@ -10089,6 +10130,8 @@ def gitea_acquire_merger_pr_lease( "session_id": sid, "comment_id": posted.get("id"), "session_lease": session_lease, + "candidate_head": head_pin, + "repo": repo_label, "reasons": [], } @@ -13951,11 +13994,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", @@ -13963,6 +14051,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/task_capability_map.py b/task_capability_map.py index e270009..d9a4789 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -64,6 +64,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", @@ -86,10 +94,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 index 2ff0488..3c9da2a 100644 --- a/tests/test_merger_lease_acquire.py +++ b/tests/test_merger_lease_acquire.py @@ -1,15 +1,22 @@ -"""Merger lease acquisition tests (#718).""" +"""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 patch +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): @@ -17,62 +24,80 @@ class TestMergerLeaseAcquireTool(unittest.TestCase): 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=("dadeschools", "org", "repo")) - @patch("gitea_mcp_server.get_profile", return_value={"username": "merger-user", "profile_name": "prgs-merger", "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"], "forbidden_operations": []}) + @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, _profile, _resolve, _auth, mock_api): - """Merger lease acquisition succeeds when no reviewer lease exists.""" - # api_request is called for PR state, then for POSTing comment + 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"}, # PR is open - {"id": 456}, # Posted comment ID + {"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"]) - self.assertTrue(res["acquired"]) - self.assertEqual(res["comment_id"], 456) - self.assertEqual(res["session_id"], "merger-session") - - # Verify lease body includes merger identity - args, kwargs = mock_api.call_args_list[-1] - self.assertEqual(args[0], "POST") - body = args[3]["body"] - self.assertIn("reviewer_identity: merger-user", body) - self.assertIn("profile: prgs-merger", body) - self.assertIn("phase: claimed", body) - - # Post-mutation readback: lease is recorded in session - session_lease = leases._SESSION_LEASE - self.assertIsNotNone(session_lease) - self.assertEqual(session_lease["phase"], "claimed") - self.assertEqual(session_lease["session_id"], "merger-session") - - # Verify lease provenance uses SOURCE_ACQUIRE_MERGER - provenance = session_lease.get("lease_provenance") or {} - self.assertEqual(provenance.get("source"), "gitea_acquire_merger_pr_lease") + 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=("dadeschools", "org", "repo")) - @patch("gitea_mcp_server.get_profile", return_value={"username": "merger-user", "profile_name": "prgs-merger", "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"], "forbidden_operations": []}) + @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, _profile, _resolve, _auth, mock_api): - """Acquisition fails closed if another session holds an active lease.""" + 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="org/repo", + repo="Scaled-Tech-Consulting/Gitea-Tools", pr_number=718, issue_number=None, reviewer_identity="other-user", @@ -80,51 +105,285 @@ class TestMergerLeaseAcquireTool(unittest.TestCase): session_id="other-session", worktree="branches/review-1", phase="claimed", - candidate_head="a" * 40, + 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"}}] - - # PR is open - mock_api.return_value = {"state": "open"} - + _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"])) - - # No POST made - post_calls = [c for c in mock_api.call_args_list if c[0][0] == "POST"] - self.assertEqual(len(post_calls), 0) - - # No partial state recorded - self.assertIsNone(leases._SESSION_LEASE) + 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", return_value={"username": "merger-user", "profile_name": "prgs-merger", "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"], "forbidden_operations": []}) + @patch("gitea_mcp_server.get_profile") @patch("gitea_mcp_server._verify_role_mutation_workspace") - def test_acquire_merger_lease_fails_workspace_binding(self, _verify, _profile): - """Fails closed if workspace verification throws RuntimeError.""" + 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.assertFalse(res["acquired"]) - self.assertIn("Branches-only mutation guard", res["reasons"][0]) - self.assertIsNone(leases._SESSION_LEASE) + self.assertFalse(res["success"]) + self.assertIn("Branches-only mutation guard", res["reasons"][0]) + self.assertIsNone(leases._SESSION_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._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()