Test runner

test_runner

Agent-optimized test runner with structured output.

Wraps pytest with pytest-json-report to produce compact, token-efficient results for AI coding agents.

FailureDetail dataclass

Structured detail for a single test failure.

Source code in packages/axm-audit/src/axm_audit/core/test_runner.py

@dataclass
class FailureDetail:
    """Structured detail for a single test failure."""

    test: str
    """Full node ID, e.g. ``tests/unit/test_x.py::TestFoo::test_bar``."""

    error_type: str
    """Exception class name, e.g. ``AssertionError``."""

    message: str
    """One-line error message."""

    file: str
    """Relative file path."""

    line: int
    """Line number of the assertion / raising statement."""

    traceback: str
    """Short traceback (truncated to ``_MAX_TB_LINES``)."""

error_type instance-attribute

Exception class name, e.g. AssertionError.

file instance-attribute

Relative file path.

line instance-attribute

Line number of the assertion / raising statement.

message instance-attribute

One-line error message.

test instance-attribute

Full node ID, e.g. tests/unit/test_x.py::TestFoo::test_bar.

traceback instance-attribute

Short traceback (truncated to _MAX_TB_LINES).

NonTestCauseDetail dataclass

Serializable projection of a classified non-test cause.

TestReport is a dataclass whose payload is produced by dataclasses.asdict: a pydantic model attached to it would survive that traversal as an object and break every JSON consumer of the report. The classifier remains the single source of the classification itself; this only mirrors its verdict in the report's own payload shape.

Source code in packages/axm-audit/src/axm_audit/core/test_runner.py

@dataclass(frozen=True)
class NonTestCauseDetail:
    """Serializable projection of a classified non-test cause.

    ``TestReport`` is a dataclass whose payload is produced by
    ``dataclasses.asdict``: a pydantic model attached to it would survive that
    traversal as an object and break every JSON consumer of the report. The
    classifier remains the single source of the classification itself; this
    only mirrors its verdict in the report's own payload shape.
    """

    code: str
    summary: str
    excerpt: str

    @classmethod
    def from_cause(cls, cause: NonTestCause | None) -> NonTestCauseDetail | None:
        """Return the projection of *cause*, or ``None`` when there is none."""
        if cause is None:
            return None
        return cls(code=cause.code, summary=cause.summary, excerpt=cause.excerpt)

from_cause(cause) classmethod

Return the projection of cause, or None when there is none.

Source code in packages/axm-audit/src/axm_audit/core/test_runner.py

@classmethod
def from_cause(cls, cause: NonTestCause | None) -> NonTestCauseDetail | None:
    """Return the projection of *cause*, or ``None`` when there is none."""
    if cause is None:
        return None
    return cls(code=cause.code, summary=cause.summary, excerpt=cause.excerpt)

TestCase dataclass

Lossless verdict for one collected pytest item.

Source code in packages/axm-audit/src/axm_audit/core/test_runner.py

@dataclass(frozen=True, slots=True)
class TestCase:
    """Lossless verdict for one collected pytest item."""

    node_id: str
    outcome: TestOutcome
    detail: str | None = field(default=None, compare=False)

TestReport dataclass

Compact test execution report.

All fields use None rather than empty containers when no data exists so that dataclasses.asdict produces a minimal payload.

Source code in packages/axm-audit/src/axm_audit/core/test_runner.py

@dataclass
class TestReport:
    """Compact test execution report.

    All fields use ``None`` rather than empty containers when no data
    exists so that ``dataclasses.asdict`` produces a minimal payload.
    """

    passed: int = 0
    failed: int = 0
    errors: int = 0
    skipped: int = 0
    warnings: int = 0
    duration: float = 0.0
    coverage: float | None = None
    failures: list[FailureDetail] | None = None
    coverage_by_file: dict[str, float] | None = None
    cases: tuple[TestCase, ...] = ()
    timed_out: bool = False
    pytest_return_code: int = 0
    collected: int | None = None
    target_statuses: list[dict[str, str]] = field(default_factory=list)
    verdict: bool | None = None
    non_test_cause: NonTestCauseDetail | None = None

    def __post_init__(self) -> None:
        """Populate additive execution metadata for directly-built reports."""
        if self.collected is None:
            self.collected = self.passed + self.failed + self.errors + self.skipped
        if self.verdict is None:
            self.verdict = self._derive_verdict()

    def model_dump(self) -> dict[str, object]:
        """Return the report's payload, every key present even when ``None``."""
        return dataclasses.asdict(self)

    def record_execution(
        self,
        *,
        return_code: int,
        collected: int,
        target_statuses: list[dict[str, str]],
    ) -> None:
        """Attach pytest evidence and compute the report's canonical verdict."""
        self.pytest_return_code = return_code
        self.collected = collected
        self.target_statuses = target_statuses
        self.verdict = self._derive_verdict()

    def _derive_verdict(self) -> bool:
        """Return the fail-closed verdict from all available evidence."""
        targets_valid = all(
            status["status"] == "validated" for status in self.target_statuses
        )
        return (
            self.pytest_return_code == 0
            and (self.collected or 0) > 0
            and self.failed == 0
            and self.errors == 0
            and not self.timed_out
            and targets_valid
        )

__post_init__()

Populate additive execution metadata for directly-built reports.

Source code in packages/axm-audit/src/axm_audit/core/test_runner.py

def __post_init__(self) -> None:
    """Populate additive execution metadata for directly-built reports."""
    if self.collected is None:
        self.collected = self.passed + self.failed + self.errors + self.skipped
    if self.verdict is None:
        self.verdict = self._derive_verdict()

model_dump()

Return the report's payload, every key present even when None.

Source code in packages/axm-audit/src/axm_audit/core/test_runner.py

def model_dump(self) -> dict[str, object]:
    """Return the report's payload, every key present even when ``None``."""
    return dataclasses.asdict(self)

record_execution(*, return_code, collected, target_statuses)

Attach pytest evidence and compute the report's canonical verdict.

Source code in packages/axm-audit/src/axm_audit/core/test_runner.py

def record_execution(
    self,
    *,
    return_code: int,
    collected: int,
    target_statuses: list[dict[str, str]],
) -> None:
    """Attach pytest evidence and compute the report's canonical verdict."""
    self.pytest_return_code = return_code
    self.collected = collected
    self.target_statuses = target_statuses
    self.verdict = self._derive_verdict()

build_pytest_cmd(*, report_path, coverage_path, files, markers, stop_on_first)

Build the pytest command line.

Source code in packages/axm-audit/src/axm_audit/core/test_runner.py

def build_pytest_cmd(
    *,
    report_path: Path,
    coverage_path: Path | None,
    files: list[str] | None,
    markers: list[str] | None,
    stop_on_first: bool,
) -> list[str]:
    """Build the pytest command line."""
    cmd = [
        "pytest",
        "--json-report",
        f"--json-report-file={report_path}",
        "--json-report-omit=log,keywords",
        "--tb=short",
        "--no-header",
        "-q",
    ]

    if coverage_path is not None:
        cmd.extend(["--cov", f"--cov-report=json:{coverage_path}"])
    elif files:
        cmd.append("--no-cov")

    if stop_on_first:
        cmd.append("-x")

    if markers:
        cmd.extend(["-m", " or ".join(markers)])

    if files:
        cmd.extend(files)

    return cmd

build_test_report(*, report_data, total_cov, per_file_cov, mode=None, last_coverage=None, include_cases=False, return_code=0, stdout='', stderr='')

Build a TestReport from pytest JSON and coverage data.

Always parses failures and populates coverage — no mode branching. The captured stdout/stderr and return_code are optional. When they are supplied, the cause of a red that no test failure accounts for is classified and attached to non_test_cause — explanatory only, it never influences the verdict, and it stays None for a green run or a run red by test failures.

Returns None for failures and coverage_by_file when no data exists.

Source code in packages/axm-audit/src/axm_audit/core/test_runner.py

def build_test_report(
    *,
    report_data: dict[str, object],
    total_cov: float | None,
    per_file_cov: dict[str, float],
    mode: str | None = None,
    last_coverage: dict[str, float] | None = None,
    include_cases: bool = False,
    return_code: int = 0,
    stdout: str = "",
    stderr: str = "",
) -> TestReport:
    """Build a ``TestReport`` from pytest JSON and coverage data.

    Always parses failures and populates coverage — no mode branching.
    The captured ``stdout``/``stderr`` and ``return_code`` are optional. When
    they are supplied, the cause of a red that no test failure accounts for is
    classified and attached to ``non_test_cause`` — explanatory only, it never
    influences the verdict, and it stays ``None`` for a green run or a run red
    by test failures.

    Returns ``None`` for ``failures`` and ``coverage_by_file`` when no
    data exists.
    """
    summary = cast("dict[str, object]", report_data.get("summary", {}))
    tests_list = cast("list[dict[str, object]]", report_data.get("tests", []))
    cases = _extract_cases(report_data) if include_cases else ()

    # Always parse failures
    failures = parse_failures(tests_list)
    collectors_list = cast("list[dict[str, object]]", report_data.get("collectors", []))
    failures.extend(parse_collector_errors(collectors_list))
    if include_cases:
        collection_failure = _collection_failure_detail(collectors_list)
        if collection_failure is not None:
            raise ValueError(collection_failure)

    return TestReport(
        passed=cast(int, summary.get("passed", 0)),
        failed=cast(int, summary.get("failed", 0)),
        errors=cast(int, summary.get("error", 0)),
        skipped=cast(int, summary.get("skipped", 0)),
        warnings=cast(int, summary.get("warnings", 0)),
        duration=cast(float, report_data.get("duration", 0.0)),
        coverage=total_cov,
        failures=failures or None,
        coverage_by_file=per_file_cov or None,
        cases=cases,
        non_test_cause=NonTestCauseDetail.from_cause(
            classify_non_test_cause(
                return_code=return_code,
                failed=cast(int, summary.get("failed", 0)),
                errors=cast(int, summary.get("error", 0)),
                stdout=stdout,
                stderr=stderr,
            )
        ),
    )

parse_collector_errors(collectors)

Extract FailureDetail items from pytest-json-report collectors list.

Collector errors occur before test discovery completes (e.g. SyntaxError in a test file, broken imports).

Source code in packages/axm-audit/src/axm_audit/core/test_runner.py

def parse_collector_errors(
    collectors: list[dict[str, object]],
) -> list[FailureDetail]:
    """Extract ``FailureDetail`` items from pytest-json-report collectors list.

    Collector errors occur before test discovery completes (e.g.
    ``SyntaxError`` in a test file, broken imports).
    """
    failures: list[FailureDetail] = []
    for collector in collectors:
        longrepr = cast(str, collector.get("longrepr", ""))
        if not longrepr:
            continue

        nodeid = cast(str, collector.get("nodeid", "unknown"))

        # Truncate traceback
        tb_lines = longrepr.strip().splitlines()
        if len(tb_lines) > _MAX_TB_LINES:
            tb_lines = tb_lines[-_MAX_TB_LINES:]
        short_tb = "\n".join(tb_lines)

        # Extract error type from last line (e.g. "SyntaxError: invalid syntax")
        last_line = longrepr.strip().splitlines()[-1] if longrepr.strip() else ""
        error_type = "CollectionError"
        message = last_line
        if ":" in last_line:
            error_type = last_line.split(":")[0].strip()
            message = last_line

        failures.append(
            FailureDetail(
                test=nodeid,
                error_type=error_type,
                message=message,
                file=nodeid if nodeid != "unknown" else "",
                line=0,
                traceback=short_tb,
            )
        )
    return failures

parse_coverage(coverage_path)

Parse coverage JSON into total % and per-file dict.

Files whose basename equals __main__.py are excluded from the per-file map (they typically contain only a python -m entry point and are not meaningfully unit-testable). The aggregate total_pct from pytest-cov is left untouched, in line with coverage.py's exclude_also convention of filtering reports rather than rewriting the underlying totals.

Source code in packages/axm-audit/src/axm_audit/core/test_runner.py

def parse_coverage(coverage_path: Path) -> tuple[float | None, dict[str, float]]:
    """Parse coverage JSON into total % and per-file dict.

    Files whose basename equals ``__main__.py`` are excluded from the
    per-file map (they typically contain only a ``python -m`` entry
    point and are not meaningfully unit-testable). The aggregate
    ``total_pct`` from pytest-cov is left untouched, in line with
    coverage.py's ``exclude_also`` convention of filtering reports
    rather than rewriting the underlying totals.
    """
    if not coverage_path.exists():
        return None, {}
    try:
        data = cast("dict[str, object]", json.loads(coverage_path.read_text()))
    except (json.JSONDecodeError, OSError):
        return None, {}

    totals = cast("dict[str, object]", data.get("totals", {}))
    total_pct = cast("float | None", totals.get("percent_covered"))
    per_file: dict[str, float] = {}
    files_map = cast("dict[str, dict[str, object]]", data.get("files", {}))
    for fpath, fdata in files_map.items():
        if Path(fpath).name == "__main__.py":
            continue
        summary = cast("dict[str, object]", fdata.get("summary", {}))
        per_file[fpath] = cast(float, summary.get("percent_covered", 0.0))

    return total_pct, per_file

parse_failures(tests)

Extract FailureDetail items from pytest-json-report tests list.

Source code in packages/axm-audit/src/axm_audit/core/test_runner.py

def parse_failures(tests: list[dict[str, object]]) -> list[FailureDetail]:
    """Extract ``FailureDetail`` items from pytest-json-report tests list."""
    failures: list[FailureDetail] = []
    for test in tests:
        outcome = test.get("outcome", "")
        if outcome not in ("failed", "error"):
            continue

        nodeid = cast(str, test.get("nodeid", "unknown"))
        call_info = cast(
            "dict[str, object]", test.get("call") or test.get("setup") or {}
        )
        crash = cast("dict[str, object]", call_info.get("crash", {}))
        tb_text = cast(str, call_info.get("longrepr", ""))

        # Truncate traceback to _MAX_TB_LINES
        tb_lines = tb_text.strip().splitlines()
        if len(tb_lines) > _MAX_TB_LINES:
            tb_lines = tb_lines[-_MAX_TB_LINES:]
        short_tb = "\n".join(tb_lines)

        # Extract error type from the crash message
        message = cast(str, crash.get("message", ""))
        error_type = _failure_error_type(message)

        failures.append(
            FailureDetail(
                test=nodeid,
                error_type=error_type,
                message=message,
                file=cast(str, crash.get("path", "")),
                line=cast(int, crash.get("lineno", 0)),
                traceback=short_tb,
            )
        )
    return failures

parse_json_report(report_path)

Read and parse a pytest-json-report JSON file.

Source code in packages/axm-audit/src/axm_audit/core/test_runner.py

def parse_json_report(report_path: Path) -> dict[str, object]:
    """Read and parse a pytest-json-report JSON file."""
    try:
        raw_report = json.loads(report_path.read_text())
    except (json.JSONDecodeError, OSError) as exc:
        msg = f"Invalid pytest JSON report at {report_path}: {exc}"
        raise ValueError(msg) from exc

    if not isinstance(raw_report, dict):
        msg = f"Invalid pytest JSON report at {report_path}: root is not an object"
        raise ValueError(msg)
    if not isinstance(raw_report.get("summary"), dict):
        msg = f"Invalid pytest JSON report at {report_path}: missing object 'summary'"
        raise ValueError(msg)
    return cast("dict[str, object]", raw_report)

run_tests(project_path, *, mode='failures', files=None, markers=None, stop_on_first=True, include_cases=False)

Run tests with agent-optimized structured output.

Parameters:

Name	Type	Description	Default
project_path	Path	Root of the project to test.	required
mode	str	Accepted for backward compatibility but ignored — all modes now produce the same unified output (failures + coverage).	'failures'
files	list[str] | None	Specific test files or paths to run.	None
markers	list[str] | None	Pytest markers to filter (-m).	None
stop_on_first	bool	Stop on first failure (-x).	True

Returns:

Type	Description
TestReport	Structured TestReport with failures and coverage populated.

Source code in packages/axm-audit/src/axm_audit/core/test_runner.py

def run_tests(
    project_path: Path,
    *,
    mode: str = "failures",
    files: list[str] | None = None,
    markers: list[str] | None = None,
    stop_on_first: bool = True,
    include_cases: bool = False,
) -> TestReport:
    """Run tests with agent-optimized structured output.

    Args:
        project_path: Root of the project to test.
        mode: Accepted for backward compatibility but ignored — all modes
            now produce the same unified output (failures + coverage).
        files: Specific test files or paths to run.
        markers: Pytest markers to filter (``-m``).
        stop_on_first: Stop on first failure (``-x``).

    Returns:
        Structured ``TestReport`` with failures and coverage populated.
    """
    # Create temp files for reports
    report_tmp = tempfile.NamedTemporaryFile(
        suffix=".json", prefix="axm_report_", delete=False
    )
    report_path = Path(report_tmp.name)
    report_tmp.close()

    cov_tmp = tempfile.NamedTemporaryFile(
        suffix=".json", prefix="axm_cov_", delete=False
    )
    coverage_path = Path(cov_tmp.name)
    cov_tmp.close()

    try:
        effective_cov_path = None if files else coverage_path
        cmd = build_pytest_cmd(
            report_path=report_path,
            coverage_path=effective_cov_path,
            files=files,
            markers=markers,
            stop_on_first=stop_on_first,
        )

        logger.debug("Running: %s", " ".join(cmd))
        result = run_in_project(
            cmd,
            project_path,
            timeout=_COVERAGE_RUN_TIMEOUT,
            with_packages=["pytest-json-report", "pytest-cov"],
            capture_output=True,
            text=True,
            check=False,
        )

        # A timed-out subprocess (synthetic returncode 124) leaves the
        # report/coverage JSON truncated. Surface the timeout explicitly
        # instead of parsing a fabricated coverage % from partial data.
        if result.returncode == _TIMEOUT_RETURNCODE:
            logger.warning(
                "Test run timed out after %ds — coverage not measured",
                _COVERAGE_RUN_TIMEOUT,
            )
            if include_cases:
                raise ValueError(
                    f"pytest subprocess timed out after {_COVERAGE_RUN_TIMEOUT}s"
                )
            return TestReport(
                timed_out=True,
                pytest_return_code=result.returncode,
                collected=0,
                verdict=False,
            )

        # Parse JSON report. A subprocess that died before pytest could write
        # the report (failed uv resolution, missing plugin, collection crash)
        # leaves it absent or empty: the bare JSON decode error then describes a
        # corrupt file and hides the real cause, which only the captured stderr
        # carries. Re-raise it enriched instead.
        try:
            report_data = parse_json_report(report_path)
        except ValueError as exc:
            raise _subprocess_failure(exc, result) from exc

        # Parse coverage
        total_cov, per_file_cov = (
            parse_coverage(coverage_path) if not files else (None, {})
        )

        report = build_test_report(
            report_data=report_data,
            total_cov=total_cov,
            per_file_cov=per_file_cov,
            include_cases=include_cases,
            return_code=result.returncode,
            stdout=result.stdout or "",
            stderr=result.stderr or "",
        )
        report.record_execution(
            return_code=result.returncode,
            collected=_collected_count(report_data),
            target_statuses=_build_target_statuses(
                project_path,
                files,
                report_data,
            ),
        )
        return report

    finally:
        report_path.unlink(missing_ok=True)
        coverage_path.unlink(missing_ok=True)