137 lines
5.1 KiB
Python
137 lines
5.1 KiB
Python
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/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()
|