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