scripts.comprehensive_review

Reference › scripts › comprehensive_review

Runs all the diagnostic tools and generates one unified report. tests/scripts/test_comprehensive_review.py imports it, and a human runs it.

Invocation: main takes --output/-o (a file for the report), --format (markdown, the default, or json), --flaky-runs N (run the flaky test hunter N times; default 0, and slow) and --skip-tests.

The sections run in a fixed order, each through run_command, which applies a timeout (default 300 seconds): run_code_review (code-review), run_isolation_audit (the test isolation audit), run_neo4j_preflight (the graph preflight), run_test_suite (unless skipped) and run_flaky_hunter (flaky_test_hunter). A ComprehensiveReport collects them with add_section, recording a status, detail lines and counts of issues and critical issues, and renders itself with to_markdown or to_json. A section with errors is marked FAIL; at most 20 warnings are listed.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Classes

ComprehensiveReport

Line 39

Unified report aggregating all diagnostic outputs.

Functions

Source

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

#!/usr/bin/env python3
"""Comprehensive Review — Runs all diagnostic tools and generates unified report.

Combines:
- Code review (ruff, mypy, AST style checks)
- Test isolation audit
- graph preflight validation
- Test suite execution with diagnostic capture
- Flaky test analysis (if --runs specified)

Usage:
    python scripts/comprehensive_review.py
    python scripts/comprehensive_review.py --output reports/comprehensive-review.md
    python scripts/comprehensive_review.py --flaky-runs 3 --output reports/review.md
"""

import argparse
import json
import os
import re
import subprocess
import sys
import tempfile
import time
from datetime import datetime
from pathlib import Path
from typing import Any

# 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


class ComprehensiveReport:
    """Unified report aggregating all diagnostic outputs."""

    def __init__(self) -> None:
        self.timestamp = datetime.now().isoformat()
        # Section payload is heterogeneous per key (str/list/int) — see
        # add_section() below — so `dict[str, Any]` is the honest value type.
        self.sections: dict[str, dict[str, Any]] = {}
        self.summary = {
            "total_issues": 0,
            "critical_issues": 0,
            "warnings": 0,
            "tests_passed": 0,
            "tests_failed": 0,
            "flaky_tests": 0,
            "isolation_violations": 0,
            "constraint_violations": 0,
        }

    def add_section(
        self, name: str, status: str, details: list, issues: int = 0, critical: int = 0
    ) -> None:
        self.sections[name] = {
            "status": status,
            "details": details,
            "issues": issues,
            "critical": critical,
        }
        self.summary["total_issues"] += issues
        self.summary["critical_issues"] += critical
        if critical > 0:
            self.summary["warnings"] += issues

    def to_markdown(self) -> str:
        lines = []
        lines.append("# Comprehensive Code Review Report")
        lines.append(f"\n**Generated:** {self.timestamp}")
        lines.append("**Project:** orchestrata-ultimate")
        lines.append("")

        # Summary table
        lines.append("## Executive Summary")
        lines.append("")
        lines.append("| Metric | Count | Status |")
        lines.append("|--------|-------|--------|")

        def status_icon(s: str) -> str:
            return "✅" if s == "PASS" else "⚠️" if s == "WARN" else "❌"

        for name, section in self.sections.items():
            icon = status_icon(section["status"])
            issues = section["issues"]
            critical = section["critical"]
            if critical > 0:
                lines.append(
                    f"| {name} | {issues} issues ({critical} critical) | {icon} {section['status']} |"
                )
            else:
                lines.append(
                    f"| {name} | {issues} issues | {icon} {section['status']} |"
                )

        lines.append("")

        # Detailed sections
        for name, section in self.sections.items():
            lines.append(f"## {name}")
            lines.append(
                f"**Status:** {status_icon(section['status'])} {section['status']}"
            )
            lines.append("")

            if section["details"]:
                lines.extend(
                    f"- {detail}" for detail in section["details"][:50]
                )  # Limit output
                if len(section["details"]) > 50:
                    lines.append(f"- ... and {len(section['details']) - 50} more")
            else:
                lines.append("No issues found.")

            lines.append("")

        # Recommendations
        lines.append("## Recommendations")
        lines.append("")

        if self.summary["critical_issues"] > 0:
            lines.append("### 🔴 Critical (Fix Before Merge)")
            for name, section in self.sections.items():
                if section["critical"] > 0:
                    lines.append(f"- **{name}:** {section['critical']} critical issues")
            lines.append("")

        if self.summary["isolation_violations"] > 0:
            lines.append("### ⚠️ Test Isolation")
            lines.append(
                "- Run `python scripts/test_isolation_audit.py tests/` to see all violations"
            )
            lines.append("- Fix CREATE vs MERGE for Neo4j File nodes")
            lines.append("- Remove pytest.skip usage")
            lines.append("")

        if self.summary["constraint_violations"] > 0:
            lines.append("### ⚠️ Neo4j Constraints")
            lines.append(
                "- Run `python scripts/graph_test_preflight.py tests/ --offline`"
            )
            lines.append("- Replace CREATE with MERGE for constrained properties")
            lines.append("")

        if self.summary["flaky_tests"] > 0:
            lines.append("### 🎯 Flaky Tests")
            lines.append(
                "- Run `python scripts/flaky_test_hunter.py --runs 5` for detailed analysis"
            )
            lines.append("")

        lines.append("### 📚 References")
        lines.append("- `docs/DIAGNOSTIC_PLAYBOOK.md` — How to debug test failures")
        lines.append("- `docs/PROPOSED_DIAGNOSTIC_TOOLS.md` — Tool documentation")
        lines.append("")

        lines.append("---")
        lines.append(
            f"*Report generated by comprehensive_review.py at {self.timestamp}*"
        )

        return "\n".join(lines)

    def to_json(self) -> dict:
        return {
            "timestamp": self.timestamp,
            "summary": self.summary,
            "sections": self.sections,
        }


def run_command(
    cmd: list[str], timeout: int = 300, capture: bool = True
) -> tuple[int, str, str]:
    """Run a command and return exit code, stdout, stderr."""
    try:
        result = subprocess.run(
            cmd,
            capture_output=capture,
            text=True,
            timeout=timeout,
        )
        return result.returncode, result.stdout, result.stderr
    except subprocess.TimeoutExpired:
        return -1, "", f"Command timed out after {timeout}s"
    except Exception as e:
        return -1, "", str(e)


def run_code_review() -> tuple[int, list]:
    """Run the existing code review script."""
    print("🔍 Running code review...")
    with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
        output_path = f.name

    cmd = [
        sys.executable,
        "scripts/code-review.py",
        "src/",
        "tests/",
        "--format",
        "json",
        "--output",
        output_path,
    ]

    rc, stdout, stderr = run_command(cmd, timeout=300)

    try:
        with open(output_path) as f:
            data = json.load(f)
        issues = data.get("issues", [])
        return rc, issues
    except Exception:
        return rc, []


def run_isolation_audit() -> tuple[int, list]:
    """Run test isolation audit."""
    print("🔍 Running test isolation audit...")
    cmd = [sys.executable, "scripts/test_isolation_audit.py", "tests/"]
    rc, stdout, stderr = run_command(cmd, timeout=60)

    # Parse output for violations
    violations = []
    current_file = None
    for line in (stdout + stderr).split("\n"):
        if line.startswith("tests/"):
            current_file = line.strip()
        elif "[CREATE_VS_MERGE]" in line or "[HARDCODED_NAME]" in line:
            violations.append(f"{current_file}: {line.strip()}")
        elif "[PYTEST_SKIP]" in line:
            violations.append(f"{current_file}: {line.strip()}")
        elif "[FIXTURE_SCOPE]" in line:
            violations.append(f"{current_file}: {line.strip()}")
        elif "[OS_ENVIRON_MUTATION]" in line:
            violations.append(f"{current_file}: {line.strip()}")

    return rc, violations


def run_neo4j_preflight() -> tuple[int, list]:
    """Run graph preflight validation."""
    print("🔍 Running graph preflight...")
    cmd = [sys.executable, "scripts/graph_test_preflight.py", "tests/", "--offline"]
    rc, stdout, stderr = run_command(cmd, timeout=60)

    violations = []
    current_file = None
    for line in (stdout + stderr).split("\n"):
        if line.startswith("tests/"):
            current_file = line.strip()
        elif "[HIGH]" in line or "[MEDIUM]" in line:
            violations.append(f"{current_file}: {line.strip()}")

    return rc, violations


def run_test_suite() -> tuple[int, int, int, str]:
    """Run the test suite and capture results."""
    print("🔍 Running test suite...")
    # No pytest-xdist (-n/--dist): it deadlocks this suite (~78% stall) — see
    # Key Rules / T15. This diagnostic must run serially like every other
    # entry point in the project.
    cmd = [
        sys.executable,
        "-m",
        "pytest",
        "tests/",
        "--timeout",
        "120",
        "-q",
        "--tb=line",
    ]

    rc, stdout, stderr = run_command(cmd, timeout=600)
    output = stdout + stderr

    # Parse summary. pytest's short summary line orders categories as e.g.
    # "1 failed, 2 passed, 1 error in 0.05s" — "failed" before "passed",
    # never "passed ... failed ... error" — so a combined regex requiring
    # that literal order can never match real pytest output. Match each
    # count independently instead, and fold setup/collection errors into
    # the failure count so they aren't silently dropped from the report.
    passed = 0
    failed = 0

    passed_match = re.search(r"(\d+) passed", output)
    if passed_match:
        passed = int(passed_match.group(1))

    fail_match = re.search(r"(\d+) failed", output)
    if fail_match:
        failed = int(fail_match.group(1))

    error_match = re.search(r"(\d+) error", output)
    if error_match:
        failed += int(error_match.group(1))

    return rc, passed, failed, output


def run_flaky_hunter(runs: int) -> tuple[int, list]:
    """Run flaky test hunter."""
    print(f"🔍 Running flaky test hunter ({runs} runs)...")
    cmd = [
        sys.executable,
        "scripts/flaky_test_hunter.py",
        "--runs",
        str(runs),
    ]
    rc, stdout, stderr = run_command(cmd, timeout=runs * 400)

    # Parse flaky tests from output
    flaky = []
    in_flaky = False
    for line in (stdout + stderr).split("\n"):
        if "FLAKY TESTS" in line:
            in_flaky = True
        elif in_flaky and line.startswith("tests/"):
            flaky.append(line.strip())
        elif in_flaky and line.startswith("="):
            in_flaky = False

    return rc, flaky


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Comprehensive code review with diagnostics"
    )
    parser.add_argument("--output", "-o", help="Output file for report")
    parser.add_argument(
        "--format",
        choices=["markdown", "json"],
        default="markdown",
        help="Output format",
    )
    parser.add_argument(
        "--flaky-runs", type=int, default=0, help="Run flaky test hunter N times (slow)"
    )
    parser.add_argument(
        "--skip-tests", action="store_true", help="Skip test suite execution"
    )
    args = parser.parse_args()

    start_time = time.time()
    report = ComprehensiveReport()

    # 1. Code Review
    rc, code_issues = run_code_review()
    errors = [i for i in code_issues if i.get("severity") == "ERROR"]
    warnings = [i for i in code_issues if i.get("severity") == "WARNING"]

    report.add_section(
        "Code Review",
        "PASS" if not errors else "FAIL",
        [
            f"{i['file']}:{i['line']} [{i['code']}] {i['message']}"
            for i in errors + warnings[:20]
        ],
        issues=len(code_issues),
        critical=len(errors),
    )

    # 2. Test Isolation Audit
    rc, isolation_violations = run_isolation_audit()
    critical = len(
        [
            v
            for v in isolation_violations
            if "CREATE_VS_MERGE" in v or "HARDCODED_NAME" in v
        ]
    )
    report.summary["isolation_violations"] = len(isolation_violations)

    report.add_section(
        "Test Isolation Audit",
        "PASS" if not isolation_violations else "WARN",
        isolation_violations[:30],
        issues=len(isolation_violations),
        critical=critical,
    )

    # 3. Neo4j Preflight
    rc, constraint_violations = run_neo4j_preflight()
    critical = len([v for v in constraint_violations if "[HIGH]" in v])
    report.summary["constraint_violations"] = len(constraint_violations)

    report.add_section(
        "Neo4j Constraint Validation",
        "PASS" if not constraint_violations else "WARN",
        constraint_violations[:30],
        issues=len(constraint_violations),
        critical=critical,
    )

    # 4. Test Suite
    if not args.skip_tests:
        rc, passed, failed, test_output = run_test_suite()
        report.summary["tests_passed"] = passed
        report.summary["tests_failed"] = failed

        # Parse failure details
        failure_details = []
        for line in test_output.split("\n"):
            if line.startswith("FAILED"):
                failure_details.append(line.strip())

        report.add_section(
            "Test Suite",
            "PASS" if failed == 0 else "FAIL",
            failure_details[:20],
            issues=failed,
            critical=failed,
        )

        # 5. Flaky Test Hunter (optional, slow)
        if args.flaky_runs > 0 and failed > 0:
            rc, flaky = run_flaky_hunter(args.flaky_runs)
            report.summary["flaky_tests"] = len(flaky)
            report.add_section(
                "Flaky Test Analysis",
                "PASS" if not flaky else "WARN",
                flaky,
                issues=len(flaky),
                critical=0,
            )
    else:
        report.add_section(
            "Test Suite",
            "SKIP",
            ["Skipped via --skip-tests"],
            issues=0,
            critical=0,
        )

    # Generate output
    elapsed = time.time() - start_time

    if args.format == "json":
        content = json.dumps(report.to_json(), indent=2)
    else:
        content = report.to_markdown()

    # Write to file or stdout
    if args.output:
        output_path = Path(args.output)
        output_path.parent.mkdir(parents=True, exist_ok=True)
        output_path.write_text(content, encoding="utf-8")
        print(f"\n✅ Report written to {output_path}")
    else:
        # Write to reviews/ directory with timestamp
        reviews_dir = Path("reviews")
        reviews_dir.mkdir(exist_ok=True)
        timestamp = datetime.now().strftime("%Y-%m-%d-%H%M%S")
        ext = "json" if args.format == "json" else "md"
        output_path = reviews_dir / f"comprehensive-review-{timestamp}.{ext}"
        output_path.write_text(content, encoding="utf-8")
        print(f"\n✅ Report written to {output_path}")

    print(f"\n⏱️  Total time: {elapsed:.1f}s")
    print(
        f"📊 Summary: {report.summary['critical_issues']} critical, {report.summary['total_issues']} total issues"
    )

    return 1 if report.summary["critical_issues"] > 0 else 0


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