feat: add call history API endpoints and TTS service client
Adds read-only access to persisted call records for the dashboard and implements a client for the Rhema text-to-speech service. - api/call_history.py: New router providing paged call lists and detailed call records with transcript metadata. - services/tts.py: Async client for OpenAI-compatible TTS endpoints (Rhema/Kokoro) used for call-flow steps.
This commit is contained in:
72
tests/test_tts.py
Normal file
72
tests/test_tts.py
Normal file
@@ -0,0 +1,72 @@
|
||||
"""Tests for the Rhema TTS client (services/tts.py)."""
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from config import TTSSettings
|
||||
from services.tts import TTSService
|
||||
|
||||
|
||||
def _settings(**overrides) -> TTSSettings:
|
||||
defaults = {
|
||||
"base_url": "http://localhost:9000",
|
||||
"model": "speaches-ai/Kokoro-82M-v1.0-ONNX",
|
||||
"voice": "af_heart",
|
||||
"api_key": "",
|
||||
"timeout": 5.0,
|
||||
}
|
||||
defaults.update(overrides)
|
||||
return TTSSettings(**defaults)
|
||||
|
||||
|
||||
class TestTTSService:
|
||||
@pytest.mark.asyncio
|
||||
async def test_synthesize_sends_expected_body(self):
|
||||
svc = TTSService(_settings())
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.content = b"RIFFfake-wav-bytes"
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client.is_closed = False
|
||||
mock_client.post = AsyncMock(return_value=mock_response)
|
||||
|
||||
with patch.object(svc, "_get_client", AsyncMock(return_value=mock_client)):
|
||||
audio = await svc.synthesize("hello world", voice="am_michael")
|
||||
|
||||
assert audio == b"RIFFfake-wav-bytes"
|
||||
mock_client.post.assert_awaited_once()
|
||||
call_args = mock_client.post.call_args
|
||||
assert call_args.args[0] == "/v1/audio/speech"
|
||||
body = call_args.kwargs["json"]
|
||||
assert body["input"] == "hello world"
|
||||
assert body["voice"] == "am_michael"
|
||||
assert body["response_format"] == "wav"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_synthesize_empty_text_returns_empty(self):
|
||||
svc = TTSService(_settings())
|
||||
assert await svc.synthesize("") == b""
|
||||
assert await svc.synthesize(" ") == b""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_synthesize_to_file_writes_audio(self, tmp_path):
|
||||
svc = TTSService(_settings())
|
||||
target = tmp_path / "out.wav"
|
||||
|
||||
with patch.object(svc, "synthesize", AsyncMock(return_value=b"WAVE-bytes")):
|
||||
ok = await svc.synthesize_to_file("hello", str(target))
|
||||
|
||||
assert ok is True
|
||||
assert target.read_bytes() == b"WAVE-bytes"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bearer_header_used_when_api_key_set(self):
|
||||
svc = TTSService(_settings(api_key="secret-token"))
|
||||
client = await svc._get_client()
|
||||
try:
|
||||
assert client.headers.get("authorization") == "Bearer secret-token"
|
||||
finally:
|
||||
await svc.close()
|
||||
Reference in New Issue
Block a user