93be60f74c
- Pipeline: skip _send_tts_audio only for non-manual when ENABLE_TTS=false; remove enable_tts early return from handle_tts_request_on_demand. - Tencent TTS: PrimaryLanguage/chunking follow user language preference only. - Expo: let manual tts_audio bypass late-segment playback gate after interrupt. - Docs: clarify ENABLE_TTS vs tts_request in api/.env.example and TTSProvider port. - Tests: add manual bypass cases; adjust pipeline language tests for en+Chinese text. Co-authored-by: Cursor <cursoragent@cursor.com>
66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
"""ENABLE_TTS=false 时仍可走喇叭按需合成；自动回复路径则被关闭。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from app.features.conversation.ws import pipeline as pl
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_tts_manual_bypasses_enable_tts_false(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(pl.settings, "enable_tts", False)
|
|
|
|
fake_tts = MagicMock()
|
|
fake_tts.synthesize = AsyncMock(return_value=b"\xff\xd3-mp3stub")
|
|
monkeypatch.setattr(pl, "get_tts_provider", lambda: fake_tts)
|
|
|
|
storage = MagicMock()
|
|
storage.upload.return_value = "https://example/public.wav"
|
|
storage.get_url.return_value = "https://example/signed.wav"
|
|
monkeypatch.setattr(pl, "get_object_storage", lambda: storage)
|
|
|
|
send_mock = AsyncMock()
|
|
monkeypatch.setattr(pl.manager, "send_message", send_mock)
|
|
monkeypatch.setattr(pl, "_tts_epoch_value", lambda _cid: 0)
|
|
|
|
cid = "c0000000-0000-4000-8000-000000000001"
|
|
out = await pl._send_tts_audio(
|
|
cid,
|
|
"hi",
|
|
chunk_index=0,
|
|
chunk_total=1,
|
|
assistant_message_id="m1",
|
|
tts_epoch_start=0,
|
|
manual=True,
|
|
language="en",
|
|
)
|
|
assert out == "https://example/public.wav"
|
|
fake_tts.synthesize.assert_awaited_once()
|
|
send_mock.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_tts_auto_blocked_when_enable_tts_false(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(pl.settings, "enable_tts", False)
|
|
|
|
fake_tts = MagicMock()
|
|
fake_tts.synthesize = AsyncMock(return_value=b"audio")
|
|
monkeypatch.setattr(pl, "get_tts_provider", lambda: fake_tts)
|
|
|
|
cid = "c0000000-0000-4000-8000-000000000002"
|
|
out = await pl._send_tts_audio(
|
|
cid,
|
|
"hi",
|
|
chunk_index=0,
|
|
chunk_total=1,
|
|
assistant_message_id="m1",
|
|
tts_epoch_start=0,
|
|
manual=False,
|
|
language="en",
|
|
)
|
|
assert out is None
|
|
fake_tts.synthesize.assert_not_called()
|