life-echo/api/tests/test_http_contract_errors.py

"""HTTP 层对外错误文案脱敏契约（响应体不含内部异常串）。"""

from io import BytesIO
from unittest.mock import MagicMock

import pytest
from httpx import ASGITransport, AsyncClient

from app.core.dependencies import get_current_user
from app.core.errors import register_exception_handlers
from app.core.middleware import RequestIdMiddleware
from app.features.auth.deps import get_auth_service
from app.features.auth.router import router as auth_router
from app.features.payment.deps import get_payment_order_service
from app.features.payment.router import router as payment_router


@pytest.mark.asyncio
async def test_wechat_notify_returns_fixed_message_on_service_error() -> None:
    from fastapi import FastAPI

    class BoomOrderService:
        async def handle_wechat_notify(self, *, headers: dict, body: str):
            raise RuntimeError("wechat_sdk_secret_123")

    app = FastAPI()
    app.add_middleware(RequestIdMiddleware)
    register_exception_handlers(app)
    app.include_router(payment_router)
    app.dependency_overrides[get_payment_order_service] = lambda: BoomOrderService()

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        r = await client.post("/api/payment/notify/wechat", content="raw")
    assert r.status_code == 200
    data = r.json()
    assert data.get("code") == "FAIL"
    assert data.get("message") == "处理失败"
    assert "wechat_sdk" not in r.text
    assert "secret" not in r.text.lower()


def _minimal_jpeg_bytes() -> bytes:
    """1x1 JPEG 最小合法文件。"""
    return (
        b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00"
        b"\xff\xdb\x00C\x00\x08\x06\x06\x07\x06\x05\x08\x07\x07\x07\t\t\x08\n\x0c\x14\r\x0c\x0b\x0b\x0c\x19\x12\x13\x0f\x14\x1d\x1a\x1f\x1e\x1d\x1a\x1c\x1c $.' \",#\x1c\x1c(7),01444\x1f'9=82<.342\xff\xc0\x00\x11\x08\x00\x01\x00\x01\x01\x01\x11\x00\xff\xc4\x00\x14\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x08\xff\xc4\x00\x14\x10\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xda\x00\x0c\x03\x01\x00\x02\x11\x03\x11\x00\x3f\x00\xaa\xff\xd9"
    )


@pytest.mark.asyncio
async def test_avatar_upload_500_detail_sanitized(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from fastapi import FastAPI

    from app.core.config import settings
    import app.core.dependencies as deps

    fake_user = MagicMock()
    fake_user.id = "user-contract-test"
    fake_user.avatar_url = None

    mock_storage = MagicMock()
    mock_storage.upload = MagicMock(
        return_value="https://test-bucket.cos.ap-shanghai.myqcloud.com/avatars/user-contract-test/abc.jpg"
    )
    monkeypatch.setattr(deps, "get_object_storage", lambda: mock_storage)
    monkeypatch.setattr(settings, "tencent_secret_id", "sid", raising=False)
    monkeypatch.setattr(settings, "tencent_secret_key", "sk", raising=False)
    monkeypatch.setattr(settings, "tencent_cos_bucket", "test-bucket", raising=False)
    monkeypatch.setattr(
        settings,
        "tencent_cos_base_url",
        "https://test-bucket.cos.ap-shanghai.myqcloud.com",
        raising=False,
    )

    class BoomAuth:
        async def upload_avatar(self, user_id, file_content, content_type, **kwargs):
            raise RuntimeError("db_connection_secret_xyz")

    app = FastAPI()
    app.add_middleware(RequestIdMiddleware)
    register_exception_handlers(app)
    app.include_router(auth_router)
    app.dependency_overrides[get_current_user] = lambda: fake_user
    app.dependency_overrides[get_auth_service] = lambda: BoomAuth()

    transport = ASGITransport(app=app, raise_app_exceptions=False)
    files = {"file": ("a.jpg", BytesIO(_minimal_jpeg_bytes()), "image/jpeg")}
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        r = await client.post("/api/auth/me/avatar", files=files)

    assert r.status_code == 500
    body = r.json()
    assert body.get("error_code") == "INTERNAL_ERROR"
    message = body.get("message", "")
    assert message == "服务器内部错误"
    assert "secret" not in str(message).lower()
    assert "db_connection" not in r.text