OBLITERATUS/scripts/check_mutation_targets.py

#!/usr/bin/env python3
"""Fail closed when configured mutmut targets produce no mutants."""

from __future__ import annotations

import argparse
from importlib import metadata
import json
from pathlib import Path
from typing import Any

try:  # Python 3.11+
    import tomllib
except ModuleNotFoundError:  # pragma: no cover - exercised on Python 3.10
    import tomli as tomllib  # type: ignore[no-redef]


GLOB_MARKERS = frozenset("*?[")
SUPPORTED_MUTMUT_VERSION = "3.7.0"
MUTMUT_NO_TEST_EXIT_CODES = frozenset({5, 33})


def load_mutmut_config(pyproject: Path) -> dict[str, Any]:
    """Return the `[tool.mutmut]` table from pyproject.toml."""
    data = tomllib.loads(pyproject.read_text(encoding="utf-8"))
    mutmut_config = data.get("tool", {}).get("mutmut", {})
    if not isinstance(mutmut_config, dict):
        raise ValueError("[tool.mutmut] must be a table")
    return mutmut_config


def exact_python_targets(mutmut_config: dict[str, Any]) -> list[Path]:
    """Return exact `.py` paths that must enumerate mutants."""
    raw_targets = mutmut_config.get("required_mutation_targets", mutmut_config.get("only_mutate", []))
    if not isinstance(raw_targets, list) or not all(isinstance(path, str) for path in raw_targets):
        raise ValueError(
            "[tool.mutmut].required_mutation_targets must be a list of strings",
        )

    targets: list[Path] = []
    strict_required = "required_mutation_targets" in mutmut_config
    for raw_path in raw_targets:
        path = Path(raw_path)
        invalid = (
            path.is_absolute()
            or ".." in path.parts
            or not raw_path.endswith(".py")
            or any(marker in raw_path for marker in GLOB_MARKERS)
        )
        if invalid:
            if strict_required:
                raise ValueError(f"invalid required mutation target: {raw_path}")
            continue
        targets.append(path)
    return sorted(targets, key=str)


def _owned_mutants_dir(project_root: Path, mutants_dir: Path) -> Path:
    if mutants_dir != Path("mutants"):
        raise ValueError(
            "mutants directory must be project-owned as the literal project-owned mutants directory",
        )
    root = project_root.resolve()
    configured = root / "mutants"
    if configured.is_symlink():
        raise ValueError(
            "mutants directory must be project-owned as the literal project-owned mutants directory",
        )
    return configured


def _owned_artifact(project_root: Path, mutants_dir: Path, target: Path, suffix: str) -> Path:
    configured = _owned_mutants_dir(project_root, mutants_dir)
    artifact = configured / f"{target}{suffix}"
    resolved = artifact.resolve() if artifact.exists() else artifact.resolve(strict=False)
    try:
        resolved.relative_to(configured)
    except ValueError as exc:
        raise ValueError(f"mutation artifact escapes project-owned mutants/: {artifact}") from exc
    return artifact


def _validate_owned_source(project_root: Path, target: Path) -> Path:
    source_path = project_root / target
    if not source_path.is_file():
        raise ValueError(f"configured mutation target does not exist: {target}")
    try:
        source_path.resolve().relative_to(project_root.resolve())
    except ValueError as exc:
        raise ValueError(f"configured mutation target escapes project root: {target}") from exc
    return source_path


def mutant_count(meta_path: Path) -> int:
    """Return the number of generated mutants recorded in a mutmut metadata file."""
    try:
        metadata = json.loads(meta_path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        return 0
    except json.JSONDecodeError as exc:
        raise ValueError(f"{meta_path} is not valid JSON: {exc}") from exc

    exit_code_by_key = metadata.get("exit_code_by_key")
    if not isinstance(exit_code_by_key, dict):
        raise ValueError(f"{meta_path} does not contain an exit_code_by_key object")
    return len(exit_code_by_key)


def _installed_mutmut_version() -> str | None:
    try:
        return metadata.version("mutmut")
    except metadata.PackageNotFoundError:
        return None


def assert_supported_mutmut_no_test_codes() -> None:
    version = _installed_mutmut_version()
    if version != SUPPORTED_MUTMUT_VERSION:
        raise ValueError(
            f"unsupported mutmut version {version!r}; expected {SUPPORTED_MUTMUT_VERSION} "
            "before interpreting no-test exit codes",
        )


def no_test_mutants(meta_path: Path) -> list[str]:
    """Return required-target mutants that mutmut marked as having no covering tests."""
    try:
        metadata = json.loads(meta_path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        return []
    except json.JSONDecodeError as exc:
        raise ValueError(f"{meta_path} is not valid JSON: {exc}") from exc

    exit_code_by_key = metadata.get("exit_code_by_key")
    if not isinstance(exit_code_by_key, dict):
        raise ValueError(f"{meta_path} does not contain an exit_code_by_key object")
    no_tests = sorted(
        key for key, exit_code in exit_code_by_key.items()
        if exit_code in MUTMUT_NO_TEST_EXIT_CODES
    )
    if no_tests:
        assert_supported_mutmut_no_test_codes()
    return no_tests


def stale_or_empty_targets(
    targets: list[Path], *, project_root: Path = Path("."), mutants_dir: Path = Path("mutants"),
) -> list[Path]:
    """Return configured targets whose mutant metadata is missing or empty."""
    stale: list[Path] = []
    _owned_mutants_dir(project_root, mutants_dir)
    for target in targets:
        _validate_owned_source(project_root, target)
        if mutant_count(_owned_artifact(project_root, mutants_dir, target, ".meta")) == 0:
            stale.append(target)
    return stale


def prepare_required_targets(
    pyproject: Path = Path("pyproject.toml"), *, mutants_dir: Path = Path("mutants"),
) -> list[Path]:
    """Remove stale copied mutant files so mutmut regenerates required target metadata."""
    targets = exact_python_targets(load_mutmut_config(pyproject))
    stale = stale_or_empty_targets(targets, project_root=pyproject.parent, mutants_dir=mutants_dir)
    for target in stale:
        for suffix in ("", ".meta", ".spans"):
            artifact = _owned_artifact(pyproject.parent, mutants_dir, target, suffix)
            if artifact.exists():
                artifact.unlink()
    return stale


def validate_required_targets(
    pyproject: Path = Path("pyproject.toml"), *, mutants_dir: Path = Path("mutants"),
) -> list[str]:
    """Return fail-closed messages for exact mutation targets with zero generated mutants."""
    targets = exact_python_targets(load_mutmut_config(pyproject))
    failures: list[str] = []
    for target in stale_or_empty_targets(targets, project_root=pyproject.parent, mutants_dir=mutants_dir):
        failures.append(f"configured mutation target produced zero mutants: {target}")
    for target in targets:
        no_tests = no_test_mutants(_owned_artifact(pyproject.parent, mutants_dir, target, ".meta"))
        if no_tests:
            failures.append(
                f"configured mutation target has {len(no_tests)} mutant(s) with no tests: {target}",
            )
    return failures


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("command", choices=("prepare", "check"))
    parser.add_argument("--pyproject", type=Path, default=Path("pyproject.toml"))
    parser.add_argument("--mutants-dir", type=Path, default=Path("mutants"))
    return parser


def main() -> int:
    args = _parser().parse_args()
    try:
        if args.command == "prepare":
            stale = prepare_required_targets(args.pyproject, mutants_dir=args.mutants_dir)
            if stale:
                joined = ", ".join(str(path) for path in stale)
                print(f"mutation target guard invalidated stale metadata for: {joined}")
            else:
                print("mutation target guard found no stale required targets")
            return 0

        failures = validate_required_targets(args.pyproject, mutants_dir=args.mutants_dir)
    except (OSError, ValueError) as exc:
        print(f"mutation target guard failed: {exc}")
        return 1

    if failures:
        for failure in failures:
            print(f"mutation target guard failed: {failure}")
        return 1
    print("mutation target guard passed")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())