"""Subprocess pytest adapter for ``muse code test``. Executes a targeted set of pytest node IDs in an isolated subprocess environment, captures structured results via pytest-json-report, and returns a typed :class:`RunResult`. Design principles ----------------- **Isolation** — Each ``run_tests`` call spawns a fresh subprocess. Only the variables named in *env_allowlist* are passed through from the parent environment. This prevents secrets (tokens, credentials, API keys) from leaking into the test environment and ensures test results are reproducible. **Parallelism** — When ``workers > 1``, the target list is split into ``workers`` partitions and executed as independent subprocesses. Results are gathered in completion order and merged into a single :class:`RunResult`. This is a coarse partition strategy; for fine-grained per-test parallelism use ``pytest-xdist`` by adding ``-x``/``--dist`` to *extra_args*. **Budget** — A wall-clock timeout (``timeout_s``) is enforced via ``subprocess.run(timeout=...)``. When exceeded the subprocess is killed and a ``timed_out=True`` flag is set on the :class:`RunResult`. Individual test budgets can be enforced with ``pytest-timeout`` in *extra_args* (``--timeout=N``). **Fallback** — When pytest-json-report is not installed, the adapter falls back to parsing pytest's ``--tb=short`` text output. The fallback extracts pass/fail counts but cannot produce per-test :class:`CaseResult` records; it returns a summary-only result with an empty ``results`` list and sets ``json_report_available=False``. Security -------- * ``subprocess.run`` is called with ``shell=False`` — the command is a ``list[str]`` never passed to a shell interpreter. * The subprocess receives no environment variables except those explicitly listed in *env_allowlist* plus the minimal variables required for Python to start (``PATH``, ``HOME``, ``PYTHONPATH``, ``VIRTUAL_ENV``). * pytest is invoked as ``sys.executable -m pytest`` so the test runner uses the same interpreter as the calling process — it is not resolved from ``PATH`` and cannot be hijacked by a malicious ``pytest`` script on the user's ``PATH``. * The JSON report is written to a ``tempfile.NamedTemporaryFile`` in the system temp directory and deleted after reading. It is not world-writable. """ from __future__ import annotations import json import logging import os import pathlib import subprocess import sys import tempfile import time import uuid from collections.abc import Callable, Sequence from typing import Literal, NotRequired, TypedDict from muse.core._types import Manifest logger = logging.getLogger(__name__) # Allowlist of flag prefixes that agents/users may pass via extra_args. # Any flag not matching a prefix here is silently dropped to prevent # injection attacks (e.g. --rootdir overrides, --plugin, --import-mode). _SAFE_PYTEST_FLAG_PREFIXES: frozenset[str] = frozenset({ "-v", "-vv", "-s", "-x", "--tb=", "-k", "-m", "--collect-only", # explicit — NOT "--co" (would match --confcutdir) "--durations=", "--durations", "--no-header", "--maxfail=", "--log-level=", "--log-cli-level=", "--timeout=", }) def _filter_extra_args(extra_args: list[str]) -> list[str]: """Return only the safe subset of *extra_args*. Strips any flag not in :data:`_SAFE_PYTEST_FLAG_PREFIXES` to prevent injection via ``--rootdir``, ``--import-mode``, ``-p``, ``--override-ini``, ``--confcutdir``, or other subprocess-escaping vectors. Positional values that immediately follow a dropped flag are also dropped, because they are the value of the banned flag (e.g. ``--rootdir /etc``). """ safe: list[str] = [] # None = neutral, True = prev flag was kept and expects a value, False = prev was dropped prev_kept_expects_value: bool | None = None for arg in extra_args: if not arg.startswith("-"): # Positional — keep only if it belongs to a kept flag, not a dropped one. if prev_kept_expects_value is True: safe.append(arg) elif prev_kept_expects_value is False: logger.debug("test_runner: dropping positional after unsafe flag %r", arg) prev_kept_expects_value = None continue allowed = any(arg == p or arg.startswith(p) for p in _SAFE_PYTEST_FLAG_PREFIXES) if allowed: safe.append(arg) # Flags that take a separate positional value. prev_kept_expects_value = arg in {"-k", "-m", "--durations", "--maxfail", "--timeout"} else: logger.debug("test_runner: dropping unsafe extra_arg %r", arg) # Mark as dropped so the next positional (if any) is also dropped. prev_kept_expects_value = False return safe # --------------------------------------------------------------------------- # Public type definitions # --------------------------------------------------------------------------- Outcome = Literal["passed", "failed", "error", "skipped"] # Environment variables always forwarded to the subprocess regardless of the # allowlist, because Python cannot start without them. _MANDATORY_ENV_VARS: frozenset[str] = frozenset( { "PATH", "HOME", "PYTHONPATH", "VIRTUAL_ENV", "PYTHONHOME", "PYTHONDONTWRITEBYTECODE", "TMPDIR", "TMP", "TEMP", "LANG", "LC_ALL", "LC_CTYPE", } ) # Default list of non-sensitive env vars allowed through. DEFAULT_ENV_ALLOWLIST: list[str] = [ "MUSE_REPO_ROOT", "MUSE_TEST_ENV", "CI", "TERM", "COLORTERM", "COLUMNS", "ROWS", "NO_COLOR", ] class RunConfig(TypedDict): """Configuration for a single ``run_tests`` invocation.""" targets: list[str] """Pytest node IDs or file paths to execute. Each entry may be: * A file path: ``"tests/test_foo.py"`` * A node ID: ``"tests/test_foo.py::TestBar::test_baz"`` * Empty list → pytest discovers all tests under ``testpaths`` from ``pytest.ini``/``pyproject.toml``. """ workers: int """Number of parallel subprocess partitions. ``1`` = single process.""" timeout_s: float """Wall-clock budget per *partition* in seconds. ``0`` = unlimited.""" extra_args: list[str] """Additional arguments forwarded verbatim to pytest after node IDs.""" env_allowlist: list[str] """Additional env var names to forward (beyond mandatory vars).""" cwd: pathlib.Path | None """Working directory for the subprocess. ``None`` = inherit CWD.""" stream_output: bool """When ``True``, pytest's stdout/stderr are inherited by the parent process so output streams live to the terminal. ``False`` (the default) captures both streams for machine-readable processing (``--json`` mode).""" class CaseResult(TypedDict): """Outcome for a single pytest test function.""" node_id: str """Pytest node ID.""" outcome: Outcome """Test outcome.""" duration_ms: float """Wall-clock execution time in milliseconds.""" stdout: str """Captured stdout (empty when pytest was not run with ``-s``).""" stderr: str """Captured stderr.""" longrepr: NotRequired[str] """Short failure representation (omitted when test passes).""" class RunResult(TypedDict): """Structured result of a ``run_tests`` call.""" run_id: str """UUID v4 run identifier (matches the :class:`RunRecord` in history).""" targets: list[str] """Targets that were passed to pytest.""" exit_code: int """Exit code from pytest (0 = all passed, 1 = failures, 2 = interrupted, …).""" duration_ms: float """Total wall-clock time for the run in milliseconds.""" results: list[CaseResult] """Per-test results (empty when JSON report is unavailable).""" total: int """Total number of collected test cases.""" passed: int """Number of passing tests.""" failed: int """Number of failing tests.""" errored: int """Number of tests that raised an unexpected exception.""" skipped: int """Number of skipped tests.""" timed_out: bool """True if the subprocess was killed due to *timeout_s*.""" json_report_available: bool """True if pytest-json-report was available and produced a report.""" stdout: str """Combined stdout from all subprocess partitions (for fallback mode).""" stderr: str """Combined stderr from all subprocess partitions.""" # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _build_env(allowlist: Sequence[str]) -> Manifest: """Build a sanitised environment dict for the pytest subprocess. Only variables in *_MANDATORY_ENV_VARS* and *allowlist* are forwarded from the current process environment. All other variables are stripped to prevent credential leakage. """ allowed: frozenset[str] = _MANDATORY_ENV_VARS | frozenset(allowlist) return { k: v for k, v in os.environ.items() if k in allowed } def _check_json_report() -> bool: """Return True if pytest-json-report is importable in the current env.""" try: import importlib.util return importlib.util.find_spec("pytest_jsonreport") is not None except Exception: return False class _ParsedReport(TypedDict): """Parsed content from a pytest JSON report file.""" results: list[CaseResult] total: int passed: int failed: int errored: int skipped: int def _parse_json_report(path: str) -> _ParsedReport: """Parse a pytest-json-report output file into a :class:`_ParsedReport`.""" try: with open(path, encoding="utf-8") as fh: doc = json.load(fh) except Exception as exc: logger.warning("⚠️ test_runner: failed to read JSON report %s: %s", path, exc) return _ParsedReport( results=[], total=0, passed=0, failed=0, errored=0, skipped=0 ) if not isinstance(doc, dict): return _ParsedReport( results=[], total=0, passed=0, failed=0, errored=0, skipped=0 ) summary = doc.get("summary", {}) if not isinstance(summary, dict): summary = {} raw_tests = doc.get("tests", []) if not isinstance(raw_tests, list): raw_tests = [] results: list[CaseResult] = [] for t in raw_tests: if not isinstance(t, dict): continue node_id = str(t.get("nodeid", "")) raw_outcome = str(t.get("outcome", "error")) if raw_outcome == "passed": outcome: Outcome = "passed" elif raw_outcome == "failed": outcome = "failed" elif raw_outcome == "skipped": outcome = "skipped" else: outcome = "error" call_info = t.get("call", {}) if not isinstance(call_info, dict): call_info = {} duration_ms = float(call_info.get("duration", 0.0)) * 1000.0 stdout = str(call_info.get("stdout", "") or "") stderr = str(call_info.get("stderr", "") or "") longrepr_raw = call_info.get("longrepr", "") longrepr = str(longrepr_raw) if longrepr_raw else "" rec = CaseResult( node_id=node_id, outcome=outcome, duration_ms=duration_ms, stdout=stdout, stderr=stderr, ) if longrepr: rec["longrepr"] = longrepr results.append(rec) passed = int(summary.get("passed", 0)) failed = int(summary.get("failed", 0)) errored = int(summary.get("error", 0)) skipped = int(summary.get("skipped", 0)) total = int(summary.get("total", len(results))) return _ParsedReport( results=results, total=total, passed=passed, failed=failed, errored=errored, skipped=skipped, ) class _FallbackCounts(TypedDict): """Counts parsed from pytest text output when JSON report is unavailable.""" total: int passed: int failed: int errored: int skipped: int def _parse_text_output(stdout: str) -> _FallbackCounts: """Extract pass/fail counts from pytest's ``--tb=short`` text output.""" passed = failed = errored = skipped = total = 0 for line in stdout.splitlines(): if " passed" in line or " failed" in line or " error" in line: parts = line.split() for i, part in enumerate(parts): # Strip trailing commas: pytest emits "2 passed, 1 failed …" bare = part.rstrip(",") if bare == "passed" and i > 0: try: passed = int(parts[i - 1].rstrip(",")) except ValueError: pass elif bare == "failed" and i > 0: try: failed = int(parts[i - 1].rstrip(",")) except ValueError: pass elif bare in {"error", "errors"} and i > 0: try: errored = int(parts[i - 1].rstrip(",")) except ValueError: pass elif bare == "skipped" and i > 0: try: skipped = int(parts[i - 1].rstrip(",")) except ValueError: pass total = passed + failed + errored + skipped return _FallbackCounts( total=total, passed=passed, failed=failed, errored=errored, skipped=skipped, ) class _PartitionResult(TypedDict): """Result from executing one subprocess partition.""" exit_code: int duration_ms: float timed_out: bool report: _ParsedReport | None stdout: str stderr: str fallback_counts: _FallbackCounts | None def _run_partition( targets: list[str], config: RunConfig, json_report: bool, report_path: str, ) -> _PartitionResult: """Execute one pytest subprocess for *targets* and return its result.""" stream = config.get("stream_output", False) cmd: list[str] = [sys.executable, "-m", "pytest"] if json_report: cmd += ["--json-report", f"--json-report-file={report_path}"] if stream: # Verbose mode: let pytest own the terminal directly. cmd += ["--tb=short", "-v"] else: cmd += ["--tb=short", "-q"] cmd += targets cmd += _filter_extra_args(config["extra_args"]) env = _build_env(config["env_allowlist"]) cwd_str: str | None = str(config["cwd"]) if config["cwd"] else None timeout: float | None = config["timeout_s"] if config["timeout_s"] > 0 else None t0 = time.monotonic() timed_out = False exit_code = 0 stdout = "" stderr = "" try: if stream: # Inherit the parent's stdout/stderr so pytest output flows live. stream_proc: subprocess.CompletedProcess[bytes] = subprocess.run( cmd, timeout=timeout, env=env, cwd=cwd_str, ) exit_code = stream_proc.returncode else: cap_proc: subprocess.CompletedProcess[str] = subprocess.run( cmd, capture_output=True, text=True, timeout=timeout, env=env, cwd=cwd_str, ) exit_code = cap_proc.returncode stdout = cap_proc.stdout or "" stderr = cap_proc.stderr or "" except subprocess.TimeoutExpired as exc: timed_out = True exit_code = 124 # same as GNU timeout convention if not stream: raw_out = exc.stdout or b"" raw_err = exc.stderr or b"" stdout = raw_out.decode(errors="replace") if isinstance(raw_out, bytes) else raw_out stderr = raw_err.decode(errors="replace") if isinstance(raw_err, bytes) else raw_err logger.warning("⚠️ test_runner: partition timed out after %.1f s", config["timeout_s"]) except OSError as exc: exit_code = 127 stderr = f"failed to launch pytest: {exc}" logger.error("❌ test_runner: subprocess error: %s", exc) duration_ms = (time.monotonic() - t0) * 1000.0 parsed_report: _ParsedReport | None = None fallback: _FallbackCounts | None = None if json_report and not timed_out: try: parsed_report = _parse_json_report(report_path) except Exception as exc: logger.debug("test_runner: json report parse failed: %s", exc) # Only parse text output when we actually captured it. Stream # mode inherits the parent's stdout/stderr, so `stdout` is always # the empty string — passing it to _parse_text_output would return # zeros and produce a misleading "0 passed 0 failed" summary. fallback = _parse_text_output(stdout) if stdout else None else: fallback = _parse_text_output(stdout) if stdout else None return _PartitionResult( exit_code=exit_code, duration_ms=duration_ms, timed_out=timed_out, report=parsed_report, stdout=stdout, stderr=stderr, fallback_counts=fallback, ) def _partition(items: list[str], n: int) -> list[list[str]]: """Split *items* into *n* roughly equal partitions.""" if n <= 1 or not items: return [items] size = max(1, len(items) // n) parts: list[list[str]] = [] for i in range(0, len(items), size): parts.append(items[i : i + size]) return parts # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def run_tests( config: RunConfig, *, progress_cb: Callable[[CaseResult], None] | None = None, ) -> RunResult: """Execute the tests described by *config* and return structured results. Args: config: See :class:`RunConfig` for full documentation. progress_cb: Optional callback invoked once per :class:`CaseResult` as results arrive (useful for streaming progress to a terminal). Called synchronously in the gathering loop. Returns: A :class:`RunResult` with per-test outcomes and aggregate counts. """ run_id = str(uuid.uuid4()) targets = config["targets"] workers = max(1, config["workers"]) json_report = _check_json_report() if not json_report: logger.info( "ℹ️ pytest-jsonreport not installed — running in fallback mode " "(aggregate counts only, no per-test results)" ) t_start = time.monotonic() partitions = _partition(targets, workers) # --- Execute partitions ------------------------------------------------ all_results: list[CaseResult] = [] all_stdout: list[str] = [] all_stderr: list[str] = [] combined_exit_code = 0 any_timed_out = False total = passed = failed = errored = skipped = 0 for part_targets in partitions: with tempfile.NamedTemporaryFile( suffix=".json", delete=False, mode="w" ) as tmp: report_path = tmp.name try: part = _run_partition(part_targets, config, json_report, report_path) finally: try: os.unlink(report_path) except OSError: pass if part["exit_code"] != 0: combined_exit_code = max(combined_exit_code, part["exit_code"]) if part["timed_out"]: any_timed_out = True all_stdout.append(part["stdout"]) all_stderr.append(part["stderr"]) if part["report"] is not None: rep = part["report"] total += rep["total"] passed += rep["passed"] failed += rep["failed"] errored += rep["errored"] skipped += rep["skipped"] for res in rep["results"]: all_results.append(res) if progress_cb is not None: progress_cb(res) elif part["fallback_counts"] is not None: fb = part["fallback_counts"] total += fb["total"] passed += fb["passed"] failed += fb["failed"] errored += fb["errored"] skipped += fb["skipped"] duration_ms = (time.monotonic() - t_start) * 1000.0 return RunResult( run_id=run_id, targets=targets, exit_code=combined_exit_code, duration_ms=duration_ms, results=all_results, total=total, passed=passed, failed=failed, errored=errored, skipped=skipped, timed_out=any_timed_out, json_report_available=json_report, stdout="\n".join(all_stdout), stderr="\n".join(all_stderr), )