53e0065e3e
配置 SSOT（TOML + .env） 统一错误契约 Auth 与事务边界 Redis / Celery 可靠性:业务 Redis（DB/0）与 Celery broker/backend（DB/1）显式拆分；连接池、sync client 可观测性（OpenTelemetry + LGTM）
113 lines
4.2 KiB
Python
113 lines
4.2 KiB
Python
"""OpenAPI 统一 ErrorResponse 组件与 router 引用。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi import FastAPI
|
|
|
|
from app.core.error_codes import AUTH_ERROR_CODES, ERROR_CODE_ENUM
|
|
from app.core.openapi import COMMON_ERROR_RESPONSES, custom_openapi, error_responses
|
|
from app.features.auth.router import router as auth_router
|
|
from app.features.memoir.router import router as memoir_router
|
|
from app.features.payment.router import router as payment_router
|
|
|
|
|
|
def _openapi_app(*routers) -> FastAPI:
|
|
app = FastAPI()
|
|
for router in routers:
|
|
app.include_router(router)
|
|
app.openapi = lambda: custom_openapi(app) # type: ignore[assignment]
|
|
return app
|
|
|
|
|
|
def test_openapi_includes_error_response_schema() -> None:
|
|
app = _openapi_app(payment_router)
|
|
app.openapi_schema = None
|
|
schema = custom_openapi(app)
|
|
|
|
error_schema = schema["components"]["schemas"]["ErrorResponse"]
|
|
assert set(error_schema["required"]) == {"error_code", "message", "request_id"}
|
|
error_code_prop = error_schema["properties"]["error_code"]
|
|
assert error_code_prop["allOf"][0]["$ref"].endswith("ErrorCode")
|
|
|
|
error_code_enum = schema["components"]["schemas"]["ErrorCode"]["enum"]
|
|
assert "NOT_FOUND" in error_code_enum
|
|
assert "PHONE_EXISTS" in error_code_enum
|
|
assert set(error_code_enum) == set(ERROR_CODE_ENUM)
|
|
|
|
|
|
def test_openapi_description_includes_error_code_table() -> None:
|
|
app = _openapi_app(auth_router)
|
|
app.openapi_schema = None
|
|
schema = custom_openapi(app)
|
|
description = schema["info"]["description"]
|
|
assert "PHONE_EXISTS" in description
|
|
assert "INVALID_SMS_CODE" in description
|
|
assert AUTH_ERROR_CODES[0]["code"] in description
|
|
|
|
|
|
def test_openapi_domain_error_code_schema() -> None:
|
|
app = _openapi_app(auth_router)
|
|
app.openapi_schema = None
|
|
schema = custom_openapi(app)
|
|
domain_enum = schema["components"]["schemas"]["DomainErrorCode"]["enum"]
|
|
assert "PHONE_EXISTS" in domain_enum
|
|
assert "NOT_FOUND" not in domain_enum
|
|
|
|
|
|
def test_error_responses_reference_error_response_component() -> None:
|
|
responses = error_responses(401, 404)
|
|
assert responses[401]["content"]["application/json"]["schema"]["$ref"].endswith(
|
|
"ErrorResponse"
|
|
)
|
|
assert 404 in responses
|
|
assert responses[404] == COMMON_ERROR_RESPONSES[404]
|
|
|
|
|
|
def test_memoir_router_openapi_uses_error_response_ref() -> None:
|
|
app = _openapi_app(memoir_router)
|
|
app.openapi_schema = None
|
|
schema = custom_openapi(app)
|
|
get_chapters = schema["paths"]["/api/chapters"]["get"]
|
|
assert "401" in get_chapters["responses"]
|
|
ref = get_chapters["responses"]["401"]["content"]["application/json"]["schema"]["$ref"]
|
|
assert ref.endswith("ErrorResponse")
|
|
|
|
|
|
def test_auth_register_openapi_uses_error_response_ref() -> None:
|
|
app = _openapi_app(auth_router)
|
|
app.openapi_schema = None
|
|
schema = custom_openapi(app)
|
|
register = schema["paths"]["/api/auth/register"]["post"]
|
|
assert "400" in register["responses"]
|
|
ref = register["responses"]["400"]["content"]["application/json"]["schema"]["$ref"]
|
|
assert ref.endswith("ErrorResponse")
|
|
|
|
|
|
def test_error_responses_includes_502() -> None:
|
|
responses = error_responses(502)
|
|
assert responses[502] == COMMON_ERROR_RESPONSES[502]
|
|
assert responses[502]["content"]["application/json"]["schema"]["$ref"].endswith(
|
|
"ErrorResponse"
|
|
)
|
|
|
|
|
|
def test_payment_router_openapi_includes_502() -> None:
|
|
app = _openapi_app(payment_router)
|
|
app.openapi_schema = None
|
|
schema = custom_openapi(app)
|
|
create_order = schema["paths"]["/api/payment/create-order"]["post"]
|
|
assert "502" in create_order["responses"]
|
|
ref = create_order["responses"]["502"]["content"]["application/json"]["schema"]["$ref"]
|
|
assert ref.endswith("ErrorResponse")
|
|
|
|
|
|
def test_payment_router_openapi_includes_504_and_500() -> None:
|
|
app = _openapi_app(payment_router)
|
|
app.openapi_schema = None
|
|
schema = custom_openapi(app)
|
|
create_order = schema["paths"]["/api/payment/create-order"]["post"]
|
|
for status in ("500", "504"):
|
|
assert status in create_order["responses"]
|
|
ref = create_order["responses"][status]["content"]["application/json"]["schema"]["$ref"]
|
|
assert ref.endswith("ErrorResponse")
|