"""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, HIGGS_TTS_ENDPOINT, HIGGS_TTS_MODEL, VoicePreset, VoiceService, assess_end_of_turn, build_higgs_prompt, build_tts_payload, resolve_voice, resolve_voice_from_map, ) 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) def test_openai_persona_voice_map_overrides_live_tts_fields(self) -> None: voice = resolve_voice_from_map( provider="openai", voice_id="voice-p1-custom", persona_code="P1", base_params={ "preset": "soft-young-fem", "openai_voice": "nova", "rate": 1.14, "instructions": "Keep the voice quiet and hesitant.", }, ) self.assertIsNotNone(voice) assert voice is not None self.assertEqual(voice.preset, "soft-young-fem") self.assertEqual(voice.openai_voice, "nova") self.assertAlmostEqual(voice.rate, 1.14) self.assertEqual(voice.instructions, "Keep the voice quiet and hesitant.") def test_openai_persona_voice_map_can_use_voice_id_as_openai_voice(self) -> None: voice = resolve_voice_from_map( provider="openai", voice_id="verse", persona_code="P2", base_params={"preset": "calm-adult-male"}, ) self.assertIsNotNone(voice) assert voice is not None self.assertEqual(voice.preset, "calm-adult-male") self.assertEqual(voice.openai_voice, "verse") def test_non_openai_persona_voice_map_returns_none_for_safe_fallback(self) -> None: self.assertIsNone( resolve_voice_from_map( provider="higgs", voice_id="p1-synthetic", persona_code="P1", base_params={"openai_voice": "coral"}, ) ) class TTSPayloadTest(unittest.TestCase): def test_higgs_prompt_keeps_first_word_before_emotion_tags(self) -> None: voice = VoicePreset(preset="soft-young-fem", openai_voice="coral", rate=0.96) prompt = build_higgs_prompt("그냥 학교 가도 아무 의미 없는 것 같아요.", voice) self.assertTrue(prompt.startswith("그냥 ")) self.assertIn("<|emotion:helplessness|>", prompt) self.assertIn("<|prosody:speed_slow|>", prompt) self.assertNotIn("<|emotion:helplessness|> ", prompt) 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_higgs_tts_uses_local_synthetic_voice_server_in_dev(self) -> None: client = _CaptureTTSClient([b"RIFF", b"synthetic-wav"]) service = VoiceService( api_key="", environment="dev", tts_provider="higgs", higgs_base_url="http://127.0.0.1:9881", ) service._higgs_client = client # type: ignore[assignment] voice = VoicePreset(preset="soft-young-fem", openai_voice="coral") chunks = [ chunk async for chunk in service.synthesize_stream( "엄마한테 말하지 않는 거죠?", voice, ) ] self.assertFalse(service.stt_available()) self.assertTrue(service.tts_available(voice)) self.assertEqual(service.tts_provider_for_voice(voice), "higgs") self.assertEqual(service.tts_model_for_voice(voice), HIGGS_TTS_MODEL) self.assertEqual(service.tts_media_type_for_voice(voice), "audio/wav") self.assertEqual(len(client.calls), 1) method, endpoint, payload = client.calls[0] self.assertEqual((method, endpoint), ("POST", HIGGS_TTS_ENDPOINT)) self.assertIn("<|emotion:fear|>", str(payload["text"])) self.assertEqual(payload["preset"], "soft-young-fem") self.assertEqual(b"".join(chunk.audio for chunk in chunks), b"RIFFsynthetic-wav") async def test_higgs_tts_is_fail_closed_outside_dev(self) -> None: service = VoiceService( api_key="", environment="prod", tts_provider="higgs", ) voice = VoicePreset(preset="soft-young-fem", openai_voice="coral") self.assertFalse(service.tts_available(voice)) self.assertEqual(service.tts_provider(), "disabled-non-dev") 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()