scripts.flaky_test_hunter

Reference › scripts › flaky_test_hunter

An automated flaky-test detector: it runs the suite several times and identifies tests whose result differs between runs. comprehensive_review runs it as an optional section, and tests/scripts/test_flaky_test_hunter.py imports it.

Invocation: main takes an optional test path (default tests/), --runs (default 5), --timeout per run in seconds (default 300) and --junit-xml to write a JUnit report. It runs pytest that many times with run_pytest, which also applies a timeout, and records per-run failures and passes (parse_failures, parse_passes) as typed records. It prints a passed and failed count for each run.

classify_flakiness compares the runs: a test that fails in every run is consistently failing, and one that passes in some runs and fails in others is flaky, with a failure rate. infer_root_cause suggests a likely category for a failure from its error type and message. generate_report renders the classification as text, and generate_junit_xml writes it as JUnit XML for CI.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Classes

_RunResult

Bases: TypedDict · Line 129

Parsed outcome of one full suite run.

_FailureRecord

Bases: TypedDict · Line 136

One per-run failure entry for a single test.

_TestOutcome

Bases: TypedDict · Line 144

Aggregated outcome for one test across all runs.

Functions

Source

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

#!/usr/bin/env python3
"""Automated flaky test detector — runs the suite N times and identifies flaky tests.

Classifies failures by error type, identifies tests that fail inconsistently,
and generates a report with root cause hypotheses.

Usage:
    python scripts/flaky_test_hunter.py --runs 5
    python scripts/flaky_test_hunter.py tests/agents/volks/architect/ --runs 10
    python scripts/flaky_test_hunter.py --runs 5 --junit-xml=flaky-report.xml
"""

import argparse
import os
import re
import subprocess
import sys
import time
from collections import defaultdict
from typing import TypedDict

# Children's output is machine-parsed (regex/substring greps): force color
# OFF for every subprocess this tool spawns. An ambient FORCE_COLOR (e.g.
# exported by the launching terminal/agent harness) made mypy/ruff/pytest
# emit ANSI into captured pipes and silently broke the parsers (gate
# 2026-08-05: mypy "0 issues" on a dirty tree, ruff int(line) crash,
# FAILED-line greps missing their lines).
os.environ["NO_COLOR"] = "1"  # gate-ok: process-bootstrap env routing
os.environ.pop("FORCE_COLOR", None)  # gate-ok: process-bootstrap env routing


def run_pytest(test_path: str, timeout: int = 600) -> tuple[str, int]:
    """Run pytest and capture output.

    Deliberately does NOT pass ``-n auto`` (pytest-xdist): this project's suite
    deadlocks under xdist (~78% of the time, see AGENTS.md / CLAUDE.md "Key
    Rules"), and this hunter is invoked by comprehensive_review.py against the
    full ``tests/`` tree by default — xdist here would hang the exact tool
    meant to find flakiness, not just risk it.
    """
    cmd = [
        sys.executable,
        "-m",
        "pytest",
        test_path,
        "--timeout",
        str(timeout),
        "-v",
        "--tb=line",
        "-q",
        "--no-header",
        # "-v -q" nets out to verbosity 0 (dot progress, no per-test lines);
        # -rA forces the short summary to list PASSED ids too, which
        # parse_passes() needs to count real passes per run.
        "-rA",
    ]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout + 60)
    return result.stdout + result.stderr, result.returncode


def parse_failures(output: str) -> list[dict]:
    """Parse pytest output into structured failure records."""
    failures = []

    # Find all FAILED lines
    failed_pattern = re.compile(r"^FAILED\s+(\S+)::(\S+)", re.MULTILINE)
    for match in failed_pattern.finditer(output):
        test_file = match.group(1)
        test_name = match.group(2)

        # Extract error context after the FAILED line
        pos = match.end()
        context = output[pos : pos + 2000]

        # Determine error type
        error_type = "Unknown"
        error_msg = ""

        # Check for common error patterns in context
        if "ConstraintError" in context:
            error_type = "ConstraintError"
        elif "EntityNotFound" in context:
            error_type = "EntityNotFound"
        elif "AssertionError" in context:
            error_type = "AssertionError"
        elif "Timeout" in context or "timeout" in context.lower():
            error_type = "Timeout"
        elif "ConnectionRefused" in context:
            error_type = "ConnectionRefused"
        elif "RuntimeError" in context:
            error_type = "RuntimeError"
        elif "ModuleNotFoundError" in context:
            error_type = "ModuleNotFoundError"

        # Extract error message
        msg_match = re.search(r"(?:Error|Exception):\s*(.+?)(?=\n|$)", context)
        if msg_match:
            error_msg = msg_match.group(1).strip()[:150]

        failures.append(
            {
                "test_file": test_file,
                "test_name": test_name,
                "error_type": error_type,
                "error_msg": error_msg,
            }
        )

    return failures


def parse_passes(output: str) -> set[str]:
    """Parse the ids of PASSED tests out of pytest output.

    Matches both short-summary format (``PASSED tests/x.py::test_y``, emitted
    under ``-rA``) and verbose progress format (``tests/x.py::test_y PASSED``),
    so the hunter keeps counting passes even if run_pytest's verbosity flags
    change.
    """
    passes: set[str] = set()
    summary_pattern = re.compile(r"^PASSED\s+(\S+::\S+)", re.MULTILINE)
    verbose_pattern = re.compile(r"^(\S+::\S+)\s+PASSED", re.MULTILINE)
    for pattern in (summary_pattern, verbose_pattern):
        for match in pattern.finditer(output):
            passes.add(match.group(1))
    return passes


class _RunResult(TypedDict):
    """Parsed outcome of one full suite run."""

    failures: list[dict]
    passes: set[str]


class _FailureRecord(TypedDict):
    """One per-run failure entry for a single test."""

    run: int
    error_type: str
    error_msg: str


class _TestOutcome(TypedDict):
    """Aggregated outcome for one test across all runs."""

    passes: int
    failures: list[_FailureRecord]
    error_types: set[str]


def _new_test_outcome() -> _TestOutcome:
    """Default-factory for ``test_outcomes`` below (a plain lambda dict literal
    doesn't get TypedDict context from mypy; a named, return-annotated
    function does)."""
    return {"passes": 0, "failures": [], "error_types": set()}


def classify_flakiness(results: list[_RunResult], total_runs: int) -> dict:
    """Classify tests by their failure patterns across runs.

    Each entry in ``results`` is one run's parsed outcome:
    ``{"failures": parse_failures(...), "passes": parse_passes(...)}``.
    Passes are counted from the actually-observed PASSED ids — never inferred
    from absence — so a test that passes early and fails late is classified
    flaky (not "consistent failure"), and never-failing tests reach
    ``consistent_passes``.
    """
    # Collect all test outcomes
    test_outcomes: defaultdict[str, _TestOutcome] = defaultdict(_new_test_outcome)

    for run_idx, run_result in enumerate(results):
        for f in run_result["failures"]:
            key = f"{f['test_file']}::{f['test_name']}"
            test_outcomes[key]["failures"].append(
                {
                    "run": run_idx + 1,
                    "error_type": f["error_type"],
                    "error_msg": f["error_msg"],
                }
            )
            test_outcomes[key]["error_types"].add(f["error_type"])

        for key in run_result["passes"]:
            test_outcomes[key]["passes"] += 1

    # Classify
    flaky = []
    consistent_failures = []
    consistent_passes = []

    for test_key, data in test_outcomes.items():
        total = data["passes"] + len(data["failures"])
        if total == 0:
            continue

        fail_rate = len(data["failures"]) / total

        if 0 < fail_rate < 1.0:
            # Flaky: fails some of the time
            flaky.append(
                {
                    "test": test_key,
                    "fail_rate": fail_rate,
                    "fails": len(data["failures"]),
                    "passes": data["passes"],
                    "error_types": list(data["error_types"]),
                    "failures": data["failures"],
                }
            )
        elif fail_rate == 1.0:
            # Consistently failing
            consistent_failures.append(
                {
                    "test": test_key,
                    "fails": len(data["failures"]),
                    "error_types": list(data["error_types"]),
                    "failures": data["failures"],
                }
            )
        else:
            # Consistently passing
            consistent_passes.append(test_key)

    return {
        "flaky": flaky,
        "consistent_failures": consistent_failures,
        "consistent_passes": consistent_passes,
    }


def infer_root_cause(error_type: str, error_msg: str) -> dict:
    """Infer root cause from error pattern."""
    causes = {
        "ConstraintError": {
            "root_cause": "Uniqueness constraint violation — test creates node with duplicate property",
            "confidence": "HIGH",
            "fix": "Use MERGE instead of CREATE, or delete existing node first",
        },
        "EntityNotFound": {
            "root_cause": "Concurrent cleanup deletes node while query is executing",
            "confidence": "HIGH",
            "fix": "Add retry logic, or filter queries by test_id",
        },
        "AssertionError": {
            "root_cause": "Test sees unexpected data (possibly from other tests)",
            "confidence": "MEDIUM",
            "fix": "Check if queries filter by test_id, or assertions are too strict",
        },
        "Timeout": {
            "root_cause": "Test hangs due to blocking I/O or infinite loop",
            "confidence": "HIGH",
            "fix": "Add timeouts to external calls, check for missing async/await",
        },
        "ConnectionRefused": {
            "root_cause": "External service (graph/Redis) not running",
            "confidence": "HIGH",
            "fix": "Start the service container",
        },
        "RuntimeError": {
            "root_cause": "Fixture setup failed (connection, config, etc.)",
            "confidence": "MEDIUM",
            "fix": "Check fixture definition and external service availability",
        },
        "ModuleNotFoundError": {
            "root_cause": "Import path issue (porting artifact)",
            "confidence": "HIGH",
            "fix": "Fix import translation from orchestrata to orchestrata_ultimate",
        },
    }

    return causes.get(
        error_type,
        {
            "root_cause": "Unknown — manual investigation needed",
            "confidence": "LOW",
            "fix": "See docs/DIAGNOSTIC_PLAYBOOK.md",
        },
    )


def generate_report(classification: dict, total_runs: int, total_time: float) -> str:
    """Generate a human-readable report."""
    lines = []
    lines.append("=" * 70)
    lines.append("FLAKY TEST HUNTER REPORT")
    lines.append("=" * 70)
    lines.append(f"Total runs: {total_runs}")
    lines.append(f"Total time: {total_time:.1f}s")
    lines.append(f"Flaky tests: {len(classification['flaky'])}")
    lines.append(f"Consistently failing: {len(classification['consistent_failures'])}")
    lines.append(f"Consistently passing: {len(classification['consistent_passes'])}")
    lines.append("=" * 70)

    if classification["flaky"]:
        lines.append("\n🎯 FLAKY TESTS (inconsistent failures)\n")
        lines.append("-" * 70)

        # Sort by fail rate descending
        for item in sorted(classification["flaky"], key=lambda x: -x["fail_rate"]):
            lines.append(f"\n{item['test']}")
            lines.append(
                f"  Fails: {item['fails']}/{item['fails'] + item['passes']} runs ({item['fail_rate']*100:.0f}%)"
            )
            lines.append(f"  Error types: {', '.join(item['error_types'])}")

            # Show failure details
            lines.extend(
                f"    Run {f['run']}: {f['error_type']} — {f['error_msg'][:80]}"
                for f in item["failures"][:3]
            )

            # Infer root cause from most common error
            most_common = max(
                item["error_types"],
                key=lambda et: sum(
                    1 for f in item["failures"] if f["error_type"] == et
                ),
            )
            cause = infer_root_cause(most_common, "")
            lines.append(f"  🔍 Likely cause: {cause['root_cause']}")
            lines.append(f"     Confidence: {cause['confidence']}")
            lines.append(f"     Fix: {cause['fix']}")

    if classification["consistent_failures"]:
        lines.append("\n❌ CONSISTENTLY FAILING TESTS (100% failure rate)\n")
        lines.append("-" * 70)

        for item in classification["consistent_failures"]:
            lines.append(f"\n{item['test']}")
            lines.append(f"  Fails: {item['fails']}/{total_runs} runs")
            lines.append(f"  Error types: {', '.join(item['error_types'])}")
            cause = infer_root_cause(item["error_types"][0], "")
            lines.append(f"  🔍 Likely cause: {cause['root_cause']}")
            lines.append(f"     Fix: {cause['fix']}")

    lines.append("\n" + "=" * 70)
    lines.append("RECOMMENDATIONS")
    lines.append("=" * 70)

    if classification["flaky"]:
        lines.append(
            "1. Fix flaky tests first — they indicate race conditions or shared state"
        )
        lines.append(
            "2. Run: python scripts/test_isolation_audit.py to find isolation violations"
        )
        lines.append(
            "3. Run: python scripts/graph_test_preflight.py to check constraint issues"
        )

    if classification["consistent_failures"]:
        lines.append(
            "4. Fix consistently failing tests — they indicate broken code or missing deps"
        )

    lines.append("5. See docs/DIAGNOSTIC_PLAYBOOK.md for detailed debugging guidance")
    lines.append("=" * 70)

    return "\n".join(lines)


def generate_junit_xml(classification: dict, output_path: str) -> None:
    """Generate JUnit XML report for CI integration."""
    import xml.etree.ElementTree as ET

    testsuites = ET.Element("testsuites")
    testsuite = ET.SubElement(testsuites, "testsuite", name="flaky-test-hunter")

    for item in classification["flaky"]:
        testcase = ET.SubElement(testsuite, "testcase", name=item["test"])
        failure = ET.SubElement(testcase, "failure", message="Flaky test detected")
        failure.text = f"Fail rate: {item['fail_rate']*100:.0f}%\n"
        failure.text += f"Error types: {', '.join(item['error_types'])}\n"
        cause = infer_root_cause(item["error_types"][0], "")
        failure.text += f"Likely cause: {cause['root_cause']}\n"
        failure.text += f"Fix: {cause['fix']}"

    tree = ET.ElementTree(testsuites)
    tree.write(output_path, encoding="utf-8", xml_declaration=True)
    print(f"JUnit XML report saved to: {output_path}")


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Detect flaky tests by running suite multiple times"
    )
    parser.add_argument("path", nargs="?", default="tests/", help="Test path to run")
    parser.add_argument(
        "--runs", type=int, default=5, help="Number of times to run the suite"
    )
    parser.add_argument(
        "--timeout", type=int, default=300, help="Timeout per run in seconds"
    )
    parser.add_argument("--junit-xml", help="Output JUnit XML report")
    args = parser.parse_args()

    print(f"Flaky Test Hunter: Running {args.path} {args.runs} times...")
    print(f"Timeout per run: {args.timeout}s")
    print("=" * 70)

    all_results: list[_RunResult] = []
    total_time: float = 0

    for run in range(1, args.runs + 1):
        print(f"\nRun {run}/{args.runs}...", end=" ", flush=True)
        start = time.monotonic()
        output, returncode = run_pytest(args.path, args.timeout)
        elapsed = time.monotonic() - start
        total_time += elapsed

        all_results.append(
            {"failures": parse_failures(output), "passes": parse_passes(output)}
        )

        # Parse pass/fail summary. pytest's own short summary line orders
        # categories as "N failed, M passed" (failed first) — matching
        # "passed" before "failed" in one pattern never fires against real
        # output, so passed/failed are searched independently.
        passed_match = re.search(r"(\d+) passed", output)
        failed_match = re.search(r"(\d+) failed", output)
        if passed_match or failed_match:
            passed = int(passed_match.group(1)) if passed_match else 0
            failed = int(failed_match.group(1)) if failed_match else 0
            print(f"{passed} passed, {failed} failed ({elapsed:.1f}s)")
        else:
            print(f"completed ({elapsed:.1f}s)")

    print(f"\n{'=' * 70}")
    print("Analyzing results...")

    classification = classify_flakiness(all_results, args.runs)
    report = generate_report(classification, args.runs, total_time)
    print(report)

    if args.junit_xml:
        generate_junit_xml(classification, args.junit_xml)

    return 1 if classification["flaky"] or classification["consistent_failures"] else 0


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