wordloop/backend/tests/test_tts_service.py

import base64
import json
import os
import tempfile
import unittest
from unittest.mock import MagicMock, patch

from services.tts_service import TtsService, TtsServiceError, tts_service


class _FakeResponse:
    def __init__(self, body: bytes, content_type: str = "application/json"):
        self._body = body
        self.headers = MagicMock()
        self.headers.get_content_type.return_value = content_type

    def read(self):
        return self._body

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, tb):
        return False


class TtsServiceTest(unittest.TestCase):
    def tearDown(self):
        get_settings = __import__("config", fromlist=["get_settings"]).get_settings
        get_settings.cache_clear()

    def test_not_configured_when_disabled(self):
        with patch.dict(
            os.environ,
            {"TKMIND_TTS_ENABLED": "false", "TKMIND_TTS_BASE_URL": "http://127.0.0.1:19200"},
            clear=False,
        ):
            __import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
            self.assertFalse(tts_service.is_configured())

    @patch("services.tts_service._open_tts_request")
    def test_request_includes_user_agent(self, urlopen_mock: MagicMock):
        wav = b"RIFFxxxx"
        payload = {"code": 0, "data": base64.b64encode(wav).decode("ascii")}
        urlopen_mock.return_value = _FakeResponse(json.dumps(payload).encode("utf-8"))

        with patch.dict(
            os.environ,
            {"TKMIND_TTS_BASE_URL": "https://asr.tkmind.cn", "TKMIND_TTS_ENABLED": "true"},
            clear=False,
        ):
            __import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
            with tempfile.TemporaryDirectory() as tmp:
                with patch("services.tts_service._CACHE_DIR", __import__("pathlib").Path(tmp)):
                    tts_service.synthesize("apple")
                    request = urlopen_mock.call_args.args[0]
                    self.assertEqual(request.get_header("User-agent"), "WordLoop-TTS/1.0")

    @patch("services.tts_service._open_tts_request")
    def test_synthesize_parses_json_base64(self, urlopen_mock: MagicMock):
        wav = b"RIFFxxxx"
        payload = {"code": 0, "message": "ok", "data": base64.b64encode(wav).decode("ascii")}
        urlopen_mock.return_value = _FakeResponse(json.dumps(payload).encode("utf-8"))

        with patch.dict(
            os.environ,
            {"TKMIND_TTS_BASE_URL": "http://127.0.0.1:19200", "TKMIND_TTS_ENABLED": "true"},
            clear=False,
        ):
            __import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
            with tempfile.TemporaryDirectory() as tmp:
                with patch("services.tts_service._CACHE_DIR", __import__("pathlib").Path(tmp)):
                    audio, content_type = tts_service.synthesize("apple")
                    self.assertEqual(audio, wav)
                    self.assertEqual(content_type, "audio/wav")

    @patch("services.tts_service._open_tts_request")
    def test_synthesize_uses_cache_on_second_call(self, urlopen_mock: MagicMock):
        wav = b"RIFFcached"
        payload = {"code": 0, "data": base64.b64encode(wav).decode("ascii")}
        urlopen_mock.return_value = _FakeResponse(json.dumps(payload).encode("utf-8"))

        with patch.dict(
            os.environ,
            {"TKMIND_TTS_BASE_URL": "http://127.0.0.1:19200", "TKMIND_TTS_ENABLED": "true"},
            clear=False,
        ):
            __import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
            with tempfile.TemporaryDirectory() as tmp:
                with patch("services.tts_service._CACHE_DIR", __import__("pathlib").Path(tmp)):
                    first, _ = tts_service.synthesize("apple")
                    second, _ = tts_service.synthesize("apple")
                    self.assertEqual(first, second)
                    self.assertEqual(urlopen_mock.call_count, 1)

    @patch("services.tts_service._open_tts_request")
    def test_upstream_error_message(self, urlopen_mock: MagicMock):
        payload = {"code": 500, "message": "TTS 未启用（ENABLE_TTS=false）", "data": None}
        urlopen_mock.return_value = _FakeResponse(json.dumps(payload).encode("utf-8"))

        with patch.dict(
            os.environ,
            {"TKMIND_TTS_BASE_URL": "http://127.0.0.1:19200", "TKMIND_TTS_ENABLED": "true"},
            clear=False,
        ):
            __import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
            with tempfile.TemporaryDirectory() as tmp:
                with patch("services.tts_service._CACHE_DIR", __import__("pathlib").Path(tmp)):
                    with self.assertRaises(TtsServiceError) as ctx:
                        tts_service.synthesize("apple")
                    self.assertIn("TTS 未启用", str(ctx.exception))

    def test_extract_audio_base64_from_dict(self):
        value = TtsService._extract_audio_base64({"audio": "abc"})
        self.assertEqual(value, "abc")