Gitea-Tools/tests/test_health.py

import os
import subprocess
import sys
import unittest
from unittest.mock import patch

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.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
    ):
        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"])