server/docker_runner.py

from __future__ import annotations

import concurrent.futures
import logging
import os
import re
import subprocess
import sys
import time
from pathlib import Path
from typing import List

try:
    from ..models import RunRecord
except ImportError:
    from models import RunRecord

logger = logging.getLogger(__name__)


class DockerTestRunner:
    """Runs pytest either locally or inside a Docker sandbox.

    Mode selection is controlled by environment variables:
    - USE_DOCKER_IMAGE=1 enables sandbox mode
    - LOCAL_IMAGE_NAME sets the image (default: flakeforge-env:latest)
    """

    # Resolve the FlakeForge repo root once. We must keep its `__init__.py`
    # off the pytest import path for seed-repo subprocesses, otherwise pytest
    # walks up to the FlakeForge `pytest.ini` and tries to import the heavy
    # FlakeForge package as a "test module" — surfacing as bogus
    # `ModuleNotFoundError` (e.g. typeguard / openenv) and a false
    # `infra_broken` preflight verdict for every seed repo.
    _FF_REPO_ROOT = Path(__file__).resolve().parent.parent

    def __init__(self, repo_path: str) -> None:
        self.repo_path = Path(repo_path)
        self.use_docker_image = os.getenv("USE_DOCKER_IMAGE", "0").strip().lower() in {"1", "true", "yes"}
        self.local_image_name = os.getenv("LOCAL_IMAGE_NAME", "flakeforge-env:latest").strip() or "flakeforge-env:latest"
        self.pytest_timeout_seconds = int(os.getenv("FF_PYTEST_TIMEOUT_SECONDS", "20") or "20")
        self._deps_checked = False
        self._deps_ready = False
        self._deps_error = ""
        self._docker_checked = False
        self._docker_available = False
        self._docker_unavailable_reason = ""
        self._warned_unavailable = False

    def _ensure_docker_available(self) -> bool:
        """One-time probe for docker CLI and image presence."""
        if self._docker_checked:
            return self._docker_available

        self._docker_checked = True
        try:
            cli = subprocess.run(
                ["docker", "version", "--format", "{{.Server.Version}}"],
                capture_output=True,
                text=True,
                timeout=5,
            )
            if cli.returncode != 0:
                self._docker_available = False
                self._docker_unavailable_reason = (cli.stderr or cli.stdout or "docker CLI probe failed").strip()
                return False

            img = subprocess.run(
                ["docker", "image", "inspect", self.local_image_name],
                capture_output=True,
                text=True,
                timeout=5,
            )
            self._docker_available = img.returncode == 0
            if not self._docker_available:
                self._docker_unavailable_reason = (
                    img.stderr or img.stdout or f"docker image '{self.local_image_name}' not found"
                ).strip()
            return self._docker_available
        except Exception as exc:
            self._docker_available = False
            self._docker_unavailable_reason = f"exception while probing docker: {exc}"
            return False

    def _maybe_warn_docker_unavailable(self) -> None:
        if self.use_docker_image and not self._warned_unavailable:
            self._warned_unavailable = True
            logger.warning(
                "USE_DOCKER_IMAGE is enabled but sandbox execution is unavailable; falling back to local pytest. reason=%s",
                self._docker_unavailable_reason or "unknown",
            )

    def _pytest_cmd(self, test_id: str) -> List[str]:
        # Pin rootdir/confcutdir to the seed repo so pytest never escapes up
        # into the FlakeForge tree. `--import-mode=importlib` skips parent-dir
        # injection into sys.path, which is what previously caused FlakeForge
        # to be imported as part of the test package chain.
        abs_repo = str(self.repo_path.resolve())
        return [
            sys.executable, "-W", "ignore::SyntaxWarning",
            "-m", "pytest", test_id,
            "--rootdir", abs_repo,
            "--confcutdir", abs_repo,
            "--import-mode=importlib",
            "-p", "no:cacheprovider",
            "-W", "ignore::SyntaxWarning",
            "--tb=short", "-q", "--no-header",
        ]

    def _isolated_env(self) -> dict[str, str]:
        """Return a subprocess env that hides the FlakeForge repo from pytest.

        Even with rootdir/confcutdir pinned, an inherited PYTHONPATH that
        contains the FlakeForge root will still let pytest discover and
        import the FlakeForge package on `import FlakeForge` style probes.
        We strip it here, and disable bytecode writes to keep seed repos
        clean (no stray `__pycache__` dirs polluting git status).
        """
        env = os.environ.copy()
        ff_root_norm = os.path.normcase(str(self._FF_REPO_ROOT.resolve()))
        pp = env.get("PYTHONPATH", "")
        if pp:
            sep = os.pathsep
            kept = []
            for part in pp.split(sep):
                if not part:
                    continue
                try:
                    norm = os.path.normcase(str(Path(part).resolve()))
                except Exception:
                    norm = os.path.normcase(part)
                if norm != ff_root_norm:
                    kept.append(part)
            if kept:
                env["PYTHONPATH"] = sep.join(kept)
            else:
                env.pop("PYTHONPATH", None)
        env["PYTHONDONTWRITEBYTECODE"] = "1"
        env["PYTHONWARNINGS"] = "ignore::SyntaxWarning"
        return env

    def _deps_marker_path(self) -> Path:
        return self.repo_path / ".flakeforge_deps_ready"

    def _ensure_local_deps(self) -> bool:
        """Best-effort install of repo-specific test dependencies (local mode only).

        Many IDoFT repos require dependencies to import modules during pytest
        collection.  We treat pytest itself as critical (hard fail) but allow
        requirements / editable-install failures to be non-fatal: most seed
        repos work fine without them, and a failed `pip install -e .` should
        never prevent the test from being attempted at all.
        """
        if self._deps_checked:
            return self._deps_ready

        self._deps_checked = True
        marker = self._deps_marker_path()
        if marker.exists():
            self._deps_ready = True
            return True

        try:
            # Step 1 (critical): ensure pytest is available.
            proc = subprocess.run(
                [sys.executable, "-m", "pip", "install", "-q", "pytest"],
                capture_output=True,
                text=True,
                cwd=self.repo_path,
                timeout=300,
            )
            if proc.returncode != 0:
                self._deps_ready = False
                combined = f"{proc.stdout}\n{proc.stderr}".strip()
                self._deps_error = (combined or "pytest install failed")[-800:]
                return False

            # Step 2 (best-effort): install declared requirements only.
            # NEVER do `pip install -e .` on seed repos — that pollutes the
            # .venv's site-packages with editable links to every repo,
            # causing massive cross-contamination between unrelated projects.
            requirements = self.repo_path / "requirements.txt"
            req_test = self.repo_path / "requirements-test.txt"

            extra_cmds: list[list[str]] = []
            if requirements.exists():
                extra_cmds.append([sys.executable, "-m", "pip", "install", "-q", "-r", "requirements.txt"])
            if req_test.exists():
                extra_cmds.append([sys.executable, "-m", "pip", "install", "-q", "-r", "requirements-test.txt"])

            for cmd in extra_cmds:
                try:
                    subprocess.run(
                        cmd,
                        capture_output=True,
                        text=True,
                        cwd=self.repo_path,
                        timeout=300,
                    )
                except Exception:
                    pass

            marker.write_text("ok\n", encoding="utf-8")
            self._deps_ready = True
            return True
        except Exception as exc:
            self._deps_ready = False
            self._deps_error = f"{type(exc).__name__}: {exc}"
            return False

    _CREATION_FLAGS = (
        (subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.CREATE_NO_WINDOW)
        if sys.platform == "win32" else 0
    )

    def _run_local_pytest(self, test_id: str, timeout_seconds: int) -> subprocess.CompletedProcess[str]:
        return subprocess.run(
            self._pytest_cmd(test_id),
            capture_output=True,
            text=True,
            timeout=timeout_seconds,
            cwd=self.repo_path,
            env=self._isolated_env(),
            creationflags=self._CREATION_FLAGS,
        )

    def _run_docker_pytest(self, test_id: str, timeout_seconds: int) -> subprocess.CompletedProcess[str]:
        mount_src = str(self.repo_path.resolve())
        cmd = [
            "docker", "run", "--rm",
            "--network", "none",
            "--cpus", os.getenv("FF_DOCKER_CPUS", "1.0"),
            "--memory", os.getenv("FF_DOCKER_MEMORY", "2g"),
            "-v", f"{mount_src}:/workspace",
            "-w", "/workspace",
            "-e", "PYTHONDONTWRITEBYTECODE=1",
            self.local_image_name,
            *self._pytest_cmd(test_id),
        ]
        return subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout_seconds,
            cwd=self.repo_path,
        )

    def run_test(self, test_id: str, timeout_seconds: int | None = None) -> RunRecord:
        start = time.perf_counter()
        timeout_seconds = int(timeout_seconds or self.pytest_timeout_seconds)
        try:
            if self.use_docker_image and self._ensure_docker_available():
                proc = self._run_docker_pytest(test_id, timeout_seconds)
            else:
                if self.use_docker_image:
                    self._maybe_warn_docker_unavailable()
                if not self._ensure_local_deps():
                    duration_ms = int((time.perf_counter() - start) * 1000)
                    return RunRecord(
                        passed=False,
                        duration_ms=duration_ms,
                        error_type="ImportError",
                        error_message="dependency_install_failed",
                        stderr_excerpt=(self._deps_error or "")[-500:],
                    )
                proc = self._run_local_pytest(test_id, timeout_seconds)
            duration_ms = int((time.perf_counter() - start) * 1000)
            output = f"{proc.stdout}\n{proc.stderr}".strip()
            passed = proc.returncode == 0
            error_type = self._extract_error_type(output)
            error_message = self._extract_error_message(output)
            return RunRecord(
                passed=passed,
                duration_ms=duration_ms,
                error_type=error_type,
                error_message=error_message,
                stderr_excerpt=(proc.stderr or "")[-500:],
            )
        except subprocess.TimeoutExpired:
            duration_ms = int((time.perf_counter() - start) * 1000)
            return RunRecord(
                passed=False,
                duration_ms=duration_ms,
                error_type="TimeoutError",
                error_message=f"pytest timed out after {timeout_seconds}s",
                stderr_excerpt=None,
            )
        except Exception as exc:  # pragma: no cover
            duration_ms = int((time.perf_counter() - start) * 1000)
            return RunRecord(
                passed=False,
                duration_ms=duration_ms,
                error_type=type(exc).__name__,
                error_message=str(exc),
                stderr_excerpt=None,
            )

    def run_test_n_times(self, test_id: str, n: int, max_workers: int = 4) -> List[RunRecord]:
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            return list(executor.map(lambda _: self.run_test(test_id), range(n)))

    def check_regressions(self, exclude_test_id: str, timeout_seconds: int = 30) -> bool:
        exclude_test_file = exclude_test_id.split("::", 1)[0]
        repo_root = self.repo_path.resolve()
        tests_root = repo_root / "tests"
        if not tests_root.exists():
            return False

        try:
            cmd = [
                sys.executable, "-W", "ignore::SyntaxWarning",
                "-m", "pytest",
                str(tests_root),
                f"--ignore={repo_root / exclude_test_file}",
                "--rootdir", str(repo_root),
                "--confcutdir", str(repo_root),
                "--import-mode=importlib",
                "-p", "no:cacheprovider",
                "-W", "ignore::SyntaxWarning",
                "-x",
                "-q",
            ]
            if self.use_docker_image and self._ensure_docker_available():
                mount_src = str(repo_root)
                proc = subprocess.run(
                    [
                        "docker", "run", "--rm",
                        "--network", "none",
                        "--cpus", os.getenv("FF_DOCKER_CPUS", "1.0"),
                        "--memory", os.getenv("FF_DOCKER_MEMORY", "2g"),
                        "-v", f"{mount_src}:/workspace",
                        "-w", "/workspace",
                        "-e", "PYTHONDONTWRITEBYTECODE=1",
                        self.local_image_name,
                        *cmd,
                    ],
                    capture_output=True,
                    text=True,
                    timeout=timeout_seconds,
                    cwd=repo_root,
                )
            else:
                if self.use_docker_image:
                    self._maybe_warn_docker_unavailable()
                proc = subprocess.run(
                    cmd,
                    capture_output=True,
                    text=True,
                    timeout=timeout_seconds,
                    cwd=repo_root,
                    env=self._isolated_env(),
                    creationflags=self._CREATION_FLAGS,
                )
            # pytest exits with code 5 when every test file was excluded and
            # no tests were collected. That is not a regression; it just means
            # this tiny target repo only has the flaky test file.
            return proc.returncode not in (0, 5)
        except subprocess.TimeoutExpired:
            return True

    @staticmethod
    def _extract_error_type(output: str) -> str | None:
        match = re.search(r"([A-Za-z_][A-Za-z0-9_]*Error|Exception)\b", output)
        return match.group(1) if match else None

    @staticmethod
    def _extract_error_message(output: str) -> str | None:
        lines = [line.strip() for line in output.splitlines() if line.strip()]
        if not lines:
            return None
        return lines[-1][:200]