"""Tests for the audio.cpp server.json generator tool.""" import io import json import sys import tempfile import unittest from contextlib import redirect_stdout from pathlib import Path from unittest.mock import MagicMock, patch from converter import config from tools import make_audiocpp_server_json as make_server FAKE_CONFIG = ( 'LANGUAGE = "English"\n' "\n" 'AUDIOCPP_API_URL = "http://127.0.0.1:9999" # audio.cpp audiocpp_server\n' "\n" "CHUNK_SIZE = 250\n" ) FAKE_CONFIG_WITH_MODEL_IDS = ( 'AUDIOCPP_API_URL = "http://127.0.0.1:9999" # audio.cpp audiocpp_server\n' "\n" 'AUDIOCPP_MODEL_ID = "qwen" # server entry for speaker mode\n' 'AUDIOCPP_CLONE_MODEL_ID = "qwen-clone"\n' ) 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() self.folder = Path(self._tmp.name) def tearDown(self): self._tmp.cleanup() def _touch(self, name): path = self.folder / name path.write_bytes(b"x") return path def test_finds_only_wavs_case_insensitive(self): self._touch("b.wav") self._touch("a.WAV") self._touch("notes.txt") (self.folder / "sub").mkdir() (self.folder / "sub" / "c.wav").write_bytes(b"x") names = [path.name for path in make_server.find_wav_files(self.folder)] self.assertEqual(names, ["a.WAV", "b.wav"]) def test_sorted_alphabetically_case_insensitive(self): for name in ("Zed.wav", "alpha.wav", "Beta.wav"): self._touch(name) names = [path.name for path in make_server.find_wav_files(self.folder)] self.assertEqual(names, ["alpha.wav", "Beta.wav", "Zed.wav"]) def test_empty_directory_returns_empty_list(self): self.assertEqual(make_server.find_wav_files(self.folder), []) class ConfigPortTests(unittest.TestCase): def test_port_parsed_from_config_url(self): with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1:8080"): self.assertEqual(make_server.config_port(), 8080) def test_missing_port_falls_back(self): with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1"): self.assertEqual(make_server.config_port(), make_server.FALLBACK_PORT) def test_invalid_url_falls_back(self): with patch.object(config, "AUDIOCPP_API_URL", "not a url"): self.assertEqual(make_server.config_port(), make_server.FALLBACK_PORT) def test_url_with_port_replaces_port(self): self.assertEqual( make_server._url_with_port("http://127.0.0.1:8080", 9000), "http://127.0.0.1:9000") def test_url_without_port_adds_port(self): self.assertEqual( make_server._url_with_port("http://localhost", 8080), "http://localhost:8080") class UpdateConfigPortTests(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.config_path = Path(self._tmp.name) / "config.py" self.config_path.write_text(FAKE_CONFIG, encoding="utf-8") def tearDown(self): self._tmp.cleanup() def test_rewrites_port_preserving_comment(self): changed = make_server.update_config_api_url_port( 8080, config_path=self.config_path) self.assertTrue(changed) text = self.config_path.read_text(encoding="utf-8") self.assertIn( 'AUDIOCPP_API_URL = "http://127.0.0.1:8080" # audio.cpp audiocpp_server', text) self.assertIn('LANGUAGE = "English"', text) self.assertIn("CHUNK_SIZE = 250", text) def test_returns_false_when_no_url_line(self): path = Path(self._tmp.name) / "other.py" path.write_text('CHUNK_SIZE = 250\n', encoding="utf-8") self.assertFalse(make_server.update_config_api_url_port( 8080, config_path=path)) def test_returns_false_when_port_unchanged(self): self.assertFalse(make_server.update_config_api_url_port( 9999, config_path=self.config_path)) self.assertEqual(self.config_path.read_text(encoding="utf-8"), FAKE_CONFIG) def test_returns_false_when_file_missing(self): self.assertFalse(make_server.update_config_api_url_port( 8080, config_path=Path(self._tmp.name) / "nope.py")) class UpdateConfigModelIdsTests(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.config_path = Path(self._tmp.name) / "config.py" self.config_path.write_text(FAKE_CONFIG_WITH_MODEL_IDS, encoding="utf-8") def tearDown(self): self._tmp.cleanup() def test_rewrites_both_ids_preserving_lines(self): changed = make_server.update_config_model_ids( "higgs", "higgs", config_path=self.config_path) self.assertTrue(changed) text = self.config_path.read_text(encoding="utf-8") self.assertIn('AUDIOCPP_MODEL_ID = "higgs" # server entry for speaker mode', text) self.assertIn('AUDIOCPP_CLONE_MODEL_ID = "higgs"', text) self.assertIn('AUDIOCPP_API_URL = "http://127.0.0.1:9999"', text) def test_clone_id_optional(self): changed = make_server.update_config_model_ids( "voxcpm2", config_path=self.config_path) self.assertTrue(changed) text = self.config_path.read_text(encoding="utf-8") self.assertIn('AUDIOCPP_MODEL_ID = "voxcpm2"', text) self.assertIn('AUDIOCPP_CLONE_MODEL_ID = "qwen-clone"', text) def test_returns_false_when_ids_unchanged(self): changed = make_server.update_config_model_ids( "qwen", "qwen-clone", config_path=self.config_path) self.assertFalse(changed) self.assertEqual(self.config_path.read_text(encoding="utf-8"), FAKE_CONFIG_WITH_MODEL_IDS) def test_returns_false_when_lines_missing(self): path = Path(self._tmp.name) / "other.py" path.write_text('CHUNK_SIZE = 250\n', encoding="utf-8") self.assertFalse(make_server.update_config_model_ids( "higgs", "higgs", config_path=path)) def test_returns_false_when_file_missing(self): self.assertFalse(make_server.update_config_model_ids( "higgs", "higgs", config_path=Path(self._tmp.name) / "nope.py")) 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._td.cleanup() 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]): result = make_server.transcribe_wav_dir( [self.narrator, self.other], "base") self.assertEqual(list(result), ["narrator", "other"]) self.assertEqual(result["narrator"], "First.") def test_failed_transcription_keeps_empty_string(self): with patch.object(make_server, "transcribe_reference_audio", return_value=None): result = make_server.transcribe_wav_dir([self.narrator], "base") self.assertEqual(result["narrator"], "") def test_whisper_model_name_passed_through(self): with patch.object(make_server, "transcribe_reference_audio", return_value="text") as mock_transcribe: make_server.transcribe_wav_dir([self.narrator], "large-v3") self.assertEqual(mock_transcribe.call_args.kwargs["model_name"], "large-v3") 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): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.folder = Path(self._tmp.name) def tearDown(self): self._tmp.cleanup() def test_ask_port_reprompts_until_valid(self): with patch("builtins.input", side_effect=["abc", "8081"]): self.assertEqual(make_server.ask_port(8080), 8081) def test_ask_port_eof_returns_default(self): with patch("builtins.input", side_effect=EOFError): self.assertEqual(make_server.ask_port(8080), 8080) def test_ask_menu_reprompts_until_valid(self): options = [("One", "one"), ("Two", "two")] with patch("builtins.input", side_effect=["9", "2"]): self.assertEqual( make_server.ask_menu("Pick:", options, default_index=1), "two") def test_ask_menu_eof_returns_default(self): options = [("One", "one"), ("Two", "two")] with patch("builtins.input", side_effect=EOFError): self.assertEqual( make_server.ask_menu("Pick:", options, default_index=1), "one") class _MainTestBase(unittest.TestCase): """Shared fixtures for end-to-end main() tests.""" def setUp(self): 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.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._td.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() 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), "--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(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"]], [config.AUDIOCPP_MODEL_ID, config.AUDIOCPP_CLONE_MODEL_ID]) self.assertEqual( [model["path"] for model in data["models"]], [make_server.DEFAULT_CUSTOM_VOICE_PATH, make_server.DEFAULT_BASE_PATH]) # 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(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"] 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")) data = json.loads(self.output.read_text(encoding="utf-8")) self.assertEqual(data["port"], 8080) 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( 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): # 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_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_existing_output_declined_keeps_file(self): self.output.write_text('{"old": true}', encoding="utf-8") 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}) def test_existing_output_accepted_overwrites(self): self.output.write_text('{"old": true}', encoding="utf-8") inputs = ["y"] + self._defaults() 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) def test_force_overwrites_without_prompt(self): self.output.write_text('{"old": true}', encoding="utf-8") inputs = self._defaults() 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): # --families qwen3_tts --models both + 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", "--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")) data = json.loads(self.output.read_text(encoding="utf-8")) self.assertEqual(data["host"], "0.0.0.0") self.assertEqual(data["port"], 9000) self.assertEqual(data["backend"], "cpu") self.assertTrue(data["lazy_load"]) def test_missing_positional_wav_dir_errors(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), "--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(_MainTestBase): """The --families flow for clone-only model families.""" def setUp(self): 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") 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") # 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( 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")) self.assertEqual(len(data["models"]), 1) entry = data["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") # 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): (self.folder / "narrator.wav").write_bytes(b"x") # path, host, port, backend, lazy, confirm, sync(n) inputs = ["", "", "", "", "", "y", "n"] exit_code = self._run( 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) self.assertIn('AUDIOCPP_CLONE_MODEL_ID = "qwen-clone"', text) data = json.loads(self.output.read_text(encoding="utf-8")) self.assertEqual(data["models"][0]["family"], "voxcpm2") def test_no_wavs_warns_and_omits_voice_dir(self): buf = io.StringIO() # 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), "--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"), \ redirect_stdout(buf): code = make_server.main() self.assertEqual(code, 0) out = buf.getvalue() 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_dir", data) def test_unknown_family_rejected(self): with self.assertRaises(SystemExit) as ctx: self._run(self._args("not_a_family"), inputs=[]) self.assertEqual(ctx.exception.code, 2) class MultiFamilyMainTests(_MainTestBase): """Hosting several families in one server.json.""" def _args(self, *extra): return [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"] 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 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=whisper), \ redirect_stdout(buf): code = make_server.main() return code, buf.getvalue() 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"] code, out = self._run_capturing( 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("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( 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("faster_whisper", out) 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( self._args(), inputs=inputs, transcribe=lambda path, model_name="base": "a real transcript", whisper="faster_whisper") self.assertEqual(code, 0) self.assertNotIn("MANUAL TRANSCRIPTION REQUIRED", out) if __name__ == "__main__": unittest.main()