From d9c9ba6630f5dba56b63eaafc5ea20550949c9f5 Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Thu, 23 Jul 2026 22:37:25 +0200 Subject: [PATCH] Phase 5.12: API health check script for KI --- scripts/ai_health_check.py | 276 ++++++++++++++++++++++++++++++++++ tests/test_ai_health_check.py | 109 ++++++++++++++ 2 files changed, 385 insertions(+) create mode 100755 scripts/ai_health_check.py create mode 100644 tests/test_ai_health_check.py diff --git a/scripts/ai_health_check.py b/scripts/ai_health_check.py new file mode 100755 index 0000000..d8d4e3c --- /dev/null +++ b/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 + python scripts/ai_health_check.py --base-url http://localhost:8000 --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()) diff --git a/tests/test_ai_health_check.py b/tests/test_ai_health_check.py new file mode 100644 index 0000000..77e9f96 --- /dev/null +++ b/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")