runner.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/mutation/runner.py
 
"""
Purpose
-------
Run a test command once per mutant and classify each outcome honestly.
 
Responsibilities
----------------
- Prove the tests are green on the pristine target before judging any mutant.
- Execute the given test path in a subprocess with a per-mutant timeout.
- Classify each mutant as killed, survived, or timeout, never conflating them.
- Stream per-mutant progress so a long run is never silent.
 
Diagnostics
-----------
Domain: OCT-MUTATION
Levels:
    L2 - lifecycle
    L3 - per-mutant execution details
    L4 - deep tracing
 
Contracts
---------
- A timeout is its own class and is never counted as a kill.
- The pristine baseline must pass; without it every mutant would look killed,
  so a red or empty baseline is a named refusal rather than a 100% kill rate.
- Each subprocess gets a fresh bytecode-cache directory, so no run can be
  decided by a stale ``.pyc`` from the previous mutant.
- Runtime is mutant count times one test run; the caller is told the count
  before the loop starts.
 
Dependencies
------------
- oct.mutation.engine (enumeration and mutant identity)
- oct.mutation.sandbox (target refusal, apply, and restore)
- subprocess (per-mutant pytest execution)
"""
 
# OCT-LINT: disable-file=dbg_usage reason="Progress is the user-facing channel here; diagnostics would duplicate it into a stream the command reserves"
# OCT-LINT: disable-file=dbg_import reason="Runner is kept import-minimal so the restore path stays dependency-light"
 
from __future__ import annotations
 
import os
import subprocess
import sys
import tempfile
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, Iterable, Sequence
 
from oct.mutation.engine import Mutant, enumerate_source, relative_id
from oct.mutation.sandbox import (
    MutationRefusal,
    ensure_recoverable_target,
    mutated_target,
    read_target,
)
 
 
STATUS_KILLED = "killed"
STATUS_SURVIVED = "survived"
STATUS_TIMEOUT = "timeout"
 
# pytest's own "no tests were collected" exit code. The baseline preflight
# turns this into a refusal, which is what makes the mutant-loop mapping of
# "any non-zero exit is a kill" honest.
_PYTEST_NO_TESTS_COLLECTED = 5
 
DEFAULT_TIMEOUT_SECONDS = 60
 
 
@dataclass(frozen=True, slots=True)
class MutantOutcome:
    """One mutant's classification and the evidence behind it."""
 
    mutant: Mutant
    status: str
    duration_seconds: float
    returncode: int | None
 
    def to_dict(self) -> dict[str, object]:
        """Return a JSON-ready record carrying the mutant and its verdict."""
        return {
            **self.mutant.to_dict(),
            "status": self.status,
            "duration_seconds": round(self.duration_seconds, 3),
            "returncode": self.returncode,
        }
 
 
@dataclass(frozen=True, slots=True)
class RunReport:
    """The full result of one mutation run over one target."""
 
    target: str
    tests: str
    timeout_seconds: int
    baseline_seconds: float
    outcomes: list[MutantOutcome] = field(default_factory=list)
 
    @property
    def total(self) -> int:
        """Return how many mutants were run."""
        return len(self.outcomes)
 
    @property
    def killed(self) -> int:
        """Return how many mutants the tests noticed."""
        return self._count(STATUS_KILLED)
 
    @property
    def survived(self) -> int:
        """Return how many mutants the tests failed to notice."""
        return self._count(STATUS_SURVIVED)
 
    @property
    def timed_out(self) -> int:
        """Return how many mutants exceeded the per-run timeout."""
        return self._count(STATUS_TIMEOUT)
 
    @property
    def survivors(self) -> list[MutantOutcome]:
        """Return the surviving outcomes in enumeration order."""
        return [item for item in self.outcomes if item.status == STATUS_SURVIVED]
 
    @property
    def kill_rate(self) -> float | None:
        """Return killed over total, or ``None`` when nothing was run.
 
        Timeouts stay in the denominator: a mutant nobody could judge is not
        evidence that the suite would have caught it.
        """
        if not self.outcomes:
            return None
        return self.killed / len(self.outcomes)
 
    def _count(self, status: str) -> int:
        """Return how many outcomes carry ``status``."""
        return sum(1 for item in self.outcomes if item.status == status)
 
    def to_dict(self) -> dict[str, object]:
        """Return the full machine-readable structure."""
        return {
            "target": self.target,
            "tests": self.tests,
            "timeout_seconds": self.timeout_seconds,
            "baseline_seconds": round(self.baseline_seconds, 3),
            "totals": {
                "mutants": self.total,
                "killed": self.killed,
                "survived": self.survived,
                "timeout": self.timed_out,
            },
            "kill_rate": None if self.kill_rate is None else round(self.kill_rate, 4),
            "mutants": [outcome.to_dict() for outcome in self.outcomes],
        }
 
 
def _pytest_argv(tests: Path) -> list[str]:
    """Return the pytest command used for both the baseline and each mutant.
 
    ``-x`` stops at the first failure because one failure already proves the
    mutant was noticed. Plugin autoloading for randomisation and the cache is
    switched off so a run is decided by the mutant and nothing else.
    """
    return [
        sys.executable,
        "-m",
        "pytest",
        str(tests),
        "-q",
        "-x",
        "-p",
        "no:randomly",
        "-p",
        "no:cacheprovider",
    ]
 
 
def _run_tests(
    tests: Path,
    cwd: Path,
    timeout: int,
    pycache_dir: Path,
) -> tuple[int | None, float]:
    """Run the tests once, returning the exit code and elapsed seconds.
 
    The exit code is ``None`` when the run exceeded ``timeout``; the child is
    killed by :func:`subprocess.run` at that point. Grandchildren the test run
    spawned are not tracked, which is a named residual rather than a guarantee.
    """
    env = dict(os.environ)
    env["PYTHONPYCACHEPREFIX"] = str(pycache_dir)
    started = time.monotonic()
    try:
        proc = subprocess.run(
            _pytest_argv(tests),
            cwd=str(cwd),
            env=env,
            shell=False,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=timeout,
        )
    except subprocess.TimeoutExpired:
        return None, time.monotonic() - started
    return proc.returncode, time.monotonic() - started
 
 
def _verify_baseline(
    tests: Path,
    cwd: Path,
    timeout: int,
    pycache_dir: Path,
) -> float:
    """Refuse unless the pristine target passes the given tests.
 
    Skipping this check is how a mutation tool reports a flattering kill rate
    it did not earn: against a red suite every mutant "fails the tests" and is
    scored as killed.
    """
    returncode, duration = _run_tests(tests, cwd, timeout, pycache_dir)
    if returncode is None:
        raise MutationRefusal(
            f"the tests did not finish within {timeout}s on the unmutated "
            "target, so no mutant could be judged against them; raise "
            "--timeout or narrow --tests"
        )
    if returncode == _PYTEST_NO_TESTS_COLLECTED:
        raise MutationRefusal(
            f"no tests were collected from {tests}; every mutant would be "
            "scored against an empty suite"
        )
    if returncode != 0:
        raise MutationRefusal(
            f"the tests fail on the unmutated target (pytest exit {returncode}); "
            "against a red suite every mutant looks killed, so the run stops "
            "before it can report a kill rate it did not earn"
        )
    return duration
 
 
def run_mutants(
    *,
    target: str | Path,
    tests: str | Path,
    timeout: int = DEFAULT_TIMEOUT_SECONDS,
    root: str | Path | None = None,
    mutant_ids: Sequence[str] | None = None,
    on_progress: Callable[[str], None] | None = None,
) -> RunReport:
    """Run every enumerated mutant of ``target`` and classify each outcome.
 
    ``mutant_ids`` narrows the run to named mutants; it exists for tests and
    is deliberately not a CLI flag in v1.
    """
    resolved_target = Path(target).resolve()
    resolved_tests = Path(tests).resolve()
    repo_root = ensure_recoverable_target(resolved_target)
    if not resolved_tests.exists():
        raise MutationRefusal(f"tests path does not exist: {resolved_tests}")
    anchor = Path(root).resolve() if root is not None else repo_root
    relpath = relative_id(resolved_target, anchor)
    _, source, _ = read_target(resolved_target)
    mutants = _select(enumerate_source(source, relpath), mutant_ids)
    report_progress = on_progress or (lambda _message: None)
    with tempfile.TemporaryDirectory(prefix="oct-mutate-") as cache_root:
        cache_dir = Path(cache_root)
        report_progress("Verifying the tests are green on the unmutated target...")
        baseline_seconds = _verify_baseline(
            resolved_tests, repo_root, timeout, cache_dir / "baseline"
        )
        report_progress(
            f"Baseline green in {baseline_seconds:.1f}s. "
            f"Running {len(mutants)} mutant(s); "
            f"about {len(mutants) * baseline_seconds:.0f}s at baseline speed."
        )
        outcomes = _run_each(
            mutants,
            resolved_target,
            resolved_tests,
            repo_root,
            relpath,
            timeout,
            cache_dir,
            report_progress,
        )
    return RunReport(
        target=relpath,
        tests=str(resolved_tests),
        timeout_seconds=timeout,
        baseline_seconds=baseline_seconds,
        outcomes=outcomes,
    )
 
 
def _select(
    mutants: list[Mutant],
    mutant_ids: Sequence[str] | None,
) -> list[Mutant]:
    """Return the requested subset, refusing an id the target does not hold."""
    if mutant_ids is None:
        return mutants
    wanted = list(mutant_ids)
    by_id = {mutant.mutant_id: mutant for mutant in mutants}
    missing = [item for item in wanted if item not in by_id]
    if missing:
        raise MutationRefusal(f"unknown mutant id(s): {', '.join(missing)}")
    return [by_id[item] for item in wanted]
 
 
def _run_each(
    mutants: Iterable[Mutant],
    target: Path,
    tests: Path,
    repo_root: Path,
    relpath: str,
    timeout: int,
    cache_dir: Path,
    report_progress: Callable[[str], None],
) -> list[MutantOutcome]:
    """Apply, run, and classify each mutant, restoring the target every time."""
    outcomes: list[MutantOutcome] = []
    ordered = list(mutants)
    for index, mutant in enumerate(ordered, start=1):
        with mutated_target(target, mutant.mutant_id, relpath):
            returncode, duration = _run_tests(
                tests, repo_root, timeout, cache_dir / f"mutant-{index}"
            )
        status = _classify(returncode)
        outcomes.append(
            MutantOutcome(
                mutant=mutant,
                status=status,
                duration_seconds=duration,
                returncode=returncode,
            )
        )
        report_progress(
            f"[{index}/{len(ordered)}] {mutant.mutant_id} -> {status} "
            f"({duration:.1f}s)"
        )
    return outcomes
 
 
def _classify(returncode: int | None) -> str:
    """Map one test run's exit code onto the three pinned outcome classes."""
    if returncode is None:
        return STATUS_TIMEOUT
    if returncode == 0:
        return STATUS_SURVIVED
    return STATUS_KILLED