"""Opt-in tracker test runner backing the dashboard Testing tab.

Security posture (dashboard-test-runner US-001):

- **Disabled by default** — the tracker only constructs a manager when started
  with ``--enable-test-runner`` / ``MESHNET_ENABLE_TEST_RUNNER=1`` or
  ``TrackerServer(enable_test_runner=True)``.
- **No arbitrary commands** — a run target must be either a pytest node ID
  returned by our own ``--collect-only`` pass or an approved named suite.
  Callers never supply command arguments.
- **No shell** — pytest runs as ``[sys.executable, "-m", "pytest", ...]`` via
  ``subprocess.Popen`` without ``shell=True``.
- **One run at a time** — a second start while a run is active is rejected.
- **Bounded logs** — stdout/stderr are retained as line deques with caps.
- **Real inference stays gated** — ``tests/test_real_*.py`` modules are never
  collected and never part of a suite unless
  ``MESHNET_ENABLE_REAL_INFERENCE_TESTS=1`` is set, and even then only via the
  dedicated ``suite:real-inference`` target, never a default suite.
"""

from __future__ import annotations

import os
import re
import subprocess
import sys
import threading
import time
import uuid
from collections import deque
from pathlib import Path

ENABLE_ENV_VAR = "MESHNET_ENABLE_TEST_RUNNER"
REAL_INFERENCE_ENV_VAR = "MESHNET_ENABLE_REAL_INFERENCE_TESTS"

# Modules that talk to live nodes / spend API credit; matched by filename.
_REAL_INFERENCE_PATTERN = re.compile(r"(^|/)test_real_[^/]*\.py")

# Fixed, reviewed suite targets. Values are pytest paths relative to the repo
# root — never influenced by API input.
APPROVED_SUITES: dict[str, list[str]] = {
    "smoke": ["tests/test_smoke.py"],
    "dashboard": ["tests/test_dashboard.py"],
    "routing": ["tests/test_tracker_routing.py", "tests/test_dynamic_routing.py"],
}

# Only exists when REAL_INFERENCE_ENV_VAR=1; kept out of APPROVED_SUITES so it
# can never appear by default.
_REAL_INFERENCE_SUITE = {
    "real-inference": [
        "tests/test_real_distributed_inference.py",
        "tests/test_real_model_backend.py",
    ]
}

DEFAULT_MAX_LOG_LINES = 4000
_MAX_LINE_CHARS = 4000
DEFAULT_COLLECT_TIMEOUT = 120.0
DEFAULT_RUN_TIMEOUT = 1800.0

_NODE_ID_RE = re.compile(r"^[\w./\[\]:,= @-]+$")


class TestRunnerError(Exception):
    """Base class; carries the HTTP status the handler should send."""

    status = 400


class UnknownTargetError(TestRunnerError):
    status = 400


class RunInProgressError(TestRunnerError):
    status = 409


class CollectionError(TestRunnerError):
    status = 500


def discover_repo_root(start: Path | None = None) -> Path:
    """Find the repository root independent of the tracker's cwd.

    Walks up from this module (or ``start``) to the first directory holding
    both a ``pyproject.toml`` and ``tests/conftest.py`` — the layout of the
    meshnet monorepo root. Package-level ``pyproject.toml`` files (which have
    no ``tests/`` dir) are skipped naturally.
    """
    current = (start or Path(__file__)).resolve()
    for candidate in [current, *current.parents]:
        if (candidate / "pyproject.toml").is_file() and (candidate / "tests" / "conftest.py").is_file():
            return candidate
    raise RuntimeError("could not locate repository root (pyproject.toml + tests/conftest.py)")


def _real_inference_enabled() -> bool:
    return os.environ.get(REAL_INFERENCE_ENV_VAR) == "1"


class TestRunManager:
    """Collects pytest node IDs and runs one fixed target at a time."""

    def __init__(
        self,
        repo_root: str | Path | None = None,
        *,
        max_log_lines: int = DEFAULT_MAX_LOG_LINES,
        collect_timeout: float = DEFAULT_COLLECT_TIMEOUT,
        run_timeout: float = DEFAULT_RUN_TIMEOUT,
    ) -> None:
        self.repo_root = Path(repo_root).resolve() if repo_root else discover_repo_root()
        self.max_log_lines = max_log_lines
        self.collect_timeout = collect_timeout
        self.run_timeout = run_timeout
        self._lock = threading.Lock()
        self._collected: set[str] = set()
        self._collected_at: float | None = None
        self._run: dict | None = None
        self._process: subprocess.Popen | None = None
        self._stdout: deque[str] = deque(maxlen=max_log_lines)
        self._stderr: deque[str] = deque(maxlen=max_log_lines)
        self._reader_threads: list[threading.Thread] = []

    # ---- collection ----

    def suites(self) -> dict[str, list[str]]:
        """Approved suites whose files exist in this repo checkout."""
        available = dict(APPROVED_SUITES)
        if _real_inference_enabled():
            available.update(_REAL_INFERENCE_SUITE)
        return {
            name: paths
            for name, paths in available.items()
            if all((self.repo_root / p).is_file() for p in paths)
        }

    def collect(self, *, refresh: bool = False) -> dict:
        """Run ``pytest --collect-only -q`` and cache the node IDs.

        Real-inference modules are ignored at collection time (they are never
        imported) unless the explicit environment gate is set.
        """
        with self._lock:
            if self._collected_at is not None and not refresh:
                return self._collection_snapshot_locked()

        cmd = [sys.executable, "-m", "pytest", "--collect-only", "-q", "-p", "no:cacheprovider"]
        if not _real_inference_enabled():
            for name in sorted(
                p.name for p in (self.repo_root / "tests").glob("test_real_*.py")
            ):
                cmd.append(f"--ignore=tests/{name}")
        cmd.append("tests")
        try:
            proc = subprocess.run(
                cmd,
                cwd=self.repo_root,
                capture_output=True,
                text=True,
                timeout=self.collect_timeout,
            )
        except subprocess.TimeoutExpired as exc:
            raise CollectionError(f"pytest collection timed out after {self.collect_timeout}s") from exc
        # Exit code 5 = no tests collected; still a valid (empty) result.
        if proc.returncode not in (0, 5):
            tail = (proc.stdout or "")[-2000:] + (proc.stderr or "")[-2000:]
            raise CollectionError(f"pytest collection failed (exit {proc.returncode}): {tail}")
        node_ids = [
            line.strip()
            for line in proc.stdout.splitlines()
            if "::" in line and not line.startswith(("=", " ", "<"))
        ]
        if not _real_inference_enabled():
            node_ids = [n for n in node_ids if not _REAL_INFERENCE_PATTERN.search(n.split("::", 1)[0])]
        with self._lock:
            self._collected = set(node_ids)
            self._collected_at = time.time()
            return self._collection_snapshot_locked()

    def _collection_snapshot_locked(self) -> dict:
        return {
            "tests": sorted(self._collected),
            "suites": [
                {"id": f"suite:{name}", "name": name, "paths": paths}
                for name, paths in sorted(self.suites().items())
            ],
            "collected_at": self._collected_at,
        }

    # ---- running ----

    def start(self, target: str) -> dict:
        """Start one collected node ID or approved ``suite:<name>`` target."""
        if not isinstance(target, str) or not target.strip():
            raise UnknownTargetError("target is required")
        target = target.strip()
        if target.startswith("-") or not _NODE_ID_RE.match(target):
            raise UnknownTargetError("target must be a collected test node ID or an approved suite")

        if target.startswith("suite:"):
            suite_name = target.removeprefix("suite:")
            paths = self.suites().get(suite_name)
            if paths is None:
                raise UnknownTargetError(f"unknown suite: {suite_name!r}")
            pytest_args = list(paths)
        else:
            with self._lock:
                known = target in self._collected
            if not known:
                raise UnknownTargetError(
                    "target is not a collected test node ID; call GET /v1/tests first"
                )
            pytest_args = [target]

        with self._lock:
            if self._run is not None and self._run["status"] == "running":
                raise RunInProgressError("a test run is already in progress")
            cmd = [sys.executable, "-m", "pytest", "-q", "-p", "no:cacheprovider", *pytest_args]
            self._stdout = deque(maxlen=self.max_log_lines)
            self._stderr = deque(maxlen=self.max_log_lines)
            process = subprocess.Popen(  # noqa: S603 — fixed argv, no shell
                cmd,
                cwd=self.repo_root,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
            )
            self._process = process
            self._run = {
                "run_id": uuid.uuid4().hex,
                "target": target,
                "args": pytest_args,
                "status": "running",
                "started_at": time.time(),
                "finished_at": None,
                "exit_code": None,
            }
            self._reader_threads = [
                threading.Thread(
                    target=self._drain, args=(process.stdout, self._stdout), daemon=True
                ),
                threading.Thread(
                    target=self._drain, args=(process.stderr, self._stderr), daemon=True
                ),
            ]
            for thread in self._reader_threads:
                thread.start()
            waiter = threading.Thread(target=self._wait, args=(process,), daemon=True)
            waiter.start()
            return self._status_locked()

    @staticmethod
    def _drain(stream, sink: deque[str]) -> None:
        try:
            for line in stream:
                sink.append(line.rstrip("\n")[:_MAX_LINE_CHARS])
        except ValueError:  # stream closed mid-read on shutdown
            pass
        finally:
            stream.close()

    def _wait(self, process: subprocess.Popen) -> None:
        timed_out = False
        try:
            exit_code = process.wait(timeout=self.run_timeout)
        except subprocess.TimeoutExpired:
            timed_out = True
            process.kill()
            exit_code = process.wait()
        for thread in self._reader_threads:
            thread.join(timeout=5)
        with self._lock:
            if self._process is not process or self._run is None:
                return
            self._run["exit_code"] = exit_code
            self._run["finished_at"] = time.time()
            if timed_out:
                self._run["status"] = "timeout"
            elif exit_code == 0:
                self._run["status"] = "passed"
            else:
                self._run["status"] = "failed"
            self._process = None

    def status(self) -> dict:
        with self._lock:
            return self._status_locked()

    def _status_locked(self) -> dict:
        if self._run is None:
            return {"run": None, "stdout": "", "stderr": ""}
        run = dict(self._run)
        if run["status"] == "running":
            run["elapsed_seconds"] = time.time() - run["started_at"]
        elif run["finished_at"] is not None:
            run["elapsed_seconds"] = run["finished_at"] - run["started_at"]
        return {
            "run": run,
            "stdout": "\n".join(self._stdout),
            "stderr": "\n".join(self._stderr),
            "log_lines_limit": self.max_log_lines,
        }

    def shutdown(self) -> None:
        """Kill any active child process (tracker stop path)."""
        with self._lock:
            process = self._process
        if process is not None and process.poll() is None:
            process.kill()
            try:
                process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                pass