"""Tests for the TTS client wrappers (language handling and payloads).""" import io import json import tempfile import unittest import wave from pathlib import Path from unittest.mock import MagicMock, patch from converter import config, tts from converter.converter import AudiobookConverter from converter.tts import ( AudioCppTTSClient, FasterTTSClient, QwenTTSClient, normalize_language, ) 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") def test_all_supported_languages_round_trip(self): for name in tts.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(**kwargs) def test_default_follows_config_for_each_mode(self): custom = self._make_client(voice_mode=tts.VOICE_MODE_CUSTOM) self.assertEqual(custom.language, config.LANGUAGE) clone = self._make_client(voice_mode=tts.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=tts.VOICE_MODE_CUSTOM, language="ja") 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(language="klingon") mock_connect.assert_not_called() 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(**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=tts.VOICE_MODE_CUSTOM) 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=tts.VOICE_MODE_CUSTOM) 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=tts.VOICE_MODE_CUSTOM) 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 = tts.VOICE_MODE_CUSTOM 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 = tts.VOICE_MODE_CUSTOM 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 = tts.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"], tts.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.tts.urllib.request.urlopen", side_effect=urllib.error.URLError("Connection refused")): with self.assertRaises(RuntimeError) as ctx: FasterTTSClient() message = str(ctx.exception) self.assertIn("not reachable", message) self.assertIn("README", message) def test_model_not_loaded_raises(self): with patch("converter.tts.urllib.request.urlopen", return_value=self._health_response(model_loaded=False)): with self.assertRaises(RuntimeError) as ctx: FasterTTSClient() self.assertIn("not loaded", str(ctx.exception)) def test_healthy_server_defaults_from_config(self): with patch("converter.tts.urllib.request.urlopen", return_value=self._health_response()): client = FasterTTSClient() self.assertEqual(client.voice, config.FASTER_VOICE) self.assertEqual(client.api_url, config.FASTER_API_URL.rstrip("/")) def test_explicit_voice_and_url_override_config(self): with patch("converter.tts.urllib.request.urlopen", return_value=self._health_response()): client = FasterTTSClient(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._chunks = patch.object(tts, "CHUNKS_FOLDER", Path(self._tmp.name)) self._chunks.start() self._sleep = patch("converter.tts.time.sleep") self._sleep.start() def tearDown(self): self._sleep.stop() self._chunks.stop() self._tmp.cleanup() def _make_client(self): client = FasterTTSClient.__new__(FasterTTSClient) 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, tts.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_is_retried(self): 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.assertIsNotNone(result) self.assertEqual(mock_pcm.call_count, 2) def test_empty_pcm_response_is_treated_as_failure(self): client = self._make_client() pcm = b"\x01\x00" * 10 def _response(body): response = MagicMock() response.__enter__.return_value = response response.read.return_value = body return response with patch("converter.tts.urllib.request.urlopen", side_effect=[_response(b""), _response(pcm)]) as mock_urlopen: result = client.generate_chunk("Hello.", 1) self.assertIsNotNone(result) self.assertEqual(mock_urlopen.call_count, 2) _, _, _, frames = self._read_wav(Path(result)) self.assertEqual(frames, pcm) def test_exhausted_subchunk_retries_fail_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, config.MAX_RETRIES) 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.tts.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") class TruncationDetectionTests(unittest.TestCase): """check_for_truncation: servers cut audio silently past their caps, so grossly short audio must fail the request (then retry / fail visibly).""" def test_duration_below_ratio_raises(self): with self.assertRaises(RuntimeError) as ctx: tts.check_for_truncation(" ".join(["w"] * 150), 25.0, "Chunk 1") self.assertIn("truncated", str(ctx.exception)) def test_duration_at_ratio_passes(self): tts.check_for_truncation(" ".join(["w"] * 150), 30.0, "Chunk 1") def test_unknown_duration_skips_check(self): tts.check_for_truncation(" ".join(["w"] * 150), None, "Chunk 1") def test_short_requests_skip_check(self): tts.check_for_truncation(" ".join(["w"] * 9), 0.0, "Chunk 1") def test_zero_duration_fails_checked_requests(self): with self.assertRaises(RuntimeError): tts.check_for_truncation(" ".join(["w"] * 150), 0.0, "Chunk 1") class FasterTTSClientTruncationTests(unittest.TestCase): """Audio far shorter than its text implies fails the faster request.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self._chunks = patch.object(tts, "CHUNKS_FOLDER", Path(self._tmp.name)) self._chunks.start() def tearDown(self): self._chunks.stop() self._tmp.cleanup() def _make_client(self): client = FasterTTSClient.__new__(FasterTTSClient) client.voice = "default" client.api_url = "http://127.0.0.1:8000" return client def test_truncated_pcm_fails_the_chunk(self): client = self._make_client() text = " ".join(f"word{i}" for i in range(12)) with patch.object(client, "_request_pcm", return_value=b"\x01\x00" * 24), \ self.assertLogs("converter.tts", level="ERROR") as logs: result = client.generate_chunk(text, 1) self.assertIsNone(result) self.assertTrue(any("truncated" in line for line in logs.output)) 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 * tts.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() self._chunks = patch.object(tts, "CHUNKS_FOLDER", Path(self._tmp.name)) self._chunks.start() def tearDown(self): self._chunks.stop() self._tmp.cleanup() def _make_client(self): client = QwenTTSClient.__new__(QwenTTSClient) client.voice_mode = tts.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(tts.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_truncated_sub_request_fails_the_chunk(self): client = self._make_client() # 0.1s of audio for 12 words (expected >= 2.4s). source = self._write_wav(Path(self._tmp.name) / "short.wav", b"\x01\x00" * int(0.1 * tts.SAMPLE_RATE)) text = " ".join(f"word{i}" for i in range(12)) with patch.object(client, "_generate_custom_voice", return_value=(str(source),)) as mock_generate, \ self.assertLogs("converter.tts", level="ERROR") as logs: result = client.generate_chunk(text, 1) self.assertIsNone(result) self.assertEqual(mock_generate.call_count, 1) self.assertTrue(any("truncated" in line for line in logs.output)) 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 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": config.AUDIOCPP_MODEL_ID}]}) 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, **kwargs): with patch("converter.tts.urllib.request.urlopen", side_effect=self._get_responses(**kwargs)): return AudioCppTTSClient(voice=voice, language=language) def test_unreachable_server_raises_with_readme_pointer(self): import urllib.error with patch("converter.tts.urllib.request.urlopen", side_effect=urllib.error.URLError("Connection refused")): with self.assertRaises(RuntimeError) as ctx: AudioCppTTSClient() 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(config.AUDIOCPP_MODEL_ID, message) self.assertIn("pocket-tts", message) self.assertIn("other", message) def test_healthy_server_speaker_mode_defaults(self): client = self._client() self.assertEqual(client.api_url, config.AUDIOCPP_API_URL.rstrip("/")) self.assertEqual(client.model_id, config.AUDIOCPP_MODEL_ID) self.assertEqual(client.language, config.LANGUAGE) self.assertEqual(client.voice, "Vivian") self.assertFalse(client.preset_mode) def test_speaker_mode_uses_configured_speaker(self): with patch.object(config, "SPEAKER", "uncle_fu"): client = self._client() self.assertEqual(client.voice, "Uncle Fu") 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.tts.urllib.request.urlopen") as mock_urlopen: with self.assertRaises(ValueError): AudioCppTTSClient(language="klingon") mock_urlopen.assert_not_called() def test_explicit_language_normalized(self): client = self._client(language="ja") 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() self.assertGreaterEqual(client._seed, 0) def test_preset_mode_routes_to_clone_model_when_configured(self): with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen3-tts"), \ patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"): client = self._client( voice="narrator", models={"data": [{"id": "qwen3-tts"}, {"id": "qwen3-tts-clone"}]}) self.assertEqual(client.model_id, "qwen3-tts-clone") def test_preset_mode_falls_back_when_clone_model_not_on_server(self): with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen3-tts"), \ patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"), \ self.assertLogs("converter.tts", level="WARNING") as logs: client = self._client( voice="narrator", models={"data": [{"id": "qwen3-tts"}, {"id": "pocket-tts"}]}) self.assertEqual(client.model_id, "qwen3-tts") self.assertTrue(any("qwen3-tts-clone" in line for line in logs.output)) def test_clone_model_id_ignored_for_speaker_mode(self): with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen3-tts"), \ patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"): client = self._client( models={"data": [{"id": "qwen3-tts"}, {"id": "qwen3-tts-clone"}]}) self.assertEqual(client.model_id, "qwen3-tts") def test_clone_model_id_equal_to_primary_is_noop(self): with patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", config.AUDIOCPP_MODEL_ID): client = self._client(voice="narrator") self.assertEqual(client.model_id, config.AUDIOCPP_MODEL_ID) def test_preset_mode_with_clone_only_server_uses_clone_model(self): with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen3-tts"), \ patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"): client = self._client( voice="narrator", models={"data": [{"id": "qwen3-tts-clone"}]}) self.assertEqual(client.model_id, "qwen3-tts-clone") def test_speaker_mode_with_clone_only_server_suggests_voice(self): with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen3-tts"), \ patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"): with self.assertRaises(RuntimeError) as ctx: self._client(models={"data": [{"id": "qwen3-tts-clone"}]}) message = str(ctx.exception) self.assertIn("qwen3-tts", message) self.assertIn("--voice", message) def test_preset_mode_with_no_matching_model_lists_both_ids(self): with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen3-tts"), \ patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"), \ self.assertLogs("converter.tts", level="WARNING"): with self.assertRaises(RuntimeError) as ctx: self._client(voice="narrator", models={"data": [{"id": "pocket-tts"}]}) message = str(ctx.exception) self.assertIn("qwen3-tts", message) self.assertIn("qwen3-tts-clone", message) self.assertIn("pocket-tts", message) class AudioCppTTSClientRequestTests(unittest.TestCase): """The /v1/audio/speech payload and response validation.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self._chunks = patch.object(tts, "CHUNKS_FOLDER", Path(self._tmp.name)) self._chunks.start() self._sleep = patch("converter.tts.time.sleep") self._sleep.start() def tearDown(self): self._sleep.stop() self._chunks.stop() self._tmp.cleanup() @staticmethod def _make_client(preset_mode=False, voice="Vivian", language="English", seed=-1, chunk_text=True): client = AudioCppTTSClient.__new__(AudioCppTTSClient) client.api_url = "http://127.0.0.1:8080" client.model_id = config.AUDIOCPP_MODEL_ID client.preset_mode = preset_mode client.voice = voice client.language = language client._seed = seed client.chunk_text = chunk_text return client @staticmethod def _wav_bytes(frames=b"\x01\x00" * 10, rate=tts.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.tts.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"], config.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.tts.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_whole_text_sent_as_one_request_without_client_chunking(self): client = self._make_client(chunk_text=False) # 9 words with CHUNK_SIZE=5 would split in two if client chunking # were on; kept under the 10-word truncation-check threshold. text = " ".join(f"word{i}" for i in range(9)) with patch.object(config, "CHUNK_SIZE", 5), \ patch.object(client, "_request_wav", return_value=self._wav_bytes()) as mock_request: result = client.generate_chunk(text, 1) self.assertIsNotNone(result) self.assertEqual(mock_request.call_count, 1) self.assertEqual(mock_request.call_args[0][0], text) def test_single_request_timeout_scales_with_text_length(self): client = self._make_client(chunk_text=False) long_text = " ".join(f"word{i}" for i in range(1500)) # ~10 min of audio with patch("converter.tts.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.assertGreater(timeout, config.API_TIMEOUT) def test_client_chunking_keeps_configured_timeout(self): client = self._make_client(chunk_text=True) long_text = " ".join(f"word{i}" for i in range(1500)) with patch("converter.tts.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_sends_instruct(self): client = self._make_client(preset_mode=False) with patch("converter.tts.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"], config.INSTRUCT) 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.tts.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.tts.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_transient_failure_is_retried(self): 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.assertIsNotNone(result) self.assertEqual(mock_request.call_count, 2) def test_exhausted_retries_fail_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, config.MAX_RETRIES) 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(), tts.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 AudioCppTTSClientTruncationTests(unittest.TestCase): """Audio far shorter than its text implies fails the request.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self._chunks = patch.object(tts, "CHUNKS_FOLDER", Path(self._tmp.name)) self._chunks.start() def tearDown(self): self._chunks.stop() self._tmp.cleanup() def _make_client(self): client = AudioCppTTSClient.__new__(AudioCppTTSClient) client.api_url = "http://127.0.0.1:8080" client.model_id = config.AUDIOCPP_MODEL_ID client.preset_mode = True client.voice = "narrator" client.language = "English" client._seed = -1 client.chunk_text = True 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(tts.SAMPLE_RATE) wav_file.writeframes(frames) return buffer.getvalue() def test_truncated_wav_fails_the_chunk(self): client = self._make_client() text = " ".join(f"word{i}" for i in range(12)) wav = self._wav_bytes(b"\x01\x00" * 24) # 0.001s for ~4.8s of speech with patch.object(client, "_request_wav", return_value=wav), \ self.assertLogs("converter.tts", level="ERROR") as logs: result = client.generate_chunk(text, 1) self.assertIsNone(result) self.assertTrue(any("truncated" in line for line in logs.output)) 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 * tts.SAMPLE_RATE)) with patch.object(client, "_request_wav", return_value=wav): result = client.generate_chunk(text, 1) self.assertIsNotNone(result) 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=tts.VOICE_MODE_CLONE, backend=tts.BACKEND_FASTER, voice="narrator") mock_faster.assert_called_once_with(voice="narrator") 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=tts.VOICE_MODE_CLONE, backend=tts.BACKEND_AUDIOCPP, voice="narrator", language="ja") mock_audiocpp.assert_called_once_with(voice="narrator", language="Japanese", chunk_text=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=tts.VOICE_MODE_CUSTOM, backend=tts.BACKEND_AUDIOCPP) mock_audiocpp.assert_called_once_with(voice=None, language=config.LANGUAGE, chunk_text=False) def test_audiocpp_backend_chunk_flag_forces_client_chunking(self): with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: converter = AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, backend=tts.BACKEND_AUDIOCPP, voice="narrator", chunk=True) mock_audiocpp.assert_called_once_with(voice="narrator", language=config.LANGUAGE, chunk_text=True) self.assertTrue(converter.client_chunks) def test_gradio_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=tts.VOICE_MODE_CUSTOM) mock_qwen.assert_called_once() mock_faster.assert_not_called() mock_audiocpp.assert_not_called() def test_gradio_clone_mode_still_requires_reference(self): with patch("converter.converter.QwenTTSClient"): with self.assertRaises(ValueError): AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE) 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=tts.VOICE_MODE_CLONE, backend=tts.BACKEND_AUDIOCPP, voice="narrator") self.assertIsNone(converter.voice_clone_ref_audio) def test_chapter_chunks_audiocpp_default_is_one_request(self): converter = self._audiocpp_converter(voice="narrator") text = " ".join(f"word{i}" for i in range(50)) with patch.object(config, "CHUNK_SIZE", 10): self.assertEqual(converter._chapter_chunks(text), [text]) def test_chapter_chunks_audiocpp_chunk_flag_splits(self): with patch("converter.converter.AudioCppTTSClient"): converter = AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, backend=tts.BACKEND_AUDIOCPP, voice="narrator", chunk=True) 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_gradio_always_splits(self): with patch("converter.converter.QwenTTSClient"): converter = AudiobookConverter(voice_mode=tts.VOICE_MODE_CUSTOM) 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=tts.BACKEND_FASTER, speed=0) with self.assertRaises(ValueError): AudiobookConverter(backend=tts.BACKEND_FASTER, language="klingon") def test_audiocpp_backend_still_validates_other_settings(self): with patch("converter.converter.AudioCppTTSClient"): with self.assertRaises(ValueError): AudiobookConverter(backend=tts.BACKEND_AUDIOCPP, speed=0) with self.assertRaises(ValueError): AudiobookConverter(backend=tts.BACKEND_AUDIOCPP, language="klingon") def _faster_converter(self, voice=None): with patch("converter.converter.FasterTTSClient"): return AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, backend=tts.BACKEND_FASTER, voice=voice) def _audiocpp_converter(self, voice=None): with patch("converter.converter.AudioCppTTSClient"): return AudiobookConverter( voice_mode=tts.VOICE_MODE_CLONE if voice else tts.VOICE_MODE_CUSTOM, backend=tts.BACKEND_AUDIOCPP, voice=voice) 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_falls_back_to_config_voice(self): converter = self._faster_converter() self.assertEqual(converter._narrator_tag(), config.FASTER_VOICE) 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_falls_back_to_speaker(self): converter = self._audiocpp_converter() self.assertEqual(converter._narrator_tag(), "Vivian") 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() converter._print_banner() 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=tts.VOICE_MODE_CLONE, voice_clone_ref_audio=str(ref)) self.assertEqual(converter._narrator_tag(), "ref") if __name__ == "__main__": unittest.main()