diff --git a/README.md b/README.md index ad80697..12850d9 100644 --- a/README.md +++ b/README.md @@ -730,26 +730,31 @@ Parameter descriptions: - `health_checks_path` - Path for health check handler. - `health_checks_include_in_schema` - Must be True to include `health_checks_path` (`/health/`) in OpenAPI schema. -### Generated OpenAPI descriptions +### API metadata -Set `openapi_descriptions_enabled=True` in `FastApiSettings`, `LitestarSettings`, -or `SwaggerConfig` to describe framework-generated validation errors and JSON -values automatically. Litestar also fills descriptions for microbootstrap health -fields. The setting is disabled by default and supports the normal environment -prefix, for example `OPENAPI_DESCRIPTIONS_ENABLED=true` without a prefix. +Set `api_metadata_enabled=True` in `FastApiSettings` or `LitestarSettings` to enable +both generated OpenAPI descriptions and HTTP error titles. It defaults to `False` +and can also be enabled with `API_METADATA_ENABLED=true` (with the configured +`ENVIRONMENT_PREFIX`, if any). No separate instrument configuration is required. ```python from microbootstrap import FastApiSettings from microbootstrap.bootstrappers.fastapi import FastApiBootstrapper -app = FastApiBootstrapper(FastApiSettings(openapi_descriptions_enabled=True)).bootstrap() -# Register application routes normally; FastAPI enriches its schema when generated. +app = FastApiBootstrapper(FastApiSettings(api_metadata_enabled=True)).bootstrap() ``` -The bootstrappers preserve existing descriptions and runtime response shapes. -Application-specific descriptions and authentication declarations remain with the -application, using native framework metadata and the existing `security_schemes` -setting. No authentication policy is inferred or enforced by this option. +Default HTTP errors receive the HTTP status phrase as `title`, alongside the +framework's existing fields, status and headers. Generated validation schemas +and examples reflect that response. Application, router and route exception +handlers retain ownership of their payloads and schemas. Debug tracebacks retain +the framework's native behavior. Error titles also work with Swagger disabled. +This setting does not document authentication, change successful responses or +certify compliance with an API standard. + +`api_metadata_enabled` replaces the description-only `openapi_descriptions_enabled` +option from 1.8.0. Migrate existing settings and environment variables to the new +name; enabling it also changes default HTTP error bodies as described above. ## Configuration diff --git a/microbootstrap/bootstrappers/fastapi.py b/microbootstrap/bootstrappers/fastapi.py index fd54101..8bf5f52 100644 --- a/microbootstrap/bootstrappers/fastapi.py +++ b/microbootstrap/bootstrappers/fastapi.py @@ -1,14 +1,20 @@ import contextlib +import http import typing import fastapi +from fastapi.encoders import jsonable_encoder +from fastapi.exceptions import RequestValidationError from fastapi.middleware.cors import CORSMiddleware from fastapi_offline_docs import enable_offline_docs from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from prometheus_fastapi_instrumentator import Instrumentator, metrics +from starlette.exceptions import HTTPException +from starlette.responses import JSONResponse from microbootstrap.bootstrappers.base import ApplicationBootstrapper from microbootstrap.config.fastapi import FastApiConfig +from microbootstrap.helpers import http_error_title from microbootstrap.instruments.cors_instrument import CorsInstrument from microbootstrap.instruments.health_checks_instrument import HealthChecksInstrument, HealthCheckTypedDict from microbootstrap.instruments.logging_instrument import LoggingInstrument @@ -62,6 +68,87 @@ def bootstrap_before(self) -> dict[str, typing.Any]: "lifespan": self._wrapped_lifespan_manager if self.application_config.lifespan else self._lifespan_manager, } + def bootstrap_before_instruments_after_app_created(self, application: ApplicationT) -> ApplicationT: + if self.settings.api_metadata_enabled: + for exception, native, replacement in ( + (HTTPException, fastapi.exception_handlers.http_exception_handler, self.http_exception_response), + ( + RequestValidationError, + fastapi.exception_handlers.request_validation_exception_handler, + self.validation_error_response, + ), + ): + if application.exception_handlers.get(exception) == native: + application.exception_handlers[exception] = replacement + if http.HTTPStatus.INTERNAL_SERVER_ERROR not in application.exception_handlers: + application.exception_handlers.setdefault(Exception, self.internal_error_response) + return application + + async def http_exception_response(self, request: fastapi.Request, exc: HTTPException) -> fastapi.Response: + response = await fastapi.exception_handlers.http_exception_handler(request, exc) + if exc.status_code >= http.HTTPStatus.BAD_REQUEST: + return JSONResponse( + {"detail": exc.detail, "title": http_error_title(exc.status_code)}, + status_code=exc.status_code, + headers=exc.headers, + ) + return response + + async def validation_error_response(self, _request: fastapi.Request, exc: RequestValidationError) -> JSONResponse: + return JSONResponse( + { + "detail": jsonable_encoder(exc.errors()), + "title": http_error_title(http.HTTPStatus.UNPROCESSABLE_ENTITY), + }, + status_code=http.HTTPStatus.UNPROCESSABLE_ENTITY, + ) + + async def internal_error_response(self, _request: fastapi.Request, _exc: Exception) -> JSONResponse: + return JSONResponse( + {"detail": "Internal Server Error", "title": http_error_title(http.HTTPStatus.INTERNAL_SERVER_ERROR)}, + status_code=http.HTTPStatus.INTERNAL_SERVER_ERROR, + ) + + def bootstrap_after(self, application: ApplicationT) -> ApplicationT: + if not self.settings.api_metadata_enabled: + return application + original_openapi: typing.Final = application.openapi + + def titled_openapi() -> dict[str, typing.Any]: + document: typing.Final = original_openapi() + self._describe_generated_schemas(document) + schema = document.get("components", {}).get("schemas", {}).get("HTTPValidationError") + if ( + schema is not None + and application.exception_handlers.get(RequestValidationError) == self.validation_error_response + ): + schema.setdefault("properties", {}).setdefault( + "title", {"type": "string", "description": "Short, human-readable summary of the HTTP error."} + ) + required = schema.setdefault("required", []) + if "title" not in required: + required.append("title") + return document + + application.openapi = titled_openapi # type: ignore[method-assign] # FastAPI's public custom OpenAPI hook. + return application + + @staticmethod + def _describe_generated_schemas(document: dict[str, typing.Any]) -> None: + schemas = document.get("components", {}).get("schemas", {}) + for name in ("HTTPValidationError", "ValidationError"): + if name not in schemas: + continue + definition = schemas[name] + definition.setdefault("description", "Request validation failure reported by FastAPI.") + for field_name, description in _VALIDATION_DESCRIPTIONS.items(): + if field_name in definition.get("properties", {}): + definition["properties"][field_name].setdefault("description", description) + if "JsonValue" in schemas: + schemas["JsonValue"].setdefault( + "description", "JSON value: an object, array, string, number, boolean or null." + ) + FastApiBootstrapper.use_instrument()(SentryInstrument) @@ -79,19 +166,13 @@ def bootstrap_before(self) -> dict[str, typing.Any]: def bootstrap_after(self, application: ApplicationT) -> ApplicationT: if self.instrument_config.swagger_offline_docs: enable_offline_docs(application, static_files_handler=self.instrument_config.service_static_path) - if ( - self.instrument_config.openapi_version_docs is None - and not self.instrument_config.security_schemes - and not self.instrument_config.openapi_descriptions_enabled - ): + if self.instrument_config.openapi_version_docs is None and not self.instrument_config.security_schemes: return application original_openapi: typing.Final = application.openapi def documented_openapi() -> dict[str, typing.Any]: openapi_schema: typing.Final = original_openapi() - if self.instrument_config.openapi_descriptions_enabled: - self._describe_generated_schemas(openapi_schema) if self.instrument_config.security_schemes: self._merge_security_schemes(openapi_schema) if self.instrument_config.openapi_version_docs is not None: @@ -101,22 +182,6 @@ def documented_openapi() -> dict[str, typing.Any]: application.openapi = documented_openapi # type: ignore[method-assign] # FastAPI's public custom OpenAPI hook. return application - @staticmethod - def _describe_generated_schemas(document: dict[str, typing.Any]) -> None: - schemas = document.get("components", {}).get("schemas", {}) - for name in ("HTTPValidationError", "ValidationError"): - if name not in schemas: - continue - definition = schemas[name] - definition.setdefault("description", "Request validation failure reported by FastAPI.") - for field_name, description in _VALIDATION_DESCRIPTIONS.items(): - if field_name in definition.get("properties", {}): - definition["properties"][field_name].setdefault("description", description) - if "JsonValue" in schemas: - schemas["JsonValue"].setdefault( - "description", "JSON value: an object, array, string, number, boolean or null." - ) - def _merge_security_schemes(self, openapi_schema: dict[str, typing.Any]) -> None: expected_schemes: typing.Final = serialize_security_schemes(self.instrument_config.security_schemes) components = openapi_schema.setdefault("components", {}) diff --git a/microbootstrap/bootstrappers/litestar.py b/microbootstrap/bootstrappers/litestar.py index df77906..67e9d16 100644 --- a/microbootstrap/bootstrappers/litestar.py +++ b/microbootstrap/bootstrappers/litestar.py @@ -1,5 +1,7 @@ from __future__ import annotations import dataclasses +import http +import json import typing import litestar @@ -11,8 +13,11 @@ OpenTelemetryConfig as LitestarOpentelemetryConfig, ) from litestar.contrib.prometheus import PrometheusConfig, PrometheusController +from litestar.exceptions import HTTPException +from litestar.exceptions.responses import create_debug_response, create_exception_response from litestar.middleware import ASGIMiddleware from litestar.openapi.plugins import SwaggerRenderPlugin +from litestar.routes import HTTPRoute from litestar.types.asgi_types import ASGIApp, Scope from litestar_offline_docs import generate_static_files_config from opentelemetry.instrumentation.asgi import OpenTelemetryMiddleware @@ -21,6 +26,7 @@ from microbootstrap.bootstrappers.base import ApplicationBootstrapper from microbootstrap.config.litestar import LitestarConfig +from microbootstrap.helpers import http_error_title from microbootstrap.instruments import opentelemetry_instrument from microbootstrap.instruments.cors_instrument import CorsInstrument from microbootstrap.instruments.health_checks_instrument import ( @@ -71,6 +77,7 @@ class AcceptVersionedOperation(openapi.spec.Operation): if typing.TYPE_CHECKING: + from litestar.config.app import AppConfig from litestar.contrib.opentelemetry import OpenTelemetryConfig from litestar.types import ASGIApp, Scope from litestar.types.asgi_types import Receive, Send @@ -87,8 +94,120 @@ def bootstrap_before(self: typing_extensions.Self) -> dict[str, typing.Any]: "debug": self.settings.service_debug, "on_shutdown": [self.teardown], "on_startup": [self.console_writer.print_bootstrap_table], + "on_app_init": [self._configure_error_titles] if self.settings.api_metadata_enabled else [], } + def exception_response( + self, request: litestar.Request[typing.Any, typing.Any, typing.Any], exc: Exception + ) -> litestar.Response[typing.Any]: + status_code = exc.status_code if isinstance(exc, HTTPException) else http.HTTPStatus.INTERNAL_SERVER_ERROR + if status_code == http.HTTPStatus.INTERNAL_SERVER_ERROR and request.app.debug: + return create_debug_response(request, exc) + response = create_exception_response(request, exc) + if status_code >= http.HTTPStatus.BAD_REQUEST: + if not isinstance(response.content, dict): + response.content = json.loads(response.content) + response.content.setdefault("title", http_error_title(status_code)) + response.media_type = litestar.MediaType.JSON + return response + + def _configure_error_titles(self, config: AppConfig) -> AppConfig: + config.exception_handlers.setdefault(Exception, self.exception_response) + return config + + @classmethod + def _title_error_schema(cls, schema: openapi.spec.Schema) -> None: + for branch in (*(schema.one_of or []), *(schema.any_of or []), *(schema.all_of or [])): + if isinstance(branch, openapi.spec.Schema): + cls._title_error_schema(branch) + properties = schema.properties or {} + if "status_code" in properties and "detail" in properties: + properties.setdefault( + "title", + openapi.spec.Schema( + type=openapi.spec.OpenAPIType.STRING, + description="Short, human-readable summary of the HTTP error.", + ), + ) + schema.required = list(dict.fromkeys([*(schema.required or []), "title"])) + for example in schema.examples or []: + if isinstance(example, dict) and isinstance(example.get("status_code"), int): + example.setdefault("title", http_error_title(example["status_code"])) + + @classmethod + def _title_operation(cls, operation: openapi.spec.Operation) -> None: + for code, response in (operation.responses or {}).items(): + if str(code).startswith(("4", "5")) and isinstance(response, openapi.spec.OpenAPIResponse): + for media_type in (response.content or {}).values(): + if isinstance(media_type.schema, openapi.spec.Schema): + cls._title_error_schema(media_type.schema) + + def bootstrap_after(self, application: ApplicationT) -> ApplicationT: + if not self.settings.api_metadata_enabled or application.openapi_config is None: + return application + self._describe_generated_schemas(application.openapi_schema) + for route in application.routes: + if not isinstance(route, HTTPRoute): + continue + path_item = (application.openapi_schema.paths or {}).get(route.path_format) + if path_item is None: + continue + for method, (handler, _) in route.route_handler_map.items(): + # A custom handler can own any error schema on this route. + if any( + value != self.exception_response + and (isinstance(key, int) or issubclass(HTTPException, key) or issubclass(key, HTTPException)) + for key, value in handler.resolve_exception_handlers().items() + ): + continue + operation = getattr(path_item, method.lower()) + if operation is None: + continue + self._title_operation(operation) + return application + + @staticmethod + def _describe_properties(schema: openapi.spec.Schema, descriptions: typing.Mapping[str, str]) -> None: + for name, description in descriptions.items(): + field = (schema.properties or {}).get(name) + if field is not None and not field.description: + field.description = description + + @classmethod + def _describe_error_schema(cls, schema: openapi.spec.Schema) -> None: + for branch in (*(schema.one_of or []), *(schema.any_of or []), *(schema.all_of or [])): + if isinstance(branch, openapi.spec.Schema): + cls._describe_error_schema(branch) + if ( + schema.type is None + and schema.one_of + and all( + isinstance(branch, openapi.spec.Schema) and branch.type == openapi.spec.OpenAPIType.OBJECT + for branch in schema.one_of + ) + ): + schema.type = openapi.spec.OpenAPIType.OBJECT + cls._describe_properties(schema, _ERROR_DESCRIPTIONS) + + @classmethod + def _describe_generated_schemas(cls, document: openapi.spec.OpenAPI) -> None: + schemas = document.components.schemas or {} + if "JsonValue" in schemas and not schemas["JsonValue"].description: + schemas["JsonValue"].description = "JSON value: an object, array, string, number, boolean or null." + if "HealthCheckTypedDict" in schemas: + cls._describe_properties(schemas["HealthCheckTypedDict"], _HEALTH_PROPERTIES) + for path_item in (document.paths or {}).values(): + for method in SUPPORTED_HTTP_METHODS: + operation = getattr(path_item, method) + if operation is None: + continue + for code, response in (operation.responses or {}).items(): + if not str(code).startswith(("4", "5")) or not isinstance(response, openapi.spec.OpenAPIResponse): + continue + for media_type in (response.content or {}).values(): + if isinstance(media_type.schema, openapi.spec.Schema): + cls._describe_error_schema(media_type.schema) + @LitestarBootstrapper.use_instrument() class LitestarSentryInstrument(SentryInstrument): @@ -136,62 +255,16 @@ def bootstrap_before(self) -> dict[str, typing.Any]: return bootstrap_result def bootstrap_after(self, application: ApplicationT) -> ApplicationT: - if ( - self.instrument_config.openapi_version_docs is None - and not self.instrument_config.security_schemes - and not self.instrument_config.openapi_descriptions_enabled - ) or (application.openapi_config is None): + if (self.instrument_config.openapi_version_docs is None and not self.instrument_config.security_schemes) or ( + application.openapi_config is None + ): return application - if self.instrument_config.openapi_descriptions_enabled: - self._describe_generated_schemas(application.openapi_schema) if self.instrument_config.security_schemes: self._merge_security_schemes(application.openapi_schema) if self.instrument_config.openapi_version_docs is not None: self._document_operations(application.openapi_schema) return application - @staticmethod - def _describe_properties(schema: openapi.spec.Schema, descriptions: typing.Mapping[str, str]) -> None: - for name, description in descriptions.items(): - field = (schema.properties or {}).get(name) - if field is not None and not field.description: - field.description = description - - @classmethod - def _describe_error_schema(cls, schema: openapi.spec.Schema) -> None: - for branch in (*(schema.one_of or []), *(schema.any_of or []), *(schema.all_of or [])): - if isinstance(branch, openapi.spec.Schema): - cls._describe_error_schema(branch) - if ( - schema.type is None - and schema.one_of - and all( - isinstance(branch, openapi.spec.Schema) and branch.type == openapi.spec.OpenAPIType.OBJECT - for branch in schema.one_of - ) - ): - schema.type = openapi.spec.OpenAPIType.OBJECT - cls._describe_properties(schema, _ERROR_DESCRIPTIONS) - - @classmethod - def _describe_generated_schemas(cls, document: openapi.spec.OpenAPI) -> None: - schemas = document.components.schemas or {} - if "JsonValue" in schemas and not schemas["JsonValue"].description: - schemas["JsonValue"].description = "JSON value: an object, array, string, number, boolean or null." - if "HealthCheckTypedDict" in schemas: - cls._describe_properties(schemas["HealthCheckTypedDict"], _HEALTH_PROPERTIES) - for path_item in (document.paths or {}).values(): - for method in SUPPORTED_HTTP_METHODS: - operation = getattr(path_item, method) - if operation is None: - continue - for code, response in (operation.responses or {}).items(): - if not str(code).startswith(("4", "5")) or not isinstance(response, openapi.spec.OpenAPIResponse): - continue - for media_type in (response.content or {}).values(): - if isinstance(media_type.schema, openapi.spec.Schema): - cls._describe_error_schema(media_type.schema) - def _merge_security_schemes(self, openapi_schema: openapi.spec.OpenAPI) -> None: expected_schemes: typing.Final = serialize_security_schemes(self.instrument_config.security_schemes) security_schemes = openapi_schema.components.security_schemes diff --git a/microbootstrap/helpers.py b/microbootstrap/helpers.py index 1164b25..aa334ac 100644 --- a/microbootstrap/helpers.py +++ b/microbootstrap/helpers.py @@ -1,4 +1,5 @@ import dataclasses +import http import re import typing @@ -108,3 +109,10 @@ def optimize_exclude_paths( endpoints_to_ignore = set(endpoints_to_ignore) return endpoints_to_ignore + + +def http_error_title(status_code: int) -> str: + try: + return http.HTTPStatus(status_code).phrase + except ValueError: + return "HTTP error" diff --git a/microbootstrap/instruments/swagger_instrument.py b/microbootstrap/instruments/swagger_instrument.py index c212b76..00efbda 100644 --- a/microbootstrap/instruments/swagger_instrument.py +++ b/microbootstrap/instruments/swagger_instrument.py @@ -27,7 +27,6 @@ class SwaggerConfig(BaseInstrumentConfig): swagger_extra_params: dict[str, typing.Any] = pydantic.Field(default_factory=dict) security_schemes: typing.Mapping[str, _OpenApiSecurityScheme] = pydantic.Field(default_factory=dict) openapi_version_docs: OpenApiVersionDocsConfig | None = None - openapi_descriptions_enabled: bool = False @pydantic.field_validator("security_schemes") @classmethod diff --git a/microbootstrap/settings.py b/microbootstrap/settings.py index f7f2400..a7cd675 100644 --- a/microbootstrap/settings.py +++ b/microbootstrap/settings.py @@ -72,6 +72,8 @@ class LitestarSettings( # type: ignore[misc] ): """Settings for a litestar botstrap.""" + api_metadata_enabled: bool = False + class FastApiSettings( # type: ignore[misc] BaseServiceSettings, @@ -87,6 +89,8 @@ class FastApiSettings( # type: ignore[misc] ): """Settings for a fastapi botstrap.""" + api_metadata_enabled: bool = False + class FastStreamSettings( # type: ignore[misc] BaseServiceSettings, diff --git a/tests/bootstrappers/test_http_error_titles.py b/tests/bootstrappers/test_http_error_titles.py new file mode 100644 index 0000000..ae16fa0 --- /dev/null +++ b/tests/bootstrappers/test_http_error_titles.py @@ -0,0 +1,327 @@ +import http +import typing + +import fastapi +import litestar +import pytest +from fastapi.testclient import TestClient as FastAPITestClient +from litestar.config.app import AppConfig +from litestar.exceptions import HTTPException as LitestarHTTPException +from litestar.testing import TestClient as LitestarTestClient + +from microbootstrap import FastApiSettings, LitestarSettings +from microbootstrap.bootstrappers.fastapi import FastApiBootstrapper +from microbootstrap.bootstrappers.litestar import LitestarBootstrapper +from microbootstrap.config.fastapi import FastApiConfig +from microbootstrap.config.litestar import LitestarConfig + + +if typing.TYPE_CHECKING: + from litestar.types import ExceptionHandlersMap + + +NONSTANDARD_STATUS: typing.Final = 599 + + +@pytest.mark.parametrize("enabled", [False, True]) +@pytest.mark.parametrize("status_code", [http.HTTPStatus.BAD_REQUEST, http.HTTPStatus.UNAUTHORIZED, NONSTANDARD_STATUS]) +def test_fastapi_errors_preserve_fields_headers_and_status(enabled: bool, status_code: int) -> None: + application: typing.Final = FastApiBootstrapper( + FastApiSettings(service_debug=False, api_metadata_enabled=enabled) + ).bootstrap() + + @application.get("/error") + def reject() -> None: + raise fastapi.HTTPException( + status_code=status_code, detail={"reason": "rejected"}, headers={"X-Reason": "test"} + ) + + with FastAPITestClient(application) as client: + response = client.get("/error") + assert response.status_code == status_code + assert response.headers["x-reason"] == "test" + assert response.json()["detail"] == {"reason": "rejected"} + assert ("title" in response.json()) is enabled + if enabled: + assert response.json()["title"] == ( + "HTTP error" if status_code == NONSTANDARD_STATUS else http.HTTPStatus(status_code).phrase + ) + + +@pytest.mark.parametrize("enabled", [False, True]) +def test_fastapi_validation_schema_matches_real_errors(enabled: bool) -> None: + application: typing.Final = FastApiBootstrapper( + FastApiSettings(service_debug=False, api_metadata_enabled=enabled) + ).bootstrap() + + @application.get("/items") + def get_items(limit: int) -> int: + return limit + + schema: typing.Final = application.openapi() + assert application.openapi() == schema + error: typing.Final = schema["components"]["schemas"]["HTTPValidationError"] + assert ("title" in error.get("required", [])) is enabled + assert ("title" in error["properties"]) is enabled + assert ("description" in error["properties"]["detail"]) is enabled + with FastAPITestClient(application) as client: + assert client.get("/items?limit=1").json() == 1 + response = client.get("/items?limit=invalid") + assert response.status_code == http.HTTPStatus.UNPROCESSABLE_ENTITY + assert isinstance(response.json()["detail"], list) + assert ("title" in response.json()) is enabled + + +def test_fastapi_empty_schema_and_non_error_exception_keep_native_behavior() -> None: + application: typing.Final = FastApiBootstrapper( + FastApiSettings(service_debug=False, api_metadata_enabled=True, health_checks_enabled=False) + ).bootstrap() + assert not application.openapi().get("components", {}).get("schemas") + + @application.get("/redirect") + def redirect() -> None: + raise fastapi.HTTPException(status_code=http.HTTPStatus.NOT_MODIFIED, headers={"ETag": "test"}) + + with FastAPITestClient(application) as client: + response = client.get("/redirect") + assert response.status_code == http.HTTPStatus.NOT_MODIFIED + assert response.content == b"" + assert response.headers["etag"] == "test" + + +def test_fastapi_custom_validation_handler_keeps_ownership() -> None: + async def custom_validation(_request: fastapi.Request, _exc: Exception) -> fastapi.responses.JSONResponse: + return fastapi.responses.JSONResponse({"owned": True}, status_code=http.HTTPStatus.BAD_REQUEST) + + application: typing.Final = ( + FastApiBootstrapper(FastApiSettings(service_debug=False, api_metadata_enabled=True)) + .configure_application( + FastApiConfig(exception_handlers={fastapi.exceptions.RequestValidationError: custom_validation}) + ) + .bootstrap() + ) + + @application.get("/items") + def get_items(limit: int) -> int: + return limit + + assert "title" not in application.openapi()["components"]["schemas"]["HTTPValidationError"].get("required", []) + with FastAPITestClient(application) as client: + assert client.get("/items?limit=1").json() == 1 + assert client.get("/items?limit=invalid").json() == {"owned": True} + + +@pytest.mark.parametrize("enabled", [False, True]) +@pytest.mark.parametrize("media_type", [litestar.MediaType.JSON, litestar.MediaType.HTML]) +def test_litestar_errors_preserve_fields_headers_and_status(enabled: bool, media_type: litestar.MediaType) -> None: + @litestar.get("/error", sync_to_thread=False, media_type=media_type) + def reject() -> None: + raise LitestarHTTPException( + status_code=http.HTTPStatus.UNAUTHORIZED, + detail="rejected", + extra={"reason": "test"}, + headers={"X-Reason": "test"}, + ) + + application: typing.Final = ( + LitestarBootstrapper(LitestarSettings(service_debug=False, api_metadata_enabled=enabled)) + .configure_application(LitestarConfig(route_handlers=[reject])) + .bootstrap() + ) + with LitestarTestClient(application) as client: + response = client.get("/error") + assert response.status_code == http.HTTPStatus.UNAUTHORIZED + assert response.headers["x-reason"] == "test" + assert response.json()["detail"] == "rejected" + assert response.json()["extra"] == {"reason": "test"} + assert ("title" in response.json()) is enabled + if enabled: + assert response.headers["content-type"].startswith("application/json") + assert response.json()["title"] == "Unauthorized" + + +@pytest.mark.parametrize("enabled", [False, True]) +def test_litestar_validation_schema_matches_real_errors(enabled: bool) -> None: + @litestar.get("/items", sync_to_thread=False) + def get_items(limit: int) -> int: + return limit + + application: typing.Final = ( + LitestarBootstrapper(LitestarSettings(service_debug=False, api_metadata_enabled=enabled)) + .configure_application(LitestarConfig(route_handlers=[get_items])) + .bootstrap() + ) + schema: typing.Final = application.openapi_schema.to_schema() + error: typing.Final = schema["paths"]["/items"]["get"]["responses"]["400"]["content"]["application/json"]["schema"] + assert ("title" in error["required"]) is enabled + assert ("description" in error["properties"]["detail"]) is enabled + with LitestarTestClient(application) as client: + assert client.get("/items?limit=1").json() == 1 + response = client.get("/items?limit=invalid") + assert response.status_code == http.HTTPStatus.BAD_REQUEST + assert ("title" in response.json()) is enabled + + +@pytest.mark.parametrize("framework", ["fastapi", "litestar"]) +def test_internal_errors_have_safe_titles(framework: str) -> None: + def fail() -> None: + raise RuntimeError("private diagnostic") + + if framework == "fastapi": + application = FastApiBootstrapper(FastApiSettings(service_debug=False, api_metadata_enabled=True)).bootstrap() + application.get("/error")(fail) + with FastAPITestClient(application, raise_server_exceptions=False) as client: + response = client.get("/error") + else: + handler: typing.Final = litestar.get("/error", sync_to_thread=False)(fail) + litestar_application: typing.Final = ( + LitestarBootstrapper(LitestarSettings(service_debug=False, api_metadata_enabled=True)) + .configure_application(LitestarConfig(route_handlers=[handler])) + .bootstrap() + ) + with LitestarTestClient(litestar_application, raise_server_exceptions=False) as litestar_client: + response = litestar_client.get("/error") + assert response.status_code == http.HTTPStatus.INTERNAL_SERVER_ERROR + assert response.json()["title"] == "Internal Server Error" + assert "private diagnostic" not in response.text + + +def test_setting_defaults_and_environment(monkeypatch: pytest.MonkeyPatch) -> None: + assert FastApiSettings().api_metadata_enabled is False + assert LitestarSettings().api_metadata_enabled is False + monkeypatch.setenv("API_METADATA_ENABLED", "true") + assert FastApiSettings().api_metadata_enabled is True + assert LitestarSettings().api_metadata_enabled is True + + +@pytest.mark.parametrize("level", ["application", "controller", "router", "route"]) +@pytest.mark.parametrize("key", [http.HTTPStatus.BAD_REQUEST, litestar.exceptions.ValidationException]) +def test_litestar_custom_validation_handler_owns_payload_and_schema(level: str, key: int | type[Exception]) -> None: + def custom_validation( + _request: litestar.Request[typing.Any, typing.Any, typing.Any], _exc: Exception + ) -> litestar.Response[typing.Any]: + return litestar.Response({"status_code": 400, "detail": "owned", "extra": None}, status_code=400) + + handlers: typing.Final[ExceptionHandlersMap] = {key: custom_validation} + + @litestar.get("/items", sync_to_thread=False, exception_handlers=handlers if level == "route" else None) + def get_items(limit: int) -> int: + return limit + + class ItemsController(litestar.Controller): + path = "/" + exception_handlers = handlers if level == "controller" else None + + @litestar.get("/items", sync_to_thread=False) + def items(self, limit: int) -> int: + return limit + + router: typing.Final = litestar.Router( + path="/", + route_handlers=[ItemsController if level == "controller" else get_items], + exception_handlers=handlers if level == "router" else None, + ) + application: typing.Final = ( + LitestarBootstrapper(LitestarSettings(service_debug=False, api_metadata_enabled=True)) + .configure_application( + LitestarConfig(route_handlers=[router], exception_handlers=handlers if level == "application" else {}) + ) + .bootstrap() + ) + error: typing.Final = application.openapi_schema.to_schema()["paths"]["/items"]["get"]["responses"]["400"][ + "content" + ]["application/json"]["schema"] + assert "title" not in error["required"] + with LitestarTestClient(application) as client: + response = client.get("/items?limit=invalid") + assert client.get("/items?limit=1").json() == 1 + assert response.status_code == http.HTTPStatus.BAD_REQUEST + assert response.json() == {"status_code": 400, "detail": "owned", "extra": None} + + +@pytest.mark.parametrize("enabled", [False, True]) +def test_litestar_debug_traceback_remains_framework_owned(enabled: bool) -> None: + @litestar.get("/error", sync_to_thread=False) + def fail() -> None: + raise RuntimeError("private diagnostic") + + application: typing.Final = ( + LitestarBootstrapper(LitestarSettings(service_debug=True, api_metadata_enabled=enabled)) + .configure_application(LitestarConfig(route_handlers=[fail])) + .bootstrap() + ) + with LitestarTestClient(application, raise_server_exceptions=False) as client: + response = client.get("/error") + assert response.status_code == http.HTTPStatus.INTERNAL_SERVER_ERROR + assert "private diagnostic" in response.text + assert response.headers["content-type"].startswith("text/plain") + + +@pytest.mark.parametrize("swagger_enabled", [False, True]) +def test_litestar_non_http_routes_and_combined_error_schemas(swagger_enabled: bool) -> None: + @litestar.get( + "/items", + sync_to_thread=False, + raises=[litestar.exceptions.ValidationException, litestar.exceptions.ClientException, LitestarHTTPException], + ) + def get_items(limit: int) -> int: + return limit + + @litestar.websocket("/ws") + async def websocket_handler(socket: litestar.WebSocket[typing.Any, typing.Any, typing.Any]) -> None: + await socket.accept() + await socket.send_text("ready") + await socket.close() + + def disable_docs(config: AppConfig) -> AppConfig: + config.openapi_config = None + return config + + application: typing.Final = ( + LitestarBootstrapper( + LitestarSettings(service_debug=False, api_metadata_enabled=True, swagger_enabled=swagger_enabled) + ) + .configure_application( + LitestarConfig( + route_handlers=[get_items, websocket_handler], on_app_init=[] if swagger_enabled else [disable_docs] + ) + ) + .bootstrap() + ) + with LitestarTestClient(application) as client: + with client.websocket_connect("/ws") as websocket: + assert websocket.receive_text() == "ready" + assert client.get("/items?limit=1").json() == 1 + response = client.get("/items?limit=invalid") + assert response.status_code == http.HTTPStatus.BAD_REQUEST + assert response.json()["title"] == "Bad Request" + if swagger_enabled: + schema: typing.Final = application.openapi_schema.to_schema()["paths"]["/items"]["get"]["responses"]["400"][ + "content" + ]["application/json"]["schema"] + assert all("title" in branch["required"] for branch in schema["oneOf"]) + assert "500" in application.openapi_schema.to_schema()["paths"]["/items"]["get"]["responses"] + + +@pytest.mark.parametrize("key", [http.HTTPStatus.INTERNAL_SERVER_ERROR, Exception]) +def test_fastapi_custom_server_error_handler_keeps_ownership(key: int | type[Exception]) -> None: + async def custom_error(_request: fastapi.Request, _exc: Exception) -> fastapi.responses.JSONResponse: + return fastapi.responses.JSONResponse( + {"detail": "owned"}, status_code=http.HTTPStatus.SERVICE_UNAVAILABLE, headers={"X-Owner": "application"} + ) + + application: typing.Final = ( + FastApiBootstrapper(FastApiSettings(service_debug=False, api_metadata_enabled=True)) + .configure_application(FastApiConfig(exception_handlers={key: custom_error})) + .bootstrap() + ) + + @application.get("/error") + def fail() -> None: + raise RuntimeError("private diagnostic") + + with FastAPITestClient(application, raise_server_exceptions=False) as client: + response = client.get("/error") + assert response.status_code == http.HTTPStatus.SERVICE_UNAVAILABLE + assert response.json() == {"detail": "owned"} + assert response.headers["x-owner"] == "application" diff --git a/tests/bootstrappers/test_openapi_descriptions.py b/tests/bootstrappers/test_openapi_descriptions.py index 2e7ac8a..f15f1b1 100644 --- a/tests/bootstrappers/test_openapi_descriptions.py +++ b/tests/bootstrappers/test_openapi_descriptions.py @@ -9,9 +9,9 @@ from litestar.openapi import spec from litestar.testing import TestClient as LitestarTestClient -from microbootstrap import FastApiSettings, LitestarSettings, SwaggerConfig +from microbootstrap import FastApiSettings, LitestarSettings from microbootstrap.bootstrappers.fastapi import FastApiBootstrapper -from microbootstrap.bootstrappers.litestar import LitestarBootstrapper, LitestarSwaggerInstrument +from microbootstrap.bootstrappers.litestar import LitestarBootstrapper from microbootstrap.config.litestar import LitestarConfig from microbootstrap.instruments.health_checks_instrument import HealthCheckTypedDict @@ -22,7 +22,7 @@ class Payload(pydantic.BaseModel): def test_fastapi_descriptions_preserve_validation_and_are_idempotent() -> None: application: typing.Final = FastApiBootstrapper( - FastApiSettings(openapi_descriptions_enabled=True, service_debug=False, health_checks_enabled=False) + FastApiSettings(api_metadata_enabled=True, service_debug=False, health_checks_enabled=False) ).bootstrap() @application.post("/items") @@ -44,19 +44,19 @@ def create_item(payload: Payload) -> Payload: assert client.post("/items", json={"value": {"nested": [1, None]}}).json() == {"value": {"nested": [1, None]}} rejected = client.post("/items", json={}) assert rejected.status_code == http.HTTPStatus.UNPROCESSABLE_ENTITY - assert set(rejected.json()) == {"detail"} - assert "title" not in schemas["HTTPValidationError"]["properties"] + assert set(rejected.json()) == {"detail", "title"} + assert schemas["HTTPValidationError"]["properties"]["title"]["description"] def test_fastapi_empty_schema_is_supported() -> None: application: typing.Final = FastApiBootstrapper( - FastApiSettings(openapi_descriptions_enabled=True, service_debug=False, health_checks_enabled=False) + FastApiSettings(api_metadata_enabled=True, service_debug=False, health_checks_enabled=False) ).bootstrap() assert "security" not in application.openapi() assert not application.openapi().get("components", {}).get("schemas") -def test_litestar_descriptions_preserve_runtime_contracts() -> None: +def test_litestar_metadata_describes_schemas_and_preserves_successful_responses() -> None: @litestar.get("/example-health", sync_to_thread=False) def get_health() -> HealthCheckTypedDict: return {"health_status": True} @@ -66,15 +66,15 @@ def get_item(item_id: int, limit: int = 1) -> int: return item_id * limit application: typing.Final = ( - LitestarBootstrapper(LitestarSettings(openapi_descriptions_enabled=True, service_debug=False)) + LitestarBootstrapper(LitestarSettings(api_metadata_enabled=True, service_debug=False)) .configure_application(LitestarConfig(route_handlers=[get_health, get_item])) .bootstrap() ) document: typing.Final = application.openapi_schema before: typing.Final = copy.deepcopy(document.to_schema()) - LitestarSwaggerInstrument(SwaggerConfig(openapi_descriptions_enabled=True)).bootstrap_after(application) + LitestarBootstrapper(LitestarSettings(api_metadata_enabled=True, service_debug=False)).bootstrap_after(application) after: typing.Final = copy.deepcopy(document.to_schema()) - LitestarSwaggerInstrument(SwaggerConfig(openapi_descriptions_enabled=True)).bootstrap_after(application) + LitestarBootstrapper(LitestarSettings(api_metadata_enabled=True, service_debug=False)).bootstrap_after(application) assert document.to_schema() == after health: typing.Final = after["components"]["schemas"]["HealthCheckTypedDict"] assert health["properties"]["health_status"]["description"].startswith("Static process health") @@ -82,7 +82,7 @@ def get_item(item_id: int, limit: int = 1) -> int: "schema" ] assert error["properties"]["detail"]["description"] - assert "title" not in error["properties"] + assert error["properties"]["title"]["description"] before_error: typing.Final = before["paths"]["/items/{item_id}"]["get"]["responses"]["400"]["content"][ "application/json" ]["schema"] @@ -94,7 +94,7 @@ def get_item(item_id: int, limit: int = 1) -> int: assert client.get(f"/items/{item_id}").json() == item_id rejected = client.get(f"/items/{item_id}?limit=invalid") assert rejected.status_code == http.HTTPStatus.BAD_REQUEST - assert set(rejected.json()) == {"status_code", "detail", "extra"} + assert set(rejected.json()) == {"status_code", "detail", "extra", "title"} def test_composite_error_schemas_and_references_keep_their_shapes() -> None: @@ -137,7 +137,7 @@ def test_composite_error_schemas_and_references_keep_their_shapes() -> None: document: typing.Final = application.openapi_schema document.paths = {"/items": spec.PathItem(get=operation)} document.components.schemas = {"JsonValue": spec.Schema(description="Owned JSON description")} - LitestarSwaggerInstrument(SwaggerConfig(openapi_descriptions_enabled=True)).bootstrap_after(application) + LitestarBootstrapper(LitestarSettings(api_metadata_enabled=True, service_debug=False)).bootstrap_after(application) assert document.components.schemas["JsonValue"].description == "Owned JSON description" assert compound.type == spec.OpenAPIType.OBJECT assert compound.one_of == [branch, spec.Schema(type=spec.OpenAPIType.OBJECT)] @@ -150,7 +150,7 @@ def test_composite_error_schemas_and_references_keep_their_shapes() -> None: def test_fastapi_preserves_owned_descriptions() -> None: application: typing.Final = FastApiBootstrapper( - FastApiSettings(openapi_descriptions_enabled=True, service_debug=False) + FastApiSettings(api_metadata_enabled=True, service_debug=False) ).bootstrap() cached: typing.Final = application.openapi() cached.clear() @@ -178,7 +178,7 @@ def test_fastapi_preserves_owned_descriptions() -> None: @pytest.mark.parametrize("enabled", [False, True]) def test_fastapi_setting_controls_generated_descriptions(enabled: bool) -> None: application: typing.Final = FastApiBootstrapper( - FastApiSettings(openapi_descriptions_enabled=enabled, service_debug=False) + FastApiSettings(api_metadata_enabled=enabled, service_debug=False) ).bootstrap() @application.post("/items") @@ -201,8 +201,7 @@ def get_items(limit: int) -> int: return limit application: typing.Final = ( - LitestarBootstrapper(LitestarSettings(service_debug=False)) - .configure_instrument(SwaggerConfig(openapi_descriptions_enabled=enabled)) + LitestarBootstrapper(LitestarSettings(service_debug=False, api_metadata_enabled=enabled)) .configure_application(LitestarConfig(route_handlers=[get_items])) .bootstrap() ) @@ -217,7 +216,7 @@ def get_items(limit: int) -> int: def test_litestar_missing_openapi_and_empty_schema_are_supported() -> None: - instrument: typing.Final = LitestarSwaggerInstrument(SwaggerConfig(openapi_descriptions_enabled=True)) + instrument: typing.Final = LitestarBootstrapper(LitestarSettings(api_metadata_enabled=True, service_debug=False)) disabled: typing.Final = litestar.Litestar(openapi_config=None) assert instrument.bootstrap_after(disabled) is disabled application: typing.Final = litestar.Litestar(route_handlers=[]) @@ -230,7 +229,10 @@ def test_litestar_missing_openapi_and_empty_schema_are_supported() -> None: def test_environment_can_enable_descriptions(monkeypatch: pytest.MonkeyPatch) -> None: - assert SwaggerConfig().openapi_descriptions_enabled is False - monkeypatch.setenv("OPENAPI_DESCRIPTIONS_ENABLED", "true") - assert FastApiSettings().openapi_descriptions_enabled is True - assert LitestarSettings().openapi_descriptions_enabled is True + assert FastApiSettings().api_metadata_enabled is False + assert LitestarSettings().api_metadata_enabled is False + assert "openapi_descriptions_enabled" not in FastApiSettings.model_fields + assert "openapi_descriptions_enabled" not in LitestarSettings.model_fields + monkeypatch.setenv("API_METADATA_ENABLED", "true") + assert FastApiSettings().api_metadata_enabled is True + assert LitestarSettings().api_metadata_enabled is True