Knowledge Wiki Orchestrata Ultimate

scripts.diagnose_test_failure

Reference › scripts › diagnose_test_failure

Auto-diagnoses a test failure by following a seven-step diagnostic process. tests/scripts/test_diagnose_test_failure.py imports it, and a human runs it after a failing test run. It uses the graph factory in graph_factory for its graph check.

Invocation: main takes either a pytest failure log as a positional argument, --run TEST to run one test and diagnose its output, or --last-failed to rerun the last failed test. With none of these it prints usage and exits 1. If no failure can be parsed it saves the output to a temporary file for inspection and exits 1.

The flow: parse_failure_log extracts the failing test, file and error from the log. read_test_code reads that test's source, extract_fixtures_from_test lists its fixtures and find_fixture_code finds each fixture's source. check_graph_constraints inspects the graph for constraint problems that could explain a failure, and generate_diagnosis assembles a text diagnosis from all of these. run_test_and_capture and run_last_failed run pytest to obtain the log.

At a glance
Source scripts/diagnose_test_failure.py
Lines of code 505
Classes 0
Functions 9
Imports (internal) orchestrata_ultimate.core.graph_factory
Imported by none
Tests that import it tests/scripts/test_diagnose_test_failure.py

Class diagram

No classes; see Functions.

Dependency diagram

Dependency diagram scripts.diagnose_test_failure

API

Functions

Function Signature Summary
parse_failure_log parse_failure_log(log_content: str) -> dict | None Parse pytest failure log to extract key information.
read_test_code read_test_code(test_file: str, test_name: str) -> tuple[str, int] | None Read the failing test function from the test file.
find_fixture_code find_fixture_code(fixture_name: str) -> str | None Find fixture definition in conftest.py files.
extract_fixtures_from_test extract_fixtures_from_test(test_code: str) -> list[str] Extract fixture names from test function parameters.
check_graph_constraints check_graph_constraints() -> list[dict] Check the operational KG's constraints if available.
generate_diagnosis generate_diagnosis(failure: dict, test_code: str | None, fixtures: list[str]) -> str Generate a diagnosis report.
run_test_and_capture run_test_and_capture(test_path: str) -> str Run a specific test and capture output.
run_last_failed run_last_failed() -> str Run pytest --lf and capture output.
main main() -> int

Source

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

#!/usr/bin/env python3
"""Auto-diagnose a test failure by performing the 7-step diagnostic process.

Reads a pytest failure log (or runs the test once) and produces a structured
diagnosis report with root cause hypothesis and recommended fixes.

Usage:
    python scripts/diagnose_test_failure.py /tmp/pytest_failure.log
    python scripts/diagnose_test_failure.py --run tests/path/test_file.py::test_name
    python scripts/diagnose_test_failure.py --last-failed
"""

import argparse
import os
import re
import subprocess
import sys
import tempfile
from pathlib import Path

# Common error patterns and their likely causes
ERROR_SIGNATURES = {
    "ConstraintError": {
        "cause": "graph uniqueness constraint violation",
        "check": "Check if another test created a node with the same unique property",
        "fix": "Use MERGE instead of CREATE, or delete existing node first",
    },
    "EntityNotFound": {
        "cause": "Node was deleted by concurrent transaction during query",
        "check": "Check if fixture cleanup (DETACH DELETE) runs while query is active",
        "fix": "Add retry logic, or filter queries by test_id",
    },
    "ConnectionRefused": {
        "cause": "External service not running",
        "check": "Check incus containers: redis-queue, neo4j-db",
        "fix": "Start the service container",
    },
    "AssertionError": {
        "cause": "Test assertion failed",
        "check": "Read the assertion to understand expected vs actual value",
        "fix": "Fix the code or update the test assertion",
    },
    "Timeout": {
        "cause": "Test took longer than --timeout limit",
        "check": "Check for infinite loops, blocking I/O, or missing async/await",
        "fix": "Profile the test, add timeouts to external calls",
    },
    "ModuleNotFoundError": {
        "cause": "Missing import or incorrect module path",
        "check": "Check import translations from orchestrata to orchestrata_ultimate",
        "fix": "Fix the import path",
    },
    "RuntimeError": {
        "cause": "Fixture setup failed",
        "check": "Read the fixture definition for connection/setup errors",
        "fix": "Check external service availability and credentials",
    },
}


# 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 parse_failure_log(log_content: str) -> dict | None:
    """Parse pytest failure log to extract key information."""
    failure: dict[str, str | int | None] = {
        "test_name": None,
        "test_file": None,
        "error_type": None,
        "error_message": None,
        "traceback": None,
        "line_number": None,
    }

    # Find FAILED line
    failed_match = re.search(r"FAILED\s+(\S+)::(\S+)", log_content)
    if failed_match:
        failure["test_file"] = failed_match.group(1)
        failure["test_name"] = failed_match.group(2)

    # Find error type and message
    # Look for the last exception in the traceback
    error_match = re.search(r"(\w+Error):\s*(.+?)(?=\n|$)", log_content)
    if error_match:
        failure["error_type"] = error_match.group(1)
        failure["error_message"] = error_match.group(2).strip()

    # Look for neo4j specific errors
    neo4j_match = re.search(r"neo4j\.exceptions\.(\w+):\s*\{([^}]+)\}", log_content)
    if neo4j_match:
        failure["error_type"] = neo4j_match.group(1)
        failure["error_message"] = neo4j_match.group(2).strip()

    # Look for AssertionError with details
    assertion_match = re.search(r"AssertionError:.*assert\s+(.+)", log_content)
    if assertion_match:
        failure["error_type"] = "AssertionError"
        failure["error_message"] = assertion_match.group(0)

    # Extract traceback section
    tb_start = log_content.find("Traceback")
    if tb_start != -1:
        tb_end = log_content.find("\n\n", tb_start + 1000)
        if tb_end == -1:
            tb_end = len(log_content)
        failure["traceback"] = log_content[tb_start:tb_end]

    # Extract line number from traceback
    test_name = failure["test_name"]
    if failure["test_file"] and isinstance(test_name, str):
        line_pattern = rf"{re.escape(test_name)}\s*\n.*?line\s+(\d+)"
        line_match = re.search(line_pattern, log_content, re.DOTALL)
        if line_match:
            failure["line_number"] = int(line_match.group(1))

    return failure if failure["test_name"] else None


def read_test_code(test_file: str, test_name: str) -> tuple[str, int] | None:
    """Read the failing test function from the test file."""
    file_path = Path(test_file)
    if not file_path.exists():
        return None

    try:
        content = file_path.read_text(encoding="utf-8")
        lines = content.splitlines()

        # Find the test function
        for i, line in enumerate(lines):
            if re.search(rf"\bdef\s+{re.escape(test_name)}\b", line):
                # Extract function body (naive: until next function or class)
                func_lines = []
                indent = len(line) - len(line.lstrip())
                for j in range(i, min(i + 50, len(lines))):
                    if j > i and lines[j].strip():
                        curr_indent = len(lines[j]) - len(lines[j].lstrip())
                        if curr_indent <= indent and (
                            lines[j].startswith("def ")
                            or lines[j].startswith("class ")
                            or lines[j].startswith("async def")
                        ):
                            break
                    func_lines.append(lines[j])
                return "\n".join(func_lines), i + 1
    except Exception:
        pass

    return None


def find_fixture_code(fixture_name: str) -> str | None:
    """Find fixture definition in conftest.py files."""
    for conftest in Path("tests").rglob("conftest.py"):
        try:
            content = conftest.read_text(encoding="utf-8")
            lines = content.splitlines()

            for i, line in enumerate(lines):
                if re.search(rf"\bdef\s+{re.escape(fixture_name)}\b", line):
                    # Extract fixture body
                    func_lines = []
                    indent = len(line) - len(line.lstrip())
                    for j in range(i, min(i + 40, len(lines))):
                        if j > i and lines[j].strip():
                            curr_indent = len(lines[j]) - len(lines[j].lstrip())
                            if curr_indent <= indent and (
                                lines[j].startswith("def ")
                                or lines[j].startswith("class ")
                                or lines[j].startswith("async def")
                            ):
                                break
                        func_lines.append(lines[j])
                    return "\n".join(func_lines)
        except Exception:
            continue

    return None


def extract_fixtures_from_test(test_code: str) -> list[str]:
    """Extract fixture names from test function parameters."""
    fixtures = []
    # Match function definition and parameters
    match = re.search(r"def\s+\w+\s*\((.*?)\)", test_code)
    if match:
        params = match.group(1)
        # Split by comma, skip 'self' and common builtins
        for param in params.split(","):
            param = param.strip().split(":")[0].split("=")[0].strip()
            if param and param not in (
                "self",
                "cls",
                "request",
                "capfd",
                "capsys",
                "tmp_path",
            ):
                fixtures.append(param)
    return fixtures


def check_graph_constraints() -> list[dict]:
    """Check the operational KG's constraints if available.

    Engine-aware (Neo4j -> orchgraph migration, plans/neo4j-to-orchgraph-
    migration-plan.md dialect gate): Neo4j's admin surface is
    ``SHOW CONSTRAINTS`` (columns ``name``/``type``/``labelsOrTypes``/
    ``properties``); Memgraph's is ``SHOW CONSTRAINT INFO`` (columns
    ``constraint type``/``label``/``properties`` — no constraint name, no
    plural ``labelsOrTypes``). Normalized to the same output shape either
    way so callers don't need to know which engine answered.
    """
    try:
        import os

        from neo4j import GraphDatabase

        from orchestrata_ultimate.core.graph_factory import (
            resolve_kg_database,
            resolve_uri,
        )

        # NEO4J_URI/USERNAME/PASSWORD were retired with the Neo4j containers
        # on 2026-07-27; resolve_uri() reads the KG_ROUTING_URI/KG_URI seam and
        # raises with the key name rather than dialling a localhost default
        # that nothing in this infrastructure has ever listened on.
        uri = resolve_uri()
        engine = os.environ.get("KG_ENGINE", "neo4j")

        driver = GraphDatabase.driver(uri, auth=None)
        # driver.session()'s keyword params carry heterogeneous types
        # (fetch_size: int, bookmarks: ..., etc.) — unpacking a plain
        # dict[str, str] via **session_kwargs is not something mypy can
        # verify against that signature, so the two engine paths call it
        # directly instead of building a kwargs dict.
        session_cm = (
            driver.session(database=resolve_kg_database())
            if engine == "memgraph"
            else driver.session()
        )
        with session_cm as session:
            if engine == "memgraph":
                result = session.run("SHOW CONSTRAINT INFO")
                constraints = [
                    {
                        "name": "",
                        "type": record.get("constraint type", ""),
                        "labels": [record.get("label", "")],
                        "properties": record.get("properties", []),
                    }
                    for record in result
                ]
            else:
                result = session.run("SHOW CONSTRAINTS")
                constraints = [
                    {
                        "name": record.get("name", ""),
                        "type": record.get("type", ""),
                        "labels": record.get("labelsOrTypes", []),
                        "properties": record.get("properties", []),
                    }
                    for record in result
                ]
            driver.close()
            return constraints
    except Exception:
        return []


def generate_diagnosis(
    failure: dict, test_code: str | None, fixtures: list[str]
) -> str:
    """Generate a diagnosis report."""
    lines = []
    lines.append("=" * 70)
    lines.append("AUTO-DIAGNOSIS REPORT")
    lines.append("=" * 70)
    lines.append("")

    # Basic info
    lines.append(f"FAILING TEST: {failure['test_file']}::{failure['test_name']}")
    if failure["line_number"]:
        lines.append(f"LINE: {failure['line_number']}")
    lines.append(f"ERROR TYPE: {failure['error_type'] or 'Unknown'}")
    if failure["error_message"]:
        msg = failure["error_message"][:200]
        lines.append(f"ERROR MESSAGE: {msg}")
    lines.append("")

    # Match against known signatures
    matched = False
    for signature, info in ERROR_SIGNATURES.items():
        if failure["error_type"] and signature in failure["error_type"]:
            lines.append(f"DIAGNOSIS: {info['cause']}")
            lines.append(f"CHECK: {info['check']}")
            lines.append(f"RECOMMENDED FIX: {info['fix']}")
            matched = True
            break

    if not matched:
        lines.append("DIAGNOSIS: Unknown error pattern")
        lines.append("CHECK: Read the full traceback for context")
        lines.append("RECOMMENDED FIX: Investigate the specific error type")

    lines.append("")

    # Test code snippet
    if test_code:
        lines.append("-" * 70)
        lines.append("TEST CODE:")
        lines.append("-" * 70)
        lines.extend(f"  {line}" for line in test_code.split("\n")[:20])
        lines.append("")

    # Fixture analysis
    if fixtures:
        lines.append("-" * 70)
        lines.append("FIXTURES USED:")
        lines.append("-" * 70)
        for fixture in fixtures:
            fixture_code = find_fixture_code(fixture)
            if fixture_code:
                # Check scope
                scope_match = re.search(r"scope\s*=\s*['\"](\w+)['\"]", fixture_code)
                scope = scope_match.group(1) if scope_match else "function"
                lines.append(f"  {fixture} (scope={scope})")

                # Check cleanup
                if "yield" in fixture_code:
                    if "DELETE" in fixture_code or "cleanup" in fixture_code.lower():
                        lines.append(
                            "    ⚠️  Has cleanup logic — check for race conditions"
                        )
                if "neo4j" in fixture_code.lower() or "redis" in fixture_code.lower():
                    lines.append(
                        "    ⚠️  Uses external service — verify service is running"
                    )
            else:
                lines.append(f"  {fixture} (builtin or not found)")
        lines.append("")

    # Check for interfering tests
    if failure["test_file"] and failure["test_name"]:
        lines.append("-" * 70)
        lines.append("SUSPECTED INTERFERENCE:")
        lines.append("-" * 70)

        # Search for tests that create similar resources
        if test_code:
            # Look for hardcoded names
            hardcoded = re.findall(r"['\"](\w+)[\"']", test_code)
            for name in set(hardcoded):
                if len(name) > 3 and name.lower() not in (
                    "true",
                    "false",
                    "none",
                    "self",
                    "test",
                ):
                    # Count how many other tests use this name
                    result = subprocess.run(
                        ["grep", "-rln", name, "tests/", "--include=*.py"],
                        capture_output=True,
                        text=True,
                    )
                    count = len(
                        [line for line in result.stdout.strip().split("\n") if line]
                    )
                    if count > 1:
                        lines.append(f"  '{name}' used in {count} test files")
                        lines.append("    → Potential shared state conflict")

        lines.append("")

    # Graph constraints check
    constraints = check_graph_constraints()
    try:
        from orchestrata_ultimate.core.graph_factory import resolve_kg_database as _rgd

        database = _rgd()
    except Exception:
        database = "<unknown>"
    lines.append("-" * 70)
    lines.append("GRAPH CONSTRAINTS:")
    lines.append("-" * 70)
    if constraints:
        lines.extend(
            f"  {c['name']}: {c['type']} on {c['labels']}.{c['properties']}"
            for c in constraints[:5]
        )
    else:
        lines.append(f"(none on database {database})")
    lines.append("")

    lines.append("=" * 70)
    lines.append("NEXT STEPS:")
    lines.append("=" * 70)
    lines.append("1. Read the full traceback in the log file")
    lines.append("2. Check the fixture definitions for cleanup logic")
    lines.append("3. Run the test in isolation: pytest <file>::<test> -v")
    lines.append("4. Check external service status if applicable")
    lines.append("5. See docs/DIAGNOSTIC_PLAYBOOK.md for detailed guidance")
    lines.append("=" * 70)

    return "\n".join(lines)


def run_test_and_capture(test_path: str) -> str:
    """Run a specific test and capture output."""
    cmd = [
        sys.executable,
        "-m",
        "pytest",
        test_path,
        "-v",
        "--tb=long",
        "--timeout",
        "120",
    ]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=180)
    return result.stdout + result.stderr


def run_last_failed() -> str:
    """Run pytest --lf and capture output."""
    cmd = [
        sys.executable,
        "-m",
        "pytest",
        "tests/",
        "--lf",
        "-v",
        "--tb=long",
        "--timeout",
        "120",
    ]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=180)
    return result.stdout + result.stderr


def main() -> int:
    parser = argparse.ArgumentParser(description="Auto-diagnose a test failure")
    parser.add_argument("logfile", nargs="?", help="Path to pytest failure log")
    parser.add_argument("--run", metavar="TEST", help="Run specific test and diagnose")
    parser.add_argument(
        "--last-failed", action="store_true", help="Diagnose last failed test"
    )
    args = parser.parse_args()

    log_content = ""

    if args.logfile:
        log_path = Path(args.logfile)
        if not log_path.exists():
            print(f"ERROR: Log file not found: {args.logfile}")
            return 1
        log_content = log_path.read_text(encoding="utf-8")
    elif args.run:
        print(f"Running test: {args.run}")
        log_content = run_test_and_capture(args.run)
    elif args.last_failed:
        print("Running last failed test...")
        log_content = run_last_failed()
    else:
        print("ERROR: Provide a log file, --run TEST, or --last-failed")
        parser.print_help()
        return 1

    failure = parse_failure_log(log_content)
    if not failure:
        print("No failure found in the provided log/output")
        # Save output for manual inspection
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".log", delete=False, prefix="pytest_diagnosis_"
        ) as f:
            f.write(log_content)
            print(f"Output saved to: {f.name}")
        return 1

    test_code = None
    if failure["test_file"]:
        test_result = read_test_code(failure["test_file"], failure["test_name"])
        if test_result:
            test_code, _ = test_result

    fixtures = []
    if test_code:
        fixtures = extract_fixtures_from_test(test_code)

    diagnosis = generate_diagnosis(failure, test_code, fixtures)
    print(diagnosis)

    return 0


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