diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_converter.py | 45 | ||||
| -rw-r--r-- | tests/test_make_audiocpp_server_json.py | 585 | ||||
| -rw-r--r-- | tests/test_tts.py | 254 |
3 files changed, 736 insertions, 148 deletions
diff --git a/tests/test_converter.py b/tests/test_converter.py index f09e151..2fe0f5d 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -125,12 +125,13 @@ class FindExistingOutputsTests(unittest.TestCase): class NarratorTagTests(unittest.TestCase): - def _converter(self, voice_mode, ref_audio=None): + def _converter(self, voice_mode, ref_audio=None, instructions=None): converter = AudiobookConverter.__new__(AudiobookConverter) converter.voice_mode = voice_mode converter.voice_clone_ref_audio = ref_audio converter.backend = tts.BACKEND_QWEN converter.voice = None + converter.instructions = instructions return converter def test_custom_voice_uses_speaker_display_name(self): @@ -158,6 +159,47 @@ class NarratorTagTests(unittest.TestCase): self.assertEqual(self._converter(tts.VOICE_MODE_CLONE, "/x/???.wav")._narrator_tag(), "narrator") + def _audiocpp_converter(self, voice=None, instructions=None): + converter = self._converter(tts.VOICE_MODE_CUSTOM, + instructions=instructions) + converter.backend = tts.BACKEND_AUDIOCPP + converter.voice = voice + return converter + + def test_audiocpp_design_run_uses_designed_tag(self): + # An instruction without a voice (voice design, or instruction- + # defined voices) must not be named after the built-in speaker. + converter = self._audiocpp_converter(instructions="A warm narrator") + self.assertEqual(converter._narrator_tag(), "designed") + + def test_audiocpp_instruction_with_voice_keeps_voice_tag(self): + converter = self._audiocpp_converter( + voice="narrator", instructions="Calm delivery") + self.assertEqual(converter._narrator_tag(), "narrator") + + def test_audiocpp_speaker_mode_keeps_speaker_tag(self): + converter = self._audiocpp_converter() + self.assertEqual(converter._narrator_tag(), "Vivian") + + def test_preflight_design_run_uses_designed_tag(self): + with tempfile.TemporaryDirectory() as books_tmp, \ + tempfile.TemporaryDirectory() as output_tmp: + original = (converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER) + converter_mod.BOOKS_FOLDER = Path(books_tmp) + converter_mod.AUDIOBOOKS_FOLDER = Path(output_tmp) + try: + (converter_mod.BOOKS_FOLDER / "book.txt").write_text( + "hello world", encoding="utf-8") + with patch("builtins.input", + side_effect=AssertionError("should not prompt")): + _, planned = AudiobookConverter.preflight_overwrites( + tts.BACKEND_AUDIOCPP, None, tts.VOICE_MODE_CUSTOM, + None, "mp3", instructions="A warm narrator") + self.assertEqual(planned, [(converter_mod.BOOKS_FOLDER / "book.txt", + "book_designed")]) + finally: + converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = original + class ChapterDebugDirTests(unittest.TestCase): """Per-chapter debug subfolder naming (chunk numbering restarts per chapter).""" @@ -538,6 +580,7 @@ class RunOverwritePromptTests(unittest.TestCase): self.converter.voice_clone_ref_audio = None self.converter.backend = tts.BACKEND_QWEN self.converter.voice = None + self.converter.instructions = None self.converter.speed = 1.0 self.converter.single_file = False self.converter.output_format = "mp3" diff --git a/tests/test_make_audiocpp_server_json.py b/tests/test_make_audiocpp_server_json.py index d2d93bb..ca17e93 100644 --- a/tests/test_make_audiocpp_server_json.py +++ b/tests/test_make_audiocpp_server_json.py @@ -1,5 +1,6 @@ """Tests for the audio.cpp server.json generator tool.""" +import argparse import io import json import sys @@ -58,11 +59,17 @@ def _make_checkout(tmp: Path) -> Path: _write_spec(checkout, "qwen3_tts", display_name="Qwen3-TTS", tasks=("tts", "clone", "design"), languages=("zh", "en", "ja"), - packages=[{ - "id": "qwen3_tts_1_7b_base_q8_0", "default": True, - "format": "gguf", - "target_directory": "Qwen3-TTS-12Hz-1.7B-Base-GGUF", - }]) + packages=[ + {"id": "qwen3_tts_1_7b_base_q8_0", "default": True, + "format": "gguf", + "target_directory": "Qwen3-TTS-12Hz-1.7B-Base-GGUF"}, + {"id": "qwen3_tts_1_7b_customvoice_q8_0", + "format": "gguf", + "target_directory": "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF"}, + {"id": "qwen3_tts_1_7b_voicedesign_q8_0", + "format": "gguf", + "target_directory": "Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF"}, + ]) _write_spec(checkout, "higgs_audio_tts", display_name="Higgs Audio v3 TTS 4B", languages=("auto",), packages=[{ @@ -286,6 +293,23 @@ class ResolveWavDirArgTests(unittest.TestCase): self.folder.resolve()) +class NormalizeDirArgTests(unittest.TestCase): + """Path normalization for the audio.cpp checkout argument.""" + + def test_expands_tilde_and_resolves(self): + with patch.object(make_server.os.path, "expanduser", + return_value="/home/u/audio.cpp") as mock_expand: + result = make_server.normalize_dir_arg("~/audio.cpp") + mock_expand.assert_called_once_with("~/audio.cpp") + self.assertEqual(result, Path("/home/u/audio.cpp").resolve()) + + def test_strips_quotes_and_whitespace(self): + with patch.object(make_server.os.path, "expanduser", + side_effect=lambda s: s): + result = make_server.normalize_dir_arg(' "/tmp/foo" ') + self.assertEqual(result, Path("/tmp/foo").resolve()) + + class DefaultModelIdTests(unittest.TestCase): def test_preferred_ids_for_tested_families(self): self.assertEqual(make_server.default_model_id("qwen3_tts"), "qwen") @@ -444,6 +468,12 @@ class BuildServerConfigTests(unittest.TestCase): self.assertEqual(entry["task"], "tts") self.assertEqual(entry["mode"], "offline") + def test_model_entry_design_task(self): + entry = make_server.build_model_entry( + "qwen3_tts", "qwen-design", "p", task="vdes") + self.assertEqual(entry["task"], "vdes") + self.assertEqual(entry["mode"], "offline") + class TranscribeWavDirTests(unittest.TestCase): def setUp(self): @@ -522,6 +552,67 @@ class PromptHelperTests(unittest.TestCase): "one") +class DesignPackageTests(unittest.TestCase): + """Voice-design package detection.""" + + def test_detects_voicedesign_in_id(self): + self.assertTrue(make_server.is_design_package( + {"id": "qwen3_tts_1_7b_voicedesign_q8_0"})) + + def test_detects_voicedesign_in_directory(self): + self.assertTrue(make_server.is_design_package( + {"target_directory": "Foo-VoiceDesign-GGUF"})) + + def test_detects_separated_voice_design(self): + self.assertTrue(make_server.is_design_package( + {"display_name": "Voice Design Q8_0"})) + + def test_ignores_other_packages(self): + self.assertFalse(make_server.is_design_package( + {"id": "higgs_audio_tts_4b_q8_0"})) + self.assertFalse(make_server.is_design_package({})) + + +class PackageDirOptionsTests(unittest.TestCase): + """Grouping a family's packages into distinct target directories.""" + + def test_groups_precisions_and_marks_recommended(self): + entry = { + "family": "qwen3_tts", + "packages": [ + {"id": "base_q8", "default": True, "format": "gguf", + "target_directory": "Base-GGUF"}, + {"id": "base_bf16", "format": "gguf", + "target_directory": "Base-GGUF"}, + {"id": "voicedesign_q8", "format": "gguf", + "target_directory": "VoiceDesign-GGUF"}, + ], + } + options = make_server.package_dir_options(entry) + self.assertEqual([o["target_directory"] for o in options], + ["Base-GGUF", "VoiceDesign-GGUF"]) + self.assertTrue(options[0]["recommended"]) + self.assertFalse(options[0]["design"]) + self.assertFalse(options[1]["recommended"]) + self.assertTrue(options[1]["design"]) + # The recommended precision inside the shared directory wins. + self.assertEqual(options[0]["install_id"], "base_q8") + + def test_recommended_comes_first_even_if_listed_later(self): + entry = { + "family": "demo_tts", + "packages": [ + {"id": "demo_other", "format": "gguf", + "target_directory": "Other-GGUF"}, + {"id": "demo_default", "default": True, "format": "gguf", + "target_directory": "Default-GGUF"}, + ], + } + options = make_server.package_dir_options(entry) + self.assertEqual([o["target_directory"] for o in options], + ["Default-GGUF", "Other-GGUF"]) + + class _MainTestBase(unittest.TestCase): """Shared fixtures for end-to-end main() tests.""" @@ -539,6 +630,12 @@ class _MainTestBase(unittest.TestCase): patcher = patch.object(make_server, "CONFIG_PATH", self.fake_config) patcher.start() self.addCleanup(patcher.stop) + # Force the line-prompt flow regardless of the test terminal, so + # the builtins.input patches below are what actually answer the + # questions (the TUI path is exercised separately). + patcher = patch.object(make_server, "_tui_enabled", return_value=False) + patcher.start() + self.addCleanup(patcher.stop) def tearDown(self): self._td.cleanup() @@ -555,30 +652,35 @@ class _MainTestBase(unittest.TestCase): return_value=whisper): return make_server.main() + # Default single-family run inputs (no flags, port matches config): + # family, host, port, backend, lazy, model-id-sync. + def _defaults(self, sync="y"): + return ["", "", "", "", "", sync] + class MainTests(_MainTestBase): - """The default Qwen3-TTS flow and shared server settings.""" + """The default single-family flow and shared server settings.""" def _args(self, *extra): - return [str(self.folder), "--output", str(self.output), + return ["--wavs", str(self.folder), "--output", str(self.output), "--audiocpp-dir", str(self.checkout)] + list(extra) - # Default Qwen3-TTS "both" run inputs (no flags, port matches config): - # families, models, custom_path, base_path, host, port, backend, lazy, confirm - def _defaults(self, confirm="y"): - return ["", "", "", "", "", "", "", "", confirm] - - def test_required_wav_dir_missing_prints_usage(self): - with self.assertRaises(SystemExit) as ctx: - self._run(["--output", str(self.output), - "--audiocpp-dir", str(self.checkout)], inputs=[]) + def test_missing_wav_dir_prompted_errors(self): + # No --wavs and EOF at the prompt -> hard error. + buf = io.StringIO() + with patch.object(sys, "argv", + ["make_audiocpp_server_json.py", + "--output", str(self.output), + "--audiocpp-dir", str(self.checkout)]), \ + patch("builtins.input", side_effect=EOFError), \ + redirect_stdout(buf): + with self.assertRaises(SystemExit) as ctx: + make_server.main() self.assertEqual(ctx.exception.code, 2) - self.assertFalse(self.output.exists()) def test_missing_audiocpp_dir_errors(self): with self.assertRaises(SystemExit) as ctx: - self._run([str(self.folder), "--output", str(self.output), - "--audiocpp-dir", str(self.root / "nope")], + self._run(self._args("--audiocpp-dir", str(self.root / "nope")), inputs=[]) self.assertEqual(ctx.exception.code, 2) @@ -587,29 +689,28 @@ class MainTests(_MainTestBase): buf = io.StringIO() with patch.object(sys, "argv", ["make_audiocpp_server_json.py", - str(self.folder), "--output", str(self.output)]), \ + "--wavs", str(self.folder), + "--output", str(self.output)]), \ patch("builtins.input", side_effect=EOFError), \ redirect_stdout(buf): with self.assertRaises(SystemExit) as ctx: make_server.main() self.assertEqual(ctx.exception.code, 2) - def test_default_run_hosts_both_models(self): + def test_default_run_hosts_recommended_entry(self): exit_code = self._run(self._args(), inputs=self._defaults()) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) self.assertEqual(data["host"], "127.0.0.1") self.assertEqual(data["port"], make_server.config_port()) self.assertEqual(data["backend"], "cuda") - # Single family (qwen3_tts) -> lazy defaults to False. + # Single family -> one entry, lazy defaults to False. self.assertFalse(data["lazy_load"]) - self.assertEqual( - [model["id"] for model in data["models"]], - [config.AUDIOCPP_MODEL_ID, config.AUDIOCPP_CLONE_MODEL_ID]) + self.assertEqual([model["id"] for model in data["models"]], ["qwen"]) self.assertEqual( [model["path"] for model in data["models"]], - [make_server.DEFAULT_CUSTOM_VOICE_PATH, - make_server.DEFAULT_BASE_PATH]) + ["models/Qwen3-TTS-12Hz-1.7B-Base-GGUF"]) + self.assertEqual(data["models"][0]["task"], "tts") # voice_dir only when wavs are present; this run has none. self.assertNotIn("voice_dir", data) @@ -617,71 +718,14 @@ class MainTests(_MainTestBase): exit_code = self._run(self._args()) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(len(data["models"]), 2) - - def test_clone_only_run(self): - (self.folder / "narrator.wav").write_bytes(b"x") - (self.folder / "alpha.wav").write_bytes(b"x") - # families=default, models=3(clone), custom_path skipped, base_path, - # host, port, backend, lazy, confirm - inputs = ["", "3", "", "", "", "", "", "y"] - exit_code = self._run( - self._args(), - inputs=inputs, - transcribe=lambda path, model_name="base": - f"transcript of {Path(path).name}") - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) self.assertEqual(len(data["models"]), 1) - clone_entry = data["models"][0] - self.assertEqual(clone_entry["id"], config.AUDIOCPP_CLONE_MODEL_ID) - # Voice presets now live in a server-level voice_dir + prompt_text, - # not per-entry voice_presets. - self.assertNotIn("voice_presets", clone_entry) - self.assertIn("voice_dir", data) - self.assertEqual(data["voice_dir"], str(self.folder.resolve())) - prompt = (self.folder / make_server.PROMPT_TEXT_FILENAME).read_text( - encoding="utf-8") - self.assertIn("narrator|transcript of narrator.wav", prompt) - self.assertIn("alpha|transcript of alpha.wav", prompt) - - def test_custom_only_single_model(self): - # families=default, models=2(custom), host, port, backend, lazy, confirm - inputs = ["", "2", "", "", "", "", "y"] - exit_code = self._run( - self._args("--models", "custom"), inputs=inputs) - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual([model["id"] for model in data["models"]], - [config.AUDIOCPP_MODEL_ID]) - - def test_duplicate_ids_prompt_for_distinct_clone_id(self): - with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen"), \ - patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen"): - # families, models(default both), distinct_clone_id, custom_path, - # base_path, host, port, backend, lazy, confirm - inputs = ["", "", "qwen-clone-2", "", "", "", "", "", "", "y"] - exit_code = self._run(self._args(), inputs=inputs) - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual([model["id"] for model in data["models"]], - ["qwen", "qwen-clone-2"]) - - def test_duplicate_ids_eof_exits(self): - with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen"), \ - patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen"): - with self.assertRaises(SystemExit) as ctx: - self._run(self._args()) - self.assertNotEqual(ctx.exception.code, 0) - self.assertFalse(self.output.exists()) def test_port_sync_accepted_updates_config(self): with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1:9999"): # --port 8080 differs from config port 9999 -> sync prompt fires. - # families, models, custom_path, base_path, host, port_sync(y), - # backend, lazy, confirm - inputs = ["", "", "", "", "", "y", "", "", "y"] + # family, host, port_sync(y), backend, lazy, sync(y) + inputs = ["", "", "y", "", "", "y"] exit_code = self._run( self._args("--port", "8080"), inputs=inputs) self.assertEqual(exit_code, 0) @@ -693,7 +737,7 @@ class MainTests(_MainTestBase): def test_port_sync_declined_keeps_config(self): with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1:9999"): - inputs = ["", "", "", "", "", "n", "", "", "y"] + inputs = ["", "", "n", "", "", "n"] exit_code = self._run( self._args("--port", "8080"), inputs=inputs) self.assertEqual(exit_code, 0) @@ -708,11 +752,12 @@ class MainTests(_MainTestBase): self.assertEqual(self.fake_config.read_text(encoding="utf-8"), FAKE_CONFIG) - def test_confirm_declined_writes_nothing(self): - inputs = self._defaults(confirm="n") - exit_code = self._run(self._args(), inputs=inputs) - self.assertEqual(exit_code, 1) - self.assertFalse(self.output.exists()) + def test_no_final_confirm_prompt_writes_file(self): + # There is no final confirmation prompt anymore; the config is always + # written once the (single) overwrite check has been passed. + exit_code = self._run(self._args(), inputs=EOFError) + self.assertEqual(exit_code, 0) + self.assertTrue(self.output.exists()) def test_existing_output_declined_keeps_file(self): self.output.write_text('{"old": true}', encoding="utf-8") @@ -727,7 +772,7 @@ class MainTests(_MainTestBase): exit_code = self._run(self._args(), inputs=inputs) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(len(data["models"]), 2) + self.assertEqual(len(data["models"]), 1) def test_force_overwrites_without_prompt(self): self.output.write_text('{"old": true}', encoding="utf-8") @@ -735,16 +780,16 @@ class MainTests(_MainTestBase): exit_code = self._run(self._args("--force"), inputs=inputs) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(len(data["models"]), 2) + self.assertEqual(len(data["models"]), 1) def test_flags_skip_prompts(self): - # --families qwen3_tts --models both + server flags; port 9000 differs - # from config port 8080 -> the port sync prompt still fires. + # --families qwen3_tts + server flags; port 9000 differs from config + # port 8080 -> the port sync prompt still fires. exit_code = self._run( - self._args("--families", "qwen3_tts", "--models", "both", + self._args("--families", "qwen3_tts", "--host", "0.0.0.0", "--port", "9000", "--backend", "cpu", "--lazy-load"), - inputs=["y", "", "", "y"]) + inputs=["y", "y"]) self.assertEqual(exit_code, 0) self.assertIn('"http://127.0.0.1:9000"', self.fake_config.read_text(encoding="utf-8")) @@ -754,11 +799,11 @@ class MainTests(_MainTestBase): self.assertEqual(data["backend"], "cpu") self.assertTrue(data["lazy_load"]) - def test_missing_positional_wav_dir_errors(self): + def test_missing_wav_dir_flag_errors_with_message(self): missing = self.root / "nope" with self.assertRaises(SystemExit) as ctx, \ patch("sys.stderr") as mock_stderr: - self._run([str(missing), "--output", str(self.output), + self._run(["--wavs", str(missing), "--output", str(self.output), "--audiocpp-dir", str(self.checkout)], inputs=self._defaults()) self.assertEqual(ctx.exception.code, 2) @@ -766,12 +811,80 @@ class MainTests(_MainTestBase): self.assertIn(f"WAV directory not found: {missing.resolve()}", shown) self.assertIn("directory containing the .wav", shown) - def test_models_flag_rejected_without_qwen(self): - with self.assertRaises(SystemExit) as ctx: - self._run(self._args("--families", "higgs_audio_tts", - "--models", "both"), - inputs=[]) - self.assertEqual(ctx.exception.code, 2) + def _run_capturing(self, argv, inputs): + argv = ["make_audiocpp_server_json.py"] + argv + buf = io.StringIO() + with patch.object(sys, "argv", argv), \ + patch("builtins.input", side_effect=inputs), \ + patch.object(make_server, "transcribe_reference_audio"), \ + patch.object(make_server, "whisper_backend_available", + return_value="faster_whisper"), \ + redirect_stdout(buf): + code = make_server.main() + return code, buf.getvalue() + + def test_all_packages_design_hosts_vdes_entry(self): + # --all-packages: pick the VoiceDesign package (menu 3) and accept the + # "design" default so it is hosted with task "vdes". + self.fake_config.write_text(FAKE_CONFIG_WITH_MODEL_IDS, + encoding="utf-8") + # family, packages(3=VoiceDesign), task(design default Enter), host, + # port, backend, lazy, sync(y) + inputs = ["", "3", "", "", "", "", "", "y"] + code, out = self._run_capturing( + self._args("--all-packages"), inputs=inputs) + self.assertEqual(code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["models"], [{ + "id": "qwen-design", + "family": "qwen3_tts", + "path": "models/Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF", + "task": "vdes", + "mode": "offline", + }]) + self.assertNotIn("voice_dir", data) + # Only the VoiceDesign package is installed (custom/base are not). + self.assertIn("install qwen3_tts_1_7b_voicedesign_q8_0", out) + self.assertNotIn("install qwen3_tts_1_7b_customvoice_q8_0", out) + self.assertNotIn("install qwen3_tts_1_7b_base_q8_0", out) + # Usage guidance points at the --instructions flow. + self.assertIn("--model qwen-design", out) + self.assertIn("--instructions", out) + # Single-entry server: the converter ids are synced to the entry. + text = self.fake_config.read_text(encoding="utf-8") + self.assertIn('AUDIOCPP_MODEL_ID = "qwen-design"', text) + self.assertIn('AUDIOCPP_CLONE_MODEL_ID = "qwen-design"', text) + + def test_all_packages_non_design_package_gets_tts_no_prompt(self): + # CustomVoice (menu 2) is not a design package -> task "tts" with no + # task prompt. + inputs = ["", "2", "", "", "", "", "y"] + code, _ = self._run_capturing( + self._args("--all-packages"), inputs=inputs) + self.assertEqual(code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["models"], [{ + "id": "qwen", + "family": "qwen3_tts", + "path": "models/Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF", + "task": "tts", + "mode": "offline", + }]) + + def test_all_packages_both_tts_and_design(self): + # Pick Base (recommended) + VoiceDesign -> two entries; the design + # package prompts for its task. + # family, packages(1,3), task(design default Enter), host, port, + # backend, lazy + inputs = ["", "1,3", "", "", "", "", ""] + code, _ = self._run_capturing( + self._args("--all-packages"), inputs=inputs) + self.assertEqual(code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual([model["id"] for model in data["models"]], + ["qwen", "qwen-design"]) + self.assertEqual([model["task"] for model in data["models"]], + ["tts", "vdes"]) class NonQwenFamilyMainTests(_MainTestBase): @@ -785,16 +898,15 @@ class NonQwenFamilyMainTests(_MainTestBase): encoding="utf-8") def _args(self, family, *extra): - return [str(self.folder), "--output", str(self.output), + return ["--wavs", str(self.folder), "--output", str(self.output), "--audiocpp-dir", str(self.checkout), "--families", family] + list(extra) def test_higgs_family_run(self): (self.folder / "narrator.wav").write_bytes(b"x") - # Single non-qwen family -> path is asked; then host, port, backend, - # lazy, confirm, model-id sync(y). prompt_text is written (no overwrite - # prompt on a fresh directory). - inputs = ["", "", "", "", "", "y", "y"] + # Single family -> path comes from the catalog (no prompt); host, port, + # backend, lazy, model-id sync(y). + inputs = ["", "", "", "", "y"] exit_code = self._run( self._args("higgs_audio_tts"), inputs=inputs, transcribe=lambda path, model_name="base": "a transcript") @@ -813,15 +925,15 @@ class NonQwenFamilyMainTests(_MainTestBase): prompt = (self.folder / make_server.PROMPT_TEXT_FILENAME).read_text( encoding="utf-8") self.assertIn("narrator|a transcript", prompt) - # Single non-qwen entry -> both converter ids are synced to it. + # Single entry -> both converter ids are synced to it. text = self.fake_config.read_text(encoding="utf-8") self.assertIn('AUDIOCPP_MODEL_ID = "higgs"', text) self.assertIn('AUDIOCPP_CLONE_MODEL_ID = "higgs"', text) def test_model_id_sync_declined_keeps_config(self): (self.folder / "narrator.wav").write_bytes(b"x") - # path, host, port, backend, lazy, confirm, sync(n) - inputs = ["", "", "", "", "", "y", "n"] + # host, port, backend, lazy, sync(n) + inputs = ["", "", "", "", "n"] exit_code = self._run( self._args("voxcpm2"), inputs=inputs, transcribe=lambda path, model_name="base": "t") @@ -834,11 +946,11 @@ class NonQwenFamilyMainTests(_MainTestBase): def test_no_wavs_warns_and_omits_voice_dir(self): buf = io.StringIO() - # path, host, port, backend, lazy, confirm, sync(y) - inputs = ["", "", "", "", "", "y", "y"] + # host, port, backend, lazy, sync(y) + inputs = ["", "", "", "", "y"] with patch.object(sys, "argv", ["make_audiocpp_server_json.py", - str(self.folder), "--output", str(self.output), + "--wavs", str(self.folder), "--output", str(self.output), "--audiocpp-dir", str(self.checkout), "--families", "index_tts2"]), \ patch("builtins.input", side_effect=inputs), \ @@ -864,16 +976,14 @@ class MultiFamilyMainTests(_MainTestBase): """Hosting several families in one server.json.""" def _args(self, *extra): - return [str(self.folder), "--output", str(self.output), + return ["--wavs", str(self.folder), "--output", str(self.output), "--audiocpp-dir", str(self.checkout)] + list(extra) def test_multiple_families_lazy_by_default_with_voice_dir(self): (self.folder / "narrator.wav").write_bytes(b"x") - # --families selects qwen3_tts + higgs_audio_tts. qwen is among them - # with others -> qwen sub-flow forced to "both" (no models prompt). - # custom_path, base_path, host, port, backend, lazy(default True->Enter), - # prompt_text overwrite(none yet->writes), confirm - inputs = ["", "", "", "", "", "", "", "y"] + # --families selects qwen3_tts + higgs_audio_tts; each hosts its + # recommended package. host, port, backend, lazy(default True->Enter). + inputs = ["", "", "", ""] exit_code = self._run( self._args("--families", "qwen3_tts,higgs_audio_tts"), inputs=inputs, @@ -881,18 +991,17 @@ class MultiFamilyMainTests(_MainTestBase): self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) ids = [model["id"] for model in data["models"]] - self.assertEqual(ids, ["qwen", "qwen-clone", "higgs"]) - # Two families -> lazy defaults to True. + self.assertEqual(ids, ["qwen", "higgs"]) + # Two entries -> lazy defaults to True. self.assertTrue(data["lazy_load"]) self.assertEqual(data["voice_dir"], str(self.folder.resolve())) - # Multi-entry -> the tool prints a --model note instead of syncing. - higgs = data["models"][2] + higgs = data["models"][1] self.assertEqual(higgs["path"], "models/Higgs-Audio-v3-TTS-4B-GGUF") def test_two_non_qwen_families_use_catalog_paths(self): - # Multiple non-qwen families -> paths are NOT prompted (catalog defaults). - # qwen absent -> no models prompt; host, port, backend, lazy, confirm - inputs = ["", "", "", "", "y"] + # Multiple families -> paths come from the catalog (no prompts). + # host, port, backend, lazy + inputs = ["", "", "", ""] exit_code = self._run( self._args("--families", "higgs_audio_tts,voxcpm2"), inputs=inputs) @@ -902,17 +1011,17 @@ class MultiFamilyMainTests(_MainTestBase): self.assertEqual(by_id["higgs"]["path"], "models/Higgs-Audio-v3-TTS-4B-GGUF") self.assertEqual(by_id["voxcpm2"]["path"], "models/VoxCPM2-GGUF") - # No wavs and both clone-capable, but no wavs present -> no voice_dir. + # No wavs present -> no voice_dir. self.assertNotIn("voice_dir", data) def test_non_clone_family_selected_warns_about_wav_dir(self): buf = io.StringIO() # supertonic is TTS-only (no clone): wav dir is ignored. - # path, host, port, backend, lazy, confirm, sync(n) - inputs = ["", "", "", "", "y", "y", "n"] + # host, port, backend, lazy, sync(n) + inputs = ["", "", "", "", "n"] with patch.object(sys, "argv", ["make_audiocpp_server_json.py", - str(self.folder), "--output", str(self.output), + "--wavs", str(self.folder), "--output", str(self.output), "--audiocpp-dir", str(self.checkout), "--families", "supertonic"]), \ patch("builtins.input", side_effect=inputs), \ @@ -929,11 +1038,43 @@ class MultiFamilyMainTests(_MainTestBase): self.assertEqual(data["models"][0]["family"], "supertonic") +class DefaultOutputTests(_MainTestBase): + """server.json defaults into the audio.cpp checkout unless declined.""" + + def test_default_output_written_into_checkout(self): + # No --output: server.json lands in the audio.cpp checkout. + argv = ["--wavs", str(self.folder), "--audiocpp-dir", str(self.checkout)] + exit_code = self._run(argv, inputs=self._defaults()) + self.assertEqual(exit_code, 0) + out = self.checkout / "server.json" + self.assertTrue(out.exists()) + data = json.loads(out.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 1) + + def test_declined_overwrite_falls_back_to_cwd(self): + # A pre-existing server.json in the checkout; declining the overwrite + # writes server.json into the current working directory instead. + checkout_out = self.checkout / "server.json" + checkout_out.write_text('{"old": true}', encoding="utf-8") + cwd = self.root / "run-cwd" + cwd.mkdir() + argv = ["--wavs", str(self.folder), "--audiocpp-dir", str(self.checkout)] + with patch.object(make_server.os, "getcwd", return_value=str(cwd)): + exit_code = self._run(argv, inputs=["n"] + self._defaults()) + self.assertEqual(exit_code, 0) + self.assertEqual(json.loads(checkout_out.read_text(encoding="utf-8")), + {"old": True}) + fallback = cwd / "server.json" + self.assertTrue(fallback.exists()) + data = json.loads(fallback.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 1) + + class TranscriptWarningTests(_MainTestBase): """Empty transcripts and a missing Whisper backend produce loud warnings.""" def _args(self, *extra): - return [str(self.folder), "--output", str(self.output), + return ["--wavs", str(self.folder), "--output", str(self.output), "--audiocpp-dir", str(self.checkout)] + list(extra) def _run_capturing(self, argv, inputs, transcribe, whisper): @@ -952,9 +1093,7 @@ class TranscriptWarningTests(_MainTestBase): def test_empty_transcript_prints_loud_end_warning(self): (self.folder / "narrator.wav").write_bytes(b"x") (self.folder / "alpha.wav").write_bytes(b"x") - # Qwen clone-only (menu 3); custom_path skipped, base_path, host, port, - # backend, lazy, prompt_text write, confirm - inputs = ["", "3", "", "", "", "", "", "", "y"] + inputs = self._defaults() code, out = self._run_capturing( self._args(), inputs=inputs, transcribe=lambda path, model_name="base": None, @@ -965,20 +1104,20 @@ class TranscriptWarningTests(_MainTestBase): self.assertIn("alpha", out) self.assertIn("prompt_text", out) - def test_missing_whisper_backend_prints_conda_warning(self): + def test_missing_whisper_backend_prints_install_warning(self): (self.folder / "narrator.wav").write_bytes(b"x") - inputs = ["", "3", "", "", "", "", "", "", "y"] + inputs = self._defaults() code, out = self._run_capturing( self._args(), inputs=inputs, transcribe=lambda path, model_name="base": "a transcript", whisper=None) self.assertEqual(code, 0) - self.assertIn("conda activate qwen3-tts", out) + self.assertIn("Install whisper", out) self.assertIn("faster_whisper", out) def test_all_transcripts_present_prints_no_end_warning(self): (self.folder / "narrator.wav").write_bytes(b"x") - inputs = ["", "3", "", "", "", "", "", "", "y"] + inputs = self._defaults() code, out = self._run_capturing( self._args(), inputs=inputs, transcribe=lambda path, model_name="base": "a real transcript", @@ -987,5 +1126,167 @@ class TranscriptWarningTests(_MainTestBase): self.assertNotIn("MANUAL TRANSCRIPTION REQUIRED", out) +class PromptTextReuseTests(_MainTestBase): + """Reusing an existing prompt_text and transcribing only new voices.""" + + def _args(self, *extra): + return ["--wavs", str(self.folder), "--output", str(self.output), + "--audiocpp-dir", str(self.checkout), + "--families", "higgs_audio_tts"] + list(extra) + + def _run_capturing(self, argv, inputs, transcribe): + argv = ["make_audiocpp_server_json.py"] + argv + buf = io.StringIO() + with patch.object(sys, "argv", argv), \ + patch("builtins.input", side_effect=inputs), \ + patch.object(make_server, "transcribe_reference_audio", + side_effect=transcribe), \ + patch.object(make_server, "whisper_backend_available", + return_value="faster_whisper"), \ + redirect_stdout(buf): + code = make_server.main() + return code, buf.getvalue() + + def _transcribe(self, called, text): + def transcribe(path, model_name="base"): + called.append(path) + return text + return transcribe + + def test_all_present_decline_keeps_file_and_skips_transcribe(self): + (self.folder / "narrator.wav").write_bytes(b"x") + prompt = self.folder / make_server.PROMPT_TEXT_FILENAME + prompt.write_text("narrator|An existing transcript.\n", + encoding="utf-8") + called = [] + # host, port, backend, lazy, re-transcribe(n), sync(y) + inputs = ["", "", "", "", "n", "y"] + code, out = self._run_capturing( + self._args(), inputs=inputs, + transcribe=self._transcribe(called, "Fresh.")) + self.assertEqual(code, 0) + self.assertEqual(called, []) + self.assertEqual(prompt.read_text(encoding="utf-8"), + "narrator|An existing transcript.\n") + self.assertIn("Kept existing", out) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["voice_dir"], str(self.folder.resolve())) + + def test_all_present_accept_retranscribes_and_overwrites(self): + (self.folder / "narrator.wav").write_bytes(b"x") + prompt = self.folder / make_server.PROMPT_TEXT_FILENAME + prompt.write_text("narrator|Old.\n", encoding="utf-8") + called = [] + # host, port, backend, lazy, re-transcribe(y), sync(y) + inputs = ["", "", "", "", "y", "y"] + code, _ = self._run_capturing( + self._args(), inputs=inputs, + transcribe=self._transcribe(called, "Fresh.")) + self.assertEqual(code, 0) + self.assertEqual(called, [str(self.folder / "narrator.wav")]) + self.assertIn("narrator|Fresh.", prompt.read_text(encoding="utf-8")) + + def test_new_voice_merges_preserving_hand_edits(self): + (self.folder / "existing.wav").write_bytes(b"x") + (self.folder / "new.wav").write_bytes(b"x") + prompt = self.folder / make_server.PROMPT_TEXT_FILENAME + prompt.write_text("existing|Hand edited transcript.\n", + encoding="utf-8") + called = [] + # host, port, backend, lazy, only-new(Enter -> y), sync(y) + inputs = ["", "", "", "", "", "y"] + code, _ = self._run_capturing( + self._args(), inputs=inputs, + transcribe=self._transcribe(called, "New transcript.")) + self.assertEqual(code, 0) + self.assertEqual(called, [str(self.folder / "new.wav")]) + text = prompt.read_text(encoding="utf-8") + self.assertIn("existing|Hand edited transcript.", text) + self.assertIn("new|New transcript.", text) + + def test_new_voice_decline_retranscribes_all(self): + (self.folder / "existing.wav").write_bytes(b"x") + (self.folder / "new.wav").write_bytes(b"x") + prompt = self.folder / make_server.PROMPT_TEXT_FILENAME + prompt.write_text("existing|Old.\n", encoding="utf-8") + called = [] + # host, port, backend, lazy, only-new(n), sync(y) + inputs = ["", "", "", "", "n", "y"] + code, _ = self._run_capturing( + self._args(), inputs=inputs, + transcribe=self._transcribe(called, "Fresh.")) + self.assertEqual(code, 0) + self.assertEqual(sorted(called), sorted([ + str(self.folder / "existing.wav"), str(self.folder / "new.wav")])) + self.assertIn("existing|Fresh.", prompt.read_text(encoding="utf-8")) + + def test_force_retranscribes_without_prompt(self): + (self.folder / "narrator.wav").write_bytes(b"x") + prompt = self.folder / make_server.PROMPT_TEXT_FILENAME + prompt.write_text("narrator|Old.\n", encoding="utf-8") + called = [] + # host, port, backend, lazy, sync(y); no re-transcribe prompt with force. + inputs = ["", "", "", "", "y"] + code, _ = self._run_capturing( + self._args("--force"), inputs=inputs, + transcribe=self._transcribe(called, "Fresh.")) + self.assertEqual(code, 0) + self.assertEqual(called, [str(self.folder / "narrator.wav")]) + self.assertIn("narrator|Fresh.", prompt.read_text(encoding="utf-8")) + + def test_empty_transcript_counts_as_missing(self): + (self.folder / "narrator.wav").write_bytes(b"x") + prompt = self.folder / make_server.PROMPT_TEXT_FILENAME + prompt.write_text("narrator|\n", encoding="utf-8") + called = [] + # Empty transcript is treated as missing -> the "only new voices" + # prompt fires (Enter -> y). + # host, port, backend, lazy, only-new(Enter), sync(y) + inputs = ["", "", "", "", "", "y"] + code, _ = self._run_capturing( + self._args(), inputs=inputs, + transcribe=self._transcribe(called, "Fresh.")) + self.assertEqual(code, 0) + self.assertEqual(called, [str(self.folder / "narrator.wav")]) + self.assertIn("narrator|Fresh.", prompt.read_text(encoding="utf-8")) + + +class ModeSelectionTests(unittest.TestCase): + """Choosing between the TUI wizard and the line prompts.""" + + def _args(self, notui=False): + return argparse.Namespace(notui=notui) + + def test_notui_flag_forces_prompt_mode(self): + # Even with a tty and an importable curses, --notui disables the TUI. + with patch.object(make_server, "_curses_importable", return_value=True), \ + patch.object(make_server.sys.stdin, "isatty", return_value=True), \ + patch.object(make_server.sys.stdout, "isatty", return_value=True): + self.assertFalse(make_server._tui_enabled(self._args(notui=True))) + + def test_non_tty_forces_prompt_mode(self): + with patch.object(make_server, "_curses_importable", return_value=True), \ + patch.object(make_server.sys.stdin, "isatty", return_value=False), \ + patch.object(make_server.sys.stdout, "isatty", return_value=True): + self.assertFalse(make_server._tui_enabled(self._args())) + + def test_tty_with_curses_uses_tui(self): + with patch.object(make_server, "_curses_importable", return_value=True), \ + patch.object(make_server.sys.stdin, "isatty", return_value=True), \ + patch.object(make_server.sys.stdout, "isatty", return_value=True): + self.assertTrue(make_server._tui_enabled(self._args())) + + def test_missing_curses_forces_prompt_mode(self): + with patch.object(make_server, "_curses_importable", return_value=False), \ + patch.object(make_server.sys.stdin, "isatty", return_value=True), \ + patch.object(make_server.sys.stdout, "isatty", return_value=True): + self.assertFalse(make_server._tui_enabled(self._args())) + + def test_curses_is_importable_on_this_platform(self): + # The TUI widget module imports without curses at module load time, + # but the wizard still needs the real curses package to run. + self.assertTrue(make_server._curses_importable()) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_tts.py b/tests/test_tts.py index 89248f2..a2df07f 100644 --- a/tests/test_tts.py +++ b/tests/test_tts.py @@ -654,6 +654,144 @@ class AudioCppTTSClientHealthTests(unittest.TestCase): 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.""" @@ -768,7 +906,8 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): @staticmethod def _make_client(preset_mode=False, voice="Vivian", language="English", seed=-1, - chunk_text=True, family="qwen3_tts"): + 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 @@ -778,8 +917,18 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): 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 @@ -862,6 +1011,81 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): 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. @@ -1112,7 +1336,9 @@ class BackendWiringTests(unittest.TestCase): backend=tts.BACKEND_AUDIOCPP, voice="narrator", language="ja") mock_audiocpp.assert_called_once_with(voice="narrator", language="Japanese", - chunk_text=False, model_id=None) + chunk_text=False, model_id=None, + instructions=None, + request_options={}) mock_faster.assert_not_called() mock_qwen.assert_not_called() @@ -1121,7 +1347,9 @@ class BackendWiringTests(unittest.TestCase): 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) + 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: @@ -1130,7 +1358,9 @@ class BackendWiringTests(unittest.TestCase): voice="narrator", chunk=True) mock_audiocpp.assert_called_once_with(voice="narrator", language=config.LANGUAGE, - chunk_text=True, model_id=None) + 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): @@ -1140,7 +1370,21 @@ class BackendWiringTests(unittest.TestCase): model_id="higgs") mock_audiocpp.assert_called_once_with( voice="narrator", language=config.LANGUAGE, - chunk_text=False, model_id="higgs") + 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, \ |