fix(api): advertise 429 on every operation in OpenAPI

SlowAPI middleware can short-circuit any request with 429 once a
per-route or per-IP rate limit fires (e.g. POST /api/v1/auth/login is
capped at 10/5min). The OpenAPI spec did not declare 429 on any
operation, so schemathesis flagged legitimate rate-limit responses as
RejectedPositiveData / status-code-nonconformance failures.

Override app.openapi to inject a generic 429 response object on every
HTTP operation in the generated schema. Add a contract test that fails
if any operation drops the 429 advertisement.
This commit is contained in:
2026-04-28 00:58:37 -04:00
parent 6b407e8c9c
commit 5d883466a2
2 changed files with 68 additions and 0 deletions

View File

@@ -258,6 +258,50 @@ if DECNET_PROFILE_REQUESTS:
app.include_router(api_router, prefix="/api/v1") app.include_router(api_router, prefix="/api/v1")
def _custom_openapi() -> dict:
"""Inject the global 429 response into every operation.
SlowAPI middleware can short-circuit any request with 429 once the
per-route or per-IP rate limit fires, so the OpenAPI spec must
advertise 429 on every operation — otherwise schemathesis flags
legitimate rate-limit responses as schema-noncompliant.
"""
from fastapi.openapi.utils import get_openapi
if app.openapi_schema:
return app.openapi_schema
schema = get_openapi(
title=app.title,
version=app.version,
routes=app.routes,
description=app.description,
)
too_many = {
"description": "Rate limit exceeded",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {"error": {"type": "string"}},
"required": ["error"],
}
}
},
}
for path_item in schema.get("paths", {}).values():
for method, op in path_item.items():
if method.lower() not in {
"get", "post", "put", "patch", "delete", "options", "head",
}:
continue
op.setdefault("responses", {}).setdefault("429", too_many)
app.openapi_schema = schema
return schema
app.openapi = _custom_openapi # type: ignore[method-assign]
@app.exception_handler(RequestValidationError) @app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError) -> ORJSONResponse: async def validation_exception_handler(request: Request, exc: RequestValidationError) -> ORJSONResponse:
""" """

View File

@@ -0,0 +1,24 @@
"""OpenAPI must advertise 429 on every operation.
SlowAPI can return 429 from any rate-limited route at any time. If the
schema doesn't list it, schemathesis (and any other contract-driven
client) treats a legitimate rate-limit response as a contract violation.
"""
from decnet.web.api import app
def test_every_operation_documents_429() -> None:
schema = app.openapi()
paths = schema.get("paths", {})
assert paths, "OpenAPI schema is empty — router not mounted"
http_methods = {"get", "post", "put", "patch", "delete", "options", "head"}
missing: list[str] = []
for path, item in paths.items():
for method, op in item.items():
if method.lower() not in http_methods:
continue
if "429" not in op.get("responses", {}):
missing.append(f"{method.upper()} {path}")
assert not missing, f"Operations missing 429 response: {missing[:5]}"