"""Tests for the TTS client wrappers (language handling and payloads).""" import io import json import tempfile import time import urllib.error import unittest import wave from contextlib import redirect_stdout from pathlib import Path from unittest.mock import MagicMock, patch from converter import config from converter import converter as converter_mod from converter.clients import ( AUDIOCPP_DEFAULT_FAMILY_PROFILE, AUDIOCPP_FAMILY_PROFILES, AUDIOCPP_LANG_OMIT, AUDIOCPP_TASK_TTS, AUDIOCPP_TASK_VDES, AUDIOCPP_VOICE_CLONE, AUDIOCPP_VOICE_DESIGN, AUDIOCPP_VOICE_SPEAKER, BACKEND_AUDIOCPP, BACKEND_FASTER, BACKEND_QWEN, LANGUAGE_CHOICES, LANGUAGE_ISO_CODES, MODEL_SIZE, SAMPLE_RATE, TTS_LANGUAGES, VOICE_MODE_CLONE, VOICE_MODE_CUSTOM, VOICE_MODE_DESIGN, VOICE_MODES, AudioCppTTSClient, FasterTTSClient, QwenTTSClient, audiocpp_entry_voice_capability, normalize_language, transcribe_reference_audio_detailed, whisper_backend_problem, ) from converter.clients.base import NonRetryableTTSError from converter.converter import AudiobookConverter # Chunks folder handed to clients whose tests never write chunk files. _DUMMY_CHUNKS = Path(tempfile.gettempdir()) / "audiobook_tts_test_chunks" # A concrete audio.cpp model entry id (no config default anymore): the # tests request it explicitly, the way --model / the Generate form does. _AUDIOCPP_MODEL_ID = "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF" class NormalizeLanguageTests(unittest.TestCase): def test_display_names_case_insensitive(self): self.assertEqual(normalize_language("english"), "English") self.assertEqual(normalize_language("ENGLISH"), "English") self.assertEqual(normalize_language(" Japanese "), "Japanese") def test_auto_accepted(self): self.assertEqual(normalize_language("auto"), "Auto") self.assertEqual(normalize_language("Auto"), "Auto") def test_iso_aliases(self): self.assertEqual(normalize_language("en"), "English") self.assertEqual(normalize_language("ja"), "Japanese") self.assertEqual(normalize_language("zh"), "Chinese") self.assertEqual(normalize_language("ko"), "Korean") self.assertEqual(normalize_language("de"), "German") self.assertEqual(normalize_language("fr"), "French") self.assertEqual(normalize_language("ru"), "Russian") self.assertEqual(normalize_language("pt"), "Portuguese") self.assertEqual(normalize_language("es"), "Spanish") self.assertEqual(normalize_language("it"), "Italian") self.assertEqual(normalize_language("ar"), "Arabic") self.assertEqual(normalize_language("hi"), "Hindi") self.assertEqual(normalize_language("vi"), "Vietnamese") def test_audio_cpp_menu_languages_accepted(self): # audio.cpp's WebUI menus add Arabic, Hindi and Vietnamese; the # MagpieTTS Arabic regional variants collapse to plain Arabic. self.assertEqual(normalize_language("arabic"), "Arabic") self.assertEqual(normalize_language("Hindi"), "Hindi") self.assertEqual(normalize_language("vietnamese"), "Vietnamese") for variant in ("ar-AE", "ar-MSA", "ar-SA"): self.assertEqual(normalize_language(variant), "Arabic") def test_language_choices_are_valid_display_names(self): # The TUI's static picker lists a permutation of TTS_LANGUAGES # (common languages first), so every entry normalizes. self.assertEqual(sorted(LANGUAGE_CHOICES), sorted(TTS_LANGUAGES)) for name in LANGUAGE_CHOICES: self.assertEqual(normalize_language(name), name) def test_all_supported_languages_round_trip(self): for name in TTS_LANGUAGES: self.assertEqual(normalize_language(name.lower()), name) def test_unknown_language_rejected_with_guidance(self): with self.assertRaises(ValueError) as ctx: normalize_language("klingon") message = str(ctx.exception) self.assertIn("klingon", message) self.assertIn("English", message) def test_none_and_empty_rejected(self): with self.assertRaises(ValueError): normalize_language(None) with self.assertRaises(ValueError): normalize_language(" ") class QwenTTSClientLanguageTests(unittest.TestCase): """Language validation and defaults, without touching the network.""" def _make_client(self, **kwargs): with patch.object(QwenTTSClient, "_connect"): return QwenTTSClient(_DUMMY_CHUNKS, **kwargs) def test_default_follows_config_for_each_mode(self): custom = self._make_client(voice_mode=VOICE_MODE_CUSTOM, voice="Vivian") self.assertEqual(custom.language, config.LANGUAGE) clone = self._make_client(voice_mode=VOICE_MODE_CLONE, voice_clone_ref_audio="ref.wav") self.assertEqual(clone.language, config.LANGUAGE) def test_explicit_language_normalized(self): client = self._make_client(voice_mode=VOICE_MODE_CUSTOM, language="ja", voice="Vivian") self.assertEqual(client.language, "Japanese") def test_invalid_language_fails_before_connect(self): with patch.object(QwenTTSClient, "_connect") as mock_connect: with self.assertRaises(ValueError): QwenTTSClient(_DUMMY_CHUNKS, language="klingon") mock_connect.assert_not_called() def test_api_url_override_stored(self): client = self._make_client(voice_mode=VOICE_MODE_CUSTOM, api_url="http://10.0.0.5:7860", voice="Vivian") self.assertEqual(client.api_url, "http://10.0.0.5:7860") def test_api_url_override_used_by_connect(self): with patch.object(QwenTTSClient, "_init_client") as mk_init: client = QwenTTSClient.__new__(QwenTTSClient) client.voice_mode = VOICE_MODE_CUSTOM client.api_url = "http://10.0.0.5:7860" client._connect() mk_init.assert_called_once_with("http://10.0.0.5:7860", clone=False) class SeedResolutionTests(unittest.TestCase): """CONSTANT_SEED: one seed per run, reused for every request, so the voice stays consistent across chunk boundaries (the servers re-sample the voice when the seed changes between generations).""" def _make_client(self, **kwargs): with patch.object(QwenTTSClient, "_connect"): return QwenTTSClient(_DUMMY_CHUNKS, **kwargs) def test_constant_seed_draws_one_nonnegative_seed(self): with patch.object(config, "CONSTANT_SEED", True), \ patch.object(config, "SEED", -1): client = self._make_client(voice_mode=VOICE_MODE_CUSTOM, voice="Vivian") self.assertGreaterEqual(client._seed, 0) def test_explicit_seed_wins_over_constant_seed(self): with patch.object(config, "CONSTANT_SEED", True), \ patch.object(config, "SEED", 42): client = self._make_client(voice_mode=VOICE_MODE_CUSTOM, voice="Vivian") self.assertEqual(client._seed, 42) def test_without_constant_seed_minus_one_is_forwarded(self): with patch.object(config, "CONSTANT_SEED", False), \ patch.object(config, "SEED", -1): client = self._make_client(voice_mode=VOICE_MODE_CUSTOM, voice="Vivian") self.assertEqual(client._seed, -1) def test_resolved_seed_is_reused_across_requests(self): api_info = { "named_endpoints": { "/run_custom_voice": { "parameters": [{"parameter_name": "seed"}] } } } client = QwenTTSClient.__new__(QwenTTSClient) client.voice_mode = VOICE_MODE_CUSTOM client.speaker = "Vivian" client.language = "English" client._seed = 1234 client.api_info = api_info client.client = MagicMock() client._generate_custom_voice("first text") client._generate_custom_voice("second text") seeds = [call.kwargs["seed"] for call in client.client.predict.call_args_list] self.assertEqual(seeds, [1234, 1234]) class PayloadLanguageTests(unittest.TestCase): """The language must reach the API payload in every endpoint variant.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.ref_audio = Path(self._tmp.name) / "reference.wav" self.ref_audio.write_bytes(b"x") def tearDown(self): self._tmp.cleanup() def _custom_client(self, language, endpoint, api_info=None): client = QwenTTSClient.__new__(QwenTTSClient) client.voice_mode = VOICE_MODE_CUSTOM client.speaker = "Vivian" client.language = language client._seed = config.SEED client.api_info = api_info if api_info is not None else { "named_endpoints": {endpoint: {}} } client.client = MagicMock() return client def _clone_client(self, language, endpoint, api_info=None, ref_text="hello"): client = QwenTTSClient.__new__(QwenTTSClient) client.voice_mode = VOICE_MODE_CLONE client.language = language client._seed = config.SEED client.voice_clone_ref_audio = str(self.ref_audio) client.voice_clone_ref_text = ref_text client.clone_api_info = api_info if api_info is not None else { "named_endpoints": {endpoint: {}} } client.clone_client = MagicMock() client._ref_audio_filedata = {"dummy": "payload"} return client def test_custom_voice_run_instruct_uses_language(self): client = self._custom_client("Japanese", "/run_instruct") client._generate_custom_voice("text") kwargs = client.client.predict.call_args.kwargs self.assertEqual(kwargs["lang_disp"], "Japanese") def test_custom_voice_alt_endpoint_uses_language(self): client = self._custom_client("French", "/run_custom_voice") client._generate_custom_voice("text") kwargs = client.client.predict.call_args.kwargs self.assertEqual(kwargs["language"], "French") def test_voice_clone_run_voice_clone_uses_language(self): client = self._clone_client("Japanese", "/run_voice_clone") client._generate_voice_clone("text") kwargs = client.clone_client.predict.call_args.kwargs self.assertEqual(kwargs["lang_disp"], "Japanese") def test_voice_clone_alt_endpoint_uses_language(self): client = self._clone_client("Korean", "/generate_voice_clone") client._generate_voice_clone("text") kwargs = client.clone_client.predict.call_args.kwargs self.assertEqual(kwargs["language"], "Korean") def test_voice_clone_alt_endpoint_includes_optional_params(self): api_info = { "named_endpoints": { "/generate_voice_clone": { "parameters": [ {"parameter_name": "model_size"}, {"parameter_name": "seed"}, ] } } } client = self._clone_client("English", "/generate_voice_clone", api_info=api_info) client._generate_voice_clone("text") kwargs = client.clone_client.predict.call_args.kwargs self.assertEqual(kwargs["model_size"], MODEL_SIZE) self.assertEqual(kwargs["seed"], config.SEED) class FasterTTSClientHealthTests(unittest.TestCase): """Connection behavior of the faster-qwen3-tts client.""" def _health_response(self, model_loaded=True): response = MagicMock() response.__enter__.return_value = response response.read.return_value = json.dumps( {"status": "ok", "model_loaded": model_loaded}).encode("utf-8") return response def test_unreachable_server_raises_with_readme_pointer(self): import urllib.error with patch("converter.clients.faster.urllib.request.urlopen", side_effect=urllib.error.URLError("Connection refused")): with self.assertRaises(RuntimeError) as ctx: FasterTTSClient(_DUMMY_CHUNKS, voice="narrator") message = str(ctx.exception) self.assertIn("not reachable", message) self.assertIn("README", message) def test_model_not_loaded_raises(self): with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._health_response(model_loaded=False)): with self.assertRaises(RuntimeError) as ctx: FasterTTSClient(_DUMMY_CHUNKS, voice="narrator") self.assertIn("not loaded", str(ctx.exception)) def test_missing_voice_raises_before_connecting(self): # There is no configured default voice: a faster run names its # voice per run (the server silently falls back when the key is # not in its voices.json). with patch("converter.clients.faster.urllib.request.urlopen") \ as mock_urlopen: with self.assertRaises(RuntimeError) as ctx: FasterTTSClient(_DUMMY_CHUNKS) self.assertIn("requires a voice", str(ctx.exception)) mock_urlopen.assert_not_called() def test_healthy_server_uses_the_requested_voice(self): with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._health_response()): client = FasterTTSClient(_DUMMY_CHUNKS, voice="narrator") self.assertEqual(client.voice, "narrator") self.assertEqual(client.api_url, config.FASTER_API_URL.rstrip("/")) def test_explicit_voice_and_url_override_config(self): with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._health_response()): client = FasterTTSClient(_DUMMY_CHUNKS, voice="narrator", api_url="http://10.0.0.5:9000/") self.assertEqual(client.voice, "narrator") self.assertEqual(client.api_url, "http://10.0.0.5:9000") class FasterTTSClientGenerateTests(unittest.TestCase): """Chunk generation: sub-chunking, WAV output, retries, bookkeeping.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self._sleep = patch("converter.clients.base.time.sleep") self._sleep.start() def tearDown(self): self._sleep.stop() self._tmp.cleanup() def _make_client(self): client = FasterTTSClient.__new__(FasterTTSClient) client.chunks_dir = Path(self._tmp.name) client.voice = "default" client.api_url = "http://127.0.0.1:8000" return client def _read_wav(self, path): with wave.open(str(path), "rb") as wav_file: return (wav_file.getnchannels(), wav_file.getsampwidth(), wav_file.getframerate(), wav_file.readframes(wav_file.getnframes())) def test_generate_chunk_writes_valid_wav(self): client = self._make_client() pcm = b"\x01\x00" * 100 with patch.object(client, "_request_pcm", return_value=pcm): result = client.generate_chunk("Hello world.", 1) self.assertIsNotNone(result) path = Path(result) self.assertEqual(path.name, "chunk_0001.wav") channels, sampwidth, framerate, frames = self._read_wav(path) self.assertEqual(channels, 1) self.assertEqual(sampwidth, 2) self.assertEqual(framerate, SAMPLE_RATE) self.assertEqual(frames, pcm) def test_long_text_is_subchunked_and_concatenated_in_order(self): client = self._make_client() sentences = [" ".join(f"word{i}" for i in range(6)) + "." for _ in range(3)] text = " ".join(sentences) pcm_parts = [b"\x01\x00" * 10, b"\x02\x00" * 20, b"\x03\x00" * 30] with patch.object(config, "CHUNK_SIZE", 10), \ patch.object(client, "_request_pcm", side_effect=pcm_parts) as mock_pcm: result = client.generate_chunk(text, 1) self.assertEqual(mock_pcm.call_count, 3) _, _, _, frames = self._read_wav(Path(result)) self.assertEqual(frames, b"".join(pcm_parts)) def test_subchunk_size_follows_config_chunk_size(self): client = self._make_client() text = " ".join(f"word{i}" for i in range(8)) pcm = b"\x01\x00" * 10 with patch.object(config, "CHUNK_SIZE", 4), \ patch.object(client, "_request_pcm", return_value=pcm) as mock_pcm: result = client.generate_chunk(text, 1) # The sub-chunk split follows config.CHUNK_SIZE, so the whole # (8-word) text needs two 4-word requests here. self.assertEqual(mock_pcm.call_count, 2) self.assertIsNotNone(result) def test_stale_chunk_files_are_removed(self): stale = Path(self._tmp.name) / "chunk_0001.mp3" stale.write_bytes(b"old") client = self._make_client() with patch.object(client, "_request_pcm", return_value=b"\x01\x00"): client.generate_chunk("Hello.", 1) remaining = sorted(path.name for path in Path(self._tmp.name).glob("chunk_0001.*")) self.assertEqual(remaining, ["chunk_0001.wav"]) def test_transient_failure_fails_the_chunk_attempt(self): # Retrying is the chunk-level policy's job # (process_chunk_with_retry); one generate_chunk call makes one # request attempt per sub-chunk. client = self._make_client() pcm = b"\x01\x00" * 10 with patch.object(client, "_request_pcm", side_effect=[RuntimeError("boom"), pcm]) as mock_pcm: result = client.generate_chunk("Hello.", 1) self.assertIsNone(result) self.assertEqual(mock_pcm.call_count, 1) def test_empty_pcm_response_fails_the_chunk(self): client = self._make_client() def _response(body): response = MagicMock() response.__enter__.return_value = response response.read.return_value = body return response with patch("converter.clients.faster.urllib.request.urlopen", side_effect=[_response(b"")]) as mock_urlopen: result = client.generate_chunk("Hello.", 1) self.assertIsNone(result) self.assertEqual(mock_urlopen.call_count, 1) def test_subchunk_request_failure_fails_the_chunk(self): client = self._make_client() with patch.object(client, "_request_pcm", side_effect=RuntimeError("down")) as mock_pcm: result = client.generate_chunk("Hello.", 1) self.assertIsNone(result) self.assertEqual(mock_pcm.call_count, 1) def test_empty_text_fails_the_chunk(self): client = self._make_client() with patch.object(client, "_request_pcm") as mock_pcm: result = client.generate_chunk(" ", 1) self.assertIsNone(result) mock_pcm.assert_not_called() def test_request_payload_includes_voice_text_and_format(self): client = self._make_client() response = MagicMock() response.__enter__.return_value = response response.read.return_value = b"\x01\x00" * 10 with patch("converter.clients.faster.urllib.request.urlopen", return_value=response) as mock_urlopen: pcm = client._request_pcm("Hello world.") self.assertEqual(pcm, b"\x01\x00" * 10) request = mock_urlopen.call_args[0][0] self.assertEqual(request.full_url, "http://127.0.0.1:8000/v1/audio/speech") payload = json.loads(request.data.decode("utf-8")) self.assertEqual(payload["input"], "Hello world.") self.assertEqual(payload["voice"], "default") self.assertEqual(payload["response_format"], "pcm") def test_full_length_pcm_passes(self): client = self._make_client() text = " ".join(f"word{i}" for i in range(12)) # 12 words -> expected 4.8s, half is 2.4s -> 2.5s of audio passes. pcm = b"\x01\x00" * int(2.5 * SAMPLE_RATE) with patch.object(client, "_request_pcm", return_value=pcm): result = client.generate_chunk(text, 1) self.assertIsNotNone(result) class QwenTTSClientGenerateTests(unittest.TestCase): """Qwen chunk generation: sub-request splitting and concatenation.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() def tearDown(self): self._tmp.cleanup() def _make_client(self): client = QwenTTSClient.__new__(QwenTTSClient) client.chunks_dir = Path(self._tmp.name) client.voice_mode = VOICE_MODE_CUSTOM return client @staticmethod def _write_wav(path: Path, frames: bytes) -> Path: with wave.open(str(path), "wb") as wav_file: wav_file.setnchannels(1) wav_file.setsampwidth(2) wav_file.setframerate(SAMPLE_RATE) wav_file.writeframes(frames) return path def _read_wav_frames(self, path: Path) -> bytes: with wave.open(str(path), "rb") as wav_file: return wav_file.readframes(wav_file.getnframes()) def test_single_request_copies_audio(self): client = self._make_client() source = self._write_wav(Path(self._tmp.name) / "server.wav", b"\x01\x00" * 50) with patch.object(client, "_generate_custom_voice", return_value=(str(source),)) as mock_generate: result = client.generate_chunk("Hello world.", 1) mock_generate.assert_called_once_with("Hello world.") path = Path(result) self.assertEqual(path.name, "chunk_0001.wav") self.assertEqual(self._read_wav_frames(path), b"\x01\x00" * 50) def test_oversized_input_is_split_and_concatenated_in_order(self): client = self._make_client() first = self._write_wav(Path(self._tmp.name) / "one.wav", b"\x01\x00" * 10) second = self._write_wav(Path(self._tmp.name) / "two.wav", b"\x02\x00" * 20) text = " ".join(f"word{i}" for i in range(12)) with patch.object(config, "CHUNK_SIZE", 5), \ patch.object(client, "_generate_custom_voice", side_effect=[(str(first),), (str(second),), (str(first),)]) as mock_generate: result = client.generate_chunk(text, 1) self.assertEqual(mock_generate.call_count, 3) path = Path(result) self.assertEqual(path.name, "chunk_0001.wav") self.assertEqual(self._read_wav_frames(path), b"\x01\x00" * 10 + b"\x02\x00" * 20 + b"\x01\x00" * 10) for call in mock_generate.call_args_list: self.assertLessEqual(len(call[0][0].split()), 5) def test_empty_text_fails_the_chunk(self): client = self._make_client() with patch.object(client, "_generate_custom_voice") as mock_generate: result = client.generate_chunk(" ", 1) self.assertIsNone(result) mock_generate.assert_not_called() class QwenTTSClientVoiceDesignTests(unittest.TestCase): """Qwen VoiceDesign mode: instructions and the /run_voice_design call.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() def tearDown(self): self._tmp.cleanup() def _client(self, instructions=None): client = QwenTTSClient.__new__(QwenTTSClient) client.chunks_dir = Path(self._tmp.name) client.voice_mode = VOICE_MODE_DESIGN client.language = config.LANGUAGE client.instructions = (instructions or "").strip() client.api_info = {"named_endpoints": {"/run_voice_design": { "parameters": [ {"parameter_name": "text"}, {"parameter_name": "lang_disp"}, {"parameter_name": "design"}, ]}}} return client def _fake_output(self) -> str: out = Path(self._tmp.name) / "server_out.wav" out.write_bytes(b"\x01\x00") return str(out) def test_voice_mode_design_is_valid(self): self.assertIn(VOICE_MODE_DESIGN, VOICE_MODES) def test_generate_payload_and_return(self): client = self._client(instructions="A warm narrator") fake = MagicMock(return_value=(self._fake_output(),)) with patch.object(client, "_generate_voice_design", fake): result = client._generate_sub_request( "Hello there.", self._tmp.name, 1, 1, 1) fake.assert_called_once_with("Hello there.") self.assertEqual(Path(result).name, "part_01.wav") def test_payload_uses_design_field_language_and_instruction(self): client = self._client(instructions="A warm narrator") captured = {} def fake_predict(**payload): captured.update(payload) return (self._fake_output(),) client.client = MagicMock() client.client.predict.side_effect = fake_predict result = client._generate_voice_design("Hi.") self.assertEqual(captured["text"], "Hi.") self.assertEqual(captured["lang_disp"], config.LANGUAGE) self.assertEqual(captured["design"], "A warm narrator") self.assertNotIn("seed", captured) # not accepted by this endpoint self.assertEqual(result, (self._fake_output(),)) def test_payload_uses_empty_design_field_when_no_instructions_given(self): # There is no configured default instruction: the client sends # whatever the run provided (empty when none). client = self._client(instructions=None) self.assertEqual(client.instructions, "") def test_unknown_api_falls_back_to_the_requested_name(self): client = self._client() client.api_info = {"named_endpoints": {}} client.client = MagicMock() client.client.predict.return_value = (self._fake_output(),) client._generate_voice_design("Hi.") _, kwargs = client.client.predict.call_args self.assertEqual(kwargs["api_name"], "/run_voice_design") class AudioCppTTSClientHealthTests(unittest.TestCase): """Connection behavior of the audio.cpp client.""" @staticmethod def _json_response(payload): response = MagicMock() response.__enter__.return_value = response response.read.return_value = json.dumps(payload).encode("utf-8") return response def _get_responses(self, health=None, models=None, voices=None): """Side effect dispatching GET responses by URL.""" def _dispatch(request, **_kwargs): url = request if isinstance(request, str) else request.full_url if url.endswith("/health"): return self._json_response(health if health is not None else {"status": "ok"}) if url.endswith("/v1/models"): return self._json_response(models if models is not None else {"data": [{"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_tts"}]}) if "/v1/audio/voices" in url: if voices is Exception: raise Exception("voices endpoint down") return self._json_response(voices if voices is not None else {"voices": ["narrator"]}) raise AssertionError(f"unexpected URL: {url}") return _dispatch def _client(self, voice=None, language=None, model_id=_AUDIOCPP_MODEL_ID, **kwargs): with patch("converter.clients.faster.urllib.request.urlopen", side_effect=self._get_responses(**kwargs)): return AudioCppTTSClient(_DUMMY_CHUNKS, voice=voice, language=language, model_id=model_id) def test_unreachable_server_raises_with_readme_pointer(self): import urllib.error with patch("converter.clients.faster.urllib.request.urlopen", side_effect=urllib.error.URLError("Connection refused")): with self.assertRaises(RuntimeError) as ctx: AudioCppTTSClient(_DUMMY_CHUNKS) message = str(ctx.exception) self.assertIn("not reachable", message) self.assertIn("README", message) def test_unhealthy_status_raises(self): with self.assertRaises(RuntimeError) as ctx: self._client(health={"status": "starting"}) self.assertIn("starting", str(ctx.exception)) def test_unknown_model_id_raises_with_configured_ids(self): with self.assertRaises(RuntimeError) as ctx: self._client(models={"data": [{"id": "pocket-tts"}, {"id": "other"}]}) message = str(ctx.exception) self.assertIn(_AUDIOCPP_MODEL_ID, message) self.assertIn("pocket-tts", message) self.assertIn("other", message) def test_healthy_server_speaker_mode_defaults(self): client = self._client(voice="Vivian") self.assertEqual(client.api_url, config.AUDIOCPP_API_URL.rstrip("/")) self.assertEqual(client.model_id, _AUDIOCPP_MODEL_ID) self.assertEqual(client.language, config.LANGUAGE) self.assertEqual(client.voice, "Vivian") self.assertFalse(client.preset_mode) self.assertTrue(client.speaker_mode) def test_speaker_mode_normalizes_the_speaker_name(self): client = self._client(voice="uncle_fu") self.assertEqual(client.voice, "Uncle Fu") self.assertTrue(client.speaker_mode) def test_no_voice_on_speaker_entry_raises(self): # There is no configured default speaker: a CustomVoice entry # without --voice fails fast instead of guessing one. with self.assertRaises(RuntimeError) as ctx: self._client() message = str(ctx.exception) self.assertIn("built-in speakers", message) self.assertIn("--voice", message) def test_voice_speaker_name_selects_speaker_mode(self): # --voice naming a built-in CustomVoice speaker selects speaker # mode; the name is normalized to its wire (display) form and no # preset validation runs. client = self._client(voice="Uncle_Fu") self.assertEqual(client.voice, "Uncle Fu") self.assertFalse(client.preset_mode) self.assertTrue(client.speaker_mode) def test_voice_speaker_name_on_clone_entry_is_a_preset(self): # --voice on a clone-only (Base) entry is a server-side preset, # not a built-in speaker, so the name is validated against the # server's voice library. with self.assertRaises(RuntimeError) as ctx: self._client(voice="Ryan", model_id="Qwen3-TTS-12Hz-1.7B-Base-GGUF", models={"data": [ {"id": "Qwen3-TTS-12Hz-1.7B-Base-GGUF", "family": "qwen3_tts"}]}) message = str(ctx.exception) self.assertIn("'Ryan'", message) self.assertIn("--voice", message) def test_speaker_mode_stays_on_the_selected_entry(self): # A built-in speaker name selects speaker mode on the entry the # run picked; no second-entry rerouting exists anymore. client = self._client( voice="Ryan", model_id="Qwen3-TTS-CustomVoice", models={"data": [{"id": "Qwen3-TTS-CustomVoice", "family": "qwen3_tts"}, {"id": "qwen3-tts-clone", "family": "qwen3_tts"}]}) self.assertEqual(client.model_id, "Qwen3-TTS-CustomVoice") self.assertTrue(client.speaker_mode) self.assertFalse(client.preset_mode) def test_no_voice_on_base_entry_raises_instead_of_silent_speaker(self): # The Base model has no built-in speakers: without --voice the run # fails fast instead of silently sending a speaker name that the # model ignores. with self.assertRaises(RuntimeError) as ctx: self._client(model_id="Qwen3-TTS-12Hz-1.7B-Base-GGUF", models={"data": [ {"id": "Qwen3-TTS-12Hz-1.7B-Base-GGUF", "family": "qwen3_tts"}]}) message = str(ctx.exception) self.assertIn("Base-GGUF", message) self.assertIn("--voice", message) def test_preset_mode_uses_requested_voice(self): client = self._client(voice="narrator") self.assertEqual(client.voice, "narrator") self.assertTrue(client.preset_mode) def test_preset_mode_validates_voice_against_server_list(self): with self.assertRaises(RuntimeError) as ctx: self._client(voice="ghost", voices={"voices": ["narrator", "obama"]}) message = str(ctx.exception) self.assertIn("ghost", message) self.assertIn("narrator", message) self.assertIn("obama", message) def test_preset_mode_skips_validation_when_voices_endpoint_fails(self): client = self._client(voice="narrator", voices=Exception) self.assertEqual(client.voice, "narrator") def test_invalid_language_fails_before_connect(self): with patch("converter.clients.faster.urllib.request.urlopen") as mock_urlopen: with self.assertRaises(ValueError): AudioCppTTSClient(_DUMMY_CHUNKS, language="klingon") mock_urlopen.assert_not_called() def test_explicit_language_normalized(self): client = self._client(language="ja", voice="Vivian") self.assertEqual(client.language, "Japanese") def test_seed_resolved_once_per_run(self): with patch.object(config, "CONSTANT_SEED", True), \ patch.object(config, "SEED", -1): client = self._client(voice="Vivian") self.assertGreaterEqual(client._seed, 0) def test_preset_mode_stays_on_the_requested_entry(self): # Preset (cloning) requests synthesize with the entry the run # selected; pick the Base entry with --model to clone on it. client = self._client( voice="narrator", model_id="qwen3-tts", models={"data": [{"id": "qwen3-tts"}, {"id": "qwen3-tts-clone"}]}) self.assertEqual(client.model_id, "qwen3-tts") self.assertTrue(client.preset_mode) def test_empty_model_id_auto_picks_single_server_entry(self): # A multi-model server used without editing config.py: an empty # --model resolves to the only hosted entry automatically. client = self._client( voice="narrator", model_id="", models={"data": [{"id": "higgs", "family": "higgs_audio_tts"}]}, voices={"voices": ["narrator"]}) self.assertEqual(client.model_id, "higgs") def test_empty_model_id_with_multiple_entries_requires_explicit_choice(self): with self.assertRaises(RuntimeError) as ctx: self._client( voice="narrator", model_id="", models={"data": [{"id": "higgs"}, {"id": "voxcpm2"}]}, voices={"voices": ["narrator"]}) message = str(ctx.exception) self.assertIn("--model", message) self.assertIn("higgs", message) self.assertIn("voxcpm2", message) def test_model_id_reaches_request(self): # The per-run --model value is what the client requests. client = self._client( voice="narrator", model_id="higgs", models={"data": [{"id": "higgs", "family": "higgs_audio_tts"}]}, voices={"voices": ["narrator"]}) self.assertEqual(client.model_id, "higgs") def test_preset_mode_on_a_single_clone_entry_server(self): # A server hosting only the Base (cloning) entry: select it with # --model and a preset voice works. client = self._client( voice="narrator", model_id="qwen3-tts-clone", models={"data": [{"id": "qwen3-tts-clone"}]}) self.assertEqual(client.model_id, "qwen3-tts-clone") self.assertTrue(client.preset_mode) def test_unknown_model_id_error_suggests_a_model(self): # Requesting an id the server does not host fails fast and names # both the requested and the hosted ids. with self.assertRaises(RuntimeError) as ctx: self._client(voice="narrator", model_id="qwen3-tts", models={"data": [{"id": "qwen3-tts-clone"}]}) message = str(ctx.exception) self.assertIn("qwen3-tts", message) self.assertIn("qwen3-tts-clone", message) self.assertIn("--model", message) def test_preset_mode_with_no_matching_model_lists_both_ids(self): with self.assertNoLogs("converter.clients.audiocpp", level="WARNING"): with self.assertRaises(RuntimeError) as ctx: self._client(voice="narrator", model_id="qwen3-tts", models={"data": [{"id": "pocket-tts"}]}) message = str(ctx.exception) self.assertIn("qwen3-tts", message) self.assertIn("pocket-tts", message) class AudioCppTaskDetectionTests(unittest.TestCase): """Task auto-detection (tts/clon/vdes) and voice design validation.""" @staticmethod def _json_response(payload): response = MagicMock() response.__enter__.return_value = response response.read.return_value = json.dumps(payload).encode("utf-8") return response def _client(self, voice=None, instructions=None, request_options=None, models=None): if models is None: models = {"data": [{"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_tts"}]} def _dispatch(request, **_kwargs): url = request if isinstance(request, str) else request.full_url if url.endswith("/health"): return self._json_response({"status": "ok"}) if url.endswith("/v1/models"): return self._json_response(models) if "/v1/audio/voices" in url: return self._json_response({"voices": ["narrator"]}) raise AssertionError(f"unexpected URL: {url}") with patch("converter.clients.faster.urllib.request.urlopen", side_effect=_dispatch): return AudioCppTTSClient(_DUMMY_CHUNKS, voice=voice, instructions=instructions, request_options=request_options, model_id=_AUDIOCPP_MODEL_ID) def test_missing_task_falls_back_to_tts(self): # Servers that predate the task field hosted plain TTS models. client = self._client(voice="Vivian", models={"data": [ {"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_tts"}]}) self.assertEqual(client.task, AUDIOCPP_TASK_TTS) self.assertFalse(client.design_mode) def test_task_detected_from_models_endpoint(self): client = self._client(models={"data": [ {"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_tts", "task": "vdes"}]}, instructions="A warm adult narrator") self.assertEqual(client.task, AUDIOCPP_TASK_VDES) self.assertTrue(client.design_mode) def test_clon_task_entry_connects_in_preset_mode(self): client = self._client(voice="narrator", models={"data": [ {"id": _AUDIOCPP_MODEL_ID, "family": "chatterbox", "task": "clon"}]}) self.assertEqual(client.task, "clon") self.assertFalse(client.design_mode) self.assertTrue(client.preset_mode) def test_unsupported_task_rejected_with_available_entries(self): with self.assertRaises(RuntimeError) as ctx: self._client(models={"data": [ {"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_asr", "task": "asr"}, {"id": "tts-1", "family": "qwen3_tts", "task": "tts"}]}, instructions="unused") message = str(ctx.exception) self.assertIn("'asr'", message) self.assertIn("--model", message) self.assertIn("tts-1", message) def test_vdes_without_instructions_requires_description(self): with self.assertRaises(RuntimeError) as ctx: self._client(models={"data": [ {"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_tts", "task": "vdes"}]}) message = str(ctx.exception) self.assertIn("voice design", message) self.assertIn("--instructions", message) def test_vdes_with_voice_rejected(self): with self.assertRaises(RuntimeError) as ctx: self._client(voice="narrator", models={"data": [ {"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_tts", "task": "vdes"}]}, instructions="A warm adult narrator") self.assertIn("--voice", str(ctx.exception)) self.assertIn("--instructions", str(ctx.exception)) def test_vdes_with_instructions_connects_in_design_mode(self): buf = io.StringIO() with redirect_stdout(buf): client = self._client(models={"data": [ {"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_tts", "task": "vdes"}]}, instructions="A warm adult narrator") self.assertTrue(client.design_mode) self.assertEqual(client.instructions, "A warm adult narrator") out = buf.getvalue() self.assertIn("voice design", out) self.assertIn("A warm adult narrator", out) def test_instructions_without_voice_on_generic_family_connects(self): # Families without built-in speakers can get their voice from the # instruction alone (e.g. OmniVoice voice design). buf = io.StringIO() with redirect_stdout(buf): client = self._client(models={"data": [ {"id": _AUDIOCPP_MODEL_ID, "family": "omnivoice", "task": "tts"}]}, instructions="female, young adult, moderate pitch") self.assertFalse(client.design_mode) self.assertTrue(client.instruction_voice) self.assertIn("instruction voice", buf.getvalue()) def test_instructions_with_builtin_speaker_family_stays_speaker_mode(self): buf = io.StringIO() with redirect_stdout(buf): client = self._client( voice="Vivian", models={"data": [ {"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_tts", "task": "tts"}]}, instructions="Very happy.") self.assertFalse(client.design_mode) self.assertFalse(client.instruction_voice) self.assertIn("speaker 'Vivian'", buf.getvalue()) def test_instructions_reach_the_client(self): client = self._client( models={"data": [ {"id": _AUDIOCPP_MODEL_ID, "family": "qwen3_tts", "task": "vdes"}]}, instructions="from flag") self.assertEqual(client.instructions, "from flag") class AudioCppFamilyDetectionTests(unittest.TestCase): """Family auto-detection and per-family adaptations.""" @staticmethod def _json_response(payload): response = MagicMock() response.__enter__.return_value = response response.read.return_value = json.dumps(payload).encode("utf-8") return response def _client(self, voice="narrator", models=None): def _dispatch(request, **_kwargs): url = request if isinstance(request, str) else request.full_url if url.endswith("/health"): return self._json_response({"status": "ok"}) if url.endswith("/v1/models"): return self._json_response(models) if "/v1/audio/voices" in url: return self._json_response({"voices": [voice] if voice else []}) raise AssertionError(f"unexpected URL: {url}") with patch("converter.clients.faster.urllib.request.urlopen", side_effect=_dispatch): return AudioCppTTSClient(_DUMMY_CHUNKS, voice=voice, model_id=_AUDIOCPP_MODEL_ID) def test_family_detected_from_models_endpoint(self): client = self._client(models={"data": [ {"id": _AUDIOCPP_MODEL_ID, "family": "higgs_audio_tts"}]}) self.assertEqual(client.family, "higgs_audio_tts") self.assertIs(client.profile, AUDIOCPP_DEFAULT_FAMILY_PROFILE) def test_missing_family_uses_generic_profile(self): # A missing family is unknown (not guessed as qwen3_tts): it falls # through to the generic clone-only profile. client = self._client(models={"data": [ {"id": _AUDIOCPP_MODEL_ID}]}) self.assertEqual(client.family, "") self.assertIs(client.profile, AUDIOCPP_DEFAULT_FAMILY_PROFILE) def test_unknown_family_uses_generic_profile(self): client = self._client(models={"data": [ {"id": _AUDIOCPP_MODEL_ID, "family": "future_tts"}]}) self.assertEqual(client.family, "future_tts") self.assertIs(client.profile, AUDIOCPP_DEFAULT_FAMILY_PROFILE) self.assertEqual(client.profile.language_style, AUDIOCPP_LANG_OMIT) def test_speaker_mode_rejected_for_clone_only_family(self): client = None try: client = self._client(voice=None, models={"data": [ {"id": _AUDIOCPP_MODEL_ID, "family": "voxcpm2"}]}) except RuntimeError as exc: message = str(exc) self.assertIn("voxcpm2", message) self.assertIn("--voice", message) self.assertIn("no built-in speakers", message) self.assertIsNone(client) def test_speaker_mode_allowed_for_customvoice_entry(self): # A Qwen3-TTS entry whose id names CustomVoice is speaker-capable; # a built-in speaker name selects speaker mode on it. client = self._client( voice="Vivian", models={"data": [ {"id": "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF", "family": "qwen3_tts"}]}) self.assertEqual(client.family, "qwen3_tts") self.assertTrue(client.speaker_mode) def test_speaker_mode_rejected_for_qwen_base_entry(self): # A Qwen3-TTS entry whose id names Base (not CustomVoice) is # clone-only, even though its family has built-in speakers on other # entries: without --voice it fails fast. client = None try: client = self._client(voice=None, models={"data": [ {"id": "Qwen3-TTS-12Hz-1.7B-Base-GGUF", "family": "qwen3_tts"}]}) except RuntimeError as exc: message = str(exc) self.assertIn("Base-GGUF", message) self.assertIn("--voice", message) self.assertIsNone(client) def test_iso_language_code_helper(self): self.assertEqual(LANGUAGE_ISO_CODES["English"], "en") self.assertIsNone(LANGUAGE_ISO_CODES.get("Auto")) def test_iso_codes_cover_the_audio_cpp_menu_languages(self): self.assertEqual(LANGUAGE_ISO_CODES["Arabic"], "ar") self.assertEqual(LANGUAGE_ISO_CODES["Hindi"], "hi") self.assertEqual(LANGUAGE_ISO_CODES["Vietnamese"], "vi") class AudiocppEntryVoiceCapabilityTests(unittest.TestCase): """The per-entry voice capability resolver (speaker/clone/design).""" def _cap(self, family="", task="tts", model_id=""): return audiocpp_entry_voice_capability(family, task, model_id) def test_vdes_task_is_design(self): self.assertEqual(self._cap("qwen3_tts", "vdes", "Qwen3-TTS-VoiceDesign-GGUF"), AUDIOCPP_VOICE_DESIGN) def test_qwen_customvoice_entry_is_speaker(self): self.assertEqual(self._cap("qwen3_tts", "tts", "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF"), AUDIOCPP_VOICE_SPEAKER) def test_qwen_base_entry_is_clone(self): self.assertEqual(self._cap("qwen3_tts", "tts", "Qwen3-TTS-12Hz-1.7B-Base-GGUF"), AUDIOCPP_VOICE_CLONE) def test_qwen_unidentified_entry_is_clone(self): self.assertEqual(self._cap("qwen3_tts", "tts", "qwen"), AUDIOCPP_VOICE_CLONE) def test_other_families_are_clone(self): self.assertEqual(self._cap("higgs_audio_tts", "tts", "higgs"), AUDIOCPP_VOICE_CLONE) def test_missing_family_is_clone(self): self.assertEqual(self._cap("", "tts", "legacy"), AUDIOCPP_VOICE_CLONE) def test_customvoice_match_is_case_insensitive(self): self.assertEqual(self._cap("qwen3_tts", "tts", "Qwen3-TTS-12Hz-1.7B-CUSTOMVOICE-GGUF"), AUDIOCPP_VOICE_SPEAKER) def test_customvoice_id_in_other_family_is_not_speaker(self): # The "customvoice" substring only marks a speaker for the qwen3_tts # family; another family with a lookalike id stays clone-only. self.assertEqual(self._cap("future_tts", "tts", "Qwen3-TTS-CustomVoice"), AUDIOCPP_VOICE_CLONE) class AudioCppTTSClientRequestTests(unittest.TestCase): """The /v1/audio/speech payload and response validation.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self._sleep = patch("converter.clients.base.time.sleep") self._sleep.start() def tearDown(self): self._sleep.stop() self._tmp.cleanup() def _make_client(self, preset_mode=False, voice="Vivian", language="English", seed=-1, family="qwen3_tts", task="tts", instructions=None, request_options=None): client = AudioCppTTSClient.__new__(AudioCppTTSClient) client.chunks_dir = Path(self._tmp.name) client.api_url = "http://127.0.0.1:8080" client.model_id = _AUDIOCPP_MODEL_ID client.preset_mode = preset_mode client.voice = voice client.language = language client._seed = seed client.family = family client.task = task client.profile = AUDIOCPP_FAMILY_PROFILES.get( family, AUDIOCPP_DEFAULT_FAMILY_PROFILE) client.instructions = instructions or "" client.request_options = dict(request_options or {}) client.design_mode = task == AUDIOCPP_TASK_VDES # Mirrors the connect-time rule: an instruction-defined voice on a # clone-capable entry with no --voice (design mode takes precedence). capability = audiocpp_entry_voice_capability( family, task, client.model_id) client.instruction_voice = ( not preset_mode and not client.design_mode and capability == AUDIOCPP_VOICE_CLONE and bool(client.instructions)) return client @staticmethod def _wav_bytes(frames=b"\x01\x00" * 10, rate=SAMPLE_RATE): buffer = io.BytesIO() with wave.open(buffer, "wb") as wav_file: wav_file.setnchannels(1) wav_file.setsampwidth(2) wav_file.setframerate(rate) wav_file.writeframes(frames) return buffer.getvalue() def _post_response(self, body): response = MagicMock() response.__enter__.return_value = response response.read.return_value = body return response def test_payload_includes_model_input_voice_language_and_seed(self): client = self._make_client(preset_mode=True, voice="narrator", language="Japanese", seed=1234) with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello world.") request = mock_urlopen.call_args[0][0] self.assertEqual(request.full_url, "http://127.0.0.1:8080/v1/audio/speech") payload = json.loads(request.data.decode("utf-8")) self.assertEqual(payload["model"], _AUDIOCPP_MODEL_ID) self.assertEqual(payload["input"], "Hello world.") self.assertEqual(payload["voice"], "narrator") self.assertEqual(payload["language"], "Japanese") self.assertEqual(payload["seed"], 1234) self.assertNotIn("instructions", payload) def test_negative_seed_omitted_from_payload(self): client = self._make_client(preset_mode=True, voice="narrator", seed=-1) with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello world.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) self.assertNotIn("seed", payload) def test_request_timeout_is_the_configured_api_timeout(self): client = self._make_client() long_text = " ".join(f"word{i}" for i in range(1500)) with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav(long_text) timeout = mock_urlopen.call_args[1]["timeout"] self.assertEqual(timeout, config.API_TIMEOUT) def test_speaker_mode_without_instructions_omits_the_field(self): # There is no configured style instruction: speaker mode sends no # instructions field unless the run provides one. client = self._make_client(preset_mode=False) with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) self.assertNotIn("instructions", payload) def test_explicit_instructions_reach_the_payload(self): client = self._make_client(preset_mode=False, instructions="Read whisper quiet.") with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) self.assertEqual(payload["instructions"], "Read whisper quiet.") def test_preset_mode_sends_instructions_alongside_voice(self): # Clone + style control: both the server-side voice and the # instruction reach the model. client = self._make_client(preset_mode=True, voice="narrator", instructions="Calm and steady.") with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) self.assertEqual(payload["voice"], "narrator") self.assertEqual(payload["instructions"], "Calm and steady.") def test_design_mode_payload_omits_voice_and_sends_instructions(self): client = self._make_client(task="vdes", instructions="A warm adult narrator") with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) self.assertNotIn("voice", payload) self.assertEqual(payload["instructions"], "A warm adult narrator") def test_design_mode_language_follows_family_profile(self): # The VoiceDesign package is family qwen3_tts, whose language field # takes Qwen display names like the other variants. client = self._make_client(task="vdes", language="Japanese", instructions="A warm adult narrator") with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) self.assertEqual(payload["language"], "Japanese") def test_instruction_voice_payload_omits_voice(self): # Instruction-defined voice on a family without built-in speakers: # no speaker name is invented, the instruction carries the voice. client = self._make_client(family="omnivoice", instructions="female, young adult") with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) self.assertNotIn("voice", payload) self.assertNotIn("language", payload) # generic profile: omitted self.assertEqual(payload["instructions"], "female, young adult") def test_request_options_forwarded_in_payload(self): client = self._make_client(preset_mode=True, voice="narrator", request_options={"emotion": "neutral", "speed": "1.1"}) with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) self.assertEqual(payload["options"], {"emotion": "neutral", "speed": "1.1"}) def test_empty_request_options_omit_options_field(self): client = self._make_client(preset_mode=True, voice="narrator") with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) self.assertNotIn("options", payload) def test_generic_family_omits_language_and_instructions(self): # Clone-only families (higgs_audio_tts, voxcpm2, ...) detect the # language themselves and take no style instruction. client = self._make_client(preset_mode=False, family="higgs_audio_tts") with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) self.assertNotIn("language", payload) self.assertNotIn("instructions", payload) def test_iso_family_sends_language_code(self): client = self._make_client(preset_mode=True, voice="narrator", language="Japanese", family="index_tts2") with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) self.assertEqual(payload["language"], "ja") def test_iso_family_auto_omits_language(self): client = self._make_client(preset_mode=True, voice="narrator", language="Auto", family="index_tts2") with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) self.assertNotIn("language", payload) def test_qwen_language_display_name_still_sent(self): client = self._make_client(preset_mode=True, voice="narrator", language="Japanese", family="qwen3_tts") with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) self.assertEqual(payload["language"], "Japanese") def test_non_wav_response_rejected(self): client = self._make_client() for body in (b"", b"RIFFxxxx", b"MP3DATA-MP3DATA", b"RIFF\x00\x00\x00\x00mpeg"): with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(body)): with self.assertRaises(RuntimeError): client._request_wav("Hello.") def test_http_error_body_surfaced(self): import urllib.error client = self._make_client() error = urllib.error.HTTPError( "http://127.0.0.1:8080/v1/audio/speech", 500, "Server Error", {}, io.BytesIO(b'{"error":"bad voice"}')) with patch("converter.clients.faster.urllib.request.urlopen", side_effect=error): with self.assertRaises(RuntimeError) as ctx: client._request_wav("Hello.") self.assertIn("500", str(ctx.exception)) self.assertIn("bad voice", str(ctx.exception)) def test_reference_text_error_is_not_retryable(self): # Qwen3-TTS Base cloning without a server-side transcript fails # identically on every attempt: the error must carry the fix # (prompt_text / x_vector_only_mode) and skip the retry budget. client = self._make_client(preset_mode=True, voice="narrator") error = urllib.error.HTTPError( "http://127.0.0.1:8080/v1/audio/speech", 500, "Server Error", {}, io.BytesIO(b'{"error":{"message":"Qwen3 voice clone ICL mode ' b'requires reference text","type":"server_error"}}')) with patch("converter.clients.faster.urllib.request.urlopen", side_effect=error): with self.assertRaises(NonRetryableTTSError) as ctx: client._request_wav("Hello.") message = str(ctx.exception) self.assertIn("requires reference text", message) self.assertIn("'narrator'", message) self.assertIn("prompt_text", message) self.assertIn("x_vector_only_mode", message) def test_model_contract_error_is_not_retryable(self): client = self._make_client(preset_mode=True, voice="narrator") error = urllib.error.HTTPError( "http://127.0.0.1:8080/v1/audio/speech", 500, "Server Error", {}, io.BytesIO(b'{"error":{"message":"model contract spec not found ' b"for family 'qwen3_tts' (provide --model-spec-override)\"}}")) with patch("converter.clients.faster.urllib.request.urlopen", side_effect=error): with self.assertRaises(NonRetryableTTSError) as ctx: client._request_wav("Hello.") message = str(ctx.exception) self.assertIn("not retryable", message) self.assertIn("model contract spec not found for family 'qwen3_tts'", message) self.assertIn("--model-spec-override", message) def test_unknown_model_id_error_is_not_retryable(self): client = self._make_client(preset_mode=True, voice="narrator") error = urllib.error.HTTPError( "http://127.0.0.1:8080/v1/audio/speech", 500, "Server Error", {}, io.BytesIO(b'{"error":{"message":"unknown model id: nope"}}')) with patch("converter.clients.faster.urllib.request.urlopen", side_effect=error): with self.assertRaises(NonRetryableTTSError) as ctx: client._request_wav("Hello.") message = str(ctx.exception) self.assertIn("not retryable", message) self.assertIn("unknown model id: nope", message) def test_unmatched_server_error_stays_retryable(self): # Only known-deterministic fragments skip the retry budget; device # hiccups, OOM, and anything unrecognized keep the plain error the # retry loop has always retried. client = self._make_client() error = urllib.error.HTTPError( "http://127.0.0.1:8080/v1/audio/speech", 500, "Server Error", {}, io.BytesIO(b'{"error":{"message":"CUDA error at ggml-cuda.cu"}}')) with patch("converter.clients.faster.urllib.request.urlopen", side_effect=error): with self.assertRaises(RuntimeError) as ctx: client._request_wav("Hello.") self.assertNotIsInstance(ctx.exception, NonRetryableTTSError) self.assertIn("CUDA error", str(ctx.exception)) def test_non_retryable_error_skips_remaining_attempts(self): client = self._make_client() with patch.object(client, "generate_chunk", side_effect=NonRetryableTTSError("nope")) as mock_gen, \ patch("converter.clients.base.time.sleep") as mock_sleep: with self.assertRaises(NonRetryableTTSError): client.process_chunk_with_retry(1, "Hello.") self.assertEqual(mock_gen.call_count, 1) mock_sleep.assert_not_called() def test_transient_failure_fails_the_chunk_attempt(self): # Retrying is the chunk-level policy's job # (process_chunk_with_retry); one generate_chunk call makes one # request attempt per sub-chunk. client = self._make_client() wav = self._wav_bytes() with patch.object(client, "_request_wav", side_effect=[RuntimeError("boom"), wav]) as mock_request: result = client.generate_chunk("Hello.", 1) self.assertIsNone(result) self.assertEqual(mock_request.call_count, 1) def test_request_failure_fails_the_chunk(self): client = self._make_client() with patch.object(client, "_request_wav", side_effect=RuntimeError("down")) as mock_request: result = client.generate_chunk("Hello.", 1) self.assertIsNone(result) self.assertEqual(mock_request.call_count, 1) def test_empty_text_fails_the_chunk(self): client = self._make_client() with patch.object(client, "_request_wav") as mock_request: result = client.generate_chunk(" ", 1) self.assertIsNone(result) mock_request.assert_not_called() def test_generate_chunk_writes_valid_wav(self): client = self._make_client() frames = b"\x01\x00" * 100 with patch.object(client, "_request_wav", return_value=self._wav_bytes(frames)): result = client.generate_chunk("Hello world.", 1) self.assertIsNotNone(result) path = Path(result) self.assertEqual(path.name, "chunk_0001.wav") with wave.open(str(path), "rb") as wav_file: self.assertEqual(wav_file.getnchannels(), 1) self.assertEqual(wav_file.getsampwidth(), 2) self.assertEqual(wav_file.getframerate(), SAMPLE_RATE) self.assertEqual(wav_file.readframes(wav_file.getnframes()), frames) def test_long_text_is_subchunked_and_concatenated_in_order(self): client = self._make_client() sentences = [" ".join(f"word{i}" for i in range(6)) + "." for _ in range(3)] text = " ".join(sentences) parts = [self._wav_bytes(b"\x01\x00" * 10), self._wav_bytes(b"\x02\x00" * 20), self._wav_bytes(b"\x03\x00" * 30)] with patch.object(config, "CHUNK_SIZE", 10), \ patch.object(client, "_request_wav", side_effect=parts) as mock_request: result = client.generate_chunk(text, 1) self.assertEqual(mock_request.call_count, 3) with wave.open(str(Path(result)), "rb") as wav_file: self.assertEqual(wav_file.readframes(wav_file.getnframes()), b"\x01\x00" * 10 + b"\x02\x00" * 20 + b"\x03\x00" * 30) def test_stale_chunk_files_are_removed(self): stale = Path(self._tmp.name) / "chunk_0001.mp3" stale.write_bytes(b"old") client = self._make_client() with patch.object(client, "_request_wav", return_value=self._wav_bytes()): client.generate_chunk("Hello.", 1) remaining = sorted(path.name for path in Path(self._tmp.name).glob("chunk_0001.*")) self.assertEqual(remaining, ["chunk_0001.wav"]) class TranscribeReasonTests(unittest.TestCase): """transcribe_reference_audio_detailed: a reason for every empty result. The audio.cpp setup prints the reason per voice, so each failure class must be distinguishable: missing package vs broken import vs transcribe error vs a silent no-speech result. """ def _transcribe_with_models(self, models, spec_present=True): """Run one detailed transcription with _cached_model stubbed. MODELS maps backend name -> model object (or exception instance to raise in its place). The whisper fallback sees its own entry or a ModuleNotFoundError so no real package import ever happens; importlib.util.find_spec is pinned so the not-installed vs installed-but-broken distinction is deterministic in any env. """ def fake_cached(key, loader): backend = key[0] entry = models.get(backend) if isinstance(entry, Exception): raise entry return entry with patch("converter.clients.transcribe._cached_model", side_effect=fake_cached), \ patch("importlib.util.find_spec", return_value=MagicMock() if spec_present else None): return transcribe_reference_audio_detailed("clip.wav") def test_success_returns_text_and_ok(self): model = MagicMock() model.transcribe.return_value = (iter([MagicMock(text=" Hello. ")]), MagicMock()) text, reason = self._transcribe_with_models( {"faster_whisper": model, "whisper": ModuleNotFoundError()}) self.assertEqual(text, "Hello.") self.assertEqual(reason, "ok") def test_missing_backend_is_not_called_broken(self): text, reason = self._transcribe_with_models({ "faster_whisper": ModuleNotFoundError( "No module named 'faster_whisper'"), "whisper": ModuleNotFoundError("No module named 'whisper'"), }, spec_present=False) self.assertIsNone(text) self.assertIn("faster_whisper is not installed", reason) self.assertIn("whisper is not installed", reason) def test_broken_import_is_distinguished_from_missing(self): text, reason = self._transcribe_with_models({ "faster_whisper": ImportError( "Error loading shared library ld-linux-x86-64.so.2"), "whisper": ModuleNotFoundError("No module named 'whisper'", name="whisper"), }) self.assertIsNone(text) self.assertIn("faster_whisper is installed but failed to import", reason) self.assertIn("ld-linux-x86-64.so.2", reason) self.assertIn("whisper is not installed", reason) def test_transcribe_error_carries_the_exception(self): model = MagicMock() model.transcribe.side_effect = RuntimeError("decode failed") text, reason = self._transcribe_with_models( {"faster_whisper": model, "whisper": ModuleNotFoundError("No module named 'whisper'")}) self.assertIsNone(text) self.assertIn("faster_whisper transcription failed: decode failed", reason) def test_empty_result_reports_no_speech(self): model = MagicMock() model.transcribe.return_value = (iter([]), MagicMock()) text, reason = self._transcribe_with_models( {"faster_whisper": model, "whisper": ModuleNotFoundError("No module named 'whisper'")}) self.assertIsNone(text) self.assertIn("faster_whisper heard no speech", reason) def test_whisper_fallback_used_when_faster_whisper_fails(self): failing = MagicMock() failing.transcribe.side_effect = RuntimeError("boom") good = MagicMock() # The openai-whisper interface returns a dict with "text". good.transcribe.return_value = {"text": " Hi. "} text, reason = self._transcribe_with_models( {"faster_whisper": failing, "whisper": good}) self.assertEqual(text, "Hi.") self.assertEqual(reason, "ok") def test_backend_problem_reports_broken_import(self): def fake_import(name, *args, **kwargs): raise ImportError("lib load failure") with patch("builtins.__import__", side_effect=fake_import), \ patch("importlib.util.find_spec", return_value=MagicMock()): problem = whisper_backend_problem() self.assertIn("faster_whisper is installed but failed to import", problem) self.assertIn("whisper is installed but failed to import", problem) def test_backend_problem_none_when_a_backend_imports(self): def fake_import(name, *args, **kwargs): if name == "faster_whisper": return MagicMock() raise ImportError("should not be probed") with patch("builtins.__import__", side_effect=fake_import): self.assertIsNone(whisper_backend_problem()) class AudioCppHeartbeatTests(unittest.TestCase): """The heartbeat reports chunk progress while a request generates.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() def tearDown(self): self._tmp.cleanup() def _client(self): client = AudioCppTTSClient.__new__(AudioCppTTSClient) client.chunks_dir = Path(self._tmp.name) client.api_url = "http://127.0.0.1:8080" client.model_id = _AUDIOCPP_MODEL_ID client.preset_mode = False client.voice = "Vivian" client.language = "English" client._seed = -1 client.family = "qwen3_tts" client.profile = AUDIOCPP_DEFAULT_FAMILY_PROFILE return client @staticmethod def _wav_bytes(): buffer = io.BytesIO() with wave.open(buffer, "wb") as wav_file: wav_file.setnchannels(1) wav_file.setsampwidth(2) wav_file.setframerate(SAMPLE_RATE) wav_file.writeframes(b"\x01\x00" * 10) return buffer.getvalue() def _run(self): client = self._client() def slow_request(*_args, **_kwargs): time.sleep(0.12) return self._wav_bytes() buf = io.StringIO() with patch.object(config, "HEARTBEAT_INTERVAL_SECONDS", 0.03), \ patch.object(client, "_request_wav", side_effect=slow_request), \ redirect_stdout(buf): result = client.generate_chunk("Hello.", 1) self.assertTrue(result) return buf.getvalue() def test_heartbeat_reports_chunk_progress(self): out = self._run() self.assertIn("Chunk 1 still generating", out) class AudioCppTTSClientTruncationTests(unittest.TestCase): """Audio far shorter than its text implies fails the request.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() def tearDown(self): self._tmp.cleanup() def _make_client(self): client = AudioCppTTSClient.__new__(AudioCppTTSClient) client.chunks_dir = Path(self._tmp.name) client.api_url = "http://127.0.0.1:8080" client.model_id = _AUDIOCPP_MODEL_ID client.preset_mode = True client.voice = "narrator" client.language = "English" client._seed = -1 client.family = "qwen3_tts" client.profile = AUDIOCPP_FAMILY_PROFILES["qwen3_tts"] return client @staticmethod def _wav_bytes(frames): buffer = io.BytesIO() with wave.open(buffer, "wb") as wav_file: wav_file.setnchannels(1) wav_file.setsampwidth(2) wav_file.setframerate(SAMPLE_RATE) wav_file.writeframes(frames) return buffer.getvalue() def test_full_length_wav_passes(self): client = self._make_client() text = " ".join(f"word{i}" for i in range(12)) # 12 words -> expected 4.8s, half is 2.4s -> 2.5s of audio passes. wav = self._wav_bytes(b"\x01\x00" * int(2.5 * SAMPLE_RATE)) with patch.object(client, "_request_wav", return_value=wav): result = client.generate_chunk(text, 1) self.assertIsNotNone(result) class AudioCppUnloadModelsTests(unittest.TestCase): """Before generating, the client asks the server to drop loaded models.""" @staticmethod def _client(): client = AudioCppTTSClient.__new__(AudioCppTTSClient) client.api_url = "http://127.0.0.1:8080" return client @staticmethod def _response(body): response = MagicMock() response.__enter__.return_value = response response.read.return_value = body return response def test_posts_to_unload_all_models(self): client = self._client() with patch("converter.clients.audiocpp.urllib.request.urlopen", return_value=self._response(b'{"unloaded": ["qwen"]}')) as mock_urlopen: client._unload_server_models() request = mock_urlopen.call_args[0][0] self.assertEqual(request.full_url, "http://127.0.0.1:8080/v1/tasks/unload_all_models") self.assertEqual(request.method, "POST") self.assertEqual(request.data, b"") def test_reports_unloaded_ids(self): client = self._client() buf = io.StringIO() with patch("converter.clients.audiocpp.urllib.request.urlopen", return_value=self._response(b'{"unloaded": ["a", "b"]}')), \ redirect_stdout(buf): client._unload_server_models() self.assertIn("Unloaded 2 model(s)", buf.getvalue()) self.assertIn("a, b", buf.getvalue()) def test_no_loaded_models_is_silent(self): client = self._client() buf = io.StringIO() with patch("converter.clients.audiocpp.urllib.request.urlopen", return_value=self._response(b'{"unloaded": []}')), \ redirect_stdout(buf): client._unload_server_models() self.assertEqual(buf.getvalue(), "") def test_http_error_warns_and_continues(self): client = self._client() buf = io.StringIO() with patch("converter.clients.audiocpp.urllib.request.urlopen", side_effect=urllib.error.HTTPError( "http://127.0.0.1:8080/v1/tasks/unload_all_models", 404, "Not Found", None, io.BytesIO())), \ redirect_stdout(buf): client._unload_server_models() out = buf.getvalue() self.assertIn("[WARNING]", out) self.assertIn("404", out) def test_connection_error_warns_and_continues(self): client = self._client() buf = io.StringIO() with patch("converter.clients.audiocpp.urllib.request.urlopen", side_effect=urllib.error.URLError("refused")), \ redirect_stdout(buf): client._unload_server_models() self.assertIn("[WARNING]", buf.getvalue()) def test_connect_unloads_before_returning(self): client = AudioCppTTSClient.__new__(AudioCppTTSClient) client.api_url = "http://127.0.0.1:8080" client.model_id = _AUDIOCPP_MODEL_ID client.preset_mode = True client.voice = "narrator" client.language = "English" client._seed = -1 client.family = "qwen3_tts" client.task = AUDIOCPP_TASK_TTS client.profile = AUDIOCPP_FAMILY_PROFILES["qwen3_tts"] client.design_mode = False client.instruction_voice = False client.speaker_mode = False client.instructions = "" with patch.object(client, "_check_health"), \ patch.object(client, "_list_models", return_value=[{"id": client.model_id, "family": "qwen3_tts", "task": "tts"}]), \ patch.object(client, "_auto_pick_model_id"), \ patch.object(client, "_require_model_id"), \ patch.object(client, "_resolve_family"), \ patch.object(client, "_resolve_task"), \ patch.object(client, "_check_voice"), \ patch.object(client, "_unload_server_models") as mock_unload: client._connect() mock_unload.assert_called_once() def test_connect_skips_unload_when_disabled(self): client = AudioCppTTSClient.__new__(AudioCppTTSClient) client.api_url = "http://127.0.0.1:8080" client.model_id = _AUDIOCPP_MODEL_ID client.preset_mode = True client.voice = "narrator" client.language = "English" client._seed = -1 client.family = "qwen3_tts" client.task = AUDIOCPP_TASK_TTS client.profile = AUDIOCPP_FAMILY_PROFILES["qwen3_tts"] client.design_mode = False client.instruction_voice = False client.speaker_mode = False client.instructions = "" with patch.object(client, "_check_health"), \ patch.object(client, "_list_models", return_value=[{"id": client.model_id, "family": "qwen3_tts", "task": "tts"}]), \ patch.object(client, "_auto_pick_model_id"), \ patch.object(client, "_require_model_id"), \ patch.object(client, "_resolve_family"), \ patch.object(client, "_resolve_task"), \ patch.object(client, "_check_voice"), \ patch.object(config, "AUDIOCPP_UNLOAD_MODELS", False), \ patch.object(client, "_unload_server_models") as mock_unload: client._connect() mock_unload.assert_not_called() class BackendWiringTests(unittest.TestCase): """AudiobookConverter wiring for the --backend selector.""" def test_faster_backend_uses_faster_client_without_reference(self): with patch("converter.converter.FasterTTSClient") as mock_faster, \ patch("converter.converter.QwenTTSClient") as mock_qwen, \ patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: AudiobookConverter(voice_mode=VOICE_MODE_CLONE, backend=BACKEND_FASTER, voice="narrator") mock_faster.assert_called_once_with(chunks_dir=converter_mod.CHUNKS_FOLDER, voice="narrator", api_url=None, quiet=False) mock_qwen.assert_not_called() mock_audiocpp.assert_not_called() def test_audiocpp_backend_with_voice_uses_audiocpp_client(self): with patch("converter.converter.FasterTTSClient") as mock_faster, \ patch("converter.converter.QwenTTSClient") as mock_qwen, \ patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: AudiobookConverter(voice_mode=VOICE_MODE_CLONE, backend=BACKEND_AUDIOCPP, voice="narrator", language="ja") mock_audiocpp.assert_called_once_with(chunks_dir=converter_mod.CHUNKS_FOLDER, voice="narrator", language="Japanese", model_id=None, instructions=None, request_options={}, api_url=None, quiet=False) mock_faster.assert_not_called() mock_qwen.assert_not_called() def test_audiocpp_backend_without_voice_uses_audiocpp_client(self): with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM, backend=BACKEND_AUDIOCPP) mock_audiocpp.assert_called_once_with(chunks_dir=converter_mod.CHUNKS_FOLDER, voice=None, language=config.LANGUAGE, model_id=None, instructions=None, request_options={}, api_url=None, quiet=False) def test_audiocpp_backend_model_id_is_wired_through(self): with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: AudiobookConverter(voice_mode=VOICE_MODE_CLONE, backend=BACKEND_AUDIOCPP, voice="narrator", model_id="higgs") mock_audiocpp.assert_called_once_with( chunks_dir=converter_mod.CHUNKS_FOLDER, voice="narrator", language=config.LANGUAGE, model_id="higgs", instructions=None, request_options={}, api_url=None, quiet=False) def test_audiocpp_backend_instructions_and_options_are_wired_through(self): with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM, backend=BACKEND_AUDIOCPP, instructions="A warm adult narrator", request_options={"emotion": "neutral", "speed": "1.1"}) mock_audiocpp.assert_called_once_with( chunks_dir=converter_mod.CHUNKS_FOLDER, voice=None, language=config.LANGUAGE, model_id=None, instructions="A warm adult narrator", request_options={"emotion": "neutral", "speed": "1.1"}, api_url=None, quiet=False) def test_qwen_backend_uses_qwen_client(self): with patch("converter.converter.FasterTTSClient") as mock_faster, \ patch("converter.converter.QwenTTSClient") as mock_qwen, \ patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM, backend=BACKEND_QWEN, voice="Vivian") _, kwargs = mock_qwen.call_args self.assertEqual(kwargs["voice"], "Vivian") mock_faster.assert_not_called() mock_audiocpp.assert_not_called() def test_qwen_clone_mode_still_requires_reference(self): with patch("converter.converter.QwenTTSClient"): with self.assertRaises(ValueError): AudiobookConverter(voice_mode=VOICE_MODE_CLONE, backend=BACKEND_QWEN) def test_qwen_design_mode_without_instructions_rejected(self): # A VoiceDesign run needs a description; an empty instructions # value (not even the config default) is refused up front. with patch("converter.converter.QwenTTSClient"): with self.assertRaises(ValueError): AudiobookConverter(voice_mode=VOICE_MODE_DESIGN, backend=BACKEND_QWEN, instructions=" ") def test_qwen_design_mode_threads_instructions_to_the_client(self): with patch("converter.converter.QwenTTSClient") as mock_qwen: AudiobookConverter(voice_mode=VOICE_MODE_DESIGN, backend=BACKEND_QWEN, instructions="A warm adult female narrator") _, kwargs = mock_qwen.call_args self.assertEqual(kwargs["instructions"], "A warm adult female narrator") def test_qwen_design_narrator_tag_uses_designed(self): self.assertEqual(AudiobookConverter.compute_narrator_tag( BACKEND_QWEN, None, VOICE_MODE_DESIGN, None, "A warm adult female narrator"), "designed") def test_api_url_override_reaches_each_client(self): # A remote conversion threads api_url through to the selected client. with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: AudiobookConverter(voice_mode=VOICE_MODE_CLONE, backend=BACKEND_AUDIOCPP, voice="narrator", api_url="http://10.0.0.5:8080") mock_audiocpp.assert_called_once_with( chunks_dir=converter_mod.CHUNKS_FOLDER, voice="narrator", language=config.LANGUAGE, model_id=None, instructions=None, request_options={}, api_url="http://10.0.0.5:8080", quiet=False) with patch("converter.converter.FasterTTSClient") as mock_faster: AudiobookConverter(voice_mode=VOICE_MODE_CLONE, backend=BACKEND_FASTER, voice="narrator", api_url="http://10.0.0.5:8000") mock_faster.assert_called_once_with(chunks_dir=converter_mod.CHUNKS_FOLDER, voice="narrator", api_url="http://10.0.0.5:8000", quiet=False) with patch("converter.converter.QwenTTSClient") as mock_qwen: AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM, backend=BACKEND_QWEN, voice="Vivian", api_url="http://10.0.0.5:7860") mock_qwen.assert_called_once_with( chunks_dir=converter_mod.CHUNKS_FOLDER, voice_mode=VOICE_MODE_CUSTOM, voice_clone_ref_audio=None, voice_clone_ref_text=None, skip_transcription=False, language=config.LANGUAGE, instructions=None, api_url="http://10.0.0.5:7860", quiet=False, voice="Vivian") def test_audiocpp_clone_mode_does_not_require_reference(self): # Cloning is server-side for the audiocpp backend, so the # clone-mode voice can be selected without local reference audio. with patch("converter.converter.AudioCppTTSClient"): converter = AudiobookConverter(voice_mode=VOICE_MODE_CLONE, backend=BACKEND_AUDIOCPP, voice="narrator") self.assertIsNone(converter.voice_clone_ref_audio) def test_chapter_chunks_audiocpp_splits(self): converter = self._audiocpp_converter(voice="narrator") text = " ".join(f"word{i}" for i in range(50)) with patch.object(config, "CHUNK_SIZE", 10): chunks = converter._chapter_chunks(text) self.assertGreater(len(chunks), 1) self.assertTrue(all(len(chunk.split()) <= 10 for chunk in chunks)) def test_chapter_chunks_qwen_always_splits(self): with patch("converter.converter.QwenTTSClient"): converter = AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM, backend=BACKEND_QWEN, voice="Vivian") text = " ".join(f"word{i}" for i in range(50)) with patch.object(config, "CHUNK_SIZE", 10): chunks = converter._chapter_chunks(text) self.assertGreater(len(chunks), 1) def test_faster_backend_still_validates_other_settings(self): with patch("converter.converter.FasterTTSClient"): with self.assertRaises(ValueError): AudiobookConverter(backend=BACKEND_FASTER, speed=0) with self.assertRaises(ValueError): AudiobookConverter(backend=BACKEND_FASTER, language="klingon") def test_audiocpp_backend_still_validates_other_settings(self): with patch("converter.converter.AudioCppTTSClient"): with self.assertRaises(ValueError): AudiobookConverter(backend=BACKEND_AUDIOCPP, speed=0) with self.assertRaises(ValueError): AudiobookConverter(backend=BACKEND_AUDIOCPP, language="klingon") def _faster_converter(self, voice=None): with patch("converter.converter.FasterTTSClient"): return AudiobookConverter(voice_mode=VOICE_MODE_CLONE, backend=BACKEND_FASTER, voice=voice) def _audiocpp_converter(self, voice=None, instructions=None): with patch("converter.converter.AudioCppTTSClient"): return AudiobookConverter( voice_mode=VOICE_MODE_CLONE if voice else VOICE_MODE_CUSTOM, backend=BACKEND_AUDIOCPP, voice=voice, instructions=instructions) def test_narrator_tag_uses_faster_voice_name(self): converter = self._faster_converter(voice="male_richard_poe") self.assertEqual(converter._narrator_tag(), "male_richard_poe") def test_narrator_tag_faster_without_voice_uses_default_key(self): # Unreachable in a valid run (--voice is required); the tag stays # stable for pre-flights of runs that will fail client-side. converter = self._faster_converter() self.assertEqual(converter._narrator_tag(), "default") def test_narrator_tag_audiocpp_uses_voice_name(self): converter = self._audiocpp_converter(voice="female_narrator") self.assertEqual(converter._narrator_tag(), "female_narrator") def test_narrator_tag_audiocpp_without_voice_uses_fallback(self): # Unreachable in a valid run (the client refuses a speaker-capable # entry without --voice); the tag stays stable for pre-flights. converter = self._audiocpp_converter() self.assertEqual(converter._narrator_tag(), "narrator") def test_banner_and_narrator_work_without_reference_audio(self): converter = self._faster_converter(voice="male_richard_poe") converter._print_banner() # must not raise (regression: Path(None)) self.assertIsNone(converter.voice_clone_ref_audio) def test_audiocpp_banner_prints_without_reference_audio(self): converter = self._audiocpp_converter(voice="narrator") converter._print_banner() # must not raise converter = self._audiocpp_converter(instructions="Calm and warm.") converter._print_banner() def test_audiocpp_banner_prints_model_family(self): from contextlib import redirect_stdout converter = self._audiocpp_converter(voice="narrator") converter.tts.family = "higgs_audio_tts" buffer = io.StringIO() with redirect_stdout(buffer): converter._print_banner() self.assertIn("higgs_audio_tts", buffer.getvalue()) def test_non_faster_narrator_tag_unchanged(self): with tempfile.TemporaryDirectory() as tmp: ref = Path(tmp) / "ref.wav" ref.write_bytes(b"x") with patch("converter.converter.QwenTTSClient"): converter = AudiobookConverter(voice_mode=VOICE_MODE_CLONE, voice_clone_ref_audio=str(ref), backend=BACKEND_QWEN) self.assertEqual(converter._narrator_tag(), "ref") if __name__ == "__main__": unittest.main()