371 lines
15 KiB
Python
371 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from services.ai_voice_runtime_service.runtime_tts_provider import RuntimeConfiguredTTSProvider
|
|
from services.ai_voice_runtime_service.providers import tts as tts_module
|
|
from services.shared.db import get_session
|
|
from services.shared.sql_init import init_sql_schema
|
|
from services.shared.sql_models import VoiceTTSSettingsRow
|
|
from services.shared.voice_tts_config import save_voice_tts_config, voice_tts_default_config
|
|
|
|
init_sql_schema()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_voice_tts_settings():
|
|
session = get_session()
|
|
try:
|
|
row = session.execute(select(VoiceTTSSettingsRow)).scalar_one_or_none()
|
|
if row is not None:
|
|
session.delete(row)
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
yield
|
|
session = get_session()
|
|
try:
|
|
row = session.execute(select(VoiceTTSSettingsRow)).scalar_one_or_none()
|
|
if row is not None:
|
|
session.delete(row)
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
class _DummyResponse:
|
|
def __init__(self, content: bytes = b"", json_payload: dict | None = None) -> None:
|
|
self.content = content
|
|
self._json_payload = json_payload or {}
|
|
|
|
def raise_for_status(self) -> None:
|
|
return None
|
|
|
|
def json(self) -> dict:
|
|
return self._json_payload
|
|
|
|
|
|
def test_openai_tts_provider_persists_cache_on_disk(tmp_path, monkeypatch):
|
|
calls: list[dict] = []
|
|
|
|
class _DummyClient:
|
|
def __init__(self, *, timeout: float) -> None:
|
|
self.timeout = timeout
|
|
|
|
def __enter__(self) -> _DummyClient:
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
def post(self, url: str, *, headers: dict[str, str], json: dict[str, str]) -> _DummyResponse:
|
|
calls.append({"url": url, "headers": headers, "json": json, "timeout": self.timeout})
|
|
return _DummyResponse(b"\x01\x00\x02\x00")
|
|
|
|
monkeypatch.setenv("AI_API_KEY", "test-key")
|
|
monkeypatch.setenv("AI_API_BASE", "https://example.test/v1")
|
|
monkeypatch.setenv("AI_VOICE_TTS_MODEL", "gpt-4o-mini-tts")
|
|
monkeypatch.setenv("AI_VOICE_TTS_VOICE", "alloy")
|
|
monkeypatch.setenv("AI_VOICE_TTS_CACHE_ENABLED", "1")
|
|
monkeypatch.setenv("AI_VOICE_TTS_CACHE_DIR", str(tmp_path))
|
|
monkeypatch.setattr(tts_module.httpx, "Client", _DummyClient)
|
|
|
|
provider = tts_module.OpenAITTSProvider()
|
|
first = provider.synthesize("Hello from cache", language="ru")
|
|
second = provider.synthesize("Hello from cache", language="ru")
|
|
|
|
assert first.audio_bytes == b"\x01\x00\x02\x00"
|
|
assert second.audio_bytes == first.audio_bytes
|
|
assert len(calls) == 1
|
|
assert list(Path(tmp_path).rglob("*.pcm"))
|
|
assert list(Path(tmp_path).rglob("*.json"))
|
|
|
|
|
|
def test_openai_tts_provider_uses_cached_audio_without_api_key(tmp_path, monkeypatch):
|
|
calls: list[dict] = []
|
|
|
|
class _DummyClient:
|
|
def __init__(self, *, timeout: float) -> None:
|
|
self.timeout = timeout
|
|
|
|
def __enter__(self) -> _DummyClient:
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
def post(self, url: str, *, headers: dict[str, str], json: dict[str, str]) -> _DummyResponse:
|
|
calls.append({"url": url, "headers": headers, "json": json, "timeout": self.timeout})
|
|
return _DummyResponse(b"\x04\x00\x08\x00")
|
|
|
|
monkeypatch.setenv("AI_API_KEY", "test-key")
|
|
monkeypatch.setenv("AI_VOICE_TTS_CACHE_ENABLED", "1")
|
|
monkeypatch.setenv("AI_VOICE_TTS_CACHE_DIR", str(tmp_path))
|
|
monkeypatch.setattr(tts_module.httpx, "Client", _DummyClient)
|
|
|
|
provider = tts_module.OpenAITTSProvider()
|
|
cached = provider.synthesize("Server-side cached prompt", language="ru")
|
|
assert cached.audio_bytes == b"\x04\x00\x08\x00"
|
|
assert len(calls) == 1
|
|
|
|
monkeypatch.delenv("AI_API_KEY", raising=False)
|
|
|
|
def _unexpected_client(*args, **kwargs):
|
|
raise AssertionError("OpenAI TTS should not be called when cached audio already exists")
|
|
|
|
monkeypatch.setattr(tts_module.httpx, "Client", _unexpected_client)
|
|
|
|
provider_without_key = tts_module.OpenAITTSProvider()
|
|
replay = provider_without_key.synthesize("Server-side cached prompt", language="ru")
|
|
|
|
assert replay.audio_bytes == cached.audio_bytes
|
|
assert len(calls) == 1
|
|
|
|
|
|
def test_yandex_tts_provider_posts_lpcm_with_api_key(tmp_path, monkeypatch):
|
|
calls: list[dict] = []
|
|
encoded_audio = base64.b64encode(b"\x10\x00\x20\x00").decode("ascii")
|
|
|
|
class _DummyClient:
|
|
def __init__(self, *, timeout: float) -> None:
|
|
self.timeout = timeout
|
|
|
|
def __enter__(self) -> _DummyClient:
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
def post(self, url: str, *, headers: dict[str, str], json: dict) -> _DummyResponse:
|
|
calls.append({"url": url, "headers": headers, "json": json, "timeout": self.timeout})
|
|
return _DummyResponse(json_payload={"result": {"audioChunk": {"data": encoded_audio}}})
|
|
|
|
monkeypatch.setenv("AI_VOICE_TTS_YANDEX_API_KEY", "yandex-test-key")
|
|
monkeypatch.setenv("AI_VOICE_TTS_YANDEX_API_BASE", "https://tts.example.test")
|
|
monkeypatch.setenv("AI_VOICE_TTS_YANDEX_VOICE", "jane")
|
|
monkeypatch.setenv("AI_VOICE_TTS_YANDEX_SAMPLE_RATE_HZ", "8000")
|
|
monkeypatch.setenv("AI_VOICE_TTS_CACHE_ENABLED", "1")
|
|
monkeypatch.setenv("AI_VOICE_TTS_CACHE_DIR", str(tmp_path))
|
|
monkeypatch.setattr(tts_module.httpx, "Client", _DummyClient)
|
|
|
|
provider = tts_module.YandexTTSProvider()
|
|
synthesis = provider.synthesize("Привет из Yandex TTS", language="ru")
|
|
|
|
assert synthesis.audio_bytes == b"\x10\x00\x20\x00"
|
|
assert synthesis.sample_rate_hz == 8000
|
|
assert len(calls) == 1
|
|
assert calls[0]["url"] == "https://tts.example.test/tts/v3/utteranceSynthesis"
|
|
assert calls[0]["headers"]["Authorization"] == "Api-Key yandex-test-key"
|
|
assert calls[0]["headers"]["Accept"] == "application/json"
|
|
assert calls[0]["json"]["hints"][0]["voice"] == "jane"
|
|
assert calls[0]["json"]["hints"][1]["speed"] == 1.1
|
|
assert calls[0]["json"]["outputAudioSpec"]["rawAudio"]["audioEncoding"] == "LINEAR16_PCM"
|
|
assert calls[0]["json"]["outputAudioSpec"]["rawAudio"]["sampleRateHertz"] == 8000
|
|
assert list(Path(tmp_path).rglob("*.pcm"))
|
|
assert list(Path(tmp_path).rglob("*.json"))
|
|
|
|
|
|
def test_yandex_tts_provider_prefers_per_utterance_role_hint_over_global_role(tmp_path, monkeypatch):
|
|
calls: list[dict] = []
|
|
encoded_audio = base64.b64encode(b"\x12\x00\x34\x00").decode("ascii")
|
|
|
|
class _DummyClient:
|
|
def __init__(self, *, timeout: float) -> None:
|
|
self.timeout = timeout
|
|
|
|
def __enter__(self) -> _DummyClient:
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
def post(self, url: str, *, headers: dict[str, str], json: dict) -> _DummyResponse:
|
|
calls.append({"url": url, "headers": headers, "json": json, "timeout": self.timeout})
|
|
return _DummyResponse(json_payload={"result": {"audioChunk": {"data": encoded_audio}}})
|
|
|
|
monkeypatch.setenv("AI_VOICE_TTS_YANDEX_API_KEY", "yandex-test-key")
|
|
monkeypatch.setenv("AI_VOICE_TTS_YANDEX_API_BASE", "https://tts.example.test")
|
|
monkeypatch.setenv("AI_VOICE_TTS_YANDEX_VOICE", "jane")
|
|
monkeypatch.setenv("AI_VOICE_TTS_YANDEX_ROLE", "neutral")
|
|
monkeypatch.setenv("AI_VOICE_TTS_CACHE_ENABLED", "0")
|
|
monkeypatch.setattr(tts_module.httpx, "Client", _DummyClient)
|
|
|
|
provider = tts_module.YandexTTSProvider()
|
|
synthesis = provider.synthesize("Эмоциональный быстрый отклик", language="ru", style_hints={"role": "good"})
|
|
|
|
assert synthesis.audio_bytes == b"\x12\x00\x34\x00"
|
|
assert len(calls) == 1
|
|
assert calls[0]["json"]["hints"][0]["voice"] == "jane"
|
|
assert calls[0]["json"]["hints"][2]["role"] == "good"
|
|
|
|
|
|
def test_yandex_tts_provider_uses_iam_token_with_folder_id(tmp_path, monkeypatch):
|
|
calls: list[dict] = []
|
|
encoded_audio = base64.b64encode(b"\x30\x00\x40\x00").decode("ascii")
|
|
|
|
class _DummyClient:
|
|
def __init__(self, *, timeout: float) -> None:
|
|
self.timeout = timeout
|
|
|
|
def __enter__(self) -> _DummyClient:
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
def post(self, url: str, *, headers: dict[str, str], json: dict) -> _DummyResponse:
|
|
calls.append({"url": url, "headers": headers, "json": json, "timeout": self.timeout})
|
|
return _DummyResponse(json_payload={"audioChunk": {"data": encoded_audio}})
|
|
|
|
monkeypatch.delenv("AI_VOICE_TTS_YANDEX_API_KEY", raising=False)
|
|
monkeypatch.setenv("AI_VOICE_TTS_YANDEX_IAM_TOKEN", "iam-test-token")
|
|
monkeypatch.setenv("AI_VOICE_TTS_YANDEX_FOLDER_ID", "folder-test")
|
|
monkeypatch.setenv("AI_VOICE_TTS_YANDEX_SAMPLE_RATE_HZ", "16000")
|
|
monkeypatch.setenv("AI_VOICE_TTS_YANDEX_ROLE", "friendly")
|
|
monkeypatch.setenv("AI_VOICE_TTS_CACHE_ENABLED", "0")
|
|
monkeypatch.setattr(tts_module.httpx, "Client", _DummyClient)
|
|
|
|
provider = tts_module.YandexTTSProvider()
|
|
synthesis = provider.synthesize("Kazakh prompt", language="kk")
|
|
|
|
assert synthesis.audio_bytes == b"\x30\x00\x40\x00"
|
|
assert synthesis.sample_rate_hz == 16000
|
|
assert len(calls) == 1
|
|
assert calls[0]["headers"]["Authorization"] == "Bearer iam-test-token"
|
|
assert calls[0]["headers"]["x-folder-id"] == "folder-test"
|
|
assert calls[0]["json"]["hints"][0]["voice"] == "amira"
|
|
assert calls[0]["json"]["hints"][2]["role"] == "friendly"
|
|
assert calls[0]["json"]["outputAudioSpec"]["rawAudio"]["sampleRateHertz"] == 16000
|
|
|
|
|
|
def test_yandex_tts_provider_uses_cached_audio_without_credentials(tmp_path, monkeypatch):
|
|
calls: list[dict] = []
|
|
encoded_audio = base64.b64encode(b"\x55\x00\x66\x00").decode("ascii")
|
|
|
|
class _DummyClient:
|
|
def __init__(self, *, timeout: float) -> None:
|
|
self.timeout = timeout
|
|
|
|
def __enter__(self) -> _DummyClient:
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
def post(self, url: str, *, headers: dict[str, str], json: dict) -> _DummyResponse:
|
|
calls.append({"url": url, "headers": headers, "json": json, "timeout": self.timeout})
|
|
return _DummyResponse(json_payload={"result": {"audioChunk": {"data": encoded_audio}}})
|
|
|
|
monkeypatch.setenv("AI_VOICE_TTS_YANDEX_API_KEY", "yandex-test-key")
|
|
monkeypatch.setenv("AI_VOICE_TTS_CACHE_ENABLED", "1")
|
|
monkeypatch.setenv("AI_VOICE_TTS_CACHE_DIR", str(tmp_path))
|
|
monkeypatch.setattr(tts_module.httpx, "Client", _DummyClient)
|
|
|
|
provider = tts_module.YandexTTSProvider()
|
|
cached = provider.synthesize("Server-side cached yandex prompt", language="ru")
|
|
assert cached.audio_bytes == b"\x55\x00\x66\x00"
|
|
assert len(calls) == 1
|
|
|
|
monkeypatch.delenv("AI_VOICE_TTS_YANDEX_API_KEY", raising=False)
|
|
monkeypatch.delenv("AI_VOICE_TTS_YANDEX_IAM_TOKEN", raising=False)
|
|
|
|
def _unexpected_client(*args, **kwargs):
|
|
raise AssertionError("Yandex TTS should not be called when cached audio already exists")
|
|
|
|
monkeypatch.setattr(tts_module.httpx, "Client", _unexpected_client)
|
|
|
|
provider_without_credentials = tts_module.YandexTTSProvider()
|
|
replay = provider_without_credentials.synthesize("Server-side cached yandex prompt", language="ru")
|
|
|
|
assert replay.audio_bytes == cached.audio_bytes
|
|
assert len(calls) == 1
|
|
|
|
|
|
def test_elevenlabs_tts_provider_posts_voice_id_model_and_language_code(tmp_path, monkeypatch):
|
|
calls: list[dict] = []
|
|
|
|
class _DummyClient:
|
|
def __init__(self, *, timeout: float) -> None:
|
|
self.timeout = timeout
|
|
|
|
def __enter__(self) -> _DummyClient:
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
def post(self, url: str, *, headers: dict[str, str], params: dict[str, str], json: dict) -> _DummyResponse:
|
|
calls.append({"url": url, "headers": headers, "params": params, "json": json, "timeout": self.timeout})
|
|
return _DummyResponse(b"\x01\x00\x02\x00")
|
|
|
|
monkeypatch.setenv("AI_VOICE_TTS_ELEVENLABS_API_KEY", "elevenlabs-key")
|
|
monkeypatch.setenv("AI_VOICE_TTS_CACHE_ENABLED", "0")
|
|
monkeypatch.setattr(tts_module.httpx, "Client", _DummyClient)
|
|
|
|
provider = tts_module.ElevenLabsTTSProvider(
|
|
api_base="https://api.elevenlabs.example",
|
|
ru_voice="nPczCjzI2devNBz1zQrb",
|
|
kz_voice="nPczCjzI2devNBz1zQrb",
|
|
ru_model="eleven_v3",
|
|
kz_model="eleven_v3",
|
|
ru_language_code="ru",
|
|
kz_language_code="kk",
|
|
output_format="pcm_16000",
|
|
)
|
|
synthesis = provider.synthesize("Привет из ElevenLabs", language="kz")
|
|
|
|
assert synthesis.audio_bytes == b"\x01\x00\x02\x00"
|
|
assert synthesis.sample_rate_hz == 16000
|
|
assert len(calls) == 1
|
|
assert calls[0]["url"] == "https://api.elevenlabs.example/v1/text-to-speech/nPczCjzI2devNBz1zQrb"
|
|
assert calls[0]["headers"]["xi-api-key"] == "elevenlabs-key"
|
|
assert calls[0]["params"]["output_format"] == "pcm_16000"
|
|
assert calls[0]["json"]["model_id"] == "eleven_v3"
|
|
assert calls[0]["json"]["language_code"] == "kk"
|
|
|
|
|
|
def test_runtime_configured_tts_provider_uses_database_selected_provider(monkeypatch):
|
|
calls: list[dict] = []
|
|
|
|
class _DummyClient:
|
|
def __init__(self, *, timeout: float) -> None:
|
|
self.timeout = timeout
|
|
|
|
def __enter__(self) -> _DummyClient:
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
def post(self, url: str, *, headers: dict[str, str], params: dict[str, str], json: dict) -> _DummyResponse:
|
|
calls.append({"url": url, "headers": headers, "params": params, "json": json, "timeout": self.timeout})
|
|
return _DummyResponse(b"\x10\x00\x20\x00")
|
|
|
|
monkeypatch.setenv("AI_VOICE_TTS_ELEVENLABS_API_KEY", "elevenlabs-key")
|
|
monkeypatch.setenv("AI_VOICE_TTS_CACHE_ENABLED", "0")
|
|
monkeypatch.setattr(tts_module.httpx, "Client", _DummyClient)
|
|
|
|
session = get_session()
|
|
try:
|
|
payload = voice_tts_default_config().model_dump()
|
|
payload["provider"] = "elevenlabs"
|
|
payload["elevenlabs"]["ru"]["voice"] = "nPczCjzI2devNBz1zQrb"
|
|
payload["elevenlabs"]["kz"]["voice"] = "nPczCjzI2devNBz1zQrb"
|
|
payload["elevenlabs"]["ru"]["model_id"] = "eleven_v3"
|
|
payload["elevenlabs"]["kz"]["model_id"] = "eleven_v3"
|
|
save_voice_tts_config(session, payload)
|
|
finally:
|
|
session.close()
|
|
|
|
provider = RuntimeConfiguredTTSProvider(default_provider_name="yandex")
|
|
synthesis = provider.synthesize("Сәлем", language="kz")
|
|
|
|
assert provider.current_provider_name() == "elevenlabs"
|
|
assert synthesis.audio_bytes == b"\x10\x00\x20\x00"
|
|
assert len(calls) == 1
|
|
assert calls[0]["json"]["language_code"] == "kk"
|