"""#723 AC3/AC4/AC5: denied resolves must not poison the session role stamp, review-submission workspace failures must fail closed with structured reasons, and runtime-context capabilities must apply the resolver's role-exclusive filter. Reproduces the PR #721 sequence: a reviewer session resolved a task whose capability-map role had drifted to ``merger``; the denied resolve stamped ``_preflight_resolved_role = "merger"`` anyway, ``mark_final`` succeeded, and ``gitea_submit_pr_review`` raised RuntimeError out of the tool as a generic ``internal_error``. """ from __future__ import annotations import os import sys import unittest from pathlib import Path from unittest.mock import patch ROOT = str(Path(__file__).resolve().parent.parent) if ROOT not in sys.path: sys.path.insert(0, ROOT) import gitea_mcp_server as mcp_server import task_capability_map REVIEWER_PROFILE = { "profile_name": "prgs-reviewer", "role": "reviewer", "allowed_operations": [ "gitea.read", "gitea.pr.review", "gitea.pr.approve", "gitea.pr.request_changes", "gitea.pr.comment", "gitea.issue.comment", ], "forbidden_operations": [ "gitea.branch.create", "gitea.branch.push", "gitea.repo.commit", "gitea.pr.create", "gitea.pr.merge", ], } CONFIG = { "profiles": { "prgs-reviewer": { "role": "reviewer", "allowed_operations": REVIEWER_PROFILE["allowed_operations"], "forbidden_operations": REVIEWER_PROFILE["forbidden_operations"], }, "prgs-merger": { "role": "merger", "allowed_operations": [ "gitea.read", "gitea.pr.merge", "gitea.pr.comment", "gitea.issue.comment", ], "forbidden_operations": [ "gitea.pr.approve", "gitea.pr.review", "gitea.pr.request_changes", ], }, } } def _reset_preflight(): mcp_server._clear_preflight_capability_state() mcp_server._preflight_whoami_called = False mcp_server._preflight_whoami_violation = False class _ResolveHarness(unittest.TestCase): """Shared patched-resolver harness.""" def setUp(self): _reset_preflight() def tearDown(self): _reset_preflight() def _resolve(self, task, profile=REVIEWER_PROFILE, required_role=None): """Run gitea_resolve_task_capability with a fixed profile/config.""" patches = [ patch.object(mcp_server, "get_profile", return_value=profile), patch.object( mcp_server.gitea_config, "load_config", return_value=CONFIG ), patch.object( mcp_server, "_authenticated_username", return_value="tester" ), patch.object( mcp_server, "_ensure_matching_profile", return_value=None ), patch.object( mcp_server, "init_review_decision_lock", return_value=None ), ] if required_role is not None: patches.append( patch.object( mcp_server.task_capability_map, "required_role", side_effect=lambda t: ( required_role if t == task else task_capability_map.TASK_CAPABILITY_MAP[t]["role"] ), ) ) for p in patches: p.__enter__() try: return mcp_server.gitea_resolve_task_capability( task=task, remote="prgs" ) finally: for p in reversed(patches): p.__exit__(None, None, None) class TestAC3DeniedResolveDoesNotStampRole(_ResolveHarness): def test_allowed_resolve_stamps_role(self): result = self._resolve("review_pr") self.assertTrue(result["allowed_in_current_session"], result) self.assertEqual(mcp_server._preflight_resolved_role, "reviewer") self.assertEqual(mcp_server._preflight_resolved_task, "review_pr") def test_denied_resolve_does_not_stamp_required_role(self): """The PR #721 poison: review_pr requiring merger under a reviewer.""" result = self._resolve("review_pr", required_role="merger") self.assertFalse(result["allowed_in_current_session"], result) self.assertIsNone( mcp_server._preflight_resolved_role, "denied resolve must not stamp the required role (#723 AC3)", ) self.assertIsNone(mcp_server._preflight_resolved_task) def test_denied_resolve_clears_prior_stale_stamp(self): allowed = self._resolve("review_pr") self.assertTrue(allowed["allowed_in_current_session"]) self.assertEqual(mcp_server._preflight_resolved_role, "reviewer") denied = self._resolve("merge_pr") # reviewer profile cannot merge self.assertFalse(denied["allowed_in_current_session"], denied) self.assertIsNone( mcp_server._preflight_resolved_role, "a denied resolve must clear the previous stamp, not keep it", ) def test_denied_resolve_keeps_effective_role_on_actual_profile(self): with patch.object( mcp_server, "get_profile", return_value=REVIEWER_PROFILE ): self._resolve("review_pr", required_role="merger") self.assertEqual( mcp_server._effective_workspace_role(), "reviewer" ) class TestAC4SubmissionFailsClosedStructured(unittest.TestCase): def setUp(self): _reset_preflight() def tearDown(self): _reset_preflight() def test_workspace_binding_failure_returns_reasons_not_raise(self): boom = RuntimeError( "namespace workspace binding blocked: role 'merger' cannot " "mutate from reviewer session workspace" ) with patch.object( mcp_server, "_verify_role_mutation_workspace", side_effect=boom, ), patch.object( mcp_server, "get_profile", return_value=REVIEWER_PROFILE ): result = mcp_server._evaluate_pr_review_submission( pr_number=721, action="approve", expected_head_sha="80f59b334e6671b08006725292c08a8e8b6c823f", remote="prgs", live=True, final_review_decision_ready=True, ) self.assertFalse(result["performed"]) self.assertEqual(result["blocker_kind"], "workspace_role_binding") self.assertTrue( any("workspace/role binding failed" in r for r in result["reasons"]), result["reasons"], ) self.assertTrue( any("cannot mutate from reviewer session" in r for r in result["reasons"]), "the underlying binding error text must be preserved", ) def test_stale_runtime_failure_also_structured(self): boom = RuntimeError( "stale-runtime: The active Gitea MCP server process is stale" ) with patch.object( mcp_server, "_verify_role_mutation_workspace", side_effect=boom, ), patch.object( mcp_server, "get_profile", return_value=REVIEWER_PROFILE ): result = mcp_server._evaluate_pr_review_submission( pr_number=721, action="approve", remote="prgs", live=True, ) self.assertFalse(result["performed"]) self.assertEqual(result["blocker_kind"], "workspace_role_binding") self.assertTrue( any("stale-runtime" in r for r in result["reasons"]), result["reasons"], ) def test_dry_run_also_fails_closed_structured(self): boom = RuntimeError("binding blocked") with patch.object( mcp_server, "_verify_role_mutation_workspace", side_effect=boom, ), patch.object( mcp_server, "get_profile", return_value=REVIEWER_PROFILE ): result = mcp_server._evaluate_pr_review_submission( pr_number=721, action="approve", remote="prgs", live=False, ) self.assertFalse(result["would_perform"]) self.assertEqual(result["blocker_kind"], "workspace_role_binding") class TestAC5RuntimeContextRoleFilter(unittest.TestCase): def test_role_filtered_view_matches_resolver_denial(self): """Reviewer session: merge_pr stays denied under the role filter even when the permission is (pathologically) present.""" allowed_ops = [ "gitea.read", "gitea.pr.review", "gitea.pr.approve", "gitea.pr.request_changes", "gitea.pr.comment", "gitea.issue.comment", "gitea.pr.merge", ] caps = mcp_server._build_runtime_task_capabilities( allowed_ops, [], CONFIG, active_role_kind="reviewer" ) merge_entry = next( t for t in caps["task_capabilities"] if t["task"] == "merge_pr" ) self.assertTrue(merge_entry["role_exclusive"]) self.assertEqual(merge_entry["capability_view"], "role_filtered") self.assertFalse( merge_entry["allowed_in_current_session"], "role-exclusive merge_pr must stay denied for a reviewer role " "even when the permission is present (#723 AC5)", ) self.assertFalse(caps["can_merge_prs"]) def test_role_filter_restricts_matching_profiles(self): caps = mcp_server._build_runtime_task_capabilities( ["gitea.read"], [], CONFIG, active_role_kind="author" ) review_entry = next( t for t in caps["task_capabilities"] if t["task"] == "review_pr" ) self.assertEqual( review_entry["matching_configured_profiles"], ["prgs-reviewer"], "role-exclusive review_pr must not list the merger profile", ) merge_entry = next( t for t in caps["task_capabilities"] if t["task"] == "merge_pr" ) self.assertEqual( merge_entry["matching_configured_profiles"], ["prgs-merger"] ) def test_permission_only_view_is_labeled(self): caps = mcp_server._build_runtime_task_capabilities( ["gitea.pr.merge", "gitea.read"], [], CONFIG ) merge_entry = next( t for t in caps["task_capabilities"] if t["task"] == "merge_pr" ) self.assertEqual(merge_entry["capability_view"], "permission_only") # Legacy permission-only semantics preserved when no role supplied. self.assertTrue(merge_entry["allowed_in_current_session"]) def test_incident_722_shape_runtime_context_agrees_with_resolver(self): """Post-970e68b shape: review_pr requires merger; a reviewer session must see review_pr denied in runtime context, matching the resolver.""" with patch.object( mcp_server.task_capability_map, "required_role", side_effect=lambda t: ( "merger" if t == "review_pr" else task_capability_map.TASK_CAPABILITY_MAP[t]["role"] ), ): caps = mcp_server._build_runtime_task_capabilities( REVIEWER_PROFILE["allowed_operations"], REVIEWER_PROFILE["forbidden_operations"], CONFIG, active_role_kind="reviewer", ) review_entry = next( t for t in caps["task_capabilities"] if t["task"] == "review_pr" ) self.assertFalse( review_entry["allowed_in_current_session"], "runtime context must not report review_pr allowed when the " "resolver would fail-close it (incident #722)", ) self.assertFalse(caps["can_review_prs"]) if __name__ == "__main__": unittest.main()