From f00249db9d1ea051d29aa1bcca869fc4b88e83eb Mon Sep 17 00:00:00 2001 From: historia Date: Mon, 24 Aug 2026 02:59:26 -0400 Subject: refactor: add app directory, dir structure change --- tests/test_tts.py | 1513 ----------------------------------------------------- 1 file changed, 1513 deletions(-) delete mode 100644 tests/test_tts.py (limited to 'tests/test_tts.py') diff --git a/tests/test_tts.py b/tests/test_tts.py deleted file mode 100644 index a2df07f..0000000 --- a/tests/test_tts.py +++ /dev/null @@ -1,1513 +0,0 @@ -"""Tests for the TTS client wrappers (language handling and payloads).""" - -import io -import json -import tempfile -import time -import unittest -import wave -from contextlib import redirect_stdout -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") - - 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_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, model_id=None, **kwargs): - with patch("converter.tts.urllib.request.urlopen", - side_effect=self._get_responses(**kwargs)): - return AudioCppTTSClient(voice=voice, language=language, - model_id=model_id) - - 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_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_override_reaches_request(self): - # --model overrides AUDIOCPP_MODEL_ID for the run. - with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen"): - 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_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 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": config.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.tts.urllib.request.urlopen", - side_effect=_dispatch): - return AudioCppTTSClient(voice=voice, instructions=instructions, - request_options=request_options) - - def test_missing_task_falls_back_to_tts(self): - # Servers that predate the task field hosted plain TTS models. - client = self._client(models={"data": [ - {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts"}]}) - self.assertEqual(client.task, tts.AUDIOCPP_TASK_TTS) - self.assertFalse(client.design_mode) - - def test_task_detected_from_models_endpoint(self): - client = self._client(models={"data": [ - {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", - "task": "vdes"}]}, - instructions="A warm adult narrator") - self.assertEqual(client.task, tts.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": config.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": config.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": config.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": config.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": config.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": config.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(models={"data": [ - {"id": config.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_config_instructions_used_when_flag_omitted(self): - with patch.object(config, "AUDIOCPP_INSTRUCTIONS", - "A calm elderly storyteller"): - client = self._client(models={"data": [ - {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", - "task": "vdes"}]}) - self.assertEqual(client.instructions, "A calm elderly storyteller") - - def test_explicit_instructions_override_config_default(self): - with patch.object(config, "AUDIOCPP_INSTRUCTIONS", "from config"): - client = self._client(models={"data": [ - {"id": config.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.tts.urllib.request.urlopen", - side_effect=_dispatch): - return AudioCppTTSClient(voice=voice) - - def test_family_detected_from_models_endpoint(self): - client = self._client(models={"data": [ - {"id": config.AUDIOCPP_MODEL_ID, "family": "higgs_audio_tts"}]}) - self.assertEqual(client.family, "higgs_audio_tts") - self.assertIs(client.profile, tts.AUDIOCPP_DEFAULT_FAMILY_PROFILE) - - def test_missing_family_falls_back_to_qwen3_tts(self): - client = self._client(models={"data": [ - {"id": config.AUDIOCPP_MODEL_ID}]}) - self.assertEqual(client.family, "qwen3_tts") - self.assertTrue(client.profile.builtin_speakers) - - def test_unknown_family_uses_generic_profile(self): - client = self._client(models={"data": [ - {"id": config.AUDIOCPP_MODEL_ID, "family": "future_tts"}]}) - self.assertEqual(client.family, "future_tts") - self.assertIs(client.profile, tts.AUDIOCPP_DEFAULT_FAMILY_PROFILE) - self.assertFalse(client.profile.builtin_speakers) - self.assertEqual(client.profile.language_style, tts.AUDIOCPP_LANG_OMIT) - - def test_speaker_mode_rejected_for_clone_only_family(self): - client = None - try: - client = self._client(voice=None, models={"data": [ - {"id": config.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_qwen_family(self): - client = self._client(voice=None, models={"data": [ - {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts"}]}) - self.assertEqual(client.family, "qwen3_tts") - - def test_clone_model_id_of_different_family_is_ignored(self): - with patch.object(config, "AUDIOCPP_MODEL_ID", "higgs"), \ - patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen-clone"), \ - self.assertLogs("converter.tts", level="WARNING") as logs: - client = self._client(models={"data": [ - {"id": "higgs", "family": "higgs_audio_tts"}, - {"id": "qwen-clone", "family": "qwen3_tts"}]}) - self.assertEqual(client.model_id, "higgs") - self.assertTrue(any("different family" in line.lower() or - "hosts family" in line.lower() - for line in logs.output)) - - def test_clone_model_id_missing_on_non_qwen_server_is_debug_only(self): - with patch.object(config, "AUDIOCPP_MODEL_ID", "higgs"), \ - patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen-clone"), \ - self.assertNoLogs("converter.tts", level="WARNING"): - client = self._client(models={"data": [ - {"id": "higgs", "family": "higgs_audio_tts"}]}) - self.assertEqual(client.model_id, "higgs") - - def test_clone_model_id_missing_on_qwen_server_still_warns(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(models={"data": [ - {"id": "qwen3-tts", "family": "qwen3_tts"}, - {"id": "pocket-tts", "family": "pocket_tts"}]}) - self.assertEqual(client.model_id, "qwen3-tts") - self.assertTrue(any("qwen3-tts-clone" in line for line in logs.output)) - - def test_iso_language_code_helper(self): - self.assertEqual(tts.LANGUAGE_ISO_CODES["English"], "en") - self.assertIsNone(tts.LANGUAGE_ISO_CODES.get("Auto")) - - -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, family="qwen3_tts", task="tts", - instructions=None, request_options=None): - 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 - client.family = family - client.task = task - client.profile = tts.AUDIOCPP_FAMILY_PROFILES.get( - family, tts.AUDIOCPP_DEFAULT_FAMILY_PROFILE) - client.instructions = instructions or "" - client.request_options = dict(request_options or {}) - client.design_mode = task == tts.AUDIOCPP_TASK_VDES - # Mirrors the connect-time rule: an instruction-defined voice on a - # family without built-in speakers (design mode takes precedence). - client.instruction_voice = ( - not preset_mode and not client.design_mode - and not client.profile.builtin_speakers - and bool(client.instructions)) - 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. - 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_explicit_instructions_replace_config_instruct(self): - # --instructions overrides the INSTRUCT default in speaker mode. - client = self._make_client(preset_mode=False, - instructions="Read whisper quiet.") - 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"], "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.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["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.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.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.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["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.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.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.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["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.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.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.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.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.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["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.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.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.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["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.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 AudioCppHeartbeatTests(unittest.TestCase): - """The heartbeat label drops 'Chunk' when the server does its own - long-form chunking (chunk_text=False, the default).""" - - 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() - - @staticmethod - def _client(chunk_text): - client = AudioCppTTSClient.__new__(AudioCppTTSClient) - client.api_url = "http://127.0.0.1:8080" - client.model_id = config.AUDIOCPP_MODEL_ID - client.preset_mode = False - client.voice = "Vivian" - client.language = "English" - client._seed = -1 - client.chunk_text = chunk_text - client.family = "qwen3_tts" - client.profile = tts.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(tts.SAMPLE_RATE) - wav_file.writeframes(b"\x01\x00" * 10) - return buffer.getvalue() - - def _run(self, chunk_text): - client = self._client(chunk_text) - - 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_with_retry", - side_effect=slow_request), \ - redirect_stdout(buf): - result = client.generate_chunk("Hello.", 1) - self.assertTrue(result) - return buf.getvalue() - - def test_server_side_chunking_heartbeat_has_no_chunk_word(self): - out = self._run(chunk_text=False) - self.assertIn("Request still generating", out) - self.assertNotIn("Chunk", out) - - def test_client_side_chunking_heartbeat_keeps_chunk_word(self): - out = self._run(chunk_text=True) - 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() - 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 - client.family = "qwen3_tts" - client.profile = tts.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(tts.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 * 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, model_id=None, - instructions=None, - request_options={}) - 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, model_id=None, - instructions=None, - request_options={}) - - 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, model_id=None, - instructions=None, - request_options={}) - self.assertTrue(converter.client_chunks) - - def test_audiocpp_backend_model_id_is_wired_through(self): - with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: - AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, - backend=tts.BACKEND_AUDIOCPP, voice="narrator", - model_id="higgs") - mock_audiocpp.assert_called_once_with( - voice="narrator", language=config.LANGUAGE, - chunk_text=False, model_id="higgs", instructions=None, - request_options={}) - - def test_audiocpp_backend_instructions_and_options_are_wired_through(self): - with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: - AudiobookConverter(voice_mode=tts.VOICE_MODE_CUSTOM, - backend=tts.BACKEND_AUDIOCPP, - instructions="A warm adult narrator", - request_options={"emotion": "neutral", - "speed": "1.1"}) - mock_audiocpp.assert_called_once_with( - voice=None, language=config.LANGUAGE, - chunk_text=False, model_id=None, - instructions="A warm adult narrator", - request_options={"emotion": "neutral", "speed": "1.1"}) - - 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=tts.VOICE_MODE_CUSTOM, - backend=tts.BACKEND_QWEN) - mock_qwen.assert_called_once() - 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=tts.VOICE_MODE_CLONE, - backend=tts.BACKEND_QWEN) - - 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_qwen_always_splits(self): - with patch("converter.converter.QwenTTSClient"): - converter = AudiobookConverter(voice_mode=tts.VOICE_MODE_CUSTOM, - backend=tts.BACKEND_QWEN) - 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_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=tts.VOICE_MODE_CLONE, - voice_clone_ref_audio=str(ref), - backend=tts.BACKEND_QWEN) - self.assertEqual(converter._narrator_tag(), "ref") - - -if __name__ == "__main__": - unittest.main() -- cgit v1.2.3