scripts.test_isolation_audit

Reference › scripts › test_isolation_audit

scripts/test_isolation_audit.py audits the test suite for isolation violations: shared state, hard-coded names and missing test_id. A human or the pre-commit review battery runs it, and it accepts individual test files so the battery can audit only the modules a change touched. It has no internal imports. See test-review, gate_parallel_safety and sweep_test_pollution, which clean up after what this audit looks for.

Each file is parsed with ast, and the checks return IsolationViolation records: check_cypher_queries (queries found in string literals, using extract_string_literals and skipping docstrings), check_fixture_scopes (fixtures that are not function-scoped), check_ast_patterns and check_module_level_state (module-level mutable state). A violation can be waived by a gate-ok comment on its line or the line above (_is_waived), by the T4 immutable opt-in marker, or file-wide by header markers in the first 30 lines (_file_waived_categories). Arguments: paths (directories or files, default tests) and --fix, which generates a fix report. A path that does not exist prints an error and exits 1; the report groups violations by file.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Classes

IsolationViolation

Line 114

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/test_isolation_audit.py

#!/usr/bin/env python3
"""Audit test suite for isolation violations: shared state, hardcoded names, missing test_id.

Scans tests/ directory and reports violations that cause flaky failures under pytest -n auto.
Returns non-zero exit code if any violations are found.

Usage:
    python scripts/test_isolation_audit.py [tests/path/] [--fix]
"""

import argparse
import ast
import re
import sys
from collections import defaultdict
from pathlib import Path

# Patterns for Cypher query analysis
CYPHER_CREATE_PATTERN = re.compile(
    r"CREATE\s*\(\s*(\w+)\s*:\s*(\w+)\s*\{([^}]*)\}\s*\)",
    re.IGNORECASE | re.MULTILINE,
)
CYPHER_MERGE_PATTERN = re.compile(
    r"MERGE\s*\(\s*(\w+)\s*:\s*(\w+)\s*\{([^}]*)\}\s*\)",
    re.IGNORECASE | re.MULTILINE,
)
HARDCODED_NAME_PATTERN = re.compile(
    r"name\s*:\s*['\"]([^'\"]+)['\"]",
    re.IGNORECASE,
)
TEST_ID_PATTERN = re.compile(
    r"test_id\s*:|\$test_id|\$tid|test_id=|\{test_id\}",
    re.IGNORECASE,
)
# A backtick-quoted label interpolated from a variable -- ``:`{seed_label}` `` --
# is the project's per-test throwaway-label discipline: the fixture derives the
# label from `test_id`, so every node created under it is unreachable by any
# other test, and a hardcoded `name:` inside is a PAYLOAD, not a collision key.
#
# The check exists for the genuinely unsafe shape, which is a LITERAL label plus
# a fixed name (`CREATE (:Volk {name: 'Junior Volk'})`) -- two concurrent tests
# writing that do collide. Without this, files that scope correctly by label
# reported violations they could not fix except by adding a redundant test_id to
# a property nothing queries, which trains readers to ignore the gate.
INTERPOLATED_LABEL_PATTERN = re.compile(r":\s*`\{[^}]+\}`")
HARDCODED_PORT = re.compile(
    r"port\s*=\s*(?:['\"])?(\d{4,5})(?:['\"])?",
    re.IGNORECASE,
)
FIXTURE_SCOPE_PATTERN = re.compile(
    r"@pytest\.fixture\s*\(\s*scope\s*=\s*['\"](module|class|session)['\"]\s*\)",
    re.IGNORECASE,
)

# ``# gate-ok: <reason>`` on the flagged line (or the line directly above it)
# waives that one finding — same convention as scripts/gate_parallel_safety.py,
# for fixture Cypher that is scanned as inert string data (never executed
# against the graph), e.g. the graph_test_preflight.py scanner's own test fixtures.
# The reason is mandatory — ``\S`` refuses a bare tag.
_WAIVER_RE = re.compile(r"#\s*gate-ok:\s*\S")


def _is_waived(lines: list[str], lineno: int) -> bool:
    """True if the 1-based *lineno* (or the line above) carries a gate-ok waiver."""
    for idx in (lineno - 1, lineno - 2):
        if 0 <= idx < len(lines) and _WAIVER_RE.search(lines[idx]):
            return True
    return False


# ``# isolation-audit-file-ok: CATEGORY: <reason>`` in a file's first 30 lines
# waives that CATEGORY for the WHOLE file. Reason mandatory. Exists for
# detector-test files whose fixtures are inert STRINGS containing the banned
# patterns (the flagged line sits inside a string literal, so a per-line
# ``gate-ok`` comment physically cannot reach it). Line-level waivers remain
# preferred everywhere a comment can sit next to the finding.
_FILE_WAIVER_RE = re.compile(r"#\s*isolation-audit-file-ok:\s*([A-Z_]+):\s*\S")

# ``# T4-immutable-module-scope: <why>`` directly above the decorator is the
# project's FORMAL opt-in for module scope (AGENTS.md § Test Standards T3/T4).
# `scripts/test-review.py` rule T4 already validates it properly — mandatory
# reason, return-only body, no yield, module scope ONLY.
#
# This auditor did not know the marker, so it re-reported seven fixtures that
# the other gate had already blessed and checked. Two gates disagreeing about
# the same rule is worse than either being absent: the isolation audit can
# never reach zero, so its output stops being read and a REAL violation lands
# unnoticed among the standing noise.
#
# Module scope only — class and session scope stay flagged whatever the comment
# says, matching T4. This is a cross-reference to the owning gate, not a second
# implementation of it: nothing here re-checks return-only-ness.
_T4_IMMUTABLE_RE = re.compile(r"#\s*T4-immutable-module-scope:\s*\S")


def _is_t4_immutable_optin(lines: list[str], lineno: int) -> bool:
    """True if the 1-based *lineno* (or the line above) carries the T4 marker."""
    for idx in (lineno - 1, lineno - 2):
        if 0 <= idx < len(lines) and _T4_IMMUTABLE_RE.search(lines[idx]):
            return True
    return False


def _file_waived_categories(lines: list[str]) -> set[str]:
    """Categories waived file-wide via header markers (first 30 lines only)."""
    waived: set[str] = set()
    for line in lines[:30]:
        match = _FILE_WAIVER_RE.search(line)
        if match:
            waived.add(match.group(1))
    return waived


class IsolationViolation:
    def __init__(self, category: str, line: int, message: str, fix: str = ""):
        self.category = category
        self.line = line
        self.message = message
        self.fix = fix

    def __repr__(self):
        return f"IsolationViolation({self.category}, line {self.line}, {self.message})"


def extract_string_literals(tree: ast.AST) -> list[tuple[int, int, str]]:
    """Extract all string literals from AST with their line numbers.

    Returns the full line-range of each string constant (multi-line strings
    like ``"a\\n" "b\\n" "c"`` are reported with their start ``lineno`` and
    ``end_lineno`` so callers can skip them — tests deliberately embed code
    samples inside string literals to feed them to ``_check()``, and those
    embedded samples must NOT trigger HARDCODED_NAME / CREATE violations.
    """
    strings = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Constant) and isinstance(node.value, str):
            lineno = node.lineno or 0
            end_lineno = node.end_lineno or lineno
            strings.append((lineno, end_lineno, node.value))
        elif isinstance(node, ast.JoinedStr):
            # f-strings: try to reconstruct
            try:
                lineno = node.lineno or 0
                end_lineno = node.end_lineno or lineno
                strings.append((lineno, end_lineno, ast.unparse(node)))
            except Exception:
                pass
    return strings


def _line_is_inside_string_literal(
    string_ranges: list[tuple[int, int, str]], lineno: int
) -> bool:
    """True iff *lineno* falls inside any AST string constant (1-based)."""
    for start, end, _value in string_ranges:
        if start <= lineno <= end:
            return True
    return False


def _is_in_docstring(lines: list[str], line_idx: int) -> bool:
    """Check if a line is inside a docstring."""
    # Simple heuristic: count triple quotes before this line
    triple_count = 0
    for i in range(line_idx):
        triple_count += lines[i].count('"""') + lines[i].count("'''")
    return triple_count % 2 == 1


def check_cypher_queries(
    content: str,
    lines: list[str],
    string_ranges: list[tuple[int, int, str]] | None = None,
) -> list[IsolationViolation]:
    """Check Cypher queries in string literals for isolation issues.

    If *string_ranges* is provided (output of :func:`extract_string_literals`,
    with each entry's second element being the AST string's ``end_lineno``),
    any 1-based ``lineno`` that falls inside one of those string constants is
    skipped — tests deliberately embed code samples inside string literals to
    feed them to ``_check()``, and those embedded samples must NOT trigger
    HARDCODED_NAME / CREATE violations here.
    """
    string_ranges = string_ranges or []
    violations = []

    # Find all multi-line strings that might contain Cypher
    for i, line in enumerate(lines, 1):
        # Skip comment lines
        stripped = line.strip()
        if stripped.startswith("#") or _is_in_docstring(lines, i - 1):
            continue
        # Skip lines that live inside an AST string literal (embedded code
        # sample fed to _check() — not real production Cypher).
        if _line_is_inside_string_literal(string_ranges, i):
            continue

        # Check for CREATE statements
        if re.search(r"CREATE\s*\(", line, re.IGNORECASE):
            # Check if test_id is present in the same query block
            context_start = max(0, i - 5)
            context_end = min(len(lines), i + 5)
            context = "\n".join(lines[context_start:context_end])

            has_test_id = TEST_ID_PATTERN.search(context) or (
                INTERPOLATED_LABEL_PATTERN.search(line)
            )

            # Check for hardcoded names
            hardcoded = HARDCODED_NAME_PATTERN.findall(line)
            for name in hardcoded:
                if name.lower() in ("neo4j", "redis", "orchestrata", "project", "test"):
                    continue  # Skip generic names
                if not has_test_id and not _is_waived(lines, i):
                    violations.append(
                        IsolationViolation(
                            category="HARDCODED_NAME",
                            line=i,
                            message=f"Hardcoded resource name '{name}' without test_id in query",
                            fix=f"Add test_id to the resource name: '{name}-' + test_id",
                        )
                    )

        # Check CREATE vs MERGE for File nodes (File.path has uniqueness constraint)
        if re.search(r"CREATE\s*\(\s*\w+\s*:\s*File\b", line, re.IGNORECASE):
            context_start = max(0, i - 5)
            context_end = min(len(lines), i + 5)
            context = "\n".join(lines[context_start:context_end])

            # Skip if path is parameterized with a variable and test_id is in context
            has_param_path = re.search(r"path\s*:\s*\$\w+", line, re.IGNORECASE)
            has_test_id = TEST_ID_PATTERN.search(context)

            if has_param_path and has_test_id:
                # Path is a parameter (e.g. $path, $fp) and test_id is used in query → likely safe
                continue

            # Check for hardcoded path string on this line
            hardcoded_path_match = re.search(
                r"path\s*:\s*['\"]([^'\"]+)['\"]", line, re.IGNORECASE
            )
            if hardcoded_path_match:
                hardcoded_path = hardcoded_path_match.group(1)
                # If hardcoded path itself contains test_id or tmp_path, it's safe
                if re.search(
                    r"test_id|tmp_path|\{test_id\}", hardcoded_path, re.IGNORECASE
                ):
                    continue
                if _is_waived(lines, i):
                    continue
                # Hardcoded path without per-test uniqueness → real violation
                violations.append(
                    IsolationViolation(
                        category="CREATE_VS_MERGE",
                        line=i,
                        message=f"CREATE (f:File ...) with hardcoded path '{hardcoded_path}' — File.path has graph uniqueness constraint",
                        fix="Use a unique path per test: f'/tmp/{{test_id}}/filename.py'",
                    )
                )
            elif not has_test_id and not _is_waived(lines, i):
                # No hardcoded path and no test_id in context → flag cautiously
                violations.append(
                    IsolationViolation(
                        category="CREATE_VS_MERGE",
                        line=i,
                        message="CREATE (f:File ...) without test_id — File.path has graph uniqueness constraint",
                        fix="Add test_id filter or use MERGE instead of CREATE",
                    )
                )

    return violations


# Whitelist of read-only session fixtures that are safe to share
_READ_ONLY_FIXTURE_PATTERNS = [
    re.compile(r"return\s+(?:int\s*\()?\s*os\.environ\["),  # redis_host, redis_port
    re.compile(r"GraphDatabase\.driver\("),  # neo4j_driver
    re.compile(r"AsyncGraphDatabase\.driver\("),  # neo4j_async_driver
]

# Patterns indicating intentional bad-connection tests (not shared state)
_BAD_CONNECTION_PATTERNS = [
    re.compile(r"bad\s+(?:client|redis|connection)", re.IGNORECASE),
    re.compile(r"no\s+(?:listener|server)", re.IGNORECASE),
    re.compile(r"invalid\s+(?:host|port)", re.IGNORECASE),
    re.compile(r"non[-]?reachable", re.IGNORECASE),
    re.compile(r"connection\s*(?:error|failure|refused)", re.IGNORECASE),
    re.compile(r"pointing\s+at\s+a\s+port", re.IGNORECASE),
]


def check_fixture_scopes(content: str, lines: list[str]) -> list[IsolationViolation]:
    """Check for non-function-scoped fixtures."""
    violations = []
    for i, line in enumerate(lines, 1):
        match = FIXTURE_SCOPE_PATTERN.search(line)
        if match:
            scope = match.group(1)
            # Check if fixture body is read-only (env vars, connection drivers)
            # Look ahead for the fixture body
            fixture_body = "\n".join(lines[i : min(i + 15, len(lines))])
            is_read_only = any(
                p.search(fixture_body) for p in _READ_ONLY_FIXTURE_PATTERNS
            )
            if is_read_only:
                continue  # Skip read-only fixtures (env vars, connection drivers)
            if _is_waived(lines, i):
                continue  # explicit `# gate-ok: <reason>` waiver
            if scope.lower() == "module" and _is_t4_immutable_optin(lines, i):
                continue  # AGENTS.md T4 opt-in, validated by test-review.py
            violations.append(
                IsolationViolation(
                    category="FIXTURE_SCOPE",
                    line=i,
                    message=f"Fixture scope='{scope}' — should be 'function' for parallel safety",
                    fix="Change to @pytest.fixture(scope='function') or omit scope",
                )
            )
    return violations


def check_ast_patterns(tree: ast.AST, lines: list[str]) -> list[IsolationViolation]:
    """Check AST-level patterns for isolation violations."""
    violations = []

    # OS_ENVIRON_MUTATION lived here until 2026-08-08. It is now
    # `violation(env_mutation, ...)` in src/orchestrata_ultimate/prolog/
    # gate_rules.pl, which infers from the AST snapshot instead of re-walking
    # the tree, honours the same `# gate-ok:` waivers (core.gate_waivers), and
    # covers src/ and scripts/ as well as tests/. Deleted only after the
    # differential: both detectors over tests/ with waivers ignored found the
    # SAME 32 sites — 32 agreed, 0 legacy-only, 0 prolog-only.
    for node in ast.walk(tree):
        # Check for monkeypatch fixture parameter
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            for arg in node.args.args + node.args.kwonlyargs:
                if arg.arg == "monkeypatch":
                    violations.append(
                        IsolationViolation(
                            category="MONKEYPATCH",
                            line=node.lineno or 0,
                            message="monkeypatch fixture — forbidden by test standards",
                            fix="Remove monkeypatch, pass config explicitly",
                        )
                    )

        # Check for hardcoded ports in function calls
        if isinstance(node, ast.Call):
            func_name = ""
            if isinstance(node.func, ast.Name):
                func_name = node.func.id
            elif isinstance(node.func, ast.Attribute):
                func_name = node.func.attr

            for kw in node.keywords:
                if kw.arg == "port" and isinstance(kw.value, ast.Constant):
                    if isinstance(kw.value.value, int) and kw.value.value > 1024:
                        port_val = kw.value.value
                        line_idx = (node.lineno or 1) - 1
                        # Check surrounding context for intentional bad-connection patterns
                        # Look back 20 lines to capture docstrings/comments above the call
                        context_start = max(0, line_idx - 20)
                        context_end = min(len(lines), line_idx + 4)
                        context = "\n".join(lines[context_start:context_end])
                        if any(p.search(context) for p in _BAD_CONNECTION_PATTERNS):
                            continue  # Intentional bad connection for error handling tests
                        # Skip well-known service ports in client connection calls
                        if port_val in (6379, 7687):
                            # Redis/graph standard ports — likely intentional client connections
                            continue
                        # Skip config/dataclass constructors
                        if func_name and any(
                            x in func_name.lower()
                            for x in ("config", "settings", "options", "params")
                        ):
                            continue
                        # Skip factory function tests that verify parameter passthrough
                        # (tests create clients but never send commands)
                        if func_name and any(
                            x in func_name.lower()
                            for x in (
                                "create_sync_redis",
                                "create_async_redis",
                                "redis",
                            )
                        ):
                            if port_val not in (6379, 7687):
                                # Non-standard port in Redis factory call → likely explicit param test
                                continue
                        violations.append(
                            IsolationViolation(
                                category="HARDCODED_PORT",
                                line=node.lineno or 0,
                                message=f"Hardcoded port {port_val} — use dynamic allocation",
                                fix="Use socket.bind(('', 0)) for dynamic port allocation",
                            )
                        )

        # Check for module-level mutable state
        if isinstance(node, ast.Assign):
            if isinstance(node.targets[0], ast.Name) if node.targets else False:
                target = node.targets[0]
                if isinstance(target, ast.Name) and target.id.isupper():
                    # Skip constants (UPPER_CASE is usually a constant)
                    continue
                if isinstance(node.value, (ast.Dict, ast.List, ast.Set)):
                    # Check if this is at module level (no parent function/class)
                    # We can't easily check parent from ast.walk, so we'll use line-based heuristic
                    pass

    return violations


def check_module_level_state(
    content: str, lines: list[str]
) -> list[IsolationViolation]:
    """Check for module-level mutable state."""
    violations = []
    in_function = False
    indent_level = 0

    for i, line in enumerate(lines, 1):
        stripped = line.lstrip()
        current_indent = len(line) - len(stripped)

        if (
            stripped.startswith("def ")
            or stripped.startswith("async def ")
            or stripped.startswith("class ")
        ):
            in_function = True
            indent_level = current_indent

        if (
            in_function
            and current_indent <= indent_level
            and stripped
            and not stripped.startswith("def ")
            and not stripped.startswith("async def ")
            and not stripped.startswith("class ")
        ):
            in_function = False

        if not in_function and current_indent == 0:
            # Module level
            if re.search(r"^\w+\s*=\s*(\{\}|\[\]|set\(\))", stripped):
                violations.append(
                    IsolationViolation(
                        category="GLOBAL_MUTABLE_STATE",
                        line=i,
                        message="Module-level mutable state — use function-scope fixtures instead",
                        fix="Move to fixture or use factory function",
                    )
                )

    return violations


def check_pytest_patterns(content: str, lines: list[str]) -> list[IsolationViolation]:
    """Check for pytest anti-patterns."""
    violations = []
    for i, line in enumerate(lines, 1):
        stripped = line.strip()
        # Skip comments and docstrings
        if stripped.startswith("#"):
            continue
        if _is_in_docstring(lines, i - 1):
            continue
        # Skip string literals that mention skip in documentation
        if (
            stripped.startswith('"""')
            or stripped.startswith("'''")
            or stripped.startswith('"')
            or stripped.startswith("'")
        ):
            continue
        # Only flag actual decorators or function calls
        if re.search(r"@pytest\.mark\.skip\b|pytest\.skip\s*\(", line):
            violations.append(
                IsolationViolation(
                    category="PYTEST_SKIP",
                    line=i,
                    message="pytest.skip / pytest.mark.skip — every test must run",
                    fix="Remove skip, fix the underlying issue",
                )
            )
    return violations


def scan_file(py_file: Path) -> list[IsolationViolation]:
    """Scan a single test file for isolation violations."""
    violations = []

    try:
        content = py_file.read_text(encoding="utf-8")
        tree = ast.parse(content)
    except Exception as e:
        return [IsolationViolation("PARSE_ERROR", 0, f"Failed to parse: {e}")]

    lines = content.splitlines()

    string_ranges = extract_string_literals(tree)
    violations.extend(check_cypher_queries(content, lines, string_ranges))
    violations.extend(check_fixture_scopes(content, lines))
    violations.extend(check_ast_patterns(tree, lines))
    violations.extend(check_module_level_state(content, lines))
    violations.extend(check_pytest_patterns(content, lines))

    file_waived = _file_waived_categories(lines)
    if file_waived:
        violations = [v for v in violations if v.category not in file_waived]

    return violations


def main() -> int:
    parser = argparse.ArgumentParser(description="Audit tests for isolation violations")
    parser.add_argument(
        "paths",
        nargs="*",
        default=["tests"],
        help="Test directories OR individual test files to audit "
        "(default: tests). Files are accepted so the pre-commit review "
        "battery can audit only the modules a change touched.",
    )
    parser.add_argument("--fix", action="store_true", help="Generate fix report")
    args = parser.parse_args()

    # Keep the caller's spelling for the error message: Path("tests/") prints
    # as "tests", and echoing back something they did not type is harder to
    # match against what they ran.
    given = args.paths or ["tests"]
    targets = [Path(p) for p in given]
    missing = [raw for raw, t in zip(given, targets) if not t.exists()]
    if missing:
        for raw in missing:
            print(f"ERROR: {raw} not found")
        return 1

    py_files: list[Path] = []
    seen: set[Path] = set()
    for target in targets:
        found = sorted(target.rglob("*.py")) if target.is_dir() else [target]
        for path in found:
            if path not in seen:
                seen.add(path)
                py_files.append(path)

    all_violations = defaultdict(list)
    file_count = 0
    total_violations = 0

    for py_file in py_files:
        if "__pycache__" in str(py_file):
            continue
        if "_generated_broken" in str(py_file) or "generated_recovery" in str(py_file):
            continue

        violations = scan_file(py_file)
        if violations:
            file_count += 1
            # Best-effort display path relative to cwd — py_file need not be
            # under cwd (e.g. an audited path outside the repo root), so fall
            # back to the absolute path rather than crashing with ValueError.
            rel_path = py_file
            if py_file.is_absolute():
                try:
                    rel_path = py_file.relative_to(Path(".").resolve())
                except ValueError:
                    rel_path = py_file
            all_violations[rel_path] = violations
            total_violations += len(violations)

    # Print report
    if total_violations == 0:
        print("✅ No isolation violations found!")
        return 0

    print(f"\n{'=' * 70}")
    print("TEST ISOLATION AUDIT REPORT")
    print(f"{'=' * 70}")
    print(f"Files scanned: {len(py_files)}")
    print(f"Files with violations: {file_count}")
    print(f"Total violations: {total_violations}")
    print(f"{'=' * 70}\n")

    # Group by category
    category_counts = defaultdict(int)
    for violations in all_violations.values():
        for v in violations:
            category_counts[v.category] += 1

    print("VIOLATIONS BY CATEGORY:")
    for cat, count in sorted(category_counts.items(), key=lambda x: -x[1]):
        print(f"  {cat:25s}: {count:4d}")
    print()

    # Print details
    for rel_path, violations in sorted(all_violations.items()):
        print(f"\n{rel_path}")
        print("-" * len(str(rel_path)))
        for v in violations:
            print(f"  Line {v.line:4d} [{v.category:20s}] {v.message}")
            if args.fix and v.fix:
                print(f"           Fix: {v.fix}")

    print(f"\n{'=' * 70}")
    print(f"SUMMARY: {file_count} files with {total_violations} total violations")
    print(f"{'=' * 70}")

    return 1 if total_violations > 0 else 0


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