scripts/mutation_test.py

#!/usr/bin/env python3
"""
Mutation testing — finds gaps in your tests.

It makes small changes to your source code (mutations), runs the tests,
and reports which mutations SURVIVED (tests didn't notice the change).

Usage:
    uv run python scripts/mutation_test.py                              # expression_parser
    uv run python scripts/mutation_test.py --quick                      # only operator/literal swaps
    uv run python scripts/mutation_test.py --target core/backtester.py  # different module
    uv run python scripts/mutation_test.py --list                       # show mutations, don't run

NOTE: On macOS + Python 3.13+, multiprocessing.Lock() is restricted.
This script runs mutations SEQUENTIALLY to avoid those limitations.
"""

import os
import re
import subprocess
import sys
import time
from dataclasses import dataclass
from pathlib import Path

# ── Config ──


@dataclass
class TextMutation:
    name: str
    lineno: int
    old_text: str  # exact text to find (substring match on the line)
    new_text: str  # replacement text
    category: str = "operator"
    desc: str = ""


DEFAULT_TARGET = "core/expression_parser.py"

# Known test files per source module
TEST_MAP = {
    "core/expression_parser.py": [
        "tests/test_filters_legacy.py",
        "tests/test_grammar_complete.py",
    ],
    "core/data_manager.py": [
        "tests/test_filters_legacy.py",
        "tests/test_data_manager_extended.py",
        "tests/test_data_providers.py",
        "tests/test_scanner_logic.py",
        "tests/test_grammar_complete.py",
    ],
}


# ── Mutation generators ──


def _is_comment_or_docstring(line: str) -> bool:
    """Check if line is a comment, docstring, or string-only line."""
    stripped = line.strip()
    if stripped.startswith("#"):
        return True
    return stripped.startswith('"""') or stripped.startswith("'''") or stripped.startswith("* ")


def _is_string_literal_in_code(lines: list[str], lineno: int) -> bool:
    """Check if a given line is inside a multi-line string."""
    in_single_quote_string = False
    in_double_quote_string = False

    for i in range(lineno - 1):  # only up to current line
        line = lines[i]
        stripped = line.strip()

        if stripped.startswith("#"):
            continue
        if not stripped:
            continue

        # Skip single-line docstrings (start and end with """ on same line)
        if stripped.startswith('"""') and stripped.endswith('"""') and len(stripped) > 3:
            continue
        if stripped.startswith("'''") and stripped.endswith("'''") and len(stripped) > 3:
            continue

        # Toggle multi-line string states
        if stripped.startswith('"""') and not in_single_quote_string:
            in_double_quote_string = not in_double_quote_string
        elif stripped.startswith("'''") and not in_double_quote_string:
            in_single_quote_string = not in_single_quote_string

    return in_double_quote_string or in_single_quote_string


def generate_text_mutations(source_path: str, quick: bool = False) -> list[TextMutation]:
    """Generate mutations using text pattern matching (reliable, no AST needed).

    Only mutates actual CODE lines — docstrings, comments, and string literals are skipped.
    """
    with open(source_path) as f:
        lines = f.readlines()

    mutations: list[TextMutation] = []
    seen: set[tuple[int, str, str]] = set()

    def add(name, lineno, old_text, new_text, category="operator", desc=""):
        key = (lineno, old_text, new_text)
        if key not in seen:
            seen.add(key)
            mutations.append(
                TextMutation(
                    name=name, lineno=lineno, old_text=old_text, new_text=new_text, category=category, desc=desc
                )
            )

    for i, line in enumerate(lines, 1):
        stripped = line.strip()
        # Skip comments, docstrings, string literals
        if _is_comment_or_docstring(line):
            continue
        if _is_string_literal_in_code(lines, i):
            continue

        # ── Comparison operators ──
        # Only match operator usage in expressions, not definitions
        # Pattern: something OP something (with spaces around operators)
        if re.search(r"[\w)\]]\s*<=", stripped) and "<=" in stripped:
            add(f"L{i}: <= → <", i, "<=", "<", "operator")
        if re.search(r"[\w)\]]\s*>=", stripped) and ">=" in stripped:
            add(f"L{i}: >= → >", i, ">=", ">", "operator")
        if " == " in stripped:
            add(f"L{i}: == → !=", i, " == ", " != ", "operator")
        if " != " in stripped:
            add(f"L{i}: != → ==", i, " != ", " == ", "operator")
        # >  with spaces around it (not ->, not >=)
        if re.search(r"[\w)\]]\s+>\s+", stripped) and ">=" not in stripped:
            add(f"L{i}: > → >=", i, " > ", " >= ", "operator")
        # <  with spaces around it (not <=)
        if re.search(r"[\w)\]]\s+<\s+", stripped) and "<=" not in stripped:
            add(f"L{i}: < → <=", i, " < ", " <= ", "operator")

        # ── Boolean operators (code only — not in strings/comments) ──
        if (
            re.search(r"\band\b", stripped)
            and "and " in stripped
            and (re.search(r"\w+\s+and\s+", stripped) or re.search(r"\)\s+and\s+", stripped))
        ):
            add(f"L{i}: and → or", i, " and ", " or ", "operator")
        if (
            re.search(r"\bor\b", stripped)
            and " or " in stripped
            and (re.search(r"\w+\s+or\s+", stripped) or re.search(r"\)\s+or\s+", stripped))
            and ("for " not in stripped.split(" or ")[0] if " or " in stripped else True)
        ):
            add(f"L{i}: or → and", i, " or ", " and ", "operator")

        # ── Boolean literals in return/assignment statements ──
        if stripped == "return lambda ctx: False" or stripped == "return lambda ctx:True":
            add(f"L{i}: False → True", i, "False", "True", "literal")
        if stripped == "return lambda ctx: True" or stripped == "return lambda ctx:True":
            add(f"L{i}: True → False", i, "True", "False", "literal")

    return mutations


def apply_mutation(source_path: str, mutation: TextMutation) -> bool:
    """Apply a single mutation. Returns True if the file was changed."""
    with open(source_path) as f:
        lines = f.readlines()

    idx = mutation.lineno - 1
    if idx < 0 or idx >= len(lines):
        return False

    old = lines[idx]
    if mutation.old_text not in old:
        return False

    new = old.replace(mutation.old_text, mutation.new_text, 1)
    if new == old:
        return False

    lines[idx] = new
    with open(source_path, "w") as f:
        f.writelines(lines)
    return True


# ── Test runner ──


def find_tests(source_path: str) -> list[str]:
    rel = os.path.normpath(source_path)
    for key, tests in TEST_MAP.items():
        if key == rel:
            return tests
    basename = os.path.basename(source_path).replace(".py", "")
    test_files = list(Path("tests").glob(f"test_*{basename}*.py"))
    return [str(f) for f in test_files] if test_files else ["tests/test_filters_legacy.py"]


def run_tests(test_files: list[str]) -> tuple[bool, str, float]:
    cmd = [
        sys.executable,
        "-m",
        "pytest",
        *test_files,
        "-x",
        "--tb=line",
        "-q",
    ]
    start = time.time()
    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        timeout=180,
        env={**os.environ, "NUMBA_DISABLE_JIT": "1", "PYTEST_ADDOPTS": ""},
    )
    duration = time.time() - start
    passed = result.returncode == 0
    return passed, result.stdout + result.stderr, duration


def summarize(results: list[dict]) -> None:
    survived = [r for r in results if r["status"] == "SURVIVED"]
    killed = [r for r in results if r["status"] == "KILLED"]
    errors = [r for r in results if r["status"] == "ERROR"]

    total = len(results)
    score = len(killed) / total * 100 if total > 0 else 0

    print()
    print("=" * 65)
    print("  MUTATION TEST RESULTS")
    print(f"  {'─' * 45}")
    print(f"  Total:    {total}")
    print(f"  Killed:   {len(killed)}")
    print(f"  Survived: {len(survived)} {'⚠️' if survived else '✅'}")
    print(f"  Errors:   {len(errors)}")
    print(f"  Score:    {score:.0f}%")

    if survived:
        print(f"\n  {'─' * 45}")
        print("  SURVIVED — these changes are NOT caught by tests:")
        print(f"  {'─' * 45}")
        for i, r in enumerate(survived, 1):
            m = r["mutation"]
            print(f"  [{i}] {m.name}")
            print(f"      Line: {m.old_text!r} → {m.new_text!r}")
            print(f"      Tests: {r['duration']:.0f}s — all passed")

    return bool(survived)


# ── Main ──


def main():
    import argparse

    parser = argparse.ArgumentParser(description="Mutation testing")
    parser.add_argument("--target", default=DEFAULT_TARGET)
    parser.add_argument("--list", action="store_true", help="List mutations without running")
    parser.add_argument("--quick", action="store_true", help="Only operator/literal mutations (default)")
    parser.add_argument("--tests", nargs="*", help="Test files (default: auto-detect)")
    args = parser.parse_args()

    source_path = args.target
    if not os.path.exists(source_path):
        print(f"File not found: {source_path}")
        return 1

    # Generate mutations
    print(f"Analyzing {source_path}...")
    mutations = generate_text_mutations(source_path, quick=args.quick)
    if not mutations:
        print("No viable mutations found.")
        return 0

    print(f"Found {len(mutations)} mutations\n")

    if args.list:
        for m in mutations:
            print(f"  L{m.lineno:4d}  [{m.category:8s}] {m.name}")
        print(f"\nTotal: {len(mutations)}")
        return 0

    # Baseline
    test_files = args.tests if args.tests else find_tests(source_path)
    print(f"Tests: {', '.join(test_files)}")
    print("Baseline check...", end=" ", flush=True)
    ok, _, dur = run_tests(test_files)
    if not ok:
        print("FAILED — fix your tests first!")
        return 1
    print(f"✅ ({dur:.0f}s)\n")

    # Read original once
    with open(source_path) as f:
        original = f.read()

    results = []
    try:
        for i, mutation in enumerate(mutations):
            label = f"[{i + 1}/{len(mutations)}] {mutation.name:50s}"
            print(f"  {label} ", end="", flush=True)

            changed = apply_mutation(source_path, mutation)
            if not changed:
                print("⚠️  SKIP (text not found)")
                results.append({"mutation": mutation, "status": "SKIPPED", "duration": 0})
                continue

            time.sleep(0.05)

            try:
                ok, out, duration = run_tests(test_files)
            except subprocess.TimeoutExpired:
                print("⏰ TIMEOUT (treated as KILLED)")
                results.append({"mutation": mutation, "status": "KILLED", "duration": 0})
                ok = True  # skip the survival check below
            except Exception as e:
                print(f"💥 ERROR: {e}")
                results.append({"mutation": mutation, "status": "ERROR", "duration": 0, "error": str(e)})
                ok = True

            if ok and not isinstance(ok, bool):
                # wasn't a timeout/error
                pass

            if ok:
                print("💥 SURVIVED")
                results.append({"mutation": mutation, "status": "SURVIVED", "duration": duration})
            else:
                print("✅ KILLED")
                results.append({"mutation": mutation, "status": "KILLED", "duration": duration})

            # Restore
            with open(source_path, "w") as f:
                f.write(original)

    finally:
        # Ensure restoration even if interrupted
        with open(source_path) as fh:
            current = fh.read()
        if current != original:
            with open(source_path, "w") as f:
                f.write(original)
            print("\n⚠️  Restored original file from backup")

    has_survived = summarize(results)
    return 1 if has_survived else 0


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