shudao-main/shudao-chat-py/tests/test_speech_router.py

import importlib.util
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import Mock, patch

from fastapi import UploadFile
from starlette.responses import JSONResponse


SPEECH_PATH = Path(__file__).resolve().parents[1] / "routers" / "speech.py"
spec = importlib.util.spec_from_file_location("speech_under_test", SPEECH_PATH)
speech = importlib.util.module_from_spec(spec)
spec.loader.exec_module(speech)


class SpeechRouterTests(unittest.IsolatedAsyncioTestCase):
    def _request(self, user_id=70430):
        return SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(user_id=user_id)))

    async def test_transcribe_audio_returns_text_payload(self):
        upload = UploadFile(filename="sample.wav", file=None)

        with patch.object(
            speech.tencent_speech_service,
            "transcribe_file",
            Mock(return_value={"text": "桥梁施工注意事项", "request_id": "req-1"}),
        ):
            response = await speech.transcribe_audio(
                self._request(),
                file=upload,
                user_id="web-user",
            )

        self.assertEqual(response["statusCode"], 200)
        self.assertEqual(response["data"]["text"], "桥梁施工注意事项")
        self.assertEqual(response["data"]["request_id"], "req-1")

    async def test_synthesize_speech_returns_audio_response(self):
        with patch.object(
            speech.tencent_speech_service,
            "synthesize_text",
            Mock(
                return_value={
                    "audio_bytes": b"fake-mp3",
                    "content_type": "audio/mpeg",
                    "request_id": "req-2",
                }
            ),
        ):
            response = await speech.synthesize_speech(
                self._request(),
                speech.SpeechSynthesizeRequest(text="测试播报"),
            )

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.media_type, "audio/mpeg")
        self.assertEqual(response.headers["X-Speech-Request-Id"], "req-2")

    async def test_synthesize_speech_returns_503_when_tencent_quota_is_exhausted(self):
        with patch.object(
            speech.tencent_speech_service,
            "synthesize_text",
            Mock(
                side_effect=RuntimeError(
                    "腾讯云语音合成失败: "
                    "[TencentCloudSDKException] code:UnsupportedOperation.PkgExhausted "
                    "message:The resource pack allowance has been exhausted"
                )
            ),
        ):
            response = await speech.synthesize_speech(
                self._request(),
                speech.SpeechSynthesizeRequest(text="测试播报"),
            )

        self.assertIsInstance(response, JSONResponse)
        self.assertEqual(response.status_code, 503)
        self.assertIn('"statusCode":503', response.body.decode("utf-8"))
        self.assertIn("PkgExhausted", response.body.decode("utf-8"))


if __name__ == "__main__":
    unittest.main()