app/tools/pytest_runner.py

"""
Runs pytest against a cloned repository inside a resource-limited sandbox.
"""
import os
import sys
import subprocess   # nosec B404
from typing import Any

from app.core.logger import get_logger
from app.tools.sandbox import run_sandboxed

logger = get_logger(__name__)

# Hard cap on how many test output lines we forward to the LLM
_MAX_OUTPUT_LINES = 100


def run_tests(local_path: str) -> dict[str, Any]:
    """
    Execute pytest in local_path inside the sandbox and return a summary.

    Returns:
        {
            "passed":    int,
            "failed":    int,
            "errors":    int,
            "output":    str,   # truncated pytest stdout/stderr
            "timed_out": bool,
            "oom_killed": bool,
        }
    """
    summary: dict[str, Any] = {
        "passed":    0,
        "failed":    0,
        "errors":    0,
        "output":    "",
        "timed_out": False,
        "oom_killed": False,
        "success":   False,
    }

    # Minimal safe env: inherit PATH so pytest is findable, but strip
    # secrets that might be present in the parent environment
    safe_env = {
        "PATH": os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin"),
        "HOME": "/tmp",  # nosec B108 — restrict home dir
        "PYTHONDONTWRITEBYTECODE": "1",
    }
    try:
        result = run_sandboxed(
            cmd=[sys.executable, "-m", "pytest", "--tb=short", "-q", "--no-header"],
            cwd=local_path,
            timeout=120,
            env=safe_env,
        )
    
    except subprocess.TimeoutExpired:
        summary["output"] = "pytest timed out"
        summary["timed_out"] = True
        return summary

    summary["timed_out"]  = result["timed_out"]
    summary["oom_killed"] = result["oom_killed"]

    raw_output = (result["stdout"] + result["stderr"]).strip()

    # Truncate to avoid flooding the LLM context
    lines = raw_output.splitlines()
    if len(lines) > _MAX_OUTPUT_LINES:
        lines = lines[:_MAX_OUTPUT_LINES] + [f"... ({len(lines) - _MAX_OUTPUT_LINES} lines truncated)"]
    summary["output"] = "\n".join(lines)

    # Parse pytest's summary line: "3 passed, 1 failed, 2 errors"
    for line in lines:
        lower = line.lower()
        if "passed" in lower or "failed" in lower or "error" in lower:
            summary["passed"] += _parse_count(line, "passed")
            summary["failed"] += _parse_count(line, "failed")
            summary["errors"] += _parse_count(line, "error")

    if result["timed_out"]:
        logger.warning("pytest timed out", extra={"path": local_path})
    elif result["oom_killed"]:
        logger.warning("pytest was OOM-killed", extra={"path": local_path})
    else:
        logger.info(
            "pytest complete",
            extra={
                "passed": summary["passed"],
                "failed": summary["failed"],
                "errors": summary["errors"],
            },
        )

    summary["success"] = (
        not summary["timed_out"]
        and not summary["oom_killed"]
        and summary["failed"] == 0
        and summary["errors"] == 0
        and summary["passed"] > 0
    )
    return summary


def _parse_count(line: str, keyword: str) -> int:
    """Extract the integer before `keyword` in a pytest summary line."""
    import re
    match = re.search(rf"(\d+)\s+{keyword}", line.lower())
    return int(match.group(1)) if match else 0