Merge remote-tracking branch 'prgs/master' into feat/issue-436-ci-coverage

tests/test_native_mcp_preference.py

@@ -1,6 +1,7 @@
 """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))
@@ -26,6 +27,90 @@ class TestShellHealthCircuitBreaker(unittest.TestCase):
         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()
@@ -118,4 +203,4 @@ class TestNativeMcpPreferenceGate(unittest.TestCase):
 
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()

tests/test_resolve_task_capability.py

@@ -96,6 +96,14 @@ class TestResolveTaskCapability(unittest.TestCase):
             "GITEA_TOKEN_MERGER": "merger-pass",
         }
 
+    def test_diagnose_terminal_success_has_no_error(self):
+        res = mcp_server.gitea_diagnose_terminal(
+            command=[sys.executable, "-c", "raise SystemExit(0)"]
+        )
+        self.assertTrue(res["healthy"])
+        self.assertIsNone(res["error_type"])
+        self.assertEqual(res["error_msg"], "")
+
     @patch("mcp_server.api_request", return_value={"login": "author-user"})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_resolve_author_capable_task(self, _auth, _api):
@@ -110,6 +118,39 @@ class TestResolveTaskCapability(unittest.TestCase):
             self.assertFalse(res["different_mcp_namespace_required"])
             self.assertIn("ready for operations", res["exact_safe_next_action"])
 
+    @patch("mcp_server.native_mcp_preference.probe_terminal_spawn")
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolve_mutation_task_blocks_when_terminal_launcher_unhealthy(
+        self,
+        _auth,
+        _api,
+        mock_probe,
+    ):
+        mock_probe.return_value = {
+            "healthy": False,
+            "error_type": "missing cwd",
+            "error_msg": "Current working directory does not exist: /missing",
+            "cwd": "/missing",
+            "command": ["git", "--version"],
+            "diagnostics": {"cwd_exists": False},
+        }
+        env = self._env("author-profile")
+        # Probe is skipped under pytest unless forced (mirrors production path).
+        env["GITEA_FORCE_TERMINAL_PROBE"] = "1"
+        with patch.dict(os.environ, env):
+            res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs")
+
+        self.assertFalse(res["allowed_in_current_session"])
+        self.assertTrue(res["stop_required"])
+        self.assertTrue(res["infra_stop"])
+        self.assertTrue(res["terminal_launcher_unhealthy"])
+        self.assertTrue(res["blocked"])
+        self.assertEqual(res["blocked_reason"], "BLOCKED + DIAGNOSE")
+        self.assertEqual(res["terminal_launcher_diagnostics"]["error_type"], "missing cwd")
+        self.assertIn("terminal-launcher-failure", res["exact_safe_next_action"])
+        self.assertIn("BLOCKED + DIAGNOSE", res["exact_safe_next_action"])
+
     @patch("mcp_server.api_request", return_value={"login": "author-user"})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_resolve_reviewer_capable_task_blocked(self, _auth, _api):