libakerror/scripts/mutation_test.py

#!/usr/bin/env python3
"""
Mutation testing harness for libakerror.

Mutation testing measures how good the test suite is at catching bugs. It works
by making many small, deliberate breakages ("mutants") to the library source --
flipping a comparison, deleting a statement, swapping true/false -- and then
running the whole CTest suite against each one. If the tests fail, the mutant is
"killed" (good: the tests noticed the bug). If the tests still pass, the mutant
"survived" (bad: a real bug of that shape would slip through unnoticed).

The mutation score is killed / (killed + survived). Surviving mutants are printed
with file:line and the exact change so they can be turned into new test cases.

This harness has no third-party dependencies (Python stdlib + the project's
normal cmake/ctest toolchain). It never mutates the real working tree: it copies
the repo to a scratch directory and mutates there.

Usage:
    scripts/mutation_test.py [options]

    --source-root DIR   repo root to copy (default: parent of this script's dir)
    --target FILE       source file to mutate, relative to root; repeatable.
                        Default: src/error.c and include/akerror.tmpl.h
    --work DIR          scratch dir for the mutated copy (default: a temp dir)
    --timeout SECONDS   per-suite ctest timeout (default: 120)
    --threshold PCT     exit non-zero if mutation score < PCT (default: 0 = off)
    --list              only list the mutants that would be run, then exit
    --keep              keep the scratch working copy on exit (for debugging)
    -j N                (reserved) currently runs sequentially
"""

import argparse
import os
import re
import shutil
import subprocess
import sys
import tempfile

# --------------------------------------------------------------------------- #
# Mutation operators
#
# Each operator yields zero or more (start, end, replacement) edits for a single
# line of source. The driver applies exactly one edit per mutant so every mutant
# differs from the original by one localized change.
# --------------------------------------------------------------------------- #

# Relational operator replacement: map each operator to the alternatives that
# meaningfully change behaviour (not merely the strict negation).
_REL = {
    "==": ["!="],
    "!=": ["=="],
    "<=": ["<", "=="],
    ">=": [">", "=="],
    "<": ["<=", ">"],
    ">": [">=", "<"],
}
# Match a relational operator that is NOT part of ->, <<, >>, =>, <=, >=, ==, !=
# unless we intend it. We tokenize the two-char operators first, then single.
_REL_TWO = re.compile(r"(==|!=|<=|>=)")
_REL_ONE = re.compile(r"(?<![-<>=!+])([<>])(?![=<>])")

_LOGICAL = {"&&": "||", "||": "&&"}
_LOG_RE = re.compile(r"(&&|\|\|)")

_BOOL = {"true": "false", "false": "true"}
_BOOL_RE = re.compile(r"\b(true|false)\b")

# Arithmetic / compound-assignment on whitespace-delimited operands only, to
# avoid touching ++, --, ->, unary signs, or pointer/format punctuation.
_ARITH_RE = re.compile(r"(?<=\s)([+\-])(?=\s)")
_ARITH = {"+": "-", "-": "+"}
_COMPOUND_RE = re.compile(r"(\+=|-=)")
_COMPOUND = {"+=": "-=", "-=": "+="}

# Integer literal replacement: 0 <-> 1 (word-bounded, not inside identifiers or
# larger numbers, not a float).
_INT_RE = re.compile(r"(?<![\w.])([01])(?![\w.])")
_INT = {"0": "1", "1": "0"}


def _op_edits(line):
    """Yield (tag, start, end, replacement) for every candidate mutation."""
    # Relational (two-char first so we don't split them with the one-char pass)
    for m in _REL_TWO.finditer(line):
        for alt in _REL[m.group(1)]:
            yield ("ROR", m.start(1), m.end(1), alt)
    for m in _REL_ONE.finditer(line):
        for alt in _REL[m.group(1)]:
            yield ("ROR", m.start(1), m.end(1), alt)
    for m in _LOG_RE.finditer(line):
        yield ("LCR", m.start(1), m.end(1), _LOGICAL[m.group(1)])
    for m in _BOOL_RE.finditer(line):
        yield ("BCR", m.start(1), m.end(1), _BOOL[m.group(1)])
    for m in _COMPOUND_RE.finditer(line):
        yield ("AOR", m.start(1), m.end(1), _COMPOUND[m.group(1)])
    for m in _ARITH_RE.finditer(line):
        yield ("AOR", m.start(1), m.end(1), _ARITH[m.group(1)])
    for m in _INT_RE.finditer(line):
        yield ("ICR", m.start(1), m.end(1), _INT[m.group(1)])


# Statement-deletion: neutralize a whole statement. We only delete statements
# that are safe to drop without guaranteeing a compile error, so a surviving
# deletion is a genuine test gap rather than compiler noise.
_STMT_DELETABLE = re.compile(
    r"""^\s*(
          break |
          return\b[^;]* |
          [A-Za-z_][-\w>().\[\]* ]*\s*=\s*[^;]* |   # assignments
          [A-Za-z_][\w]*\s*\([^;]*\)                # bare function calls
        )\s*;\s*(\\?)\s*$""",
    re.VERBOSE,
)


# --------------------------------------------------------------------------- #
# Deciding which lines are eligible to mutate
# --------------------------------------------------------------------------- #

# Skip preprocessor control and the block of constant/error-code #defines in the
# template header: mutating buffer sizes or renumbering error codes produces
# equivalent or uninteresting mutants that swamp the signal.
_SKIP_LINE = re.compile(
    r"""^\s*(
          \#\s*(include|ifn?def|ifdef|if|elif|else|endif|error|pragma|undef) |
          \#\s*define\s+AKERR_(MAX|LAST|NULLPOINTER|OUTOFBOUNDS|API|ATTRIBUTE|
              TYPE|KEY|INDEX|FORMAT|IO|VALUE|RELATIONSHIP|EOF|CIRCULAR_REFERENCE|
              ITERATOR_BREAK|NOT_IMPLEMENTED|BADEXC|NOIGNORE|USE_STDLIB)\b |
          \*  | //                                   # comment bodies / line comments
        )""",
    re.VERBOSE,
)


def _is_comment_or_blank(line):
    s = line.strip()
    return (not s) or s.startswith("//") or s.startswith("/*") or s.startswith("*")


def eligible(line):
    if _is_comment_or_blank(line):
        return False
    if _SKIP_LINE.match(line):
        return False
    return True


class Mutant:
    __slots__ = ("path", "lineno", "op", "before", "after", "col")

    def __init__(self, path, lineno, op, before, after, col):
        self.path = path
        self.lineno = lineno
        self.op = op
        self.before = before
        self.after = after
        self.col = col

    def describe(self):
        return (f"{self.path}:{self.lineno} [{self.op}] "
                f"col{self.col}: {self.before.strip()}  ->  {self.after.strip()}")


def generate_mutants(root, rel_target):
    """Enumerate all mutants for one target file."""
    abspath = os.path.join(root, rel_target)
    with open(abspath, "r") as fh:
        lines = fh.readlines()

    mutants = []
    for i, line in enumerate(lines, start=1):
        if not eligible(line):
            continue
        # substitution operators
        seen = set()
        for tag, s, e, repl in _op_edits(line):
            key = (s, e, repl)
            if key in seen:
                continue
            seen.add(key)
            mutated = line[:s] + repl + line[e:]
            if mutated == line:
                continue
            mutants.append(Mutant(rel_target, i, tag, line, mutated, s))
        # statement deletion
        m = _STMT_DELETABLE.match(line)
        if m:
            indent = line[: len(line) - len(line.lstrip())]
            cont = "\\" if line.rstrip().endswith("\\") else ""
            deleted = f"{indent}/* mutant: deleted */ {cont}\n" if cont else f"{indent};\n"
            mutants.append(Mutant(rel_target, i, "SDL", line, deleted, 0))
    return mutants


# --------------------------------------------------------------------------- #
# Build / test orchestration against a scratch copy
# --------------------------------------------------------------------------- #

class Runner:
    def __init__(self, work, timeout):
        self.work = work
        self.build = os.path.join(work, "build")
        self.timeout = timeout

    def _run(self, cmd, timeout=None):
        return subprocess.run(
            cmd, cwd=self.work, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
            timeout=timeout,
        )

    def configure(self):
        r = self._run(["cmake", "-S", ".", "-B", "build"], timeout=self.timeout)
        return r.returncode == 0, r.stdout

    def build_and_test(self):
        """Return ('killed-compile' | 'killed-test' | 'killed-timeout' | 'survived')."""
        try:
            b = self._run(["cmake", "--build", "build"], timeout=self.timeout)
        except subprocess.TimeoutExpired:
            return "killed-timeout"
        if b.returncode != 0:
            return "killed-compile"
        try:
            t = subprocess.run(
                ["ctest", "--test-dir", "build", "--output-on-failure",
                 "--stop-on-failure"],
                cwd=self.work, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                timeout=self.timeout,
            )
        except subprocess.TimeoutExpired:
            return "killed-timeout"
        return "survived" if t.returncode == 0 else "killed-test"


def copy_tree(src, dst):
    ignore = shutil.ignore_patterns("build", ".git", "*.o", "*.so", "*~",
                                    "#*#", "*.iso", "*.png")
    shutil.copytree(src, dst, ignore=ignore, symlinks=True)


def read_lines(path):
    with open(path) as fh:
        return fh.readlines()


def write_lines(path, lines):
    with open(path, "w") as fh:
        fh.writelines(lines)


def main():
    # Line-buffer stdout so progress is visible live under CI / the cmake target.
    try:
        sys.stdout.reconfigure(line_buffering=True)
    except (AttributeError, ValueError):
        pass
    here = os.path.dirname(os.path.abspath(__file__))
    default_root = os.path.dirname(here)

    ap = argparse.ArgumentParser(description="Mutation testing for libakerror")
    ap.add_argument("--source-root", default=default_root)
    ap.add_argument("--target", action="append", default=None)
    ap.add_argument("--work", default=None)
    ap.add_argument("--timeout", type=int, default=120)
    ap.add_argument("--threshold", type=float, default=0.0)
    ap.add_argument("--max-mutants", type=int, default=0,
                    help="cap the run at N evenly-sampled mutants (0 = all)")
    ap.add_argument("--list", action="store_true")
    ap.add_argument("--keep", action="store_true")
    ap.add_argument("-j", type=int, default=1)
    args = ap.parse_args()

    root = os.path.abspath(args.source_root)
    targets = args.target or ["src/error.c", "include/akerror.tmpl.h"]

    # Enumerate mutants from the pristine sources.
    all_mutants = []
    for t in targets:
        all_mutants.extend(generate_mutants(root, t))

    print(f"Generated {len(all_mutants)} mutants across {len(targets)} file(s):")
    for t in targets:
        n = sum(1 for m in all_mutants if m.path == t)
        print(f"    {t}: {n}")

    # Optional even-strided sampling to bound run time (CI / smoke tests).
    if args.max_mutants and len(all_mutants) > args.max_mutants:
        step = len(all_mutants) / args.max_mutants
        sampled = [all_mutants[int(i * step)] for i in range(args.max_mutants)]
        print(f"Sampling {len(sampled)} of {len(all_mutants)} mutants "
              f"(--max-mutants {args.max_mutants}).")
        all_mutants = sampled

    if args.list:
        for m in all_mutants:
            print("  " + m.describe())
        return 0

    if not all_mutants:
        print("No mutants generated; nothing to do.")
        return 0

    # Scratch working copy.
    work_parent = args.work or tempfile.mkdtemp(prefix="akerr_mut_")
    work = os.path.join(work_parent, "src") if args.work else work_parent
    if os.path.exists(work):
        shutil.rmtree(work)
    print(f"\nCopying sources to scratch dir: {work}")
    copy_tree(root, work)

    runner = Runner(work, args.timeout)

    print("Configuring baseline ...")
    ok, out = runner.configure()
    if not ok:
        sys.stderr.write(out.decode(errors="replace"))
        sys.stderr.write("\nBaseline configure FAILED; aborting.\n")
        return 2

    print("Verifying baseline is green (no mutation) ...")
    baseline = runner.build_and_test()
    if baseline != "survived":
        sys.stderr.write(f"Baseline is not green ({baseline}); aborting. "
                         "Fix the suite before mutation testing.\n")
        return 2
    print("Baseline OK.\n")

    # Group mutants by file so we mutate one file at a time and restore it.
    killed = {"killed-compile": 0, "killed-test": 0, "killed-timeout": 0}
    survivors = []
    total = len(all_mutants)

    # Cache pristine contents per target.
    pristine = {t: read_lines(os.path.join(work, t)) for t in targets}

    for idx, m in enumerate(all_mutants, start=1):
        tgt_abs = os.path.join(work, m.path)
        lines = list(pristine[m.path])
        lines[m.lineno - 1] = m.after
        write_lines(tgt_abs, lines)
        try:
            result = runner.build_and_test()
        finally:
            write_lines(tgt_abs, pristine[m.path])  # always restore

        if result == "survived":
            survivors.append(m)
            mark = "SURVIVED"
        else:
            killed[result] += 1
            mark = result.upper()
        print(f"[{idx}/{total}] {mark:16} {m.describe()}")

    total_killed = sum(killed.values())
    score = 100.0 * total_killed / total if total else 100.0

    print("\n" + "=" * 72)
    print("MUTATION TESTING SUMMARY")
    print("=" * 72)
    print(f"  total mutants   : {total}")
    print(f"  killed (test)   : {killed['killed-test']}")
    print(f"  killed (compile): {killed['killed-compile']}")
    print(f"  killed (timeout): {killed['killed-timeout']}")
    print(f"  survived        : {len(survivors)}")
    print(f"  mutation score  : {score:.1f}%")
    if survivors:
        print("\nSurviving mutants (test-suite gaps -- turn these into tests):")
        for m in survivors:
            print("  " + m.describe())

    if not args.keep and not args.work:
        shutil.rmtree(work_parent, ignore_errors=True)
    else:
        print(f"\nScratch working copy kept at: {work}")

    if args.threshold > 0 and score < args.threshold:
        print(f"\nFAIL: mutation score {score:.1f}% < threshold {args.threshold:.1f}%")
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())