coverage.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/complexity/coverage.py
 
"""
Purpose
-------
Run pytest under coverage.py and map executed statements to function spans.
 
Responsibilities
----------------
- Execute subproject suites in process for witnesses or in a subprocess for probes.
- Preserve suite return codes and both output streams.
- Mark functions in unexecuted files as zero-coverage ``never-imported`` records.
- Intersect coverage.py statement data with the engine's pinned source spans.
 
Diagnostics
-----------
Domain: OCT-COMPLEXITY
Levels:
    L2 - suite lifecycle
    L3 - coverage mapping details
    L4 - deep tracing
 
Contracts
---------
- Third-party imports remain lazy so :mod:`oct.complexity` stays zero-dependency.
- Missing coverage.py produces an explicit error rather than an estimate.
- Failed suites retain their output for UNMEASURED probe reporting.
 
Dependencies
------------
- coverage (not imported; optional statement measurement and data reading, loaded by name through importlib and run as python -m coverage)
- pytest (not imported; test runner, loaded by name for in-process witnesses and run under coverage in probes)
- oct.complexity.engine (function spans)
"""
 
# OCT-LINT: disable-file=dbg_usage reason="Measurement output must remain deterministic and importing the package root must stay zero-dependency under Plan 214 P1"
# OCT-LINT: disable-file=dbg_import reason="Measurement output and lazy dependency boundaries must remain deterministic"
 
from __future__ import annotations
 
import contextlib
import importlib
import io
import os
import subprocess
import sys
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any, Sequence
 
from oct.complexity.engine import FunctionComplexity
 
 
class CoverageUnavailableError(RuntimeError):
    """Raised when a coverage-backed operation lacks coverage.py."""
 
 
@dataclass(frozen=True, slots=True)
class CoverageRun:
    """Captured outcome of one pytest suite executed under coverage.py."""
 
    returncode: int
    data_file: str
    command: tuple[str, ...]
    stdout: str
    stderr: str
 
    @property
    def output(self) -> str:
        """Return stdout and stderr without discarding either stream."""
        parts = [part.rstrip() for part in (self.stdout, self.stderr) if part]
        return "\n".join(parts)
 
 
@dataclass(frozen=True, slots=True)
class CoveredFunction:
    """A P1 complexity record enriched with pinned P2 coverage fields."""
 
    file: str
    qualname: str
    lineno: int
    end_lineno: int
    complexity: int
    coverage: float
    coverage_status: str
    covered_statements: int
    statement_count: int
 
    def to_dict(self) -> dict[str, str | int | float]:
        """Return every P1/P2 field in a JSON-ready dictionary."""
        return asdict(self)
 
 
def _import_coverage() -> Any:
    """Import coverage.py lazily and normalize the missing-extra error."""
    try:
        return importlib.import_module("coverage")
    except ImportError as exc:
        raise CoverageUnavailableError(
            "coverage.py is not installed; install oct[metrics] to map coverage"
        ) from exc
 
 
def _import_pytest() -> Any:
    """Import pytest lazily for the in-process witness runner."""
    try:
        return importlib.import_module("pytest")
    except ImportError as exc:
        raise CoverageUnavailableError(
            "pytest is not installed; install oct[test] to run a suite"
        ) from exc
 
 
def run_pytest_with_coverage(
    project_root: str | Path,
    pytest_args: Sequence[str] = ("tests",),
    data_file: str | Path = ".coverage",
    source: Sequence[str | Path] = (),
    *,
    in_process: bool = False,
    python_executable: str | Path | None = None,
    env: dict[str, str] | None = None,
) -> CoverageRun:
    """Execute a subproject's pytest suite and retain its coverage data.
 
    Production probes use the subprocess path so each subproject has clean
    import state. The in-process path exists for worktree tests, where the
    repository's editable install points at MAIN and subprocess assertions
    would otherwise verify the wrong source tree.
    """
    root = Path(project_root).resolve()
    resolved_data = Path(data_file)
    if not resolved_data.is_absolute():
        resolved_data = root / resolved_data
    resolved_data = resolved_data.resolve()
    resolved_data.parent.mkdir(parents=True, exist_ok=True)
    source_paths = tuple(str(Path(item).resolve()) for item in source)
    if in_process:
        return _run_pytest_in_process(
            root,
            tuple(pytest_args),
            resolved_data,
            source_paths,
        )
    return _run_pytest_subprocess(
        root,
        tuple(pytest_args),
        resolved_data,
        source_paths,
        python_executable=python_executable,
        env=env,
    )
 
 
def _run_pytest_in_process(
    project_root: Path,
    pytest_args: tuple[str, ...],
    data_file: Path,
    source: tuple[str, ...],
) -> CoverageRun:
    """Run pytest under a Coverage instance in the current interpreter."""
    coverage_module = _import_coverage()
    pytest_module = _import_pytest()
    coverage = coverage_module.Coverage(
        data_file=str(data_file),
        source=list(source) or None,
    )
    stdout = io.StringIO()
    stderr = io.StringIO()
    old_cwd = Path.cwd()
    sys.path.insert(0, str(project_root))
    command = ("pytest", *pytest_args)
    returncode = 1
    try:
        os.chdir(project_root)
        coverage.start()
        with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
            returncode = int(pytest_module.main(list(pytest_args)))
    finally:
        coverage.stop()
        coverage.save()
        os.chdir(old_cwd)
        _remove_first(sys.path, str(project_root))
    return CoverageRun(
        returncode=returncode,
        data_file=str(data_file),
        command=command,
        stdout=stdout.getvalue(),
        stderr=stderr.getvalue(),
    )
 
 
def _remove_first(items: list[str], value: str) -> None:
    """Remove the first matching value without disturbing prior path state."""
    try:
        items.remove(value)
    except ValueError:
        pass
 
 
def _run_pytest_subprocess(
    project_root: Path,
    pytest_args: tuple[str, ...],
    data_file: Path,
    source: tuple[str, ...],
    *,
    python_executable: str | Path | None,
    env: dict[str, str] | None,
) -> CoverageRun:
    """Run pytest under the coverage CLI in an isolated child process."""
    _import_coverage()
    command = [
        str(python_executable or sys.executable),
        "-m",
        "coverage",
        "run",
        f"--data-file={data_file}",
    ]
    if source:
        command.append(f"--source={','.join(source)}")
    command.extend(["-m", "pytest", *pytest_args])
    completed = subprocess.run(
        command,
        cwd=project_root,
        env=env,
        capture_output=True,
        text=True,
        errors="replace",
    )
    return CoverageRun(
        returncode=completed.returncode,
        data_file=str(data_file),
        command=tuple(command),
        stdout=completed.stdout,
        stderr=completed.stderr,
    )
 
 
def map_coverage(
    records: Sequence[FunctionComplexity],
    data_file: str | Path,
) -> list[CoveredFunction]:
    """Intersect coverage.py statements with each function's source span."""
    coverage_module = _import_coverage()
    resolved_data = Path(data_file).resolve()
    if not resolved_data.is_file():
        raise FileNotFoundError(f"coverage data file not found: {resolved_data}")
    coverage = coverage_module.Coverage(data_file=str(resolved_data))
    coverage.load()
    measured_files = {
        _normalized_path(filename): filename
        for filename in coverage.get_data().measured_files()
    }
    return [
        _map_one(coverage, measured_files, record)
        for record in records
    ]
 
 
def _map_one(
    coverage: Any,
    measured_files: dict[str, str],
    record: FunctionComplexity,
) -> CoveredFunction:
    """Map one function record, preserving unseen files as zero coverage."""
    measured_name = measured_files.get(_normalized_path(record.file))
    executed = (
        set(coverage.get_data().lines(measured_name) or ())
        if measured_name is not None
        else set()
    )
    status = "measured" if executed else "never-imported"
    statements = _statement_lines(coverage, record.file)
    span_statements = {
        line
        for line in statements
        if record.lineno <= line <= record.end_lineno
    }
    if status == "never-imported":
        covered_count = 0
        fraction = 0.0
    else:
        covered_count = len(span_statements & executed)
        fraction = (
            covered_count / len(span_statements)
            if span_statements
            else 1.0
        )
    return CoveredFunction(
        file=record.file,
        qualname=record.qualname,
        lineno=record.lineno,
        end_lineno=record.end_lineno,
        complexity=record.complexity,
        coverage=fraction,
        coverage_status=status,
        covered_statements=covered_count,
        statement_count=len(span_statements),
    )
 
 
def _statement_lines(coverage: Any, filename: str) -> set[int]:
    """Return executable statement lines using coverage.py's own parser."""
    _filename, statements, _excluded, _missing, _branches = coverage.analysis2(
        filename
    )
    return set(statements)
 
 
def _normalized_path(filename: str | Path) -> str:
    """Normalize measured filenames for case-insensitive Windows matching."""
    return os.path.normcase(str(Path(filename).resolve()))