geointel/scripts/audit_api_contracts.py

from __future__ import annotations

import re
import sys
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
BACKEND = ROOT / "backend"
DOCS = ROOT / "docs" / "API_CONTRACTS.md"  # docs/API_CONTRACTS.md

ALLOWED_NON_ENVELOPE_ENDPOINTS = {
    ("GET", "/health"),
    ("GET", "/health/live"),
    ("GET", "/health/ready"),
    ("GET", "/api/v1/auth/authentik/start"),
    ("GET", "/api/v1/auth/authentik/callback"),
    ("GET", "/api/v1/exports/{export_id}/download"),
    ("GET", "/api/v1/projects/{project_id}/datasets/{dataset_id}/raster/image"),
    ("GET", "/api/v1/projects/{project_id}/datasets/{dataset_id}/raster/terrain/image"),
    ("GET", "/api/v1/projects/{project_id}/datasets/{dataset_id}/raster/bathymetry/image"),
    ("GET", "/api/v1/projects/{project_id}/datasets/{dataset_id}/raster/flood-hazard/image"),
    ("GET", "/api/v1/projects/{project_id}/datasets/{dataset_id}/raster/thematic/image"),
    ("GET", "/api/v1/projects/{project_id}/datasets/{dataset_id}/raster/walous/image"),
}

IGNORED_OPENAPI_PATHS = {
    "/openapi.json",
    "/docs",
    "/docs/oauth2-redirect",
    "/redoc",
}


def _load_openapi() -> dict:
    sys.path.insert(0, str(BACKEND))
    from app.main import create_app

    return create_app().openapi()


def _load_app_routes(schema: dict) -> set[tuple[str, str]]:
    routes: set[tuple[str, str]] = set()
    for path, path_item in schema.get("paths", {}).items():
        if path in IGNORED_OPENAPI_PATHS or not isinstance(path_item, dict):
            continue
        for method in path_item:
            normalized_method = method.upper()
            if normalized_method in {"GET", "POST", "PATCH", "DELETE"}:
                routes.add((normalized_method, path))
    return routes


def _resolve_schema(openapi: dict, schema: dict) -> dict:
    reference = schema.get("$ref")
    if not isinstance(reference, str):
        return schema
    prefix = "#/components/schemas/"
    if not reference.startswith(prefix):
        return {}
    return openapi.get("components", {}).get("schemas", {}).get(reference[len(prefix):], {})


def _validate_response_contracts(openapi: dict) -> list[str]:
    errors: list[str] = []
    for path, path_item in openapi.get("paths", {}).items():
        if path in IGNORED_OPENAPI_PATHS or not isinstance(path_item, dict):
            continue
        for method, operation in path_item.items():
            normalized_method = method.upper()
            if normalized_method not in {"GET", "POST", "PATCH", "DELETE"}:
                continue
            if (normalized_method, path) in ALLOWED_NON_ENVELOPE_ENDPOINTS:
                continue
            responses = operation.get("responses", {})
            success_responses = [
                response
                for status_code, response in responses.items()
                if str(status_code).startswith("2")
            ]
            if not success_responses:
                errors.append(f"Missing successful OpenAPI response: {normalized_method} {path}")
                continue
            for response in success_responses:
                schema = (
                    response.get("content", {})
                    .get("application/json", {})
                    .get("schema", {})
                )
                if not schema:
                    errors.append(f"Missing concrete JSON response schema: {normalized_method} {path}")
                    continue
                resolved = _resolve_schema(openapi, schema)
                if resolved.get("additionalProperties") is True:
                    errors.append(f"Untyped dictionary response schema: {normalized_method} {path}")
                    continue
                if path not in {"/health", "/health/live", "/health/ready"}:
                    properties = resolved.get("properties", {})
                    if "data" not in properties:
                        errors.append(f"Missing canonical data envelope: {normalized_method} {path}")
    return errors


def _load_documented_routes() -> set[tuple[str, str]]:
    docs = DOCS.read_text(encoding="utf-8")
    return set(re.findall(r"###\s+(GET|POST|PATCH|DELETE)\s+`([^`]+)`", docs))


def main() -> None:
    openapi = _load_openapi()
    implemented_routes = _load_app_routes(openapi)
    documented_routes = _load_documented_routes()

    missing_docs = sorted(implemented_routes - documented_routes)
    stale_docs = sorted(documented_routes - implemented_routes)
    missing_non_envelope = sorted(ALLOWED_NON_ENVELOPE_ENDPOINTS - implemented_routes)

    errors: list[str] = []
    for method, path in missing_docs:
        errors.append(f"Missing documented API route: {method} {path}")
    for method, path in stale_docs:
        errors.append(f"Documented API route is not implemented: {method} {path}")
    for method, path in missing_non_envelope:
        errors.append(f"Allowed non-envelope endpoint is not implemented: {method} {path}")
    errors.extend(_validate_response_contracts(openapi))

    if errors:
        raise SystemExit("\n".join(errors))

    print(
        "API contract audit OK: "
        f"{len(implemented_routes)} implemented routes match docs; "
        f"{len(ALLOWED_NON_ENVELOPE_ENDPOINTS)} explicit non-envelope endpoints tracked."
    )


if __name__ == "__main__":
    main()