feat: scaffold stentor-gateway with FastAPI voice pipeline
Initialize the stentor-gateway project with WebSocket-based voice pipeline orchestrating STT → Agent → TTS via OpenAI-compatible APIs. - Add FastAPI app with WebSocket endpoint for audio streaming - Add pipeline orchestration (stt_client, tts_client, agent_client) - Add Pydantic Settings configuration and message models - Add audio utilities for PCM/WAV conversion and resampling - Add health check endpoints - Add Dockerfile and pyproject.toml with dependencies - Add initial test suite (pipeline, STT, TTS, WebSocket) - Add comprehensive README covering gateway and ESP32 ear design - Clean up .gitignore for Python/uv project
This commit is contained in:
78
stentor-gateway/tests/test_tts_client.py
Normal file
78
stentor-gateway/tests/test_tts_client.py
Normal file
@@ -0,0 +1,78 @@
|
||||
"""Tests for the Speaches TTS client."""
|
||||
|
||||
import struct
|
||||
|
||||
import httpx
|
||||
|
||||
from stentor.tts_client import TTSClient
|
||||
|
||||
|
||||
class TestTTSClient:
|
||||
"""Tests for TTSClient."""
|
||||
|
||||
async def test_synthesize_success(self, settings, httpx_mock):
|
||||
"""Test successful TTS synthesis."""
|
||||
# Generate fake PCM audio (100 samples of silence)
|
||||
fake_pcm = struct.pack("<100h", *([0] * 100))
|
||||
|
||||
httpx_mock.add_response(
|
||||
url=f"{settings.tts_url}/v1/audio/speech",
|
||||
method="POST",
|
||||
content=fake_pcm,
|
||||
)
|
||||
|
||||
async with httpx.AsyncClient() as http_client:
|
||||
client = TTSClient(settings, http_client)
|
||||
result = await client.synthesize("Hello world")
|
||||
|
||||
assert result == fake_pcm
|
||||
|
||||
async def test_synthesize_uses_correct_params(self, settings, httpx_mock):
|
||||
"""Test that TTS requests include correct parameters."""
|
||||
httpx_mock.add_response(
|
||||
url=f"{settings.tts_url}/v1/audio/speech",
|
||||
method="POST",
|
||||
content=b"\x00\x00",
|
||||
)
|
||||
|
||||
async with httpx.AsyncClient() as http_client:
|
||||
client = TTSClient(settings, http_client)
|
||||
await client.synthesize("Test text")
|
||||
|
||||
request = httpx_mock.get_request()
|
||||
assert request is not None
|
||||
|
||||
import json
|
||||
body = json.loads(request.content)
|
||||
assert body["model"] == settings.tts_model
|
||||
assert body["voice"] == settings.tts_voice
|
||||
assert body["input"] == "Test text"
|
||||
assert body["response_format"] == "pcm"
|
||||
assert body["speed"] == 1.0
|
||||
|
||||
async def test_is_available_success(self, settings, httpx_mock):
|
||||
"""Test availability check when service is up."""
|
||||
httpx_mock.add_response(
|
||||
url=f"{settings.tts_url}/v1/models",
|
||||
method="GET",
|
||||
json={"models": []},
|
||||
)
|
||||
|
||||
async with httpx.AsyncClient() as http_client:
|
||||
client = TTSClient(settings, http_client)
|
||||
available = await client.is_available()
|
||||
|
||||
assert available is True
|
||||
|
||||
async def test_is_available_failure(self, settings, httpx_mock):
|
||||
"""Test availability check when service is down."""
|
||||
httpx_mock.add_exception(
|
||||
httpx.ConnectError("Connection refused"),
|
||||
url=f"{settings.tts_url}/v1/models",
|
||||
)
|
||||
|
||||
async with httpx.AsyncClient() as http_client:
|
||||
client = TTSClient(settings, http_client)
|
||||
available = await client.is_available()
|
||||
|
||||
assert available is False
|
||||
Reference in New Issue
Block a user