import os import shutil import subprocess import sys import tempfile import unittest from unittest.mock import patch import capability_stop_terminal import role_session_router from role_session_router import python_bytes_have_conflict_markers from mcp_server import gitea_route_task_session, gitea_resolve_task_capability _HEALTH_SUBPROCESS_TIMEOUT_SEC = 30 def _health_test_python(): """Portable interpreter for subprocess health checks (#245). Prefer the active pytest interpreter, then ``GITEA_TOOLS_TEST_PYTHON``, else skip — never hard-code ``/venv/bin/python``. """ if ( sys.executable and os.path.isfile(sys.executable) and os.access(sys.executable, os.X_OK) ): return sys.executable override = (os.environ.get("GITEA_TOOLS_TEST_PYTHON") or "").strip() if override and os.path.isfile(override) and os.access(override, os.X_OK): return override raise unittest.SkipTest( "repo venv not available; run under a python interpreter or set " "GITEA_TOOLS_TEST_PYTHON" ) def _run_python_script(cwd, script_path, *, timeout=_HEALTH_SUBPROCESS_TIMEOUT_SEC): """Run a script in a child process with timeout and guaranteed teardown.""" python = _health_test_python() proc = subprocess.Popen( [python, script_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd, ) try: stdout, stderr = proc.communicate(timeout=timeout) except subprocess.TimeoutExpired: proc.kill() stdout, stderr = proc.communicate() raise finally: if proc.poll() is None: proc.terminate() try: proc.wait(timeout=5) except subprocess.TimeoutExpired: proc.kill() proc.wait() return proc.returncode, stdout, stderr, proc class TestMCPHealth(unittest.TestCase): def setUp(self): self.project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) self.temp_file = os.path.join(self.project_root, "test_temp_conflict.py") if os.path.exists(self.temp_file): os.remove(self.temp_file) def tearDown(self): if os.path.exists(self.temp_file): os.remove(self.temp_file) def test_health_test_python_prefers_sys_executable(self): resolved = _health_test_python() self.assertEqual(resolved, sys.executable) def test_health_test_python_skips_when_no_interpreter(self): with patch.object(sys, "executable", ""), patch.dict( os.environ, {}, clear=True ): with self.assertRaises(unittest.SkipTest): _health_test_python() def test_conflict_marker_helper_ignores_decorative_equals_border(self): sample = b'banner = """\n===========================================\n"""\n' self.assertFalse(python_bytes_have_conflict_markers(sample)) def test_conflict_marker_helper_detects_real_markers(self): sample = ( b"<" * 7 + b" HEAD\n" b"print('hello')\n" b"=" * 7 + b"\n" b"print('world')\n" b">" * 7 + b" main\n" ) self.assertTrue(python_bytes_have_conflict_markers(sample)) def test_startup_conflict_detection(self): # Create a Python file with conflict markers constructed dynamically with open(self.temp_file, "w") as f: f.write("<" * 7 + " HEAD\n") f.write("print('hello')\n") f.write("=" * 7 + "\n") f.write("print('world')\n") f.write(">" * 7 + " main\n") script_path = os.path.join(self.project_root, "mcp_server.py") returncode, _stdout, stderr, _proc = _run_python_script( self.project_root, script_path ) self.assertEqual(returncode, 1) stderr_text = stderr.decode() self.assertIn("infra_stop", stderr_text) self.assertIn( "Unresolved merge conflict detected in test_temp_conflict.py", stderr_text, ) @patch("role_session_router.assess_infra_stop") @patch( "mcp_server.get_profile", return_value={ "profile_name": "prgs-reviewer", "allowed_operations": ["gitea.pr.review"], }, ) def test_route_task_session_blocks_during_merge(self, mock_profile, mock_assess): mock_assess.return_value = { "infra_stop": True, "project_root": "/repo", "conflict_file": None, "mid_merge": True, "infra_stop_reasons": [ "mid-merge markers detected in /repo/.git/MERGE_HEAD (project_root=/repo)" ], } res = gitea_route_task_session("review_pr") self.assertEqual(res["route_result"], "infra_stop") self.assertIn("infra_stop", res["message"]) self.assertTrue(res.get("infra_stop_assessment", {}).get("infra_stop")) mock_assess.assert_called() @patch("role_session_router.assess_infra_stop") @patch( "mcp_server.get_profile", return_value={ "profile_name": "prgs-reviewer", "allowed_operations": ["gitea.pr.review"], }, ) def test_resolve_task_capability_blocks_during_merge( self, mock_profile, mock_assess ): mock_assess.return_value = { "infra_stop": True, "project_root": "/repo", "conflict_file": "/repo/gitea_mcp_server.py", "mid_merge": False, "infra_stop_reasons": [ "conflict markers detected in /repo/gitea_mcp_server.py (project_root=/repo)" ], } res = gitea_resolve_task_capability("review_pr") self.assertTrue(res["infra_stop"]) self.assertFalse(res["allowed_in_current_session"]) self.assertIn("infra_stop", res["exact_safe_next_action"]) self.assertEqual( res["infra_stop_assessment"]["conflict_file"], "/repo/gitea_mcp_server.py", ) @patch("role_session_router.assess_infra_stop") @patch("mcp_server._authenticated_username", return_value="reviewer-user") @patch( "mcp_server.get_profile", return_value={ "profile_name": "prgs-reviewer", "allowed_operations": ["gitea.pr.review", "gitea.pr.approve"], }, ) def test_resolve_task_capability_clears_stale_terminal_when_infra_clean( self, mock_profile, _username, mock_assess ): mock_assess.return_value = { "infra_stop": False, "project_root": "/repo", "conflict_file": None, "mid_merge": False, "infra_stop_reasons": [], } capability_stop_terminal.enter_from_capability_result({ "requested_task": "review_pr", "required_role_kind": "reviewer", "stop_required": True, "active_profile": "prgs-reviewer", "active_identity": "reviewer-user", "exact_safe_next_action": "blocked", }) self.assertTrue(capability_stop_terminal.is_active()) res = gitea_resolve_task_capability("review_pr", remote="prgs") self.assertFalse(capability_stop_terminal.is_active()) self.assertTrue(res["allowed_in_current_session"]) class TestInfraStopLiveAssessment(unittest.TestCase): def setUp(self): self.tempdir = tempfile.mkdtemp() def tearDown(self): shutil.rmtree(self.tempdir, ignore_errors=True) def test_clean_project_root_is_not_infra_stop(self): infra = role_session_router.assess_infra_stop(self.tempdir) self.assertFalse(infra["infra_stop"]) self.assertIsNone(infra["conflict_file"]) def test_conflict_present_then_resolved_recomputes_allowed(self): conflict_path = os.path.join(self.tempdir, "conflicted.py") with open(conflict_path, "wb") as handle: handle.write(b"<<<<<<< HEAD\n") blocked = role_session_router.assess_infra_stop(self.tempdir) self.assertTrue(blocked["infra_stop"]) self.assertEqual( os.path.realpath(blocked["conflict_file"]), os.path.realpath(conflict_path), ) os.remove(conflict_path) cleared = role_session_router.assess_infra_stop(self.tempdir) self.assertFalse(cleared["infra_stop"]) self.assertIsNone(cleared["conflict_file"])