sysadminandClaude Opus 4.8
5af1b796ef
Add assess_infra_stop() to scan git state and conflict markers on each reviewer capability check, return the exact conflict path in responses, and clear stale capability-stop terminal state once the worktree is clean. Closes #285 Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
224 lines
7.9 KiB
Python
224 lines
7.9 KiB
Python
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 ``<repo>/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.check_mid_merge", return_value=True)
|
|
@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_check):
|
|
res = gitea_route_task_session("review_pr")
|
|
self.assertEqual(res["route_result"], "infra_stop")
|
|
self.assertIn("infra_stop", res["message"])
|
|
|
|
@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"]) |