221 lines
9.0 KiB
Python
221 lines
9.0 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
from pathlib import Path
|
|
|
|
from services.ai_voice_runtime_service.providers import tts as tts_module
|
|
|
|
|
|
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_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
|