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

Use sys.executable with optional GITEA_TOOLS_TEST_PYTHON override instead of
hard-coded venv/bin/python. Add subprocess timeouts/teardown and skip when no
interpreter is available. Note scratch-clone norm in review-pr template.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
2026-07-06 11:42:34 -04:00
co-authored by Claude Opus 4.8
parent bab803ff3d
commit d422bc0978
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
tracked files exist outside the PR scope, STOP or run
`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`,
or `git clean` to manage another session's dirty files.
- Final report must state: Worktree path, Worktree dirty (yes/no),
+89 -15
View File
@@ -3,9 +3,60 @@ import subprocess
import sys
import unittest
from unittest.mock import patch
import role_session_router
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):
@@ -18,6 +69,17 @@ class TestMCPHealth(unittest.TestCase):
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_startup_conflict_detection(self):
# Create a Python file with conflict markers constructed dynamically
with open(self.temp_file, "w") as f:
@@ -27,32 +89,44 @@ class TestMCPHealth(unittest.TestCase):
f.write("print('world')\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")
res = subprocess.run(
[venv_python, script_path],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
cwd=self.project_root
returncode, _stdout, stderr, _proc = _run_python_script(
self.project_root, script_path
)
self.assertEqual(res.returncode, 1)
stderr = res.stderr.decode()
self.assertIn("infra_stop", stderr)
self.assertIn("Unresolved merge conflict detected in test_temp_conflict.py", stderr)
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"]})
@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.check_mid_merge", return_value=True)
@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_check):
@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_check
):
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.assertIn("infra_stop", res["exact_safe_next_action"])