Harden RC7 API response contracts

scripts/audit_api_contracts.py

@@ -15,6 +15,9 @@ ALLOWED_NON_ENVELOPE_ENDPOINTS = {
     ("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/flood-hazard/image"),
+    ("GET", "/api/v1/projects/{project_id}/datasets/{dataset_id}/raster/thematic/image"),
 }
 
 IGNORED_OPENAPI_PATHS = {
@@ -25,11 +28,14 @@ IGNORED_OPENAPI_PATHS = {
 }
 
 
-def _load_app_routes() -> set[tuple[str, str]]:
+def _load_openapi() -> dict:
     sys.path.insert(0, str(BACKEND))
     from app.main import create_app
 
-    schema = create_app().openapi()
+    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):
@@ -41,13 +47,64 @@ def _load_app_routes() -> set[tuple[str, str]]:
     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:
-    implemented_routes = _load_app_routes()
+    openapi = _load_openapi()
+    implemented_routes = _load_app_routes(openapi)
     documented_routes = _load_documented_routes()
 
     missing_docs = sorted(implemented_routes - documented_routes)
@@ -61,6 +118,7 @@ def main() -> None:
         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))