orchestrata_ultimate.factory.evidence.mutation

Reference › orchestrata_ultimate › factory › evidence › mutation

mutation provides affected-set mutation testing, evidence that a change's tests can actually fail (SPEC-033 section 9, row B6). Mutating a whole codebase for every change is neither needed nor affordable, so the module mutates only the units a change touched. affected_units lists the functions and methods defined in the changed files as UnitRef records, calling list_symbols from scripts.serialize_codebase_ast as an ordinary import instead of a subprocess. _generate_for_unit applies a small, closed set of operators to a copy of each unit's file, never to the working tree: bool_flip (flip a True or False literal), compare_swap (swap < with <= and == with !=) and early_return_none (insert return None as the first statement). Each Mutant is a full re-serialisation of the file with exactly one site changed, deep-copied from the parsed original.

run_mutations runs the tests chosen by a caller-supplied test_selector for each mutant, using the caller's runner_cmd, a per-mutant timeout and a cap limit on the number of mutants. It classifies each mutant as killed, survived or timed out and returns a MutationReport with counts, the Survivor records (mutants the tests did not catch) and any SkippedMutant records left over once the limit was reached. The scripts.factory_evidence CLI imports it; tests are in tests/factory/test_mutation.py.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

UnitRef (dataclass)

Decorators: dataclass(frozen=True) · Line 72

One mutation-testable unit: a function or method.

Mutant (dataclass)

Decorators: dataclass(frozen=True) · Line 82

One generated mutant: one operator applied at one site in one unit.

Survivor (dataclass)

Decorators: dataclass(frozen=True) · Line 93

A mutant whose selected tests passed unchanged -- the tests don't pin this unit's behaviour at this site.

SkippedMutant (dataclass)

Decorators: dataclass(frozen=True) · Line 103

A generated mutant never run because limit was reached.

MutationReport (dataclass)

Decorators: dataclass(frozen=True) · Line 112

Aggregate result of a mutation run.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/factory/evidence/mutation.py

"""Affected-set mutation testing (SPEC-033 §9, B6).

Mutating the WHOLE codebase per change is neither necessary nor affordable:
only the units a change actually touched need pinning tests. This module
(1) derives the affected units from the AST toolkit's existing serializer,
(2) applies a small, closed set of mutation operators to a COPY of each
unit's file (never the working tree), and (3) runs the caller-selected
tests for each mutant under a hard per-mutant timeout, classifying each as
killed / survived / timeout.

Affected-set source
--------------------
``affected_units`` calls ``scripts.serialize_codebase_ast.list_symbols``
directly (an ordinary Python import -- ``scripts`` is an importable package,
see ``src/orchestrata_ultimate/verification/spec_completeness.py`` for
precedent) rather than shelling out: the function already accepts a list of
individual file paths (mixed files/dirs, see ``_iter_py_files``), so no
subprocess or full-project JSON serialization is needed for a handful of
changed files.

Operator set (deliberately small and closed -- ADR-discipline: no
configurable operator plugins)
--------------------------------------------------------------------------
- ``bool_flip``: flip a boolean literal (``True``/``False``) constant.
- ``compare_swap``: swap a comparison operator (``<``<->``<=``,
  ``==``<->``!=``).
- ``early_return_none``: inject ``return None`` as the first statement of
  the unit's body (a canonical "the unit never ran" mutation).

Each mutant is a full re-serialization (``ast.unparse``) of the unit's file
AST with exactly one site changed, deep-copied from the parsed original so
concurrent mutants never share mutable AST state.
"""

from __future__ import annotations

import ast
import copy
import difflib
import logging
import shutil
import subprocess
import tempfile
from collections.abc import Callable, Sequence
from dataclasses import dataclass, field
from pathlib import Path
from typing import Literal

from scripts.serialize_codebase_ast import list_symbols

logger = logging.getLogger(__name__)

Operator = Literal["bool_flip", "compare_swap", "early_return_none"]

_OPERATORS: tuple[Operator, ...] = ("bool_flip", "compare_swap", "early_return_none")

# Directories never copied into a mutant's tmp workdir -- large/irrelevant
# to test execution and, for .git, actively wrong to duplicate.
_COPY_IGNORE = shutil.ignore_patterns(
    ".git", ".venv", "__pycache__", "*.pyc", ".pytest_cache", "*.egg-info"
)

_COMPARE_SWAP: dict[type[ast.cmpop], type[ast.cmpop]] = {
    ast.Lt: ast.LtE,
    ast.LtE: ast.Lt,
    ast.Eq: ast.NotEq,
    ast.NotEq: ast.Eq,
}


@dataclass(frozen=True)
class UnitRef:
    """One mutation-testable unit: a function or method."""

    file: str
    name: str  # qualified: "Class.method" for methods, bare name for funcs
    kind: Literal["func", "method"]
    line: int


@dataclass(frozen=True)
class Mutant:
    """One generated mutant: one operator applied at one site in one unit."""

    unit: UnitRef
    operator: Operator
    mutated_source: str
    diff_snippet: str
    site_line: int


@dataclass(frozen=True)
class Survivor:
    """A mutant whose selected tests passed unchanged -- the tests don't
    pin this unit's behaviour at this site."""

    unit: UnitRef
    operator: Operator
    diff_snippet: str


@dataclass(frozen=True)
class SkippedMutant:
    """A generated mutant never run because ``limit`` was reached."""

    unit: UnitRef
    operator: Operator
    reason: str


@dataclass(frozen=True)
class MutationReport:
    """Aggregate result of a mutation run."""

    killed: int
    survived: int
    timeout: int
    total_generated: int
    survivors: list[Survivor] = field(default_factory=list)
    skipped: list[SkippedMutant] = field(default_factory=list)


def affected_units(changed_files: Sequence[str]) -> list[UnitRef]:
    """Return the functions/methods defined in ``changed_files``.

    Uses the AST toolkit's ``list_symbols`` (an in-process import, not a
    subprocess -- see module docstring); class-definition rows are excluded
    since a class itself is not an executable mutation target.
    """
    rows = list_symbols(list(changed_files))
    return [
        UnitRef(file=r["file"], name=r["name"], kind=r["kind"], line=r["line"])
        for r in rows
        if r["kind"] in ("func", "method")
    ]


def _find_function(
    tree: ast.AST, name: str, kind: str
) -> ast.FunctionDef | ast.AsyncFunctionDef | None:
    """Locate the function/method node for a :class:`UnitRef` inside a
    freshly parsed (or deep-copied) tree of its file."""
    if kind == "method" and "." in name:
        cls_name, meth_name = name.split(".", 1)
        for node in ast.walk(tree):
            if isinstance(node, ast.ClassDef) and node.name == cls_name:
                for item in node.body:
                    if (
                        isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef))
                        and item.name == meth_name
                    ):
                        return item
        return None
    for node in ast.walk(tree):
        if (
            isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
            and node.name == name
        ):
            return node
    return None


def _unparse_diff(original_tree: ast.AST, mutated_tree: ast.AST) -> str:
    """A compact unified diff between the unparsed original and mutated
    trees -- shows only the lines the single-site mutation actually changed."""
    orig_lines = ast.unparse(original_tree).splitlines(keepends=True)
    mut_lines = ast.unparse(mutated_tree).splitlines(keepends=True)
    return "".join(difflib.unified_diff(orig_lines, mut_lines, lineterm="", n=1))


def _bool_flip_sites(func_node: ast.AST) -> list[ast.Constant]:
    sites: list[ast.Constant] = []
    for node in ast.walk(func_node):
        if isinstance(node, ast.Constant) and isinstance(node.value, bool):
            sites.append(node)
    return sites


def _compare_swap_sites(func_node: ast.AST) -> list[tuple[ast.Compare, int]]:
    sites: list[tuple[ast.Compare, int]] = []
    for node in ast.walk(func_node):
        if isinstance(node, ast.Compare):
            for i, op in enumerate(node.ops):
                if type(op) in _COMPARE_SWAP:
                    sites.append((node, i))
    return sites


def _generate_for_unit(unit: UnitRef, source: str) -> list[Mutant]:
    """Generate every mutant for ``unit`` given its file's source text."""
    original_tree = ast.parse(source)
    func = _find_function(original_tree, unit.name, unit.kind)
    if func is None:
        return []

    mutants: list[Mutant] = []

    # bool_flip: one mutant per boolean-literal site.
    for site in _bool_flip_sites(func):
        tree_copy = copy.deepcopy(original_tree)
        func_copy = _find_function(tree_copy, unit.name, unit.kind)
        assert func_copy is not None
        for cand in ast.walk(func_copy):
            if (
                isinstance(cand, ast.Constant)
                and cand.lineno == site.lineno
                and cand.col_offset == site.col_offset
                and isinstance(cand.value, bool)
            ):
                cand.value = not cand.value
                break
        ast.fix_missing_locations(tree_copy)
        mutants.append(
            Mutant(
                unit=unit,
                operator="bool_flip",
                mutated_source=ast.unparse(tree_copy),
                diff_snippet=_unparse_diff(original_tree, tree_copy),
                site_line=site.lineno,
            )
        )

    # compare_swap: one mutant per swappable comparison-operator site.
    for site_node, op_index in _compare_swap_sites(func):
        tree_copy = copy.deepcopy(original_tree)
        func_copy = _find_function(tree_copy, unit.name, unit.kind)
        assert func_copy is not None
        for cand in ast.walk(func_copy):
            if (
                isinstance(cand, ast.Compare)
                and cand.lineno == site_node.lineno
                and cand.col_offset == site_node.col_offset
            ):
                cand.ops[op_index] = _COMPARE_SWAP[type(cand.ops[op_index])]()
                break
        ast.fix_missing_locations(tree_copy)
        mutants.append(
            Mutant(
                unit=unit,
                operator="compare_swap",
                mutated_source=ast.unparse(tree_copy),
                diff_snippet=_unparse_diff(original_tree, tree_copy),
                site_line=site_node.lineno,
            )
        )

    # early_return_none: exactly one mutant, at the function head.
    tree_copy = copy.deepcopy(original_tree)
    func_copy = _find_function(tree_copy, unit.name, unit.kind)
    assert func_copy is not None
    func_copy.body.insert(0, ast.Return(value=ast.Constant(value=None)))
    ast.fix_missing_locations(tree_copy)
    mutants.append(
        Mutant(
            unit=unit,
            operator="early_return_none",
            mutated_source=ast.unparse(tree_copy),
            diff_snippet=_unparse_diff(original_tree, tree_copy),
            site_line=func.lineno,
        )
    )

    return mutants


def run_mutations(
    units: Sequence[UnitRef],
    test_selector: Callable[[UnitRef], Sequence[str]],
    runner_cmd: Sequence[str],
    limit: int = 50,
    timeout: float = 30.0,
    repo_root: str | Path = ".",
) -> MutationReport:
    """Generate and run mutants for ``units``, capped at ``limit`` total.

    ``test_selector(unit)`` returns the pytest test ids/paths to run for
    that unit -- selection is entirely the caller's responsibility (see
    module docstring: no ``TestMatcher`` reuse here, its contract needs a
    full serialized ``ASTCodebase`` pair via ``JSONASTLoader``, a mismatch
    for this task's tiny-fixture / real-subprocess design).

    ``runner_cmd`` is an argv prefix (e.g. ``["uv", "run", "pytest", "-q"]``)
    -- the selected test ids are appended and the whole command runs with
    ``cwd`` set to a tmp COPY of ``repo_root`` (the real working tree is
    never touched), under a hard ``timeout`` per mutant.

    Mutants beyond ``limit`` are never silently dropped: each is recorded in
    ``MutationReport.skipped`` and logged.
    """
    repo_root = Path(repo_root)

    all_mutants: list[Mutant] = []
    for unit in units:
        source = Path(unit.file).read_text(encoding="utf-8")
        all_mutants.extend(_generate_for_unit(unit, source))

    to_run = all_mutants[:limit]
    skipped_mutants = all_mutants[limit:]
    skipped = [
        SkippedMutant(
            unit=m.unit,
            operator=m.operator,
            reason=f"mutation limit ({limit}) reached",
        )
        for m in skipped_mutants
    ]
    for s in skipped:
        logger.warning(
            "mutation skipped (limit=%d): %s %s at %s",
            limit,
            s.unit.name,
            s.operator,
            s.unit.file,
        )

    killed = survived = timed_out = 0
    survivors: list[Survivor] = []

    if not to_run:
        return MutationReport(
            killed=0,
            survived=0,
            timeout=0,
            total_generated=len(all_mutants),
            survivors=[],
            skipped=skipped,
        )

    with tempfile.TemporaryDirectory(prefix="mutation_") as tmpdir:
        workdir = Path(tmpdir) / "repo"
        shutil.copytree(repo_root, workdir, symlinks=True, ignore=_COPY_IGNORE)

        for mutant in to_run:
            rel = Path(mutant.unit.file).resolve().relative_to(repo_root.resolve())
            target = workdir / rel
            original_bytes = target.read_bytes()
            target.write_text(mutant.mutated_source, encoding="utf-8")
            try:
                tests = list(test_selector(mutant.unit))
                if not tests:
                    survived += 1
                    survivors.append(
                        Survivor(
                            unit=mutant.unit,
                            operator=mutant.operator,
                            diff_snippet=mutant.diff_snippet,
                        )
                    )
                    continue
                try:
                    result = subprocess.run(
                        [*runner_cmd, *tests],
                        cwd=workdir,
                        capture_output=True,
                        timeout=timeout,
                        check=False,
                    )
                except subprocess.TimeoutExpired:
                    timed_out += 1
                    continue
                if result.returncode == 0:
                    survived += 1
                    survivors.append(
                        Survivor(
                            unit=mutant.unit,
                            operator=mutant.operator,
                            diff_snippet=mutant.diff_snippet,
                        )
                    )
                else:
                    killed += 1
            finally:
                target.write_bytes(original_bytes)

    return MutationReport(
        killed=killed,
        survived=survived,
        timeout=timed_out,
        total_generated=len(all_mutants),
        survivors=survivors,
        skipped=skipped,
    )