life-echo/api/tests/evaluation/test_asr_router.py

"""内部评测 ASR 转写路由。"""

import pytest
from httpx import ASGITransport, AsyncClient

from app.features.evaluation.internal_auth import get_internal_eval_principal


@pytest.mark.asyncio
async def test_eval_asr_transcribe_returns_text():
    from fastapi import FastAPI

    from tests.conftest import install_test_error_handlers

    class _FakeAsr:
        async def transcribe(self, audio: bytes, format: str = "m4a") -> str:
            assert audio == b"fake-audio"
            assert format == "wav"
            return "你好世界"

    from app.features.evaluation.asr_service import EvalAsrService
    from app.features.evaluation.deps import get_eval_asr_service
    from app.features.evaluation.router import router

    app = install_test_error_handlers(FastAPI())
    app.include_router(router, prefix="/internal/api/evaluation")

    async def _override_auth():
        from app.features.evaluation.internal_auth import InternalEvalPrincipal

        return InternalEvalPrincipal()

    app.dependency_overrides[get_internal_eval_principal] = _override_auth
    app.dependency_overrides[get_eval_asr_service] = lambda: EvalAsrService(_FakeAsr())

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://t") as client:
        r = await client.post(
            "/internal/api/evaluation/asr/transcribe?format=wav",
            files={"file": ("sample.wav", b"fake-audio", "audio/wav")},
            headers={"X-Internal-Eval-Key": "secret"},
        )
    assert r.status_code == 200
    body = r.json()
    assert body["text"] == "你好世界"
    assert body["format"] == "wav"
    assert body["audio_bytes"] == len(b"fake-audio")


@pytest.mark.asyncio
async def test_eval_asr_transcribe_rejects_empty_file():
    from fastapi import FastAPI

    from tests.conftest import install_test_error_handlers

    class _FakeAsr:
        async def transcribe(self, audio: bytes, format: str = "m4a") -> str:
            return "unused"

    from app.features.evaluation.asr_service import EvalAsrService
    from app.features.evaluation.deps import get_eval_asr_service
    from app.features.evaluation.router import router

    app = install_test_error_handlers(FastAPI())
    app.include_router(router, prefix="/internal/api/evaluation")

    async def _override_auth():
        from app.features.evaluation.internal_auth import InternalEvalPrincipal

        return InternalEvalPrincipal()

    app.dependency_overrides[get_internal_eval_principal] = _override_auth
    app.dependency_overrides[get_eval_asr_service] = lambda: EvalAsrService(_FakeAsr())

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://t") as client:
        r = await client.post(
            "/internal/api/evaluation/asr/transcribe?format=wav",
            files={"file": ("empty.wav", b"", "audio/wav")},
            headers={"X-Internal-Eval-Key": "secret"},
        )
    assert r.status_code == 400
    assert r.json()["error_code"] == "BAD_REQUEST"


@pytest.mark.asyncio
async def test_read_limited_upload_rejects_oversized_body():
    from io import BytesIO

    from starlette.datastructures import UploadFile as StarletteUploadFile

    from app.features.evaluation.asr_service import read_limited_upload
    from app.features.evaluation.errors import EvaluationBadRequestError

    upload = StarletteUploadFile(
        file=BytesIO(b"x" * 11),
        filename="tiny.wav",
    )
    with pytest.raises(EvaluationBadRequestError, match="音频过大"):
        await read_limited_upload(upload, max_bytes=10)