diff options
Diffstat (limited to 'tests/test_make_audiocpp_server_json.py')
| -rw-r--r-- | tests/test_make_audiocpp_server_json.py | 585 |
1 files changed, 443 insertions, 142 deletions
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() |