Merge pull request 'feat: add native MCP preference gate and shell circuit breaker (Closes #270)' (#377) from feat/issue-270-native-mcp-preference into master

2026-07-07 08:54:03 -05:00
parent 1c506fa779 6743cc11b9
commit 47b4ee4791
3 changed files with 563 additions and 0 deletions
@@ -0,0 +1,121 @@
+"""Tests for native MCP preference gate and shell circuit breaker (#270)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import native_mcp_preference as nmp  # noqa: E402
+
+
+class TestShellHealthCircuitBreaker(unittest.TestCase):
+    def setUp(self):
+        nmp.clear_shell_health_for_tests()
+
+    def test_two_spawn_failures_hard_stop(self):
+        nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
+        status = nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
+        self.assertTrue(status["hard_stopped"])
+        self.assertFalse(status["shell_use_allowed"])
+        self.assertEqual(status["consecutive_spawn_failures"], 2)
+
+    def test_success_resets_breaker(self):
+        nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
+        status = nmp.record_shell_spawn_outcome(exit_code=0, stdout="ok", stderr="")
+        self.assertFalse(status["hard_stopped"])
+        self.assertEqual(status["consecutive_spawn_failures"], 0)
+
+
+class TestNativeMcpPreferenceGate(unittest.TestCase):
+    def setUp(self):
+        nmp.clear_shell_health_for_tests()
+
+    def test_mcp_available_blocks_local_script(self):
+        result = nmp.assess_gitea_operation_path(
+            task="create_pr",
+            command_or_detail="python create_pr.py --remote prgs --title T --head h",
+            mcp_available=True,
+            mcp_tool_visible=True,
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["path_kind"], "shell_script")
+
+    def test_mcp_native_path_allowed(self):
+        result = nmp.assess_gitea_operation_path(
+            task="comment_issue",
+            path_kind="mcp_native",
+            mcp_available=True,
+            mcp_tool_visible=True,
+        )
+        self.assertFalse(result["block"])
+        self.assertTrue(result["allowed"])
+
+    def test_mcp_unavailable_requires_terminal_report(self):
+        result = nmp.assess_gitea_operation_path(
+            task="merge_pr",
+            path_kind="shell_script",
+            mcp_available=False,
+            mcp_tool_visible=False,
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("terminal recovery report" in r for r in result["reasons"])
+        )
+        self.assertIn(nmp.TERMINAL_REPORT_HEADING, result["terminal_report"])
+
+    def test_recovery_fallback_allowed_with_proof(self):
+        result = nmp.assess_gitea_operation_path(
+            task="merge_pr",
+            path_kind="shell_script",
+            command_or_detail="Recovery mode: MCP tools unavailable in this session.",
+            mcp_available=False,
+            recovery_mode=True,
+            recovery_proof_complete=True,
+        )
+        self.assertFalse(result["block"])
+
+    def test_cli_auth_divergence_blocked(self):
+        result = nmp.assess_gitea_operation_path(
+            task="create_pr",
+            command_or_detail="GITEA_MCP_PROFILE=prgs-reviewer python create_pr.py",
+            mcp_available=True,
+            active_profile="prgs-author",
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(any("CLI auth divergence" in r for r in result["reasons"]))
+
+    def test_unsafe_helper_fallback_denied(self):
+        result = nmp.assess_gitea_operation_path(
+            task="commit_files",
+            command_or_detail="python _encode_payload.py",
+            mcp_available=True,
+            mcp_tool_visible=True,
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["path_kind"], "unsafe_helper")
+
+    def test_reviewer_mcp_server_touch_blocked(self):
+        result = nmp.assess_gitea_operation_path(
+            task="review_pr",
+            command_or_detail="pkill -f gitea_mcp_server.py",
+            mcp_available=True,
+            role="reviewer",
+            active_profile="prgs-reviewer",
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["path_kind"], "mcp_server_touch")
+
+    def test_hard_stopped_shell_blocks_non_mcp_path(self):
+        nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
+        nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
+        result = nmp.assess_gitea_operation_path(
+            task="lock_issue",
+            path_kind="shell_script",
+            mcp_available=True,
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(any("circuit breaker" in r for r in result["reasons"]))
+
+
+if __name__ == "__main__":
+    unittest.main()