scripts/validate_tests.py

#!/usr/bin/env python3
"""
validate_tests.py — validates that tests are actually meaningful.

Pipeline:
  1. Static scan (50ms) — finds tests with no/trivial assertions
  2. Targeted mutation testing (~1-2min) — finds tests that don't catch real changes

No humanos. No revisión. 100% automatizado.

Usage:
    uv run python scripts/validate_tests.py
    uv run python scripts/validate_tests.py --quick          # static scan only
    uv run python scripts/validate_tests.py --target core/expression_parser.py
    uv run python scripts/validate_tests.py --list           # show mutations without running
"""

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

# ── Config ──

DEFAULT_TARGET = "core/expression_parser.py"

TEST_MAP = {
    "core/expression_parser.py": [
        "tests/test_filters_legacy.py",
        "tests/test_grammar_complete.py",
    ],
}

# Pytest flags for fast failure
PYTEST_FLAGS = ["-x", "--tb=line", "-q", "--no-header"]

ENV = {
    **os.environ,
    "NUMBA_DISABLE_JIT": "1",
    "PYTEST_ADDOPTS": "",
    "PYTHONWARNINGS": "ignore",
}


# ═══════════════════════════════════════════════════════════════
#  Mutation primitives
# ═══════════════════════════════════════════════════════════════


@dataclass
class TextMutation:
    name: str
    lineno: int
    old_text: str
    new_text: str
    category: str = "operator"
    desc: str = ""


def generate_text_mutations(source_path: str) -> list[TextMutation]:
    """Generate mutations using text pattern matching.

    Only mutates actual CODE lines — docstrings, comments, string literals skipped.
    Same generator as mutation_test.py.
    """
    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
                )
            )

    def _skip_line(stripped: str) -> bool:
        """Skip comments, docstrings, rst bullets, and string-only lines."""
        if not stripped:
            return True
        if stripped.startswith("#"):
            return True
        if stripped.startswith("* "):
            return True
        # Single-line triple-quoted strings
        if stripped.startswith('"""') and stripped.endswith('"""') and len(stripped) > 3:
            return True
        if stripped.startswith("'''") and stripped.endswith("'''") and len(stripped) > 3:
            return True
        # String-only lines (full-line string literals — not f-strings)
        return (stripped.startswith('"') and stripped.endswith('"') and not stripped.startswith('f"')) or (
            stripped.startswith("'") and stripped.endswith("'") and not stripped.startswith("f'")
        )

    def _code_part(line: str) -> str:
        """Return code before inline comment."""
        in_single = False
        in_double = False
        for j, ch in enumerate(line):
            if ch == '"' and not in_single:
                in_double = not in_double
            elif ch == "'" and not in_double:
                in_single = not in_single
            elif ch == "#" and not in_single and not in_double:
                return line[:j]
        return line

    # Pre-scan for multi-line string regions (docstrings, multi-line strings)
    in_multi = False
    multi_delim: str | None = None
    multi_lines: set[int] = set()
    for i, line in enumerate(lines, 1):
        stripped = line.strip()
        if in_multi:
            multi_lines.add(i)
            # Close on line that ends with the delimiter (or is just the delimiter)
            if multi_delim and stripped.rstrip().endswith(multi_delim):
                in_multi = False
                multi_delim = None
            continue
        if not stripped:
            continue
        # Line starts with """ or ''' but doesn't end with it → multi-line string start
        if stripped.startswith('"""') and not stripped.rstrip().endswith('"""'):
            in_multi = True
            multi_delim = '"""'
            multi_lines.add(i)
        elif stripped.startswith("'''") and not stripped.rstrip().endswith("'''"):
            in_multi = True
            multi_delim = "'''"
            multi_lines.add(i)

    for i, line in enumerate(lines, 1):
        # Get code part (strip inline comments, but only for operator matching)
        code = _code_part(line)
        stripped = code.strip()

        if _skip_line(stripped):
            continue

        if i in multi_lines:
            continue

        # Comparison operators — only check the code part, not comments
        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")
        if re.search(r"[\w)\]]\s+>\s+", stripped) and ">=" not in stripped:
            add(f"L{i}: > → >=", i, " > ", " >= ", "operator")
        if re.search(r"[\w)\]]\s+<\s+", stripped) and "<=" not in stripped:
            add(f"L{i}: < → <=", i, " < ", " <= ", "operator")

        # Boolean operators
        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 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")

        # Numeric boundary mutations
        # 0 → 1 (empty/null checks)
        if re.search(r"\b==\s*0\b", stripped) and "0" in stripped:
            add(f"L{i}: == 0 → == 1", i, "== 0", "== 1", "boundary")
        if re.search(r"\b!=\s*0\b", stripped) and "0" in stripped:
            add(f"L{i}: != 0 → != 1", i, "!= 0", "!= 1", "boundary")

        # len(X) == N → len(X) == N+1  (off-by-one)
        # (only for N > 0 to avoid 0→1 overlap)
        len_match = re.search(r"\blen\((\w+)\)\s*==\s*(\d+)\b", stripped)
        if len_match and int(len_match.group(2)) > 0:
            n = int(len_match.group(2))
            old = f"len({len_match.group(1)}) == {n}"
            new = f"len({len_match.group(1)}) == {n + 1}"
            add(f"L{i}: len() == {n} → == {n + 1}", i, old, new, "boundary")

    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


# ═══════════════════════════════════════════════════════════════
#  Stage 1: Static analysis
# ═══════════════════════════════════════════════════════════════


@dataclass
class TestIssue:
    test_file: str
    test_name: str
    lineno: int
    issue_type: str  # NO_ASSERT, TRIVIAL_ASSERT
    detail: str = ""


def analyze_test_quality(test_files: list[str]) -> list[TestIssue]:
    """Find tests with quality issues via AST analysis (no test execution)."""
    issues = []

    for tf in test_files:
        if not os.path.exists(tf):
            continue

        with open(tf) as f:
            try:
                source = f.read()
                tree = ast.parse(source)
            except SyntaxError as e:
                issues.append(TestIssue(tf, "<file>", 0, "PARSE_ERROR", str(e)))
                continue

        for node in ast.walk(tree):
            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue

            # Skip non-test functions (helpers with "assert" in name are OK)
            if not (node.name.startswith("test_") or node.name.startswith("async_test_")):
                # Check if this is a method inside a Test class
                for parent in ast.walk(tree):
                    if isinstance(parent, ast.ClassDef) and parent.name.startswith("Test"):
                        # Check if node is a direct child of this class
                        # We can't easily check parentage in ast.walk, so skip this heuristic
                        pass
                # Simple heuristic: only flag functions that start with test_
                # Methods inside Test* classes that don't start with test_ are still
                # run by pytest but are unusual — skip them for now
                if not node.name.startswith("test_"):
                    continue

            # Check for pytest.raises/pytest.warns context managers (valid assertions)
            has_pytest_raises = False
            for w in ast.walk(node):
                if isinstance(w, ast.With):
                    for item in w.items:
                        ctx = item.context_expr
                        if (
                            isinstance(ctx, ast.Call)
                            and isinstance(ctx.func, ast.Attribute)
                            and isinstance(ctx.func.value, ast.Name)
                            and ctx.func.value.id == "pytest"
                            and ctx.func.attr in ("raises", "warns")
                        ):
                            has_pytest_raises = True

            # Count assert statements
            assert_nodes = [n for n in ast.walk(node) if isinstance(n, ast.Assert)]

            if len(assert_nodes) == 0 and not has_pytest_raises:
                issues.append(TestIssue(tf, node.name, node.lineno, "NO_ASSERT"))
                continue

            # Check for trivial assertions (assert True/False/None)
            for n in assert_nodes:
                if isinstance(n.test, ast.Constant) and n.test.value in (True, False, None):
                    issues.append(
                        TestIssue(tf, node.name, node.lineno, "TRIVIAL_ASSERT", f"L{n.lineno}: assert {n.test.value}")
                    )
                    break  # one flag per test is enough

            # Check for single assert that is just "assert X is not None"  (weak)
            if len(assert_nodes) == 1:
                only = assert_nodes[0]
                if isinstance(only.test, ast.IsNot):
                    issues.append(
                        TestIssue(
                            tf, node.name, node.lineno, "WEAK_ASSERT", f"L{only.lineno}: assert X is not None only"
                        )
                    )
                elif isinstance(only.test, ast.Is):
                    issues.append(
                        TestIssue(tf, node.name, node.lineno, "WEAK_ASSERT", f"L{only.lineno}: assert X is None only")
                    )

    return issues


# ═══════════════════════════════════════════════════════════════
#  Stage 2: Targeted mutation testing
# ═══════════════════════════════════════════════════════════════


def get_function_at_line(source_path: str, lineno: int) -> str | None:
    """Find the function/method name that contains the given line number."""
    with open(source_path) as f:
        try:
            tree = ast.parse(f.read())
        except SyntaxError:
            return None

    best = None
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            end = getattr(node, "end_lineno", None)
            if (
                end is not None
                and node.lineno <= lineno <= end
                and (best is None or node.lineno > getattr(best, "lineno", 0))
            ):
                # Return the innermost function (the one with the highest start line)
                best = node

    return best.name if best else None


def collect_node_ids(test_files: list[str], timeout: int = 30) -> list[str]:
    """Get all pytest test node IDs for the given files."""
    cmd = [sys.executable, "-m", "pytest", *test_files, "--collect-only", "-q", "--no-header"]
    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        timeout=timeout,
        env=ENV,
    )

    node_ids = []
    for line in result.stdout.splitlines():
        line = line.strip()
        if not line:
            continue
        # Filter out noise lines
        if line.startswith("no tests ran") or line.startswith("=="):
            continue
        # It's a valid node ID if it contains "::"
        if "::" in line and not line.startswith("<"):
            node_ids.append(line)

    return node_ids


def select_tests_for_line(
    all_node_ids: list[str],
    source_path: str,
    lineno: int,
) -> list[str] | None:
    """Select test node IDs relevant to a mutation at the given line.

    Uses function name matching: if the mutation is in `compile_safe()`,
    it finds all tests with `compile_safe` or `TestCompileSafe` in their names.

    Returns None if no specific match found (fall back to all tests).
    """
    func_name = get_function_at_line(source_path, lineno)
    if not func_name:
        return None

    # Normalize for matching: remove underscores, lowercase
    norm = func_name.lower().replace("_", "")
    matching = [nid for nid in all_node_ids if norm in nid.lower().replace("_", "")]

    if not matching:
        return None  # fall back to all tests

    return matching


def run_tests_baseline(test_files: list[str]) -> tuple[bool, str, float]:
    """Run all tests once to verify they pass."""
    cmd = [sys.executable, "-m", "pytest", *test_files, *PYTEST_FLAGS]
    start = time.time()
    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        timeout=180,
        env=ENV,
    )
    duration = time.time() - start
    passed = result.returncode == 0
    return passed, result.stdout + result.stderr, duration


def run_tests_by_node_ids(node_ids: list[str]) -> tuple[bool, float]:
    """Run specific tests by node IDs."""
    cmd = [sys.executable, "-m", "pytest", *node_ids, *PYTEST_FLAGS]
    start = time.time()
    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        timeout=60,
        env=ENV,
    )
    duration = time.time() - start
    passed = result.returncode == 0
    return passed, duration


def run_mutation_test(
    mutation: TextMutation,
    source_path: str,
    original_source: str,
    all_node_ids: list[str],
) -> dict:
    """Apply a single mutation, run relevant tests, and report result."""
    # Select relevant tests
    matching = select_tests_for_line(all_node_ids, source_path, mutation.lineno)
    node_ids = matching if matching else all_node_ids

    label = f"  {mutation.name:55s}"
    print(f"{label} ", end="", flush=True)

    # Apply mutation
    changed = apply_mutation(source_path, mutation)
    if not changed:
        print("⚠️  SKIP (text not found)")
        return {"mutation": mutation, "status": "SKIPPED", "duration": 0, "tests_run": 0}

    time.sleep(0.05)

    # Run tests
    try:
        ok, duration = run_tests_by_node_ids(node_ids)
    except subprocess.TimeoutExpired:
        print("⏰  TIMEOUT")
        # Timeout means tests are slow but not broken — count as KILLED
        ok, duration = True, 0
    except Exception as e:
        print(f"💥  ERROR: {e}")
        return {"mutation": mutation, "status": "ERROR", "duration": 0, "tests_run": len(node_ids)}

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

    if ok:
        status = "SURVIVED"
        icon = "💥"
    else:
        status = "KILLED"
        icon = "✅"

    print(f"{icon} {status} ({len(node_ids)} tests in {duration:.0f}s)")
    return {
        "mutation": mutation,
        "status": status,
        "duration": duration,
        "tests_run": len(node_ids),
        "matched_tests": len(matching) if matching else "ALL",
    }


# ═══════════════════════════════════════════════════════════════
#  Report
# ═══════════════════════════════════════════════════════════════


def print_banner(text: str) -> None:
    width = 65
    print(f"\n{'─' * width}")
    print(f"  {text}")
    print(f"{'─' * width}")


def print_report(
    source_path: str,
    static_issues: list[TestIssue],
    mutation_results: list[dict],
    total_duration: float,
) -> None:
    print()
    print("=" * 65)
    print("  TEST VALIDATION REPORT")
    print(f"  Target: {source_path}")
    print(f"{'=' * 65}")

    # ── Static analysis ──
    print_banner("Stage 1: Static analysis")

    no_assert = [i for i in static_issues if i.issue_type == "NO_ASSERT"]
    trivial = [i for i in static_issues if i.issue_type == "TRIVIAL_ASSERT"]
    weak = [i for i in static_issues if i.issue_type == "WEAK_ASSERT"]

    if not static_issues:
        print("  ✅ No issues found — all tests have assertions")
    else:
        if no_assert:
            print(f"  ❌ {len(no_assert)} test(s) with NO assertions:")
            for i in no_assert:
                print(f"     - {Path(i.test_file).name}::{i.test_name} (L{i.lineno})")

        if trivial:
            print(f"  ⚠️  {len(trivial)} test(s) with TRIVIAL assertions (assert True/False/None):")
            for i in trivial:
                print(f"     - {Path(i.test_file).name}::{i.test_name} (L{i.lineno}) — {i.detail}")

        if weak:
            print(f"  ⚠️  {len(weak)} test(s) with WEAK assertions (single is/is-not check):")
            for i in weak:
                print(f"     - {Path(i.test_file).name}::{i.test_name} (L{i.lineno}) — {i.detail}")

    # ── Mutation testing ──
    print_banner("Stage 2: Targeted mutation testing")

    survived = [r for r in mutation_results if r["status"] == "SURVIVED"]
    killed = [r for r in mutation_results if r["status"] == "KILLED"]
    errors = [r for r in mutation_results if r["status"] == "ERROR"]

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

    if not mutation_results:
        print("  (no mutations generated)")
    else:
        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}%")
        print(f"  Duration: {total_duration:.0f}s")

        if survived:
            print(f"\n  {'─' * 55}")
            print("  SURVIVED — these changes are NOT caught by tests:")
            print(f"  {'─' * 55}")
            for i, r in enumerate(survived, 1):
                m = r["mutation"]
                tests_label = (
                    f"{r['tests_run']} tests matched"
                    if isinstance(r.get("matched_tests"), int)
                    else f"ALL {r['tests_run']} tests"
                )
                print(f"  [{i}] {m.name}")
                print(f"      {m.old_text!r} → {m.new_text!r}")
                print(f"      Ran {tests_label} — all passed in {r['duration']:.0f}s")

        if killed:
            print(f"\n  ✅ {len(killed)} mutation(s) killed — these tests work correctly")

    # ── Summary ──
    print_banner("Summary")

    if score >= 90 and not static_issues:
        print("  ✅ HIGH — tests are solid")
    elif score >= 70 and not static_issues:
        print("  ⚠️  MODERATE — some gaps found")
    elif score >= 50:
        print("  ⚠️  LOW — significant gaps found")
    else:
        print("  ❌ CRITICAL — tests miss most changes")

    quality_issues = len(no_assert) + len(trivial)
    if quality_issues > 0:
        print(f"  ❌ {quality_issues} test(s) with static quality issues (NO_ASSERT or TRIVIAL_ASSERT)")
    if survived:
        print(f"  ❌ {len(survived)} behavioral gap(s) — no tests catch these changes")
        print("  → Add tests for the specific edge cases listed above")

    print(f"\n  Total time: {total_duration:.0f}s")
    print(f"  {'=' * 65}")
    print()


# ═══════════════════════════════════════════════════════════════
#  CLI
# ═══════════════════════════════════════════════════════════════


def find_test_files(source_path: str) -> list[str]:
    """Find test files for a given source module."""
    rel = os.path.normpath(source_path)
    for key, tests in TEST_MAP.items():
        if key == rel:
            return tests
    # Fallback: glob for test files
    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 []


def main():
    import argparse

    parser = argparse.ArgumentParser(description="Validate that tests are meaningful")
    parser.add_argument(
        "--target", default=DEFAULT_TARGET, help="Source module to test (default: core/expression_parser.py)"
    )
    parser.add_argument("--quick", action="store_true", help="Skip mutation testing, static analysis only")
    parser.add_argument("--list", action="store_true", help="List mutations without running them")
    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

    test_files = args.tests if args.tests else find_test_files(source_path)
    if not test_files:
        print(f"❌ No test files found for {source_path}")
        return 1

    print(f"\n  Source: {source_path}")
    print(f"  Tests:  {', '.join(test_files)}")

    start_all = time.time()

    # ── Stage 1: Static analysis ──
    print()
    print("  ── Stage 1: Static analysis ──")
    static_issues = analyze_test_quality(test_files)

    if static_issues:
        print(f"  Found {len(static_issues)} issue(s)")
        for i in static_issues:
            label = {
                "NO_ASSERT": "no assert",
                "TRIVIAL_ASSERT": "trivial assert",
                "WEAK_ASSERT": "weak assert",
                "PARSE_ERROR": "parse error",
            }.get(i.issue_type, i.issue_type)
            print(f"    ⚠️  {Path(i.test_file).name}::{i.test_name} — {label}")
    else:
        print("  ✅ All tests have assertions")

    if args.quick:
        total_duration = time.time() - start_all
        print()
        print("  --quick: skipping mutation testing")
        print(f"  Total: {total_duration:.0f}s")
        return 0

    # ── Prepare mutation testing ──
    print()
    print("  ── Stage 2: Targeted mutation testing ──")

    # Generate mutations
    mutations = generate_text_mutations(source_path)
    if not mutations:
        print("  No viable mutations found.")
        total_duration = time.time() - start_all
        print(f"\n  Total: {total_duration:.0f}s")
        return 0

    print(f"  Generated {len(mutations)} mutations")

    if args.list:
        for m in mutations:
            func = get_function_at_line(source_path, m.lineno) or "?"
            print(f"    L{m.lineno:4d}  [{m.category:8s}] {m.name:45s} in {func}()")
        return 0

    # Baseline check
    print("  Baseline check...", end=" ", flush=True)
    ok, out, dur = run_tests_baseline(test_files)
    if not ok:
        print("FAILED — fix your tests first!")
        print(out[:500])
        return 1
    print(f"✅ ({dur:.0f}s)")

    # Collect all test node IDs (once)
    print("  Collecting test node IDs...", end=" ", flush=True)
    all_node_ids = collect_node_ids(test_files)
    print(f"{len(all_node_ids)} tests collected")

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

    # Run each mutation
    mutation_results = []
    try:
        for mutation in mutations:
            result = run_mutation_test(mutation, source_path, original_source, all_node_ids)
            mutation_results.append(result)
    finally:
        # Ensure restoration
        with open(source_path) as fh:
            current = fh.read()
        if current != original_source:
            with open(source_path, "w") as f:
                f.write(original_source)
            print("\n  ⚠️  Restored original file from backup")

    total_duration = time.time() - start_all

    # ── Report ──
    print_report(source_path, static_issues, mutation_results, total_duration)

    has_survived = any(r["status"] == "SURVIVED" for r in mutation_results)
    return 1 if has_survived else 0


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