268 lines
9.3 KiB
Python
268 lines
9.3 KiB
Python
"""Deterministic tests for voice preset, TTS payload, and EOT helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
|
|
from .services.voice import (
|
|
DEFAULT_OPENAI_VOICE,
|
|
EOT_SILENCE_THRESHOLD_MS,
|
|
TTS_ENDPOINT,
|
|
TTS_MODEL,
|
|
TTS_MODEL_FALLBACK,
|
|
VoicePreset,
|
|
VoiceService,
|
|
assess_end_of_turn,
|
|
build_tts_payload,
|
|
resolve_voice,
|
|
)
|
|
|
|
|
|
class VoicePresetResolutionTest(unittest.TestCase):
|
|
def test_persona_codes_resolve_distinct_voice_presets(self) -> None:
|
|
expected = {
|
|
"P1": ("soft-young-fem", "coral", 0.96),
|
|
"P2": ("calm-adult-male", "ash", 1.0),
|
|
"P3": ("warm-adult-fem", "shimmer", 0.98),
|
|
}
|
|
|
|
resolved = {code: resolve_voice(persona_code=code) for code in expected}
|
|
|
|
self.assertEqual(
|
|
{voice.openai_voice for voice in resolved.values()},
|
|
{"coral", "ash", "shimmer"},
|
|
)
|
|
for code, (preset, openai_voice, rate) in expected.items():
|
|
with self.subTest(code=code):
|
|
voice = resolved[code]
|
|
self.assertEqual(voice.preset, preset)
|
|
self.assertEqual(voice.openai_voice, openai_voice)
|
|
self.assertAlmostEqual(voice.rate, rate)
|
|
|
|
self.assertEqual(resolve_voice(persona_code="p2").preset, "calm-adult-male")
|
|
|
|
def test_invalid_explicit_preset_falls_back_to_default_openai_voice(self) -> None:
|
|
voice = resolve_voice(persona_code="P1", preset="not-a-real-preset")
|
|
|
|
self.assertEqual(voice.preset, "not-a-real-preset")
|
|
self.assertEqual(voice.openai_voice, DEFAULT_OPENAI_VOICE)
|
|
self.assertEqual(voice.rate, 1.0)
|
|
|
|
|
|
class TTSPayloadTest(unittest.TestCase):
|
|
def test_payload_contains_openai_tts_fields_and_clamps_high_speed(self) -> None:
|
|
voice = VoicePreset(
|
|
preset="soft-young-fem",
|
|
openai_voice="coral",
|
|
rate=9.5,
|
|
instructions="Speak gently with low intensity.",
|
|
)
|
|
|
|
payload = build_tts_payload("Client reply", voice)
|
|
|
|
self.assertEqual(
|
|
payload,
|
|
{
|
|
"model": TTS_MODEL,
|
|
"voice": "coral",
|
|
"input": "Client reply",
|
|
"response_format": "mp3",
|
|
"speed": 4.0,
|
|
"instructions": "Speak gently with low intensity.",
|
|
},
|
|
)
|
|
|
|
def test_payload_omits_instructions_for_fallback_model_and_clamps_low_speed(self) -> None:
|
|
voice = VoicePreset(
|
|
preset="neutral",
|
|
openai_voice="sage",
|
|
rate=0.1,
|
|
instructions="This should not be sent to tts-1.",
|
|
)
|
|
|
|
payload = build_tts_payload(
|
|
"Fallback reply",
|
|
voice,
|
|
model=TTS_MODEL_FALLBACK,
|
|
response_format="opus",
|
|
)
|
|
|
|
self.assertEqual(payload["model"], TTS_MODEL_FALLBACK)
|
|
self.assertEqual(payload["voice"], "sage")
|
|
self.assertEqual(payload["input"], "Fallback reply")
|
|
self.assertEqual(payload["response_format"], "opus")
|
|
self.assertEqual(payload["speed"], 0.25)
|
|
self.assertNotIn("instructions", payload)
|
|
|
|
def test_payload_uses_default_speed_for_invalid_rate(self) -> None:
|
|
voice = VoicePreset(
|
|
preset="neutral",
|
|
openai_voice="sage",
|
|
rate="fast", # type: ignore[arg-type]
|
|
)
|
|
|
|
payload = build_tts_payload("Client reply", voice)
|
|
|
|
self.assertEqual(payload["speed"], 1.0)
|
|
|
|
|
|
class _FakeTTSStream:
|
|
def __init__(self, chunks: list[bytes]) -> None:
|
|
self.status_code = 200
|
|
self._chunks = chunks
|
|
self.closed = False
|
|
|
|
async def __aenter__(self) -> "_FakeTTSStream":
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb) -> bool: # noqa: ANN001
|
|
return False
|
|
|
|
async def aclose(self) -> None:
|
|
self.closed = True
|
|
|
|
def raise_for_status(self) -> None:
|
|
return None
|
|
|
|
async def aiter_bytes(self, chunk_size: int = 4096):
|
|
self.chunk_size = chunk_size
|
|
for chunk in self._chunks:
|
|
yield chunk
|
|
|
|
|
|
class _CaptureTTSClient:
|
|
def __init__(self, chunks: list[bytes]) -> None:
|
|
self.chunks = chunks
|
|
self.calls: list[tuple[str, str, dict[str, object]]] = []
|
|
|
|
def stream(self, method: str, endpoint: str, *, json: dict[str, object]) -> _FakeTTSStream:
|
|
self.calls.append((method, endpoint, dict(json)))
|
|
return _FakeTTSStream(self.chunks)
|
|
|
|
|
|
class VoiceServiceStreamTest(unittest.IsolatedAsyncioTestCase):
|
|
async def test_synthesize_stream_uses_payload_with_fake_client(self) -> None:
|
|
client = _CaptureTTSClient([b"\x80\x80", b"\xff\x00"])
|
|
service = VoiceService(api_key="test-key")
|
|
service._client = client # type: ignore[assignment]
|
|
voice = VoicePreset(
|
|
preset="neutral",
|
|
openai_voice="sage",
|
|
rate=0.1,
|
|
instructions="Keep the tone grounded.",
|
|
)
|
|
|
|
chunks = [
|
|
chunk
|
|
async for chunk in service.synthesize_stream(
|
|
"Spoken client reply",
|
|
voice,
|
|
response_format="opus",
|
|
)
|
|
]
|
|
|
|
self.assertEqual(len(client.calls), 1)
|
|
method, endpoint, payload = client.calls[0]
|
|
self.assertEqual(method, "POST")
|
|
self.assertEqual(endpoint, TTS_ENDPOINT)
|
|
self.assertEqual(payload["model"], TTS_MODEL)
|
|
self.assertEqual(payload["voice"], "sage")
|
|
self.assertEqual(payload["input"], "Spoken client reply")
|
|
self.assertEqual(payload["response_format"], "opus")
|
|
self.assertEqual(payload["speed"], 0.25)
|
|
self.assertEqual(payload["instructions"], "Keep the tone grounded.")
|
|
self.assertEqual([chunk.audio for chunk in chunks], [b"\x80\x80", b"\xff\x00"])
|
|
|
|
async def test_dev_p1_sample_tts_streams_local_mp3_without_openai_key(self) -> None:
|
|
with TemporaryDirectory() as tmp:
|
|
sample_dir = Path(tmp)
|
|
default_audio = b"default-mp3"
|
|
anxious_audio = (b"anxious-mp3-" * 500)
|
|
(sample_dir / "p1_seoyeon_01_depressed_slow.mp3").write_bytes(default_audio)
|
|
(sample_dir / "p1_seoyeon_03_anxious_guarded.mp3").write_bytes(anxious_audio)
|
|
service = VoiceService(
|
|
api_key="",
|
|
poc_sample_tts_enabled=True,
|
|
environment="dev",
|
|
poc_sample_tts_dir=sample_dir,
|
|
)
|
|
voice = VoicePreset(preset="soft-young-fem", openai_voice="coral")
|
|
|
|
chunks = [
|
|
chunk
|
|
async for chunk in service.synthesize_stream(
|
|
"엄마한테 말하지 않는 거죠? 진짜 불안해요.",
|
|
voice,
|
|
)
|
|
]
|
|
|
|
self.assertFalse(service.is_available())
|
|
self.assertTrue(service.poc_sample_tts_available())
|
|
self.assertEqual(service.tts_provider(), "p1-sample-poc")
|
|
self.assertEqual(b"".join(chunk.audio for chunk in chunks), anxious_audio)
|
|
|
|
async def test_p1_sample_tts_is_disabled_outside_dev(self) -> None:
|
|
with TemporaryDirectory() as tmp:
|
|
sample_dir = Path(tmp)
|
|
(sample_dir / "p1_seoyeon_01_depressed_slow.mp3").write_bytes(b"default-mp3")
|
|
service = VoiceService(
|
|
api_key="",
|
|
poc_sample_tts_enabled=True,
|
|
environment="prod",
|
|
poc_sample_tts_dir=sample_dir,
|
|
)
|
|
|
|
self.assertFalse(service.is_available())
|
|
self.assertFalse(service.poc_sample_tts_available())
|
|
self.assertEqual(service.tts_provider(), "disabled-non-dev")
|
|
|
|
|
|
class EndOfTurnDecisionTest(unittest.TestCase):
|
|
def test_end_of_turn_requires_silence_threshold(self) -> None:
|
|
pending = assess_end_of_turn(
|
|
transcript_text="I am still talking",
|
|
transcript_final=True,
|
|
silence_ms=EOT_SILENCE_THRESHOLD_MS - 1,
|
|
)
|
|
ready = assess_end_of_turn(
|
|
transcript_text="I am done",
|
|
transcript_final=True,
|
|
silence_ms=EOT_SILENCE_THRESHOLD_MS,
|
|
)
|
|
|
|
self.assertFalse(pending.ready)
|
|
self.assertTrue(pending.transcript_ready)
|
|
self.assertFalse(pending.silence_ready)
|
|
self.assertEqual(pending.reason, "silence_threshold_pending")
|
|
self.assertTrue(ready.ready)
|
|
self.assertEqual(ready.reason, "ready")
|
|
|
|
def test_end_of_turn_waits_for_final_transcript_even_after_silence(self) -> None:
|
|
decision = assess_end_of_turn(
|
|
transcript_text="Interim transcript",
|
|
transcript_final=False,
|
|
silence_ms=2500,
|
|
)
|
|
|
|
self.assertFalse(decision.ready)
|
|
self.assertFalse(decision.transcript_ready)
|
|
self.assertTrue(decision.silence_ready)
|
|
self.assertEqual(decision.reason, "final_transcript_pending")
|
|
|
|
def test_end_of_turn_rejects_empty_final_transcript(self) -> None:
|
|
decision = assess_end_of_turn(
|
|
transcript_text=" ",
|
|
transcript_final=True,
|
|
silence_ms=2500,
|
|
)
|
|
|
|
self.assertFalse(decision.ready)
|
|
self.assertFalse(decision.transcript_ready)
|
|
self.assertTrue(decision.silence_ready)
|
|
self.assertEqual(decision.reason, "empty_transcript")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|