test: make test_health portable for scratch clones (closes #245) #248

Merged
sysadmin merged 1 commits from feat/issue-245-test-health-portable into master 2026-07-06 11:20:21 -05:00
2 changed files with 91 additions and 15 deletions
@@ -36,6 +36,8 @@ Rules (llm-project-workflow):
report the starting worktree path and whether it was dirty. If unrelated report the starting worktree path and whether it was dirty. If unrelated
tracked files exist outside the PR scope, STOP or run tracked files exist outside the PR scope, STOP or run
`scripts/worktree-review <pr-head-branch>` and validate in the scratch path. `scripts/worktree-review <pr-head-branch>` and validate in the scratch path.
Scratch-clone validation is the norm; tests must not assume the shared
development worktree or a repo-local ``venv/`` (#245).
NEVER run `git stash`, `git stash pop/drop`, `git checkout --`, `git reset`, NEVER run `git stash`, `git stash pop/drop`, `git checkout --`, `git reset`,
or `git clean` to manage another session's dirty files. or `git clean` to manage another session's dirty files.
- Final report must state: Worktree path, Worktree dirty (yes/no), - Final report must state: Worktree path, Worktree dirty (yes/no),
+88 -14
View File
@@ -3,9 +3,60 @@ import subprocess
import sys import sys
import unittest import unittest
from unittest.mock import patch from unittest.mock import patch
import role_session_router import role_session_router
from mcp_server import gitea_route_task_session, gitea_resolve_task_capability 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): class TestMCPHealth(unittest.TestCase):
def setUp(self): def setUp(self):
@@ -18,6 +69,17 @@ class TestMCPHealth(unittest.TestCase):
if os.path.exists(self.temp_file): if os.path.exists(self.temp_file):
os.remove(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_startup_conflict_detection(self): def test_startup_conflict_detection(self):
# Create a Python file with conflict markers constructed dynamically # Create a Python file with conflict markers constructed dynamically
with open(self.temp_file, "w") as f: with open(self.temp_file, "w") as f:
@@ -27,31 +89,43 @@ class TestMCPHealth(unittest.TestCase):
f.write("print('world')\n") f.write("print('world')\n")
f.write(">" * 7 + " main\n") f.write(">" * 7 + " main\n")
# Run mcp_server.py
venv_python = os.path.join(self.project_root, "venv", "bin", "python")
script_path = os.path.join(self.project_root, "mcp_server.py") script_path = os.path.join(self.project_root, "mcp_server.py")
res = subprocess.run( returncode, _stdout, stderr, _proc = _run_python_script(
[venv_python, script_path], self.project_root, script_path
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
cwd=self.project_root
) )
self.assertEqual(res.returncode, 1) self.assertEqual(returncode, 1)
stderr = res.stderr.decode() stderr_text = stderr.decode()
self.assertIn("infra_stop", stderr) self.assertIn("infra_stop", stderr_text)
self.assertIn("Unresolved merge conflict detected in test_temp_conflict.py", stderr) self.assertIn(
"Unresolved merge conflict detected in test_temp_conflict.py",
stderr_text,
)
@patch("role_session_router.check_mid_merge", return_value=True) @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"]}) @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): def test_route_task_session_blocks_during_merge(self, mock_profile, mock_check):
res = gitea_route_task_session("review_pr") res = gitea_route_task_session("review_pr")
self.assertEqual(res["route_result"], "infra_stop") self.assertEqual(res["route_result"], "infra_stop")
self.assertIn("infra_stop", res["message"]) self.assertIn("infra_stop", res["message"])
@patch("role_session_router.check_mid_merge", return_value=True) @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"]}) @patch(
def test_resolve_task_capability_blocks_during_merge(self, mock_profile, mock_check): "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_check
):
res = gitea_resolve_task_capability("review_pr") res = gitea_resolve_task_capability("review_pr")
self.assertTrue(res["infra_stop"]) self.assertTrue(res["infra_stop"])
self.assertFalse(res["allowed_in_current_session"]) self.assertFalse(res["allowed_in_current_session"])