"""Tests for native MCP preference gate and shell circuit breaker (#270).""" import sys import unittest import shutil 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 TestTerminalDiagnostics(unittest.TestCase): def test_diagnose_missing_cwd(self): res = nmp.diagnose_terminal_failure("/nonexistent_directory_for_test", ["git", "status"]) self.assertFalse(res["cwd_exists"]) self.assertEqual(res["error_type"], "missing cwd") self.assertIn("does not exist", res["error_msg"]) def test_diagnose_cwd_is_not_dir(self): import tempfile import os with tempfile.NamedTemporaryFile() as tmp: res = nmp.diagnose_terminal_failure(tmp.name, ["git", "status"]) self.assertFalse(res["cwd_is_dir"]) self.assertEqual(res["error_type"], "cwd is not a directory") def test_diagnose_missing_executable(self): res = nmp.diagnose_terminal_failure(None, ["nonexistent_exec_for_test"]) self.assertFalse(res["executable_exists"]) self.assertEqual(res["error_type"], "missing executable") def test_diagnose_missing_runtime_wrapper(self): res = nmp.diagnose_terminal_failure("/tmp", ["venv/bin/pytest"]) self.assertFalse(res["executable_exists"]) self.assertEqual(res["error_type"], "missing runtime wrapper") def test_diagnose_relative_executable_from_cwd(self): import os import tempfile with tempfile.TemporaryDirectory() as tmp: script = Path(tmp) / "tool" script.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") os.chmod(script, 0o755) res = nmp.diagnose_terminal_failure(tmp, ["./tool"]) self.assertTrue(res["executable_exists"]) self.assertEqual(res["resolved_executable"], str(script)) def test_probe_terminal_spawn_success(self): res = nmp.probe_terminal_spawn() self.assertTrue(res["healthy"]) self.assertIn("command", res) self.assertIn("diagnostics", res) def test_probe_terminal_spawn_missing_cwd(self): res = nmp.probe_terminal_spawn("/nonexistent_directory_for_test") self.assertFalse(res["healthy"]) self.assertEqual(res["error_type"], "missing cwd") def test_probe_terminal_spawn_honors_custom_command(self): res = nmp.probe_terminal_spawn( command=[sys.executable, "-c", "raise SystemExit(3)"] ) self.assertTrue(res["healthy"]) self.assertEqual(res["command"][0], sys.executable) self.assertEqual(res["exit_code"], 3) @unittest.skipUnless(shutil.which("git"), "git is required for git -C probe") def test_probe_terminal_spawn_supports_explicit_git_c_worktree(self): import subprocess import tempfile with tempfile.TemporaryDirectory() as tmp: subprocess.run( ["git", "init", tmp], check=True, capture_output=True, text=True, ) res = nmp.probe_terminal_spawn( cwd="/", command=["git", "-C", tmp, "status", "--short"], ) self.assertTrue(res["healthy"]) self.assertEqual(res["command"][1], "-C") def test_probe_terminal_spawn_blocks_missing_custom_command(self): res = nmp.probe_terminal_spawn(command=["nonexistent_exec_for_test"]) self.assertFalse(res["healthy"]) self.assertEqual(res["error_type"], "missing executable") self.assertEqual(res["command"], ["nonexistent_exec_for_test"]) 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()