Phase 5.12: API health check script for KI

scripts/ai_health_check.py

@@ -0,0 +1,276 @@
+#!/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())

tests/test_ai_health_check.py

@@ -0,0 +1,109 @@
+"""Tests for the AI health check script (scripts/ai_health_check.py)."""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+# Add scripts dir to path
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))
+
+
+def test_enumerate_routes_returns_api_routes():
+    """enumerate_routes should return only /api/ routes with methods."""
+    from ai_health_check import enumerate_routes
+
+    routes = enumerate_routes()
+    assert isinstance(routes, list)
+    assert len(routes) > 0
+    for route in routes:
+        assert route["path"].startswith("/api/")
+        assert isinstance(route["methods"], list)
+        assert len(route["methods"]) > 0
+
+
+def test_enumerate_routes_excludes_non_api():
+    """enumerate_routes should not include non-API paths like /docs or /openapi.json."""
+    from ai_health_check import enumerate_routes
+
+    routes = enumerate_routes()
+    paths = [r["path"] for r in routes]
+    assert not any(p.startswith("/docs") for p in paths)
+    assert not any(p.startswith("/openapi") for p in paths)
+
+
+def test_format_json_report_structure():
+    """JSON report should have summary and endpoints keys."""
+    from ai_health_check import format_json_report
+
+    results = [
+        {"endpoint": "/api/v1/health", "method": "GET", "status": 200, "response_time_ms": 5.0, "healthy": True, "tags": ["health"], "note": ""},
+        {"endpoint": "/api/v1/users", "method": "POST", "status": "skipped", "response_time_ms": 0, "healthy": True, "tags": ["users"], "note": "Non-GET method — documented only"},
+    ]
+    report = json.loads(format_json_report(results))
+    assert "summary" in report
+    assert "endpoints" in report
+    assert report["summary"]["total_endpoints"] == 2
+    assert report["summary"]["healthy"] == 1
+    assert report["summary"]["skipped"] == 1
+    assert report["summary"]["all_healthy"] is True
+
+
+def test_format_json_report_with_failures():
+    """JSON report should mark all_healthy=False when failures exist."""
+    from ai_health_check import format_json_report
+
+    results = [
+        {"endpoint": "/api/v1/health", "method": "GET", "status": 200, "response_time_ms": 5.0, "healthy": True, "tags": [], "note": ""},
+        {"endpoint": "/api/v1/broken", "method": "GET", "status": 500, "response_time_ms": 10.0, "healthy": False, "tags": [], "note": "HTTP 500"},
+    ]
+    report = json.loads(format_json_report(results))
+    assert report["summary"]["failed"] == 1
+    assert report["summary"]["all_healthy"] is False
+
+
+def test_format_csv_report_has_header():
+    """CSV report should have a header row with expected columns."""
+    from ai_health_check import format_csv_report
+
+    results = [
+        {"endpoint": "/api/v1/health", "method": "GET", "status": 200, "response_time_ms": 5.0, "healthy": True, "tags": ["health"], "note": ""},
+    ]
+    csv_output = format_csv_report(results)
+    lines = csv_output.strip().split("\n")
+    assert "endpoint" in lines[0]
+    assert "method" in lines[0]
+    assert "status" in lines[0]
+    assert "/api/v1/health" in lines[1]
+
+
+@pytest.mark.asyncio
+async def test_probe_all_endpoints_skips_non_get():
+    """probe_all_endpoints should skip POST/PUT/DELETE methods."""
+    from ai_health_check import probe_all_endpoints
+
+    routes = [
+        {"path": "/api/v1/users", "methods": ["POST"], "tags": ["users"]},
+    ]
+    results = await probe_all_endpoints("http://localhost:8000", "fake-token", routes)
+    assert len(results) == 1
+    assert results[0]["status"] == "skipped"
+    assert results[0]["healthy"] is True
+
+
+@pytest.mark.asyncio
+async def test_probe_all_endpoints_handles_connection_error():
+    """probe_all_endpoints should handle connection errors gracefully."""
+    from ai_health_check import probe_all_endpoints
+
+    routes = [
+        {"path": "/api/v1/health", "methods": ["GET"], "tags": ["health"]},
+    ]
+    results = await probe_all_endpoints("http://nonexistent-host:9999", "fake-token", routes)
+    assert len(results) == 1
+    assert results[0]["healthy"] is False
+    assert results[0]["status"] in ("connection_error", "error")