From f7021704b6b26ee747558d9ad701c2b25baedd2a Mon Sep 17 00:00:00 2001 From: historia Date: Fri, 21 Aug 2026 02:31:54 -0400 Subject: feat: catalog-driven audio.cpp server.json creation --- tests/test_make_audiocpp_server_json.py | 809 ++++++++++++++++++++------------ tests/test_tts.py | 49 +- 2 files changed, 562 insertions(+), 296 deletions(-) (limited to 'tests') diff --git a/tests/test_make_audiocpp_server_json.py b/tests/test_make_audiocpp_server_json.py index 2bd262a..d2d93bb 100644 --- a/tests/test_make_audiocpp_server_json.py +++ b/tests/test_make_audiocpp_server_json.py @@ -28,6 +28,81 @@ FAKE_CONFIG_WITH_MODEL_IDS = ( ) +def _write_spec(checkout: Path, family: str, *, display_name=None, + tasks=("tts", "clone"), languages=("en",), packages=None, + category="tts"): + """Write a minimal model_specs/.json into a fake checkout.""" + specs = checkout / "model_specs" + specs.mkdir(parents=True, exist_ok=True) + if packages is None: + packages = [{ + "id": f"{family}_q8_0", "default": True, "format": "gguf", + "target_directory": f"{family}-GGUF", + }] + spec = { + "family": family, + "display_name": display_name or family, + "category": category, + "tasks": list(tasks), + "languages": list(languages), + "packages": packages, + } + (specs / f"{family}.json").write_text(json.dumps(spec), encoding="utf-8") + return spec + + +def _make_checkout(tmp: Path) -> Path: + """Create a fake audio.cpp checkout with a realistic model_specs set.""" + checkout = tmp / "audio.cpp" + checkout.mkdir() + _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", + }]) + _write_spec(checkout, "higgs_audio_tts", display_name="Higgs Audio v3 TTS 4B", + languages=("auto",), + packages=[{ + "id": "higgs_audio_tts_4b_q8_0", "default": True, + "format": "gguf", + "target_directory": "Higgs-Audio-v3-TTS-4B-GGUF", + }]) + _write_spec(checkout, "voxcpm2", display_name="VoxCPM2-2B", + languages=("en", "zh"), + packages=[{ + "id": "voxcpm2_q8_0", "default": True, "format": "gguf", + "target_directory": "VoxCPM2-GGUF", + }]) + _write_spec(checkout, "index_tts2", display_name="IndexTTS-2", + languages=("zh", "en"), + packages=[{ + "id": "index_tts2_q8_0", "default": True, "format": "gguf", + "target_directory": "IndexTTS2-GGUF", + }]) + _write_spec(checkout, "pocket_tts", display_name="PocketTTS-100M", + tasks=("tts", "clone"), languages=("en", "de"), + packages=[{ + "id": "pocket_tts_q8_0", "default": True, "format": "gguf", + "target_directory": "PocketTTS-GGUF", + }]) + _write_spec(checkout, "supertonic", display_name="Supertonic 3", + tasks=("tts",), languages=("en", "ko"), + packages=[{ + "id": "supertonic_q8_0", "default": True, "format": "gguf", + "target_directory": "Supertonic-GGUF", + }]) + # An ASR family that must be filtered out. + _write_spec(checkout, "qwen3_asr", display_name="Qwen3-ASR", + tasks=("asr",), category="asr") + # A TTS family with no installable packages (must be skipped). + _write_spec(checkout, "empty_tts", display_name="Empty TTS", + tasks=("tts",), packages=[]) + return checkout + + class FindWavFilesTests(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() @@ -171,129 +246,249 @@ class UpdateConfigModelIdsTests(unittest.TestCase): config_path=Path(self._tmp.name) / "nope.py")) -class BuildSingleFamilyServerConfigTests(unittest.TestCase): - def test_single_entry_with_presets(self): - presets = {"narrator": {"voice_ref": "/x.wav", - "reference_text": "hi"}} - server_config = make_server.build_single_family_server_config( - host="127.0.0.1", port=8080, backend="cuda", lazy_load=False, - family="higgs_audio_tts", model_id="higgs", - model_path="models/Higgs-Audio-v3-TTS-4B-GGUF", - voice_presets=presets) - self.assertEqual(server_config["host"], "127.0.0.1") - self.assertEqual(server_config["port"], 8080) - self.assertEqual(server_config["backend"], "cuda") - self.assertFalse(server_config["lazy_load"]) - self.assertEqual(len(server_config["models"]), 1) - entry = server_config["models"][0] - self.assertEqual(entry["id"], "higgs") - self.assertEqual(entry["family"], "higgs_audio_tts") - self.assertEqual(entry["path"], "models/Higgs-Audio-v3-TTS-4B-GGUF") - self.assertEqual(entry["task"], "tts") - self.assertEqual(entry["mode"], "offline") - self.assertEqual(entry["voice_presets"], presets) - - def test_no_presets_omits_key(self): - server_config = make_server.build_single_family_server_config( - host="127.0.0.1", port=8080, backend="cpu", lazy_load=True, - family="index_tts2", model_id="indextts2", - model_path="models/IndexTTS2-GGUF", voice_presets={}) - self.assertNotIn("voice_presets", server_config["models"][0]) - - def test_family_entries_reference_real_families(self): - for entry in make_server.FAMILY_ENTRIES: - if entry["key"] == make_server.FAMILY_QWEN3_TTS: - continue - self.assertIn("install", entry) - self.assertIn("default_id", entry) - self.assertIn("default_path", entry) - self.assertIn("family", entry) - - -class BuildVoicePresetsTests(unittest.TestCase): +class ResolveWavDirArgTests(unittest.TestCase): + """Path normalization for the required WAV_DIR argument.""" + def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.folder = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def test_resolves_to_absolute(self): + self.assertEqual(make_server.resolve_wav_dir_arg(str(self.folder)), + self.folder.resolve()) + + def test_strips_surrounding_quotes(self): + quoted = f'"{self.folder}"' + self.assertEqual(make_server.resolve_wav_dir_arg(quoted), + self.folder.resolve()) + + def test_strips_single_quotes(self): + quoted = f"'{self.folder}'" + self.assertEqual(make_server.resolve_wav_dir_arg(quoted), + self.folder.resolve()) + + def test_strips_whitespace(self): + self.assertEqual(make_server.resolve_wav_dir_arg(f" {self.folder} "), + self.folder.resolve()) + + def test_expands_tilde(self): + with patch.object(make_server.os.path, "expanduser", + return_value=str(self.folder)) as mock_expand: + result = make_server.resolve_wav_dir_arg("~/voices") + mock_expand.assert_called_once_with("~/voices") + self.assertEqual(result, self.folder.resolve()) + + def test_trailing_slash_preserved_as_dir(self): + self.assertEqual(make_server.resolve_wav_dir_arg(f"{self.folder}/"), + self.folder.resolve()) + + +class DefaultModelIdTests(unittest.TestCase): + def test_preferred_ids_for_tested_families(self): + self.assertEqual(make_server.default_model_id("qwen3_tts"), "qwen") + self.assertEqual(make_server.default_model_id("higgs_audio_tts"), "higgs") + self.assertEqual(make_server.default_model_id("voxcpm2"), "voxcpm2") + self.assertEqual(make_server.default_model_id("index_tts2"), "indextts2") + + def test_derived_id_strips_trailing_tts_and_underscores(self): + self.assertEqual(make_server.default_model_id("pocket_tts"), "pocket") + self.assertEqual(make_server.default_model_id("dots_tts"), "dots") + # Families without a _tts suffix just drop underscores. + self.assertEqual(make_server.default_model_id("moss_tts_local"), + "mossttslocal") + + +class LoadModelCatalogTests(unittest.TestCase): + def setUp(self): + self._tmp = list(tempfile._mkdtemp() and 0 for _ in range(0)) # noqa + self._td = tempfile.TemporaryDirectory() + self.checkout = _make_checkout(Path(self._td.name)) + + def tearDown(self): + self._td.cleanup() + + def test_includes_tts_families_excludes_asr(self): + catalog = make_server.load_model_catalog(self.checkout) + families = [entry["family"] for entry in catalog] + self.assertIn("qwen3_tts", families) + self.assertIn("higgs_audio_tts", families) + self.assertIn("pocket_tts", families) + self.assertIn("supertonic", families) + self.assertNotIn("qwen3_asr", families) + + def test_skips_families_with_no_packages(self): + catalog = make_server.load_model_catalog(self.checkout) + self.assertNotIn("empty_tts", + [entry["family"] for entry in catalog]) + + def test_tested_families_come_first_in_order(self): + catalog = make_server.load_model_catalog(self.checkout) + tested = [entry["family"] for entry in catalog + if entry["tested"]] + self.assertEqual(tested, list(make_server.TESTED_FAMILIES)) + + def test_non_tested_families_follow_alphabetically(self): + catalog = make_server.load_model_catalog(self.checkout) + non_tested = [entry["family"] for entry in catalog + if not entry["tested"]] + self.assertEqual(non_tested, sorted(non_tested)) + + def test_default_package_and_target_directory_resolved(self): + catalog = make_server.load_model_catalog(self.checkout) + by_family = {entry["family"]: entry for entry in catalog} + higgs = by_family["higgs_audio_tts"] + self.assertEqual(higgs["install_id"], "higgs_audio_tts_4b_q8_0") + self.assertEqual(higgs["default_path"], + "models/Higgs-Audio-v3-TTS-4B-GGUF") + + def test_picks_first_gguf_when_no_default_flag(self): + # Rewrite the voxcpm2 spec so no package is flagged default. + _write_spec(self.checkout, "voxcpm2", display_name="VoxCPM2-2B", + packages=[ + {"id": "voxcpm2_bf16", "format": "gguf", + "target_directory": "VoxCPM2-GGUF"}, + {"id": "voxcpm2_q8_0", "format": "gguf", + "target_directory": "VoxCPM2-GGUF"}, + ]) + catalog = make_server.load_model_catalog(self.checkout) + by_family = {entry["family"]: entry for entry in catalog} + # No default:true -> first gguf package wins. + self.assertEqual(by_family["voxcpm2"]["install_id"], "voxcpm2_bf16") + + def test_clone_capability_from_tasks(self): + catalog = make_server.load_model_catalog(self.checkout) + by_family = {entry["family"]: entry for entry in catalog} + self.assertTrue(by_family["higgs_audio_tts"]["clone_capable"]) + self.assertFalse(by_family["supertonic"]["clone_capable"]) + + def test_missing_model_specs_dir_raises(self): + empty = Path(self._td.name) / "empty" + empty.mkdir() + with self.assertRaises(NotADirectoryError): + make_server.load_model_catalog(empty) + + +class AskFamiliesTests(unittest.TestCase): + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.checkout = _make_checkout(Path(self._td.name)) + self.catalog = make_server.load_model_catalog(self.checkout) + + def tearDown(self): + self._td.cleanup() + + def _ids(self): + return [entry["family"] for entry in self.catalog] + + def test_enter_selects_first_family(self): + with patch("builtins.input", side_effect=[""]): + self.assertEqual(make_server.ask_families(self.catalog), + [self.catalog[0]["family"]]) + + def test_eof_selects_first_family(self): + with patch("builtins.input", side_effect=EOFError): + self.assertEqual(make_server.ask_families(self.catalog), + [self.catalog[0]["family"]]) + + def test_comma_separated_numbers(self): + # 1 and 3 (qwen3_tts and voxcpm2 in the tested-first ordering). + with patch("builtins.input", side_effect=["1,3"]): + chosen = make_server.ask_families(self.catalog) + self.assertEqual(chosen, ["qwen3_tts", "voxcpm2"]) + + def test_space_separated_numbers(self): + with patch("builtins.input", side_effect=["2 4"]): + chosen = make_server.ask_families(self.catalog) + self.assertEqual(chosen, ["higgs_audio_tts", "index_tts2"]) + + def test_dedupes_repeated_choices(self): + with patch("builtins.input", side_effect=["1,1,2"]): + chosen = make_server.ask_families(self.catalog) + self.assertEqual(chosen, ["qwen3_tts", "higgs_audio_tts"]) + + def test_invalid_input_reprompts(self): + with patch("builtins.input", side_effect=["foo", "0", "2"]): + chosen = make_server.ask_families(self.catalog) + self.assertEqual(chosen, ["higgs_audio_tts"]) + + +class BuildServerConfigTests(unittest.TestCase): + def test_single_entry_without_voice_dir(self): + entry = make_server.build_model_entry( + "higgs_audio_tts", "higgs", "models/Higgs-GGUF") + cfg = make_server.build_server_config( + "127.0.0.1", 8080, "cuda", False, [entry]) + self.assertEqual(cfg["host"], "127.0.0.1") + self.assertEqual(cfg["port"], 8080) + self.assertEqual(cfg["backend"], "cuda") + self.assertFalse(cfg["lazy_load"]) + self.assertEqual(cfg["models"], [entry]) + self.assertNotIn("voice_dir", cfg) + + def test_voice_dir_added_when_given(self): + entry = make_server.build_model_entry("voxcpm2", "voxcpm2", "models/V") + cfg = make_server.build_server_config( + "0.0.0.0", 9000, "cpu", True, [entry], + voice_dir="/abs/voices") + self.assertTrue(cfg["lazy_load"]) + self.assertEqual(cfg["voice_dir"], "/abs/voices") + + def test_model_entry_shape(self): + entry = make_server.build_model_entry("index_tts2", "indextts2", "p") + self.assertEqual(entry["id"], "indextts2") + self.assertEqual(entry["family"], "index_tts2") + self.assertEqual(entry["path"], "p") + self.assertEqual(entry["task"], "tts") + self.assertEqual(entry["mode"], "offline") + + +class TranscribeWavDirTests(unittest.TestCase): + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.folder = Path(self._td.name) self.narrator = self.folder / "narrator.wav" self.narrator.write_bytes(b"x") self.other = self.folder / "other.wav" self.other.write_bytes(b"x") def tearDown(self): - self._tmp.cleanup() + self._td.cleanup() - def test_presets_named_after_basenames_with_absolute_paths(self): - transcripts = {str(self.narrator): "First transcript.", - str(self.other): "Second transcript."} + def test_transcribes_to_stem_map_with_absolute_paths(self): + transcripts = {str(self.narrator): "First.", + str(self.other): "Second."} with patch.object(make_server, "transcribe_reference_audio", side_effect=lambda path, model_name="base": transcripts[path]): - presets = make_server.build_voice_presets( + result = make_server.transcribe_wav_dir( [self.narrator, self.other], "base") - self.assertEqual(list(presets), ["narrator", "other"]) - self.assertEqual(presets["narrator"]["reference_text"], - "First transcript.") - self.assertEqual(Path(presets["narrator"]["voice_ref"]), - self.narrator.resolve()) + self.assertEqual(list(result), ["narrator", "other"]) + self.assertEqual(result["narrator"], "First.") - def test_failed_transcription_keeps_entry_with_empty_text(self): + def test_failed_transcription_keeps_empty_string(self): with patch.object(make_server, "transcribe_reference_audio", return_value=None): - presets = make_server.build_voice_presets([self.narrator], "base") - self.assertEqual(presets["narrator"]["reference_text"], "") + result = make_server.transcribe_wav_dir([self.narrator], "base") + self.assertEqual(result["narrator"], "") - def test_whisper_model_name_is_passed_through(self): + def test_whisper_model_name_passed_through(self): with patch.object(make_server, "transcribe_reference_audio", return_value="text") as mock_transcribe: - make_server.build_voice_presets([self.narrator], "large-v3") + make_server.transcribe_wav_dir([self.narrator], "large-v3") self.assertEqual(mock_transcribe.call_args.kwargs["model_name"], "large-v3") - -class BuildServerConfigTests(unittest.TestCase): - def test_both_models_with_presets(self): - presets = {"narrator": {"voice_ref": "/x.wav", - "reference_text": "hi"}} - server_config = make_server.build_server_config( - host="127.0.0.1", port=8080, backend="cuda", lazy_load=False, - include_custom=True, include_clone=True, - custom_voice_id="qwen", clone_model_id="qwen-clone", - custom_voice_path="models/custom", base_path="models/base", - voice_presets=presets) - self.assertEqual(server_config["host"], "127.0.0.1") - self.assertEqual(server_config["port"], 8080) - self.assertEqual(server_config["backend"], "cuda") - self.assertFalse(server_config["lazy_load"]) - self.assertEqual([model["id"] for model in server_config["models"]], - ["qwen", "qwen-clone"]) - custom_entry, clone_entry = server_config["models"] - self.assertNotIn("voice_presets", custom_entry) - self.assertEqual(custom_entry["family"], "qwen3_tts") - self.assertEqual(custom_entry["path"], "models/custom") - self.assertEqual(clone_entry["path"], "models/base") - self.assertEqual(clone_entry["voice_presets"], presets) - - def test_custom_only_has_single_entry(self): - server_config = make_server.build_server_config( - host="0.0.0.0", port=9000, backend="cpu", lazy_load=True, - include_custom=True, include_clone=False, - custom_voice_id="qwen", clone_model_id="qwen-clone", - custom_voice_path="models/custom", base_path=None, - voice_presets={}) - self.assertEqual(len(server_config["models"]), 1) - self.assertEqual(server_config["models"][0]["id"], "qwen") - self.assertNotIn("voice_presets", server_config["models"][0]) - - def test_clone_only_without_presets_omits_key(self): - server_config = make_server.build_server_config( - host="127.0.0.1", port=8080, backend="vulkan", lazy_load=False, - include_custom=False, include_clone=True, - custom_voice_id="qwen", clone_model_id="qwen-clone", - custom_voice_path=None, base_path="models/base", - voice_presets={}) - self.assertEqual(len(server_config["models"]), 1) - self.assertEqual(server_config["models"][0]["id"], "qwen-clone") - self.assertNotIn("voice_presets", server_config["models"][0]) + def test_write_prompt_text_format(self): + path = make_server.write_prompt_text( + self.folder, {"narrator": "Hello.", "other": "World."}) + self.assertEqual(path, self.folder / make_server.PROMPT_TEXT_FILENAME) + text = path.read_text(encoding="utf-8") + # One "name|transcript" line per voice, in insertion order. + self.assertIn("narrator|Hello.", text) + self.assertIn("other|World.", text) class PromptHelperTests(unittest.TestCase): @@ -327,61 +522,26 @@ class PromptHelperTests(unittest.TestCase): "one") -class ResolveWavDirArgTests(unittest.TestCase): - """Path normalization for the required WAV_DIR argument.""" - - def setUp(self): - self._tmp = tempfile.TemporaryDirectory() - self.folder = Path(self._tmp.name) - - def tearDown(self): - self._tmp.cleanup() - - def test_resolves_to_absolute(self): - self.assertEqual(make_server.resolve_wav_dir_arg(str(self.folder)), - self.folder.resolve()) - - def test_strips_surrounding_quotes(self): - quoted = f'"{self.folder}"' - self.assertEqual(make_server.resolve_wav_dir_arg(quoted), - self.folder.resolve()) - - def test_strips_single_quotes(self): - quoted = f"'{self.folder}'" - self.assertEqual(make_server.resolve_wav_dir_arg(quoted), - self.folder.resolve()) - - def test_strips_whitespace(self): - self.assertEqual(make_server.resolve_wav_dir_arg(f" {self.folder} "), - self.folder.resolve()) - - def test_expands_tilde(self): - with patch.object(make_server.os.path, "expanduser", - return_value=str(self.folder)) as mock_expand: - result = make_server.resolve_wav_dir_arg("~/voices") - mock_expand.assert_called_once_with("~/voices") - self.assertEqual(result, self.folder.resolve()) - - def test_trailing_slash_preserved_as_dir(self): - self.assertEqual(make_server.resolve_wav_dir_arg(f"{self.folder}/"), - self.folder.resolve()) - +class _MainTestBase(unittest.TestCase): + """Shared fixtures for end-to-end main() tests.""" -class MainTests(unittest.TestCase): def setUp(self): - self._tmp = tempfile.TemporaryDirectory() - self.folder = Path(self._tmp.name) - self.output = self.folder / "server.json" + self._td = tempfile.TemporaryDirectory() + self.root = Path(self._td.name) + self.folder = self.root / "wavs" + self.folder.mkdir() + self.output = self.root / "server.json" + self.checkout = _make_checkout(self.root) # Isolate the config.py rewrite target so no test can ever # modify the repository's real converter/config.py. - self.fake_config = self.folder / "config.py" + self.fake_config = self.root / "config.py" self.fake_config.write_text(FAKE_CONFIG, encoding="utf-8") patcher = patch.object(make_server, "CONFIG_PATH", self.fake_config) patcher.start() self.addCleanup(patcher.stop) def tearDown(self): - self._tmp.cleanup() + self._td.cleanup() def _run(self, argv, inputs=None, transcribe=None, whisper="faster_whisper"): argv = ["make_audiocpp_server_json.py"] + argv @@ -395,28 +555,53 @@ class MainTests(unittest.TestCase): return_value=whisper): return make_server.main() - def _defaults(self, models="", host="", port="", backend="", - lazy="", custom_path="", clone_path="", - confirm="y", prefix=()): - # First input selects the model family (default: Qwen3-TTS). The - # wav directory is always a positional argument, never prompted. - return list(prefix) + ["", models, host, port, backend, lazy, - custom_path, clone_path, confirm] + +class MainTests(_MainTestBase): + """The default Qwen3-TTS flow and shared server settings.""" + + def _args(self, *extra): + return [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)], inputs=[]) + self._run(["--output", str(self.output), + "--audiocpp-dir", str(self.checkout)], inputs=[]) 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")], + inputs=[]) + self.assertEqual(ctx.exception.code, 2) + + def test_empty_audiocpp_dir_prompted_errors(self): + # No --audiocpp-dir and EOF at the prompt -> hard error. + buf = io.StringIO() + with patch.object(sys, "argv", + ["make_audiocpp_server_json.py", + 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): - exit_code = self._run([str(self.folder), "--output", str(self.output)], - inputs=self._defaults()) + 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. self.assertFalse(data["lazy_load"]) self.assertEqual( [model["id"] for model in data["models"]], @@ -425,24 +610,23 @@ class MainTests(unittest.TestCase): [model["path"] for model in data["models"]], [make_server.DEFAULT_CUSTOM_VOICE_PATH, make_server.DEFAULT_BASE_PATH]) - self.assertNotIn("voice_presets", data["models"][1]) + # voice_dir only when wavs are present; this run has none. + self.assertNotIn("voice_dir", data) def test_eof_uses_all_defaults(self): - exit_code = self._run([str(self.folder), "--output", str(self.output)]) + exit_code = self._run(self._args()) 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") - self.assertFalse(data["lazy_load"]) self.assertEqual(len(data["models"]), 2) - def test_clone_only_with_positional_wav_dir(self): + 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( - [str(self.folder), "--output", str(self.output)], + self._args(), inputs=inputs, transcribe=lambda path, model_name="base": f"transcript of {Path(path).name}") @@ -451,17 +635,21 @@ class MainTests(unittest.TestCase): self.assertEqual(len(data["models"]), 1) clone_entry = data["models"][0] self.assertEqual(clone_entry["id"], config.AUDIOCPP_CLONE_MODEL_ID) - self.assertEqual(sorted(clone_entry["voice_presets"]), - ["alpha", "narrator"]) - self.assertEqual(clone_entry["voice_presets"]["narrator"], - {"voice_ref": str((self.folder / "narrator.wav").resolve()), - "reference_text": "transcript of narrator.wav"}) + # 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): - inputs = ["", "", "", "", "", "", "y"] + # families=default, models=2(custom), host, port, backend, lazy, confirm + inputs = ["", "2", "", "", "", "", "y"] exit_code = self._run( - [str(self.folder), "--output", str(self.output), "--models", "custom"], - inputs=inputs) + 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"]], @@ -470,9 +658,10 @@ class MainTests(unittest.TestCase): 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"): - inputs = ["", "1", "qwen-clone-2", "", "", "", "", "", "", "y"] - exit_code = self._run([str(self.folder), "--output", str(self.output)], - inputs=inputs) + # 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"]], @@ -482,17 +671,19 @@ class MainTests(unittest.TestCase): with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen"), \ patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen"): with self.assertRaises(SystemExit) as ctx: - self._run([str(self.folder), "--output", str(self.output)]) + 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"): - inputs = ["", "", "", "y", "", "", "", "", "", "y"] - exit_code = self._run([str(self.folder), "--output", str(self.output), - "--port", "8080"], - inputs=inputs) + # --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"] + exit_code = self._run( + self._args("--port", "8080"), inputs=inputs) self.assertEqual(exit_code, 0) self.assertIn('"http://127.0.0.1:8080"', self.fake_config.read_text(encoding="utf-8")) @@ -502,44 +693,30 @@ class MainTests(unittest.TestCase): def test_port_sync_declined_keeps_config(self): with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1:9999"): - inputs = ["", "", "", "n", "", "", "", "", "", "y"] - exit_code = self._run([str(self.folder), "--output", str(self.output), - "--port", "8080"], - inputs=inputs) + inputs = ["", "", "", "", "", "n", "", "", "y"] + exit_code = self._run( + self._args("--port", "8080"), inputs=inputs) self.assertEqual(exit_code, 0) self.assertIn('"http://127.0.0.1:9999"', self.fake_config.read_text(encoding="utf-8")) def test_matching_port_does_not_prompt_for_sync(self): - with patch.object(config, "AUDIOCPP_API_URL", - "http://127.0.0.1:8080"): - inputs = self._defaults() - exit_code = self._run([str(self.folder), "--output", str(self.output)], - inputs=inputs) + # config_port() is 8080 (real config); default port matches -> no sync. + inputs = self._defaults() + exit_code = self._run(self._args(), inputs=inputs) self.assertEqual(exit_code, 0) self.assertEqual(self.fake_config.read_text(encoding="utf-8"), FAKE_CONFIG) - def test_invalid_menu_choice_reprompts(self): - # Family menu default, then an invalid models-menu choice retried. - inputs = ["", "9", "", "", "", "", "", "", "", "y"] - exit_code = self._run([str(self.folder), "--output", str(self.output)], - inputs=inputs) - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(len(data["models"]), 2) - def test_confirm_declined_writes_nothing(self): inputs = self._defaults(confirm="n") - exit_code = self._run([str(self.folder), "--output", str(self.output)], - inputs=inputs) + exit_code = self._run(self._args(), inputs=inputs) self.assertEqual(exit_code, 1) self.assertFalse(self.output.exists()) def test_existing_output_declined_keeps_file(self): self.output.write_text('{"old": true}', encoding="utf-8") - exit_code = self._run([str(self.folder), "--output", str(self.output)], - inputs=["n"]) + exit_code = self._run(self._args(), inputs=["n"]) self.assertEqual(exit_code, 1) self.assertEqual(json.loads(self.output.read_text(encoding="utf-8")), {"old": True}) @@ -547,8 +724,7 @@ class MainTests(unittest.TestCase): def test_existing_output_accepted_overwrites(self): self.output.write_text('{"old": true}', encoding="utf-8") inputs = ["y"] + self._defaults() - exit_code = self._run([str(self.folder), "--output", str(self.output)], - inputs=inputs) + 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) @@ -556,22 +732,19 @@ class MainTests(unittest.TestCase): def test_force_overwrites_without_prompt(self): self.output.write_text('{"old": true}', encoding="utf-8") inputs = self._defaults() - exit_code = self._run([str(self.folder), "--output", str(self.output), - "--force"], - inputs=inputs) + 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) def test_flags_skip_prompts(self): - # Family still asked (no --family flag); port 9000 differs from the - # config port so its sync prompt fires; custom/clone paths use - # their defaults. + # --families qwen3_tts --models both + server flags; port 9000 differs + # from config port 8080 -> the port sync prompt still fires. exit_code = self._run( - [str(self.folder), "--output", str(self.output), "--models", "both", - "--host", "0.0.0.0", "--port", "9000", "--backend", "cpu", - "--lazy-load"], - inputs=["", "y", "", "", "y"]) + self._args("--families", "qwen3_tts", "--models", "both", + "--host", "0.0.0.0", "--port", "9000", + "--backend", "cpu", "--lazy-load"), + inputs=["y", "", "", "y"]) self.assertEqual(exit_code, 0) self.assertIn('"http://127.0.0.1:9000"', self.fake_config.read_text(encoding="utf-8")) @@ -582,55 +755,48 @@ class MainTests(unittest.TestCase): self.assertTrue(data["lazy_load"]) def test_missing_positional_wav_dir_errors(self): - missing = self.folder / "nope" + 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([str(missing), "--output", str(self.output), + "--audiocpp-dir", str(self.checkout)], inputs=self._defaults()) self.assertEqual(ctx.exception.code, 2) shown = "".join(call[0][0] for call in mock_stderr.write.call_args_list) 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) + -class NonQwenFamilyMainTests(unittest.TestCase): - """The --family flow for clone-only model families.""" +class NonQwenFamilyMainTests(_MainTestBase): + """The --families flow for clone-only model families.""" def setUp(self): - self._tmp = tempfile.TemporaryDirectory() - self.folder = Path(self._tmp.name) - self.output = self.folder / "server.json" - self.fake_config = self.folder / "config.py" + super().setUp() + # These tests exercise AUDIOCPP_MODEL_ID rewriting, so the fake + # config must contain the model id lines to rewrite. self.fake_config.write_text(FAKE_CONFIG_WITH_MODEL_IDS, encoding="utf-8") - patcher = patch.object(make_server, "CONFIG_PATH", self.fake_config) - patcher.start() - self.addCleanup(patcher.stop) - def tearDown(self): - self._tmp.cleanup() - - def _run(self, argv, inputs=None, transcribe=None, whisper="faster_whisper"): - argv = ["make_audiocpp_server_json.py"] + argv - input_effect = inputs if inputs is not None else EOFError - transcribe_effect = transcribe if transcribe is not None else MagicMock() - with patch.object(sys, "argv", argv), \ - patch("builtins.input", side_effect=input_effect), \ - patch.object(make_server, "transcribe_reference_audio", - side_effect=transcribe_effect), \ - patch.object(make_server, "whisper_backend_available", - return_value=whisper): - return make_server.main() + def _args(self, family, *extra): + return [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") - # Inputs: model-id sync accepted, host, port, backend, lazy, confirm. - inputs = ["y", "", "", "", "", "y"] + # 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"] exit_code = self._run( - [str(self.folder), "--output", str(self.output), - "--family", "higgs_audio_tts", "--model-id", "higgs", - "--model-path", "models/Higgs-Audio-v3-TTS-4B-GGUF"], - inputs=inputs, + self._args("higgs_audio_tts"), inputs=inputs, transcribe=lambda path, model_name="base": "a transcript") self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) @@ -641,22 +807,24 @@ class NonQwenFamilyMainTests(unittest.TestCase): self.assertEqual(entry["path"], "models/Higgs-Audio-v3-TTS-4B-GGUF") self.assertEqual(entry["task"], "tts") self.assertEqual(entry["mode"], "offline") - self.assertEqual(entry["voice_presets"]["narrator"], - {"voice_ref": str((self.folder / "narrator.wav").resolve()), - "reference_text": "a transcript"}) - # Both converter model ids point at the single server entry. - self.assertIn('AUDIOCPP_MODEL_ID = "higgs"', - self.fake_config.read_text(encoding="utf-8")) - self.assertIn('AUDIOCPP_CLONE_MODEL_ID = "higgs"', - self.fake_config.read_text(encoding="utf-8")) + # Voice presets live in the server-level voice_dir, not per entry. + self.assertNotIn("voice_presets", entry) + 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|a transcript", prompt) + # Single non-qwen 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): - # sync declined, host, port, backend, lazy, confirm - inputs = ["n", "", "", "", "", "y"] + (self.folder / "narrator.wav").write_bytes(b"x") + # path, host, port, backend, lazy, confirm, sync(n) + inputs = ["", "", "", "", "", "y", "n"] exit_code = self._run( - [str(self.folder), "--output", str(self.output), "--family", "voxcpm2", - "--model-id", "voxcpm2", "--model-path", "models/VoxCPM2-GGUF"], - inputs=inputs) + self._args("voxcpm2"), inputs=inputs, + transcribe=lambda path, model_name="base": "t") self.assertEqual(exit_code, 0) text = self.fake_config.read_text(encoding="utf-8") self.assertIn('AUDIOCPP_MODEL_ID = "qwen"', text) @@ -664,16 +832,16 @@ class NonQwenFamilyMainTests(unittest.TestCase): data = json.loads(self.output.read_text(encoding="utf-8")) self.assertEqual(data["models"][0]["family"], "voxcpm2") - def test_no_voice_presets_warns(self): + def test_no_wavs_warns_and_omits_voice_dir(self): buf = io.StringIO() - # sync accepted, host, port, backend, lazy, confirm + # path, host, port, backend, lazy, confirm, sync(y) + inputs = ["", "", "", "", "", "y", "y"] with patch.object(sys, "argv", ["make_audiocpp_server_json.py", - str(self.folder), - "--output", str(self.output), - "--family", "index_tts2", "--model-id", "indextts2", - "--model-path", "models/IndexTTS2-GGUF"]), \ - patch("builtins.input", side_effect=["y", "", "", "", "", "y"]), \ + str(self.folder), "--output", str(self.output), + "--audiocpp-dir", str(self.checkout), + "--families", "index_tts2"]), \ + patch("builtins.input", side_effect=inputs), \ patch.object(make_server, "transcribe_reference_audio"), \ patch.object(make_server, "whisper_backend_available", return_value="faster_whisper"), \ @@ -681,33 +849,92 @@ class NonQwenFamilyMainTests(unittest.TestCase): code = make_server.main() self.assertEqual(code, 0) out = buf.getvalue() - self.assertIn("No voice presets were configured", out) + self.assertIn("No .wav files found", out) self.assertIn("model_manager_v2.py install index_tts2_q8_0", out) data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertNotIn("voice_presets", data["models"][0]) + self.assertNotIn("voice_dir", data) - def test_models_flag_rejected_for_non_qwen_family(self): + def test_unknown_family_rejected(self): with self.assertRaises(SystemExit) as ctx: - self._run([str(self.folder), "--output", str(self.output), - "--family", "higgs_audio_tts", "--models", "both"]) + self._run(self._args("not_a_family"), inputs=[]) self.assertEqual(ctx.exception.code, 2) -class TranscriptWarningTests(unittest.TestCase): - """Empty transcripts and a missing Whisper backend produce loud warnings.""" +class MultiFamilyMainTests(_MainTestBase): + """Hosting several families in one server.json.""" - def setUp(self): - self._tmp = tempfile.TemporaryDirectory() - self.folder = Path(self._tmp.name) - self.output = self.folder / "server.json" - self.fake_config = self.folder / "config.py" - self.fake_config.write_text(FAKE_CONFIG, encoding="utf-8") - patcher = patch.object(make_server, "CONFIG_PATH", self.fake_config) - patcher.start() - self.addCleanup(patcher.stop) + def _args(self, *extra): + return [str(self.folder), "--output", str(self.output), + "--audiocpp-dir", str(self.checkout)] + list(extra) - def tearDown(self): - self._tmp.cleanup() + 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"] + exit_code = self._run( + self._args("--families", "qwen3_tts,higgs_audio_tts"), + inputs=inputs, + transcribe=lambda path, model_name="base": "a transcript") + 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.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] + 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"] + exit_code = self._run( + self._args("--families", "higgs_audio_tts,voxcpm2"), + inputs=inputs) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + by_id = {model["id"]: model for model in data["models"]} + 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. + 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"] + with patch.object(sys, "argv", + ["make_audiocpp_server_json.py", + str(self.folder), "--output", str(self.output), + "--audiocpp-dir", str(self.checkout), + "--families", "supertonic"]), \ + 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() + self.assertEqual(code, 0) + out = buf.getvalue() + self.assertIn("no clone-capable family selected", out) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertNotIn("voice_dir", data) + self.assertEqual(data["models"][0]["family"], "supertonic") + + +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), + "--audiocpp-dir", str(self.checkout)] + list(extra) def _run_capturing(self, argv, inputs, transcribe, whisper): argv = ["make_audiocpp_server_json.py"] + argv @@ -725,24 +952,24 @@ class TranscriptWarningTests(unittest.TestCase): 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 family default, clone-only run (menu choice 3); transcribe - # returns None (empty). + # Qwen clone-only (menu 3); custom_path skipped, base_path, host, port, + # backend, lazy, prompt_text write, confirm + inputs = ["", "3", "", "", "", "", "", "", "y"] code, out = self._run_capturing( - [str(self.folder), "--output", str(self.output)], - inputs=["", "3", "", "", "", "", "", "y"], + self._args(), inputs=inputs, transcribe=lambda path, model_name="base": None, whisper="faster_whisper") self.assertEqual(code, 0) self.assertIn("MANUAL TRANSCRIPTION REQUIRED", out) self.assertIn("narrator", out) self.assertIn("alpha", out) - self.assertIn("will NOT work", out) + self.assertIn("prompt_text", out) def test_missing_whisper_backend_prints_conda_warning(self): (self.folder / "narrator.wav").write_bytes(b"x") + inputs = ["", "3", "", "", "", "", "", "", "y"] code, out = self._run_capturing( - [str(self.folder), "--output", str(self.output)], - inputs=["", "3", "", "", "", "", "", "y"], + self._args(), inputs=inputs, transcribe=lambda path, model_name="base": "a transcript", whisper=None) self.assertEqual(code, 0) @@ -751,9 +978,9 @@ class TranscriptWarningTests(unittest.TestCase): def test_all_transcripts_present_prints_no_end_warning(self): (self.folder / "narrator.wav").write_bytes(b"x") + inputs = ["", "3", "", "", "", "", "", "", "y"] code, out = self._run_capturing( - [str(self.folder), "--output", str(self.output)], - inputs=["", "3", "", "", "", "", "", "y"], + self._args(), inputs=inputs, transcribe=lambda path, model_name="base": "a real transcript", whisper="faster_whisper") self.assertEqual(code, 0) diff --git a/tests/test_tts.py b/tests/test_tts.py index a0828de..33834d2 100644 --- a/tests/test_tts.py +++ b/tests/test_tts.py @@ -508,10 +508,11 @@ class AudioCppTTSClientHealthTests(unittest.TestCase): raise AssertionError(f"unexpected URL: {url}") return _dispatch - def _client(self, voice=None, language=None, **kwargs): + 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) + return AudioCppTTSClient(voice=voice, language=language, + model_id=model_id) def test_unreachable_server_raises_with_readme_pointer(self): import urllib.error @@ -600,6 +601,35 @@ class AudioCppTTSClientHealthTests(unittest.TestCase): 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"): @@ -1101,7 +1131,7 @@ 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) + chunk_text=False, model_id=None) mock_faster.assert_not_called() mock_qwen.assert_not_called() @@ -1110,7 +1140,7 @@ 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) + chunk_text=False, model_id=None) def test_audiocpp_backend_chunk_flag_forces_client_chunking(self): with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: @@ -1119,9 +1149,18 @@ class BackendWiringTests(unittest.TestCase): voice="narrator", chunk=True) mock_audiocpp.assert_called_once_with(voice="narrator", language=config.LANGUAGE, - chunk_text=True) + chunk_text=True, model_id=None) 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") + def test_gradio_backend_uses_qwen_client(self): with patch("converter.converter.FasterTTSClient") as mock_faster, \ patch("converter.converter.QwenTTSClient") as mock_qwen, \ -- cgit v1.2.3