leocrm/scripts/ai_health_check.py

#!/usr/bin/env python3
"""API Health Check — enumerates all registered routes and probes GET endpoints.

Usage:
    python scripts/ai_health_check.py --base-url http://localhost:8000 --token <session-token>
    python scripts/ai_health_check.py --base-url http://localhost:8000 --token <token> --output csv --report-file report.csv

Checks all GET endpoints with an auth token and reports status codes + response times.
POST/PUT/DELETE endpoints are documented but not executed (safe mode).

Exit codes:
    0 — all probed endpoints healthy (2xx/3xx)
    1 — one or more endpoints failed
    2 — script error (misconfiguration, connection failure)
"""

from __future__ import annotations

import argparse
import csv
import io
import json
import os
import sys
import time
from typing import Any

# Add project root to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import httpx

# ─── Route Enumeration ────────────────────────────────────────────────

SAFE_METHODS = {"GET", "HEAD", "OPTIONS"}


def enumerate_routes() -> list[dict[str, Any]]:
    """Enumerate all registered API routes from the FastAPI app.

    Imports the app and extracts route metadata without starting a server.
    Returns a list of dicts: {path, methods, name, tags}.
    """
    routes: list[dict[str, Any]] = []
    try:
        from app.main import app

        for route in app.routes:
            # Skip static mounts and SPA catch-all
            if not hasattr(route, "methods") or not hasattr(route, "path"):
                continue
            path = route.path
            # Only include API routes
            if not path.startswith("/api/"):
                continue
            methods = sorted(route.methods - {"HEAD", "OPTIONS"}) if route.methods else []
            if not methods:
                continue
            tags = getattr(route, "tags", []) or []
            name = getattr(route, "name", "") or ""
            routes.append({
                "path": path,
                "methods": methods,
                "name": name,
                "tags": tags,
            })
    except Exception as exc:
        print(f"ERROR: Failed to enumerate routes: {exc}", file=sys.stderr)
        # Return empty list — caller can still probe base_url manually
    return routes


# ─── Endpoint Probing ────────────────────────────────────────────────


async def probe_all_endpoints(
    base_url: str,
    token: str,
    routes: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    """Probe all GET endpoints and document non-GET ones."""
    results: list[dict[str, Any]] = []
    headers = {"Authorization": f"Bearer {token}"}

    async with httpx.AsyncClient(base_url=base_url.rstrip("/")) as client:
        for route in routes:
            path = route["path"]
            for method in route["methods"]:
                if method in SAFE_METHODS:
                    # Probe GET endpoints
                    start = time.perf_counter()
                    try:
                        response = await client.get(path, headers=headers, timeout=30.0)
                        elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
                        status_code = response.status_code
                        healthy = status_code < 400
                        results.append({
                            "endpoint": path,
                            "method": method,
                            "status": status_code,
                            "response_time_ms": elapsed_ms,
                            "healthy": healthy,
                            "tags": route.get("tags", []),
                            "note": "" if healthy else f"HTTP {status_code}",
                        })
                    except httpx.TimeoutException:
                        results.append({
                            "endpoint": path,
                            "method": method,
                            "status": "timeout",
                            "response_time_ms": round((time.perf_counter() - start) * 1000, 2),
                            "healthy": False,
                            "tags": route.get("tags", []),
                            "note": "Request timed out",
                        })
                    except httpx.ConnectError as exc:
                        results.append({
                            "endpoint": path,
                            "method": method,
                            "status": "connection_error",
                            "response_time_ms": round((time.perf_counter() - start) * 1000, 2),
                            "healthy": False,
                            "tags": route.get("tags", []),
                            "note": str(exc),
                        })
                    except Exception as exc:
                        results.append({
                            "endpoint": path,
                            "method": method,
                            "status": "error",
                            "response_time_ms": round((time.perf_counter() - start) * 1000, 2),
                            "healthy": False,
                            "tags": route.get("tags", []),
                            "note": str(exc),
                        })
                else:
                    # Document POST/PUT/DELETE — do not execute
                    results.append({
                        "endpoint": path,
                        "method": method,
                        "status": "skipped",
                        "response_time_ms": 0,
                        "healthy": True,
                        "tags": route.get("tags", []),
                        "note": "Non-GET method — documented only",
                    })

    return results


# ─── Report Formatting ───────────────────────────────────────────────


def format_json_report(results: list[dict[str, Any]]) -> str:
    """Format results as structured JSON report."""
    total = len(results)
    healthy = sum(1 for r in results if r["healthy"] and r["status"] != "skipped")
    failed = sum(1 for r in results if not r["healthy"])
    skipped = sum(1 for r in results if r["status"] == "skipped")

    report = {
        "summary": {
            "total_endpoints": total,
            "healthy": healthy,
            "failed": failed,
            "skipped": skipped,
            "all_healthy": failed == 0,
        },
        "endpoints": results,
    }
    return json.dumps(report, indent=2, ensure_ascii=False)


def format_csv_report(results: list[dict[str, Any]]) -> str:
    """Format results as CSV."""
    output = io.StringIO()
    writer = csv.DictWriter(
        output,
        fieldnames=["endpoint", "method", "status", "response_time_ms", "healthy", "tags", "note"],
    )
    writer.writeheader()
    for r in results:
        writer.writerow({
            "endpoint": r["endpoint"],
            "method": r["method"],
            "status": r["status"],
            "response_time_ms": r["response_time_ms"],
            "healthy": r["healthy"],
            "tags": ",".join(r.get("tags", [])),
            "note": r.get("note", ""),
        })
    return output.getvalue()


# ─── CLI Entry Point ─────────────────────────────────────────────────


def main() -> int:
    parser = argparse.ArgumentParser(
        description="LeoCRM API Health Check — probes all GET endpoints and reports status."
    )
    parser.add_argument(
        "--base-url",
        default=os.environ.get("LEOCRM_BASE_URL", "http://localhost:8000"),
        help="Base URL of the LeoCRM API (default: %(default)s)",
    )
    parser.add_argument(
        "--token",
        default=os.environ.get("LEOCRM_AUTH_TOKEN", ""),
        help="Auth token (Bearer token) for authenticated requests",
    )
    parser.add_argument(
        "--output",
        choices=["json", "csv"],
        default="json",
        help="Output format (default: json)",
    )
    parser.add_argument(
        "--report-file",
        default="-",
        help="Output file path (default: stdout)",
    )
    parser.add_argument(
        "--enumerate-only",
        action="store_true",
        help="Only enumerate routes without probing (list all endpoints)",
    )
    args = parser.parse_args()

    # Step 1: Enumerate routes
    print(f"Enumerating routes from FastAPI app...", file=sys.stderr)
    routes = enumerate_routes()
    print(f"Found {len(routes)} API route(s)", file=sys.stderr)

    if args.enumerate_only:
        for route in routes:
            methods = ", ".join(route["methods"])
            tags = ", ".join(route.get("tags", []))
            print(f"  [{methods}] {route['path']}  tags=[{tags}]")
        return 0

    if not args.token:
        print("ERROR: --token is required for probing (or set LEOCRM_AUTH_TOKEN env)", file=sys.stderr)
        return 2

    # Step 2: Probe endpoints
    import asyncio

    print(f"Probing GET endpoints against {args.base_url}...", file=sys.stderr)
    results = asyncio.run(probe_all_endpoints(args.base_url, args.token, routes))

    # Step 3: Format and output report
    if args.output == "json":
        report = format_json_report(results)
    else:
        report = format_csv_report(results)

    if args.report_file == "-":
        print(report)
    else:
        with open(args.report_file, "w", encoding="utf-8") as f:
            f.write(report)
        print(f"Report written to {args.report_file}", file=sys.stderr)

    # Step 4: Determine exit code
    failed = sum(1 for r in results if not r["healthy"])
    if failed > 0:
        print(f"\n{failed} endpoint(s) failed!", file=sys.stderr)
        return 1

    print(f"\nAll endpoints healthy.", file=sys.stderr)
    return 0


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