engine.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/mutation/engine.py
 
"""
Purpose
-------
Enumerate and apply deterministic source mutants over the pinned v1 operator set.
 
Responsibilities
----------------
- Derive every mutant of a module purely from its syntax tree, in source order.
- Give each mutant a stable ``relpath:line:col:op-code`` identity.
- Render a readable before/after preview for each mutant without touching disk.
- Apply exactly one named mutant to a source string and return the new text.
 
Diagnostics
-----------
Domain: OCT-MUTATION
Levels:
    L2 - lifecycle
    L3 - enumeration details
    L4 - deep tracing
 
Contracts
---------
- The same source text always yields the same mutants with the same ids.
- Enumeration is read-only; nothing in this module writes to the filesystem.
- A mutant never reproduces the original source, so no enumerated mutant is a
  guaranteed survivor by construction.
- ``SyntaxError`` propagates so callers report an unparsable target honestly.
 
Dependencies
------------
- ast (syntax tree, mutation, and source rendering)
- tokenize (declared source encodings)
"""
 
# OCT-LINT: disable-file=dbg_usage reason="Plan 227 P1 pins the engine as stdlib-only, so importing oc_diagnostics is forbidden"
# OCT-LINT: disable-file=dbg_import reason="Plan 227 P1 pins the engine as stdlib-only"
 
from __future__ import annotations
 
import ast
import copy
import tokenize
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Callable
 
 
class MutantNotFoundError(LookupError):
    """A named mutant id does not exist in the given source."""
 
 
# The pinned v1 operator set (plan 227 P2). Growth of these tables is future,
# evidence-driven work: string mutation and statement deletion are excluded
# deliberately because both generate more noise than signal.
_COMPARE_SWAPS: dict[type[ast.cmpop], tuple[type[ast.cmpop], str]] = {
    ast.Eq: (ast.NotEq, "EQ_NE"),
    ast.NotEq: (ast.Eq, "NE_EQ"),
    ast.Lt: (ast.GtE, "LT_GE"),
    ast.GtE: (ast.Lt, "GE_LT"),
    ast.Gt: (ast.LtE, "GT_LE"),
    ast.LtE: (ast.Gt, "LE_GT"),
}
 
_BINOP_SWAPS: dict[type[ast.operator], tuple[type[ast.operator], str]] = {
    ast.Add: (ast.Sub, "ADD_SUB"),
    ast.Sub: (ast.Add, "SUB_ADD"),
    ast.Mult: (ast.FloorDiv, "MUL_FLOORDIV"),
    ast.FloorDiv: (ast.Mult, "FLOORDIV_MUL"),
}
 
_BOOLOP_SWAPS: dict[type[ast.boolop], tuple[type[ast.boolop], str]] = {
    ast.And: (ast.Or, "AND_OR"),
    ast.Or: (ast.And, "OR_AND"),
}
 
 
@dataclass(frozen=True, slots=True)
class Mutant:
    """One enumerated mutant with its stable identity and readable preview."""
 
    mutant_id: str
    file: str
    lineno: int
    col_offset: int
    op_code: str
    before: str
    after: str
    line_text: str
 
    def to_dict(self) -> dict[str, str | int]:
        """Return a JSON-ready representation of every field."""
        return asdict(self)
 
 
@dataclass(slots=True)
class _Site:
    """An enumerated mutant bound to the in-tree edit that produces it."""
 
    mutant: Mutant
    apply: Callable[[], None]
 
 
class _SiteCollector(ast.NodeVisitor):
    """Walk a tree once, recording every v1 mutation site it contains.
 
    The collector records an ``apply`` closure per site instead of a second
    matching pass, so enumeration and application share one definition of
    where a mutant lives. Previews are rendered from throwaway copies, which
    keeps the visited tree pristine until a caller deliberately applies a site.
    """
 
    def __init__(self, relpath: str, lines: list[str]) -> None:
        self.relpath = relpath
        self.lines = lines
        self.sites: list[_Site] = []
 
    def visit_Compare(self, node: ast.Compare) -> None:
        """Record one swap per comparison operator in a (possibly chained) test.
 
        The anchor is the operand that FOLLOWS the operator rather than the
        ``Compare`` node itself: in ``a < b < c`` both operators share the
        node's coordinate, and two sites at one coordinate with one op-code
        would produce colliding ids.
        """
        for index, operator in enumerate(node.ops):
            swap = _COMPARE_SWAPS.get(type(operator))
            if swap is None:
                continue
            replacement, op_code = swap
            anchor = node.comparators[index]
            self._record(
                op_code=op_code,
                lineno=anchor.lineno,
                col_offset=anchor.col_offset,
                node=node,
                mutate_copy=lambda preview, i=index, r=replacement: (
                    preview.ops.__setitem__(i, r())
                ),
                apply_in_place=lambda i=index, r=replacement: (
                    node.ops.__setitem__(i, r())
                ),
            )
        self.generic_visit(node)
 
    def visit_BoolOp(self, node: ast.BoolOp) -> None:
        """Record the ``and``/``or`` swap for one boolean chain."""
        swap = _BOOLOP_SWAPS.get(type(node.op))
        if swap is not None:
            replacement, op_code = swap
            self._record(
                op_code=op_code,
                lineno=node.lineno,
                col_offset=node.col_offset,
                node=node,
                mutate_copy=lambda preview, r=replacement: setattr(
                    preview, "op", r()
                ),
                apply_in_place=lambda r=replacement: setattr(node, "op", r()),
            )
        self.generic_visit(node)
 
    def visit_BinOp(self, node: ast.BinOp) -> None:
        """Record an arithmetic swap, anchored on the right-hand operand.
 
        ``a + b + c`` parses as a left-nested pair of ``BinOp`` nodes that
        share a column, so the right operand is the only coordinate that
        distinguishes the two additions.
        """
        swap = _BINOP_SWAPS.get(type(node.op))
        if swap is not None:
            replacement, op_code = swap
            self._record(
                op_code=op_code,
                lineno=node.right.lineno,
                col_offset=node.right.col_offset,
                node=node,
                mutate_copy=lambda preview, r=replacement: setattr(
                    preview, "op", r()
                ),
                apply_in_place=lambda r=replacement: setattr(node, "op", r()),
            )
        self.generic_visit(node)
 
    def visit_Constant(self, node: ast.Constant) -> None:
        """Record numeric tweaks, excluding booleans, strings, and complex values.
 
        ``n -> 0`` is skipped when ``n`` is already zero: that mutant would be
        byte-identical to the original and would therefore survive every test
        suite, inflating the survivor list with a finding no test could ever
        kill.
        """
        value = node.value
        if isinstance(value, bool) or not isinstance(value, (int, float)):
            self.generic_visit(node)
            return
        self._record(
            op_code="NUM_INCR",
            lineno=node.lineno,
            col_offset=node.col_offset,
            node=node,
            mutate_copy=lambda preview: setattr(preview, "value", value + 1),
            apply_in_place=lambda: setattr(node, "value", value + 1),
        )
        if value != 0:
            # Zero keeps the literal's own type, so a float constant is
            # mutated to ``0.0`` rather than silently changing type as well.
            zero = type(value)(0)
            self._record(
                op_code="NUM_ZERO",
                lineno=node.lineno,
                col_offset=node.col_offset,
                node=node,
                mutate_copy=lambda preview: setattr(preview, "value", zero),
                apply_in_place=lambda: setattr(node, "value", zero),
            )
        self.generic_visit(node)
 
    def visit_If(self, node: ast.If) -> None:
        """Record a ``not`` insertion on an ``if`` (or ``elif``) test."""
        self._record_negation(node, "NOT_IF")
        self.generic_visit(node)
 
    def visit_While(self, node: ast.While) -> None:
        """Record a ``not`` insertion on a ``while`` test."""
        self._record_negation(node, "NOT_WHILE")
        self.generic_visit(node)
 
    def _record_negation(self, node: ast.If | ast.While, op_code: str) -> None:
        """Record the negation site unless the test is already negated.
 
        ``if not not value:`` has exactly the truth value of ``if value:``, so
        negating an existing ``not`` yields an equivalent mutant that can never
        be killed. Excluding it keeps the survivor list honest.
        """
        test = node.test
        if isinstance(test, ast.UnaryOp) and isinstance(test.op, ast.Not):
            return
        rendered = ast.unparse(test)
        self.sites.append(
            _Site(
                mutant=self._build_mutant(
                    op_code=op_code,
                    lineno=test.lineno,
                    col_offset=test.col_offset,
                    before=rendered,
                    after=f"not ({rendered})",
                ),
                apply=lambda: setattr(
                    node,
                    "test",
                    ast.fix_missing_locations(
                        ast.copy_location(
                            ast.UnaryOp(op=ast.Not(), operand=node.test), node.test
                        )
                    ),
                ),
            )
        )
 
    def _record(
        self,
        *,
        op_code: str,
        lineno: int,
        col_offset: int,
        node: ast.expr,
        mutate_copy: Callable[[ast.expr], None],
        apply_in_place: Callable[[], None],
    ) -> None:
        """Append one site, rendering its preview from a throwaway copy."""
        preview = copy.deepcopy(node)
        mutate_copy(preview)
        self.sites.append(
            _Site(
                mutant=self._build_mutant(
                    op_code=op_code,
                    lineno=lineno,
                    col_offset=col_offset,
                    before=ast.unparse(node),
                    after=ast.unparse(preview),
                ),
                apply=apply_in_place,
            )
        )
 
    def _build_mutant(
        self,
        *,
        op_code: str,
        lineno: int,
        col_offset: int,
        before: str,
        after: str,
    ) -> Mutant:
        """Assemble one mutant record with its stable id and source line."""
        line_text = self.lines[lineno - 1].strip() if lineno <= len(self.lines) else ""
        return Mutant(
            mutant_id=f"{self.relpath}:{lineno}:{col_offset}:{op_code}",
            file=self.relpath,
            lineno=lineno,
            col_offset=col_offset,
            op_code=op_code,
            before=before,
            after=after,
            line_text=line_text,
        )
 
 
def _collect_tree_and_sites(
    source: str, relpath: str
) -> tuple[ast.Module, list[_Site]]:
    """Return the parsed tree alongside the sites that edit it in place."""
    tree = ast.parse(source, filename=relpath)
    collector = _SiteCollector(relpath, source.splitlines())
    collector.visit(tree)
    sites = sorted(
        collector.sites,
        key=lambda site: (
            site.mutant.lineno,
            site.mutant.col_offset,
            site.mutant.op_code,
        ),
    )
    return tree, sites
 
 
def enumerate_source(source: str, relpath: str) -> list[Mutant]:
    """Return every v1 mutant of ``source`` in deterministic source order."""
    return [site.mutant for site in _collect_tree_and_sites(source, relpath)[1]]
 
 
def enumerate_file(path: str | Path, root: str | Path | None = None) -> list[Mutant]:
    """Return every v1 mutant of one file, identified relative to ``root``."""
    resolved = Path(path).resolve()
    with tokenize.open(resolved) as source_file:
        source = source_file.read()
    return enumerate_source(source, relative_id(resolved, root))
 
 
def relative_id(path: str | Path, root: str | Path | None) -> str:
    """Return the forward-slash relative path used inside mutant ids.
 
    Ids are anchored to ``root`` so they stay stable when the same tree is
    checked out at a different absolute location; a target outside ``root``
    falls back to its bare filename rather than leaking an absolute path.
    """
    resolved = Path(path).resolve()
    if root is None:
        return resolved.name
    try:
        return resolved.relative_to(Path(root).resolve()).as_posix()
    except ValueError:
        return resolved.name
 
 
def apply_mutant(source: str, mutant_id: str, relpath: str) -> str:
    """Return ``source`` with exactly the named mutant applied.
 
    The result is re-rendered by :func:`ast.unparse`, so comments and original
    formatting are not preserved. That is safe only because the runner writes
    this text transiently and always restores the pristine bytes afterwards.
    """
    tree, sites = _collect_tree_and_sites(source, relpath)
    for site in sites:
        if site.mutant.mutant_id == mutant_id:
            site.apply()
            return ast.unparse(ast.fix_missing_locations(tree))
    raise MutantNotFoundError(f"no mutant with id {mutant_id!r} in {relpath}")