life-echo/api/tests/test_app_error_contract.py

"""AppError 统一错误契约 HTTP 场景测试。"""

import pytest
from fastapi import FastAPI, HTTPException
from httpx import ASGITransport, AsyncClient

from app.core.errors import (
    AuthenticationError,
    NotFoundError,
    QuotaExceededError,
    RateLimitedError,
    ValidationError,
    register_exception_handlers,
)
from app.core.middleware import RequestIdMiddleware
from app.features.auth.service import AuthError
from app.features.payment.payment_exceptions import PaymentError


def _test_app() -> FastAPI:
    app = FastAPI()
    app.add_middleware(RequestIdMiddleware)
    register_exception_handlers(app)

    @app.get("/not-found")
    async def _not_found():
        raise NotFoundError("资源不存在")

    @app.get("/auth-failed")
    async def _auth_failed():
        raise AuthenticationError("无法验证凭据")

    @app.get("/quota")
    async def _quota():
        raise QuotaExceededError("配额已用尽")

    @app.get("/rate-limited")
    async def _rate_limited():
        raise RateLimitedError("发送过于频繁，请30秒后再试")

    @app.get("/validation")
    async def _validation():
        raise ValidationError("参数无效")

    @app.get("/auth-domain")
    async def _auth_domain():
        raise AuthError("该邮箱已被注册", "EMAIL_EXISTS")

    @app.get("/payment-domain")
    async def _payment_domain():
        raise PaymentError("支付配置错误", code="PAYMENT_CONFIG_ERROR")

    @app.get("/http-string")
    async def _http_string():
        raise HTTPException(status_code=400, detail="请求无效")

    @app.get("/http-429")
    async def _http_429():
        raise HTTPException(status_code=429, detail="发送过于频繁")

    @app.get("/http-list")
    async def _http_list():
        raise HTTPException(
            status_code=422,
            detail=[{"loc": ["body", "phone"], "msg": "field required"}],
        )

    @app.get("/http-unknown-status")
    async def _http_unknown_status():
        raise HTTPException(status_code=418, detail="teapot")

    @app.get("/http-unknown-5xx")
    async def _http_unknown_5xx():
        raise HTTPException(status_code=599, detail="upstream glitch")

    @app.get("/boom")
    async def _boom():
        raise RuntimeError("secret_internal_xyz")

    return app


@pytest.mark.asyncio
async def test_not_found_error_contract() -> None:
    app = _test_app()
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        r = await client.get("/not-found")
    assert r.status_code == 404
    body = r.json()
    assert body["error_code"] == "NOT_FOUND"
    assert body["message"] == "资源不存在"
    assert "request_id" in body


@pytest.mark.asyncio
async def test_authentication_error_contract() -> None:
    app = _test_app()
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        r = await client.get("/auth-failed")
    assert r.status_code == 401
    body = r.json()
    assert body["error_code"] == "AUTHENTICATION_FAILED"
    assert body["message"] == "无法验证凭据"
    assert r.headers.get("www-authenticate") == "Bearer"


@pytest.mark.asyncio
async def test_quota_exceeded_error_contract() -> None:
    app = _test_app()
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        r = await client.get("/quota")
    assert r.status_code == 429
    body = r.json()
    assert body["error_code"] == "QUOTA_EXCEEDED"
    assert body["message"] == "配额已用尽"


@pytest.mark.asyncio
async def test_rate_limited_error_contract() -> None:
    app = _test_app()
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        r = await client.get("/rate-limited")
    assert r.status_code == 429
    body = r.json()
    assert body["error_code"] == "RATE_LIMITED"
    assert body["message"] == "发送过于频繁，请30秒后再试"


@pytest.mark.asyncio
async def test_validation_error_contract() -> None:
    app = _test_app()
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        r = await client.get("/validation")
    assert r.status_code == 422
    body = r.json()
    assert body["error_code"] == "VALIDATION_ERROR"
    assert body["message"] == "参数无效"


@pytest.mark.asyncio
async def test_auth_error_domain_contract() -> None:
    app = _test_app()
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        r = await client.get("/auth-domain")
    assert r.status_code == 400
    body = r.json()
    assert body["error_code"] == "EMAIL_EXISTS"
    assert body["message"] == "该邮箱已被注册"


@pytest.mark.asyncio
async def test_payment_error_domain_contract() -> None:
    app = _test_app()
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        r = await client.get("/payment-domain")
    assert r.status_code == 502
    body = r.json()
    assert body["error_code"] == "PROVIDER_ERROR"
    assert body["message"] == "支付配置错误"


@pytest.mark.asyncio
async def test_http_exception_string_detail_contract() -> None:
    app = _test_app()
    transport = ASGITransport(app=app, raise_app_exceptions=False)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        r = await client.get("/http-string")
    assert r.status_code == 400
    body = r.json()
    assert body["error_code"] == "BAD_REQUEST"
    assert body["message"] == "请求无效"
    assert "request_id" in body


@pytest.mark.asyncio
async def test_http_exception_429_maps_to_rate_limited() -> None:
    app = _test_app()
    transport = ASGITransport(app=app, raise_app_exceptions=False)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        r = await client.get("/http-429")
    assert r.status_code == 429
    body = r.json()
    assert body["error_code"] == "RATE_LIMITED"
    assert body["message"] == "发送过于频繁"


@pytest.mark.asyncio
async def test_http_exception_list_detail_contract() -> None:
    app = _test_app()
    transport = ASGITransport(app=app, raise_app_exceptions=False)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        r = await client.get("/http-list")
    assert r.status_code == 422
    body = r.json()
    assert body["error_code"] == "VALIDATION_ERROR"
    assert "body.phone" in body["message"]
    assert "field required" in body["message"]


@pytest.mark.asyncio
async def test_http_exception_unknown_5xx_maps_to_internal_error() -> None:
    app = _test_app()
    transport = ASGITransport(app=app, raise_app_exceptions=False)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        r = await client.get("/http-unknown-5xx")
    assert r.status_code == 599
    body = r.json()
    assert body["error_code"] == "INTERNAL_ERROR"
    assert body["message"] == "upstream glitch"


@pytest.mark.asyncio
async def test_http_exception_unknown_status_maps_to_bad_request() -> None:
    app = _test_app()
    transport = ASGITransport(app=app, raise_app_exceptions=False)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        r = await client.get("/http-unknown-status")
    assert r.status_code == 418
    body = r.json()
    assert body["error_code"] == "BAD_REQUEST"
    assert body["message"] == "teapot"


@pytest.mark.asyncio
async def test_unhandled_exception_contract() -> None:
    app = _test_app()
    transport = ASGITransport(app=app, raise_app_exceptions=False)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        r = await client.get("/boom")
    assert r.status_code == 500
    body = r.json()
    assert body["error_code"] == "INTERNAL_ERROR"
    assert body["message"] == "服务器内部错误"
    assert "secret_internal_xyz" not in r.text