"""Tests for the audio.cpp backend setup module (backends/audiocpp.py).""" import contextlib import io import json import sys import tempfile import threading import unittest from contextlib import redirect_stdout from pathlib import Path from unittest.mock import MagicMock, patch from converter import config from backends import audiocpp as make_server from ui import tui 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"}, {"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=[{ "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 DetectWavDirTests(unittest.TestCase): """Shallow .wav-directory discovery across the two checkout roots.""" def setUp(self): self._td = tempfile.TemporaryDirectory() self.root = Path(self._td.name) self.audiocpp = self.root / "audio.cpp" self.tts_root = self.root / "tts-audiobook-generator" self.audiocpp.mkdir() self.tts_root.mkdir() def tearDown(self): self._td.cleanup() def _wav_dir(self, where, name="voices"): directory = where / name directory.mkdir(parents=True, exist_ok=True) (directory / "voice.wav").write_bytes(b"x") return directory def test_unique_wav_dir_in_tts_root_returned(self): found = self._wav_dir(self.tts_root, "voices") self.assertEqual(make_server.detect_wav_dir(self.audiocpp, self.tts_root), found) def test_unique_wav_dir_in_audiocpp_root_returned(self): found = self._wav_dir(self.audiocpp, "reference") self.assertEqual(make_server.detect_wav_dir(self.audiocpp, self.tts_root), found) def test_root_itself_containing_wavs_returned(self): (self.tts_root / "direct.wav").write_bytes(b"x") self.assertEqual(make_server.detect_wav_dir(self.audiocpp, self.tts_root), self.tts_root) def test_multiple_wav_dirs_returns_none(self): self._wav_dir(self.tts_root, "one") self._wav_dir(self.audiocpp, "two") self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, self.tts_root)) def test_output_dir_of_tts_root_excluded(self): self._wav_dir(self.tts_root, "output") self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, self.tts_root)) def test_no_wavs_returns_none(self): self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, self.tts_root)) def test_nested_wav_dir_not_seen(self): nested = self.tts_root / "outer" / "inner" nested.mkdir(parents=True) (nested / "voice.wav").write_bytes(b"x") self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, self.tts_root)) 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): 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 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") 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") self.assertEqual(make_server.default_model_id("moss_tts_local"), "mossttslocal") class LoadModelCatalogTests(unittest.TestCase): def setUp(self): 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_families_sorted_alphabetically_by_display_name(self): catalog = make_server.load_model_catalog(self.checkout) names = [entry["display_name"].lower() for entry in catalog] self.assertEqual(names, sorted(names)) self.assertNotIn("tested", catalog[0]) self.assertNotIn("TESTED_FAMILIES", dir(make_server)) 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): _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} 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 DetectBackendTests(unittest.TestCase): """Backend detection from audio.cpp build directory names.""" def setUp(self): self._td = tempfile.TemporaryDirectory() self.checkout = Path(self._td.name) / "audio.cpp" self.checkout.mkdir() def tearDown(self): self._td.cleanup() def _build(self, name, binary="audiocpp_server"): build_dir = self.checkout / "build" / name bin_dir = build_dir / "bin" bin_dir.mkdir(parents=True) (bin_dir / binary).write_bytes(b"x") return build_dir def test_no_build_dir_returns_none(self): self.assertIsNone(make_server.detect_backend(self.checkout)) def test_unique_linux_backend_detected(self): self._build("linux-cuda-release") self.assertEqual(make_server.detect_backend(self.checkout), "cuda") def test_windows_exe_backend_detected(self): self._build("windows-vulkan-debug", binary="audiocpp_server.exe") self.assertEqual(make_server.detect_backend(self.checkout), "vulkan") def test_hip_backend_detected(self): self._build("linux-hip-release") self.assertEqual(make_server.detect_backend(self.checkout), "hip") def test_cpu_backend_detected(self): self._build("linux-cpu-release") self.assertEqual(make_server.detect_backend(self.checkout), "cpu") def test_metal_maps_to_cpu(self): self._build("macos-metal-release") self.assertEqual(make_server.detect_backend(self.checkout), "cpu") def test_multiple_backends_returns_none(self): self._build("linux-cuda-release") self._build("linux-cpu-release") self.assertIsNone(make_server.detect_backend(self.checkout)) def test_multiple_builds_same_backend_detected(self): self._build("linux-cuda-release") self._build("windows-cuda-debug") self.assertEqual(make_server.detect_backend(self.checkout), "cuda") def test_build_dir_without_binary_ignored(self): (self.checkout / "build" / "linux-cuda-release").mkdir(parents=True) self.assertIsNone(make_server.detect_backend(self.checkout)) def test_non_matching_build_dir_name_ignored(self): self._build("linux-mybuild-release") self.assertIsNone(make_server.detect_backend(self.checkout)) class BackendOptionsTests(unittest.TestCase): """Aligned backend menu labels and the [auto-detected] default.""" def test_options_have_aligned_dashes(self): options, default_index = make_server._backend_options() dash_columns = {label.index(" - ") for label, _ in options} self.assertEqual(len(dash_columns), 1) self.assertEqual(default_index, 0) def test_detected_backend_marked_and_defaulted(self): options, default_index = make_server._backend_options("vulkan") labels = [label for label, _ in options] self.assertEqual(default_index, labels.index(next( label for label, value in options if value == "vulkan" and label.endswith("[auto-detected]")))) self.assertTrue(labels[default_index].endswith("[auto-detected]")) self.assertEqual(options[default_index][1], "vulkan") def test_unknown_detected_backend_is_ignored(self): options, default_index = make_server._backend_options("opencl") self.assertEqual(default_index, 0) self.assertFalse(any("[auto-detected]" in label for label, _ in options)) def test_labels_keep_backend_values(self): options, _ = make_server._backend_options() self.assertEqual([value for _, value in options], list(make_server.BACKENDS)) 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") 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 InstallModelsTests(unittest.TestCase): """Printing or auto-running the model install commands.""" def setUp(self): self._td = tempfile.TemporaryDirectory() self.checkout = Path(self._td.name) / "audio.cpp" self.checkout.mkdir() self.manager = self.checkout / "tools" / "model_manager_v2.py" self.manager.parent.mkdir() self.manager.write_text("#!/usr/bin/env python3\n", encoding="utf-8") self.guidance = [("Higgs Audio v3 TTS 4B", "higgs_audio_tts_4b_q8_0"), ("Qwen3-TTS", "qwen3_tts_1_7b_base_q8_0"), ("Qwen3-TTS", "qwen3_tts_1_7b_base_q8_0")] def tearDown(self): self._td.cleanup() def test_declined_download_prints_commands_deduped(self): buf = io.StringIO() with redirect_stdout(buf), \ patch.object(make_server.subprocess, "run") as run: make_server._install_models(self.checkout, self.guidance, download=False) out = buf.getvalue() self.assertEqual(out.count("install higgs_audio_tts_4b_q8_0"), 1) self.assertEqual(out.count("install qwen3_tts_1_7b_base_q8_0"), 1) run.assert_not_called() def test_accepted_download_runs_each_command(self): with patch.object(make_server.subprocess, "run", return_value=MagicMock(returncode=0)) as run: make_server._install_models(self.checkout, self.guidance, download=True) self.assertEqual(run.call_count, 2) commands = [call[0][0] for call in run.call_args_list] self.assertEqual(commands[0], [sys.executable, str(self.manager), "install", "higgs_audio_tts_4b_q8_0"]) self.assertEqual(commands[1], [sys.executable, str(self.manager), "install", "qwen3_tts_1_7b_base_q8_0"]) for call in run.call_args_list: self.assertEqual(call[1]["cwd"], str(self.checkout)) def test_missing_manager_falls_back_to_printing(self): self.manager.unlink() buf = io.StringIO() with redirect_stdout(buf), \ patch.object(make_server.subprocess, "run") as run: make_server._install_models(self.checkout, self.guidance, download=True) self.assertIn("install higgs_audio_tts_4b_q8_0", buf.getvalue()) run.assert_not_called() def test_failed_install_reports_warning_and_continues(self): results = iter([MagicMock(returncode=1), MagicMock(returncode=0)]) buf = io.StringIO() with redirect_stdout(buf), \ patch.object(make_server.subprocess, "run", side_effect=lambda *a, **k: next(results)) as run: make_server._install_models(self.checkout, self.guidance, download=True) self.assertEqual(run.call_count, 2) self.assertIn("exited with code 1", buf.getvalue()) def test_decide_download_skips_prompt_without_manager(self): self.manager.unlink() confirm = MagicMock() self.assertFalse(make_server._decide_download(self.checkout, confirm)) confirm.assert_not_called() def test_decide_download_asks_when_manager_present(self): confirm = MagicMock(return_value=True) self.assertTrue(make_server._decide_download(self.checkout, confirm)) confirm.assert_called_once() def test_decide_download_defaults_to_yes(self): confirm = MagicMock(return_value=True) make_server._decide_download(self.checkout, confirm) self.assertIs(confirm.call_args[0][1], True) 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") self.assertIn("narrator|Hello.", text) self.assertIn("other|World.", text) 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"]) 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 FindAudiocppServerBinTests(unittest.TestCase): """Locating the built audiocpp_server binary.""" def setUp(self): self._td = tempfile.TemporaryDirectory() self.checkout = Path(self._td.name) / "audio.cpp" self.checkout.mkdir() def tearDown(self): self._td.cleanup() def _build(self, name, binary="audiocpp_server"): bin_dir = self.checkout / "build" / name / "bin" bin_dir.mkdir(parents=True) (bin_dir / binary).write_bytes(b"x") def test_no_build_dir_returns_none(self): self.assertIsNone(make_server.find_audiocpp_server_bin(self.checkout)) def test_finds_built_binary(self): self._build("linux-cuda-release") self.assertEqual( make_server.find_audiocpp_server_bin(self.checkout), self.checkout / "build" / "linux-cuda-release" / "bin" / "audiocpp_server") def test_finds_windows_exe(self): self._build("windows-vulkan-debug", binary="audiocpp_server.exe") self.assertEqual( make_server.find_audiocpp_server_bin(self.checkout).name, "audiocpp_server.exe") def test_build_dir_without_binary_returns_none(self): (self.checkout / "build" / "linux-cuda-release" / "bin").mkdir( parents=True) self.assertIsNone(make_server.find_audiocpp_server_bin(self.checkout)) class BuiltServerBinaryTests(unittest.TestCase): """built_server_binary: locating a specific backend's build.""" def setUp(self): self._td = tempfile.TemporaryDirectory() self.checkout = Path(self._td.name) / "audio.cpp" self.checkout.mkdir() def tearDown(self): self._td.cleanup() def _build(self, name, binary="audiocpp_server"): bin_dir = self.checkout / "build" / name / "bin" bin_dir.mkdir(parents=True) (bin_dir / binary).write_bytes(b"x") def test_returns_the_matching_backend_binary(self): self._build("linux-cuda-release") self._build("linux-cpu-release") self.assertEqual( make_server.built_server_binary(self.checkout, "cpu"), self.checkout / "build" / "linux-cpu-release" / "bin" / "audiocpp_server") def test_returns_none_for_unbuilt_backend(self): self._build("linux-cuda-release") self.assertIsNone( make_server.built_server_binary(self.checkout, "vulkan")) def test_metal_counts_as_cpu(self): self._build("macos-metal-release") self.assertEqual( make_server.built_server_binary(self.checkout, "cpu"), self.checkout / "build" / "macos-metal-release" / "bin" / "audiocpp_server") def test_no_build_dir_returns_none(self): self.assertIsNone(make_server.built_server_binary(self.checkout, "cpu")) class BuildAudiocppTests(unittest.TestCase): """Running the audio.cpp build helper script.""" def setUp(self): self._td = tempfile.TemporaryDirectory() self.checkout = Path(self._td.name) / "audio.cpp" self.checkout.mkdir() self.scripts = self.checkout / "scripts" self.scripts.mkdir() (self.scripts / "build_linux.sh").write_text("#!/bin/sh\n", encoding="utf-8") self.log_dir = Path(self._td.name) / "logs" self.addCleanup(make_server.common.drain_post_tui_notices) def tearDown(self): self._td.cleanup() def _emit(self): lines = [] def emit(line): lines.append(line) return lines, emit def _log_files(self): return sorted(self.log_dir.glob("audiocpp_build_*.log")) def test_runs_build_script_with_backend_and_target(self): with patch.object(make_server.common, "run_console_subprocess", return_value=0) as run: rc = make_server.build_audiocpp(self.checkout, "cuda") self.assertEqual(rc, 0) argv = run.call_args[0][0] self.assertEqual(argv[:3], ["sh", str(self.scripts / "build_linux.sh"), "--backend"]) self.assertIn("cuda", argv) self.assertIn("--target", argv) self.assertIn("audiocpp_server", argv) self.assertEqual(run.call_args[1]["cwd"], self.checkout) def test_missing_script_returns_nonzero(self): for f in self.scripts.iterdir(): f.unlink() rc = make_server.build_audiocpp(self.checkout, "cuda") self.assertNotEqual(rc, 0) def test_console_path_writes_no_log_and_no_notice(self): with patch.object(make_server.common, "LOG_DIR", self.log_dir), \ patch.object(make_server.common, "run_console_subprocess", return_value=0): rc = make_server.build_audiocpp(self.checkout, "cuda") self.assertEqual(rc, 0) self.assertEqual(self._log_files(), []) self.assertEqual(make_server.common.drain_post_tui_notices(), []) def test_tui_success_writes_log_and_no_notice(self): emitted, emit = self._emit() with patch.object(make_server.common, "LOG_DIR", self.log_dir), \ patch.object(make_server.common, "run_console_subprocess", return_value=0): rc = make_server.build_audiocpp(self.checkout, "cuda", emit=emit) self.assertEqual(rc, 0) self.assertEqual(len(self._log_files()), 1) log_text = self._log_files()[0].read_text(encoding="utf-8") self.assertIn("[INFO] Building audiocpp_server", log_text) self.assertIn("--backend cuda", log_text) self.assertTrue(emitted) self.assertEqual(make_server.common.drain_post_tui_notices(), []) def test_tui_failure_writes_log_and_records_notice(self): emitted, emit = self._emit() with patch.object(make_server.common, "LOG_DIR", self.log_dir), \ patch.object(make_server.common, "run_console_subprocess", return_value=3): rc = make_server.build_audiocpp(self.checkout, "cuda", emit=emit) self.assertEqual(rc, 3) logs = self._log_files() self.assertEqual(len(logs), 1) log_text = logs[0].read_text(encoding="utf-8") self.assertIn("failed (exit code 3)", log_text) notices = make_server.common.drain_post_tui_notices() self.assertEqual(len(notices), 1) notice = notices[0] self.assertIn("failed (exit code 3)", notice) self.assertIn(f"Build log: {logs[0]}", notice) command = (f"cd {self.checkout} && sh " f"{self.scripts / 'build_linux.sh'} --backend cuda " "--target audiocpp_server") self.assertIn(command, notice) self.assertIn("Troubleshoot by re-running this command", notice) self.assertTrue(any("failed (exit code 3)" in line for line in emitted)) def test_tui_cancel_suppresses_notice_but_writes_log(self): emitted, emit = self._emit() cancel = threading.Event() cancel.set() with patch.object(make_server.common, "LOG_DIR", self.log_dir), \ patch.object(make_server.common, "run_console_subprocess", return_value=130): rc = make_server.build_audiocpp(self.checkout, "cuda", emit=emit, cancel=cancel) self.assertEqual(rc, 130) self.assertEqual(len(self._log_files()), 1) self.assertEqual(make_server.common.drain_post_tui_notices(), []) def test_tui_missing_script_records_guidance_notice(self): for f in self.scripts.iterdir(): f.unlink() emitted, emit = self._emit() with patch.object(make_server.common, "LOG_DIR", self.log_dir): rc = make_server.build_audiocpp(self.checkout, "cuda", emit=emit) self.assertNotEqual(rc, 0) self.assertEqual(self._log_files(), []) notices = make_server.common.drain_post_tui_notices() self.assertEqual(len(notices), 1) self.assertIn("No build script found", notices[0]) class AudiocppDetectTests(unittest.TestCase): """backends.audiocpp.detect() status reporting.""" def setUp(self): self._td = tempfile.TemporaryDirectory() self.root = Path(self._td.name) self.checkout = _make_checkout(self.root) def tearDown(self): self._td.cleanup() def test_not_cloned(self): with patch.object(make_server, "find_local_checkout", return_value=None): status = make_server.detect() self.assertFalse(status.installed) self.assertFalse(status.configured) self.assertIn("not cloned", status.details[0]) def test_cloned_not_built_not_configured(self): with patch.object(make_server, "find_local_checkout", return_value=self.checkout), \ patch.object(make_server, "find_audiocpp_server_bin", return_value=None): status = make_server.detect() self.assertFalse(status.installed) self.assertFalse(status.configured) self.assertEqual(status.launch_hint, "") self.assertEqual(status.partial, "downloaded (not built)") def test_built_and_configured_ready(self): binary = self.checkout / "build" / "linux-cuda-release" / "bin" \ / "audiocpp_server" binary.parent.mkdir(parents=True) binary.write_bytes(b"x") server_json = self.checkout / "server.json" server_json.write_text('{"models":[]}', encoding="utf-8") with patch.object(make_server, "find_local_checkout", return_value=self.checkout): status = make_server.detect() self.assertTrue(status.installed) self.assertTrue(status.configured) self.assertIn(str(binary), status.launch_hint) self.assertIn(str(server_json), status.launch_hint) self.assertEqual(status.partial, "") def test_built_not_configured(self): binary = self.checkout / "build" / "linux-cuda-release" / "bin" \ / "audiocpp_server" binary.parent.mkdir(parents=True) binary.write_bytes(b"x") with patch.object(make_server, "find_local_checkout", return_value=self.checkout): status = make_server.detect() self.assertTrue(status.installed) self.assertFalse(status.configured) self.assertEqual(status.partial, "built (not configured)") class NonInteractiveMainTests(unittest.TestCase): """The flag-only (non-TUI) path through main(), end to end.""" 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 config.py rewrites so no test touches the real one. 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) # Tests run without a tty -> main() takes the non-interactive path. patcher = patch.object(make_server, "_interactive", return_value=False) patcher.start() self.addCleanup(patcher.stop) def tearDown(self): self._td.cleanup() def _run(self, argv, transcribe=None, whisper="faster_whisper"): argv = ["backends/audiocpp.py"] + argv transcribe_effect = transcribe if transcribe is not None \ else MagicMock() with patch.object(sys, "argv", argv), \ 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, *extra): return ["--wavs", str(self.folder), "--output", str(self.output), "--audiocpp-dir", str(self.checkout)] + list(extra) def test_default_run_hosts_recommended_entry(self): exit_code = self._run( self._args("--families", "higgs_audio_tts", "--no-sync-model-ids")) 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.assertTrue(data["lazy_load"]) self.assertEqual([m["id"] for m in data["models"]], ["higgs"]) self.assertNotIn("voice_dir", data) def test_port_sync_accepted_updates_config(self): with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1:9999"): exit_code = self._run( self._args("--families", "higgs_audio_tts", "--port", "8080", "--no-sync-model-ids")) 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"): exit_code = self._run( self._args("--families", "higgs_audio_tts", "--port", "8080", "--no-sync-port", "--no-sync-model-ids")) self.assertEqual(exit_code, 0) self.assertIn('"http://127.0.0.1:9999"', self.fake_config.read_text(encoding="utf-8")) def test_model_id_sync_accepted_updates_config(self): self.fake_config.write_text(FAKE_CONFIG_WITH_MODEL_IDS, encoding="utf-8") exit_code = self._run(self._args("--families", "higgs_audio_tts")) self.assertEqual(exit_code, 0) 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_multi_family_lazy_with_voice_dir(self): (self.folder / "narrator.wav").write_bytes(b"x") exit_code = self._run( self._args("--families", "qwen3_tts,higgs_audio_tts", "--no-sync-model-ids"), 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([m["id"] for m in data["models"]], ["qwen", "higgs"]) self.assertTrue(data["lazy_load"]) 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) def test_force_overwrites_existing_output(self): self.output.write_text('{"old": true}', encoding="utf-8") exit_code = self._run( self._args("--families", "higgs_audio_tts", "--force", "--no-sync-model-ids")) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) self.assertEqual(len(data["models"]), 1) def test_existing_output_declined_keeps_file(self): self.output.write_text('{"old": true}', encoding="utf-8") exit_code = self._run( self._args("--families", "higgs_audio_tts", "--no-sync-model-ids")) self.assertEqual(exit_code, 1) self.assertEqual(json.loads(self.output.read_text(encoding="utf-8")), {"old": True}) def test_all_packages_hosts_design_as_vdes(self): exit_code = self._run( self._args("--families", "qwen3_tts", "--all-packages", "--no-sync-model-ids")) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) by_id = {m["id"]: m for m in data["models"]} self.assertIn("qwen-design", by_id) self.assertEqual(by_id["qwen-design"]["task"], "vdes") # The non-design packages are hosted with task "tts". self.assertTrue(any(m["id"] in ("qwen", "qwen-2") and m["task"] == "tts" for m in data["models"])) def test_unknown_family_rejected(self): with self.assertRaises(SystemExit) as ctx: self._run(self._args("--families", "not_a_family", "--no-sync-model-ids")) self.assertEqual(ctx.exception.code, 2) def test_missing_checkout_rejected(self): with patch.object(make_server, "find_local_checkout", return_value=None), \ self.assertRaises(SystemExit) as ctx: self._run(["--families", "higgs_audio_tts", "--output", str(self.output), "--no-sync-model-ids"]) self.assertEqual(ctx.exception.code, 2) def test_missing_wav_dir_rejected(self): missing = self.root / "nope" with self.assertRaises(SystemExit) as ctx: self._run(["--wavs", str(missing), "--output", str(self.output), "--audiocpp-dir", str(self.checkout), "--families", "higgs_audio_tts", "--no-sync-model-ids"]) self.assertEqual(ctx.exception.code, 2) def test_families_required_in_noninteractive_run(self): with self.assertRaises(SystemExit) as ctx: self._run(self._args("--no-sync-model-ids")) self.assertEqual(ctx.exception.code, 2) class FetchServerEndpointsTests(unittest.TestCase): """fetch_server_models / fetch_server_voices: live queries against a running audiocpp_server (urlopen mocked).""" @staticmethod def _urlopen_responding(bodies, errors=None): """A urlopen stub returning successive BODIES (bytes) or raising successive ERRORS; records every requested URL.""" calls = [] def fake_urlopen(url, timeout=10): calls.append(url) if errors: raise errors.pop(0) body = bodies.pop(0) context = MagicMock() context.__enter__.return_value = context context.__exit__.return_value = False context.read.return_value = body return context return fake_urlopen, calls def test_fetch_models_parses_id_family_task(self): urlopen, calls = self._urlopen_responding([json.dumps({ "data": [{"id": "qwen", "family": "qwen3_tts", "task": "tts"}, {"id": "legacy"}], }).encode("utf-8")]) with patch.object(make_server.urllib.request, "urlopen", urlopen): models = make_server.fetch_server_models("http://127.0.0.1:8080") # Missing fields mirror the converter's client: empty strings. self.assertEqual(models, [ {"id": "qwen", "family": "qwen3_tts", "task": "tts"}, {"id": "legacy", "family": "", "task": ""}, ]) self.assertEqual(calls, ["http://127.0.0.1:8080/v1/models"]) def test_fetch_models_trailing_slash_url(self): urlopen, calls = self._urlopen_responding( [b'{"data": [{"id": "m"}]}']) with patch.object(make_server.urllib.request, "urlopen", urlopen): make_server.fetch_server_models("http://host:8080/") self.assertEqual(calls, ["http://host:8080/v1/models"]) def test_fetch_models_connection_error_returns_none(self): import urllib.error urlopen, _ = self._urlopen_responding( [], errors=[urllib.error.URLError("Connection refused")]) with patch.object(make_server.urllib.request, "urlopen", urlopen): self.assertIsNone( make_server.fetch_server_models("http://127.0.0.1:8080")) def test_fetch_models_non_json_body_returns_none(self): # A port answering TCP but not speaking audiocpp_server JSON. urlopen, _ = self._urlopen_responding([b"not json"]) with patch.object(make_server.urllib.request, "urlopen", urlopen): self.assertIsNone( make_server.fetch_server_models("http://127.0.0.1:8080")) def test_fetch_models_unexpected_document_yields_empty_list(self): urlopen, _ = self._urlopen_responding([b'{"foo": 1}']) with patch.object(make_server.urllib.request, "urlopen", urlopen): self.assertEqual( make_server.fetch_server_models("http://127.0.0.1:8080"), []) def test_fetch_voices_parses_names_and_encodes_model(self): urlopen, calls = self._urlopen_responding( [b'{"voices": ["narrator", "obama"]}']) with patch.object(make_server.urllib.request, "urlopen", urlopen): voices = make_server.fetch_server_voices( "http://127.0.0.1:8080", "qwen") self.assertEqual(voices, ["narrator", "obama"]) self.assertEqual(calls, ["http://127.0.0.1:8080/v1/audio/voices?model=qwen"]) def test_fetch_voices_error_returns_none(self): import urllib.error urlopen, _ = self._urlopen_responding( [], errors=[urllib.error.URLError("boom")]) with patch.object(make_server.urllib.request, "urlopen", urlopen): self.assertIsNone( make_server.fetch_server_voices("http://h", "qwen")) def test_fetch_voices_non_list_shape_returns_none(self): urlopen, _ = self._urlopen_responding([b'{"voices": 5}']) with patch.object(make_server.urllib.request, "urlopen", urlopen): self.assertIsNone( make_server.fetch_server_voices("http://h", "qwen")) class MissingModelEntriesTests(unittest.TestCase): """missing_model_entries: server.json paths vs. files on disk.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.dir = Path(self._tmp.name) def tearDown(self): self._tmp.cleanup() def _server_json(self, models): path = self.dir / "server.json" path.write_text(json.dumps({"models": models}), encoding="utf-8") return path def test_relative_path_resolves_against_config_dir(self): (self.dir / "models" / "present").mkdir(parents=True) (self.dir / "models" / "present" / "m.gguf").write_bytes(b"x") path = self._server_json([ {"id": "a", "path": "models/present"}, {"id": "b", "path": "models/absent"}, ]) missing = make_server.missing_model_entries(path) self.assertEqual([m["id"] for m in missing], ["b"]) def test_empty_directory_counts_as_missing(self): (self.dir / "models" / "empty").mkdir(parents=True) path = self._server_json([{"id": "a", "path": "models/empty"}]) self.assertEqual(len(make_server.missing_model_entries(path)), 1) def test_absolute_paths_honored(self): target = self.dir / "absolute" target.mkdir() (target / "m.gguf").write_bytes(b"x") path = self._server_json([{"id": "a", "path": str(target)}]) self.assertEqual(make_server.missing_model_entries(path), []) def test_unreadable_json_returns_empty(self): path = self.dir / "server.json" path.write_text("not json", encoding="utf-8") self.assertEqual(make_server.missing_model_entries(path), []) def test_no_models_returns_empty(self): path = self._server_json([]) self.assertEqual(make_server.missing_model_entries(path), []) class ModelInstallHintsTests(unittest.TestCase): """model_install_hints: maps missing paths to the install command.""" def test_maps_path_to_install_id_via_catalog(self): import tempfile with tempfile.TemporaryDirectory() as td: checkout = Path(td) specs = checkout / "model_specs" specs.mkdir() (specs / "qwen3_tts.json").write_text(json.dumps({ "family": "qwen3_tts", "category": "tts", "tasks": ["tts"], "packages": [{ "id": "qwen3_tts_0_6b_base_q8_0", "format": "gguf", "target_directory": "Qwen3-TTS-12Hz-0.6B-Base-GGUF", }], }), encoding="utf-8") missing = [{"id": "qwen", "rel": "models/Qwen3-TTS-12Hz-0.6B-Base-GGUF"}] hints = make_server.model_install_hints(checkout, missing) self.assertEqual(len(hints), 1) self.assertIn("qwen3_tts_0_6b_base_q8_0", hints[0]) def test_unmapped_path_names_the_path(self): import tempfile with tempfile.TemporaryDirectory() as td: checkout = Path(td) (checkout / "model_specs").mkdir() hints = make_server.model_install_hints( checkout, [{"id": "x", "rel": "models/nope"}]) self.assertIn("models/nope", hints[0]) self.assertNotIn("install", hints[0]) class DetectServerSpecTests(unittest.TestCase): """detect(): the server spec carries the checkout cwd + identity.""" def _checkout(self): tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) checkout = Path(tmp.name) (checkout / "model_specs").mkdir() build = checkout / "build" / "linux-cuda-release" / "bin" build.mkdir(parents=True) (build / "audiocpp_server").write_bytes(b"x") (checkout / "server.json").write_text(json.dumps({ "models": [{"id": "qwen", "family": "qwen3_tts", "path": "models/Qwen3-TTS-12Hz-0.6B-Base-GGUF"}], }), encoding="utf-8") return checkout def test_spec_has_cwd_and_identity(self): checkout = self._checkout() with patch.object(make_server, "find_local_checkout", return_value=checkout), \ patch.object(make_server, "_detect_remote", return_value=(False, {})): status = make_server.detect() self.assertEqual(len(status.servers), 1) spec = status.servers[0] self.assertEqual(spec.cwd, checkout) self.assertEqual(spec.identity, "audiocpp") self.assertIn("--config", spec.argv) def test_models_missing_flag_and_details(self): checkout = self._checkout() with patch.object(make_server, "find_local_checkout", return_value=checkout), \ patch.object(make_server, "_detect_remote", return_value=(False, {})): status = make_server.detect() self.assertTrue(status.models_missing) self.assertTrue(any("not downloaded" in line for line in status.details)) class InstalledModelEntriesTests(unittest.TestCase): """installed_model_entries: the complement of missing_model_entries.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.dir = Path(self._tmp.name) def tearDown(self): self._tmp.cleanup() def _server_json(self, models): path = self.dir / "server.json" path.write_text(json.dumps({"models": models}), encoding="utf-8") return path def test_lists_entries_whose_files_are_on_disk(self): (self.dir / "models" / "present").mkdir(parents=True) (self.dir / "models" / "present" / "m.gguf").write_bytes(b"x") path = self._server_json([ {"id": "a", "path": "models/present"}, {"id": "b", "path": "models/absent"}, ]) installed = make_server.installed_model_entries(path) self.assertEqual([m["id"] for m in installed], ["a"]) def test_unreadable_json_returns_empty(self): path = self.dir / "server.json" path.write_text("not json", encoding="utf-8") self.assertEqual(make_server.installed_model_entries(path), []) class MissingModelInstallGuidanceTests(unittest.TestCase): """missing_model_install_guidance: missing paths -> (id, install_id).""" def test_maps_paths_and_skips_unmapped(self): with tempfile.TemporaryDirectory() as td: checkout = Path(td) specs = checkout / "model_specs" specs.mkdir() (specs / "qwen3_tts.json").write_text(json.dumps({ "family": "qwen3_tts", "category": "tts", "tasks": ["tts"], "packages": [{ "id": "qwen3_tts_0_6b_base_q8_0", "format": "gguf", "target_directory": "Qwen3-TTS-12Hz-0.6B-Base-GGUF", }], }), encoding="utf-8") missing = [ {"id": "qwen", "rel": "models/Qwen3-TTS-12Hz-0.6B-Base-GGUF"}, {"id": "x", "rel": "models/nope"}, ] guidance = make_server.missing_model_install_guidance( checkout, missing) self.assertEqual(guidance, [("qwen", "qwen3_tts_0_6b_base_q8_0")]) class LoadServerConfigTests(unittest.TestCase): """load_server_config: read server.json, or None when unusable.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.dir = Path(self._tmp.name) def tearDown(self): self._tmp.cleanup() def test_reads_dict_document(self): path = self.dir / "server.json" path.write_text(json.dumps({"host": "0.0.0.0", "models": []}), encoding="utf-8") self.assertEqual(make_server.load_server_config(path), {"host": "0.0.0.0", "models": []}) def test_missing_file_returns_none(self): self.assertIsNone(make_server.load_server_config( self.dir / "nope.json")) def test_unreadable_json_returns_none(self): path = self.dir / "server.json" path.write_text("not json", encoding="utf-8") self.assertIsNone(make_server.load_server_config(path)) def test_non_dict_document_returns_none(self): path = self.dir / "server.json" path.write_text("[1, 2, 3]", encoding="utf-8") self.assertIsNone(make_server.load_server_config(path)) class ServerConfigSelectionsTests(unittest.TestCase): """server_config_selections: map server.json models back to the catalog.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.checkout = _make_checkout(Path(self._tmp.name)) self.catalog = make_server.load_model_catalog(self.checkout) def tearDown(self): self._tmp.cleanup() def test_maps_paths_to_family_dirs_and_tasks(self): config = {"models": [ {"id": "qwen", "family": "qwen3_tts", "path": "models/Qwen3-TTS-12Hz-1.7B-Base-GGUF", "task": "tts"}, {"id": "qwen-design", "family": "qwen3_tts", "path": "models/Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF", "task": "vdes"}, {"id": "higgs", "family": "higgs_audio_tts", "path": "models/Higgs-Audio-v3-TTS-4B-GGUF", "task": "tts"}, ]} selected, tasks = make_server.server_config_selections(config, self.catalog) self.assertEqual(selected["qwen3_tts"], ["Qwen3-TTS-12Hz-1.7B-Base-GGUF", "Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF"]) self.assertEqual(selected["higgs_audio_tts"], ["Higgs-Audio-v3-TTS-4B-GGUF"]) self.assertEqual(tasks[("qwen3_tts", "Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF")], "vdes") self.assertEqual(tasks[("qwen3_tts", "Qwen3-TTS-12Hz-1.7B-Base-GGUF")], "tts") def test_unknown_family_ignored(self): config = {"models": [ {"id": "x", "family": "not_a_family", "path": "models/x"}, ]} selected, tasks = make_server.server_config_selections(config, self.catalog) self.assertEqual(selected, {}) self.assertEqual(tasks, {}) def test_absolute_and_unprefixed_paths_kept_as_targets(self): config = {"models": [ {"id": "qwen", "family": "qwen3_tts", "path": "/abs/Qwen3-TTS-12Hz-1.7B-Base-GGUF", "task": "tts"}, ]} selected, tasks = make_server.server_config_selections(config, self.catalog) self.assertEqual(selected["qwen3_tts"], ["/abs/Qwen3-TTS-12Hz-1.7B-Base-GGUF"]) def test_empty_models_yield_empty_selections(self): selected, tasks = make_server.server_config_selections({"models": []}, self.catalog) self.assertEqual(selected, {}) self.assertEqual(tasks, {}) class UnusedInstalledEntriesTests(unittest.TestCase): """unused_installed_entries: installed models dropped by a new selection.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.dir = Path(self._tmp.name) (self.dir / "models" / "kept").mkdir(parents=True) (self.dir / "models" / "kept" / "m.gguf").write_bytes(b"x") (self.dir / "models" / "dropped").mkdir() (self.dir / "models" / "dropped" / "m.gguf").write_bytes(b"x") (self.dir / "models" / "missing").mkdir() # empty: not installed def tearDown(self): self._tmp.cleanup() def _server_json(self, models): path = self.dir / "server.json" path.write_text(json.dumps({"models": models}), encoding="utf-8") return path def test_returns_installed_entries_not_in_new_paths(self): path = self._server_json([ {"id": "kept", "path": "models/kept"}, {"id": "dropped", "path": "models/dropped"}, {"id": "missing", "path": "models/missing"}, ]) unused = make_server.unused_installed_entries( path, {"models/kept"}) self.assertEqual([entry["id"] for entry in unused], ["dropped"]) def test_nothing_unused_when_all_kept(self): path = self._server_json([ {"id": "kept", "path": "models/kept"}, ]) unused = make_server.unused_installed_entries( path, {"models/kept"}) self.assertEqual(unused, []) class DeleteModelFilesTests(unittest.TestCase): """delete_model_files: remove on-disk model files for {id, rel} entries.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.dir = Path(self._tmp.name) (self.dir / "models" / "a").mkdir(parents=True) (self.dir / "models" / "a" / "m.gguf").write_bytes(b"x") (self.dir / "models" / "b").mkdir() (self.dir / "models" / "b" / "m.gguf").write_bytes(b"x") (self.dir / "models" / "c").mkdir(parents=True) self.server_json = self.dir / "server.json" self.server_json.write_text(json.dumps({ "models": [ {"id": "a", "path": "models/a"}, {"id": "b", "path": "models/b"}, {"id": "c", "path": "models/c"}, ], }), encoding="utf-8") def tearDown(self): self._tmp.cleanup() def test_removes_dirs_and_counts(self): removed = make_server.delete_model_files( self.server_json, [{"id": "a", "rel": "models/a"}, {"id": "b", "rel": "models/b"}]) self.assertEqual(removed, 2) self.assertFalse((self.dir / "models" / "a").exists()) self.assertFalse((self.dir / "models" / "b").exists()) self.assertTrue((self.dir / "models" / "c").exists()) def test_missing_paths_ignored(self): removed = make_server.delete_model_files( self.server_json, [{"id": "ghost", "rel": "models/ghost"}]) self.assertEqual(removed, 0) def test_removes_single_file(self): file_path = self.dir / "models" / "single.gguf" file_path.write_bytes(b"x") removed = make_server.delete_model_files( self.server_json, [{"id": "s", "rel": "models/single.gguf"}]) self.assertEqual(removed, 1) self.assertFalse(file_path.exists()) def test_absolute_rel_path_honored(self): target = self.dir / "absolute" target.mkdir() (target / "m.gguf").write_bytes(b"x") removed = make_server.delete_model_files( self.server_json, [{"id": "a", "rel": str(target)}]) self.assertEqual(removed, 1) self.assertFalse(target.exists()) class InstallModelsTests(unittest.TestCase): """install_models: runs the install helper with download=True.""" def test_downloads_delegating_to_install_models(self): with tempfile.TemporaryDirectory() as td: checkout = Path(td) guidance = [("qwen", "qwen3_tts_0_6b_base_q8_0")] with patch.object(make_server, "_install_models") as mk: make_server.install_models(checkout, guidance) mk.assert_called_once_with(checkout, guidance, download=True, emit=None, cancel=None) class HandInstallGuidanceTests(unittest.TestCase): """hand_install_guidance: explains how to install models by hand.""" def test_lists_each_model_and_its_path(self): with tempfile.TemporaryDirectory() as td: checkout = Path(td) message = make_server.hand_install_guidance(checkout, [ {"id": "qwen", "rel": "models/Qwen3-TTS-12Hz-0.6B-Base-GGUF"}, {"id": "higgs", "rel": "models/Higgs-Audio-4B-GGUF"}, ]) self.assertIn("qwen", message) self.assertIn("models/Qwen3-TTS-12Hz-0.6B-Base-GGUF", message) self.assertIn("higgs", message) self.assertIn("models/Higgs-Audio-4B-GGUF", message) self.assertIn("download", message.lower()) class WizardNavigationTests(unittest.TestCase): """Esc in the audio.cpp wizard goes back one screen (via tui.Wizard).""" def _args(self): return make_server.build_parser().parse_args([]) def _checkout(self): tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) return _make_checkout(Path(tmp.name)) def test_esc_on_first_screen_aborts(self): # Configure audio.cpp (modify flow): the families tree is the first # screen, so Esc on it must abort the wizard — not re-show itself. checkout = self._checkout() with patch.object(make_server, "find_local_checkout", return_value=checkout), \ patch.object(tui, "checkbox_tree", return_value=make_server._GO_BACK): settings = make_server._wizard(None, self._args(), make_server.build_parser()) self.assertIsNone(settings) def test_modify_flow_offers_build_when_not_built(self): # A server.json recording "vulkan" exists, but nothing is built: the # wizard must still reach the backend menu (pre-selecting vulkan) and # offer the build — instead of silently skipping it because the # existing server.json already records a backend. checkout = self._checkout() (checkout / "server.json").write_text( json.dumps({"models": [], "backend": "vulkan"}), encoding="utf-8") catalog = make_server.load_model_catalog(checkout) supertonic = next(i for i, entry in enumerate(catalog) if entry["family"] == "supertonic") confirm_questions = [] def fake_tree(*args, **kwargs): return [(supertonic, "Supertonic-GGUF")] def fake_line_edit(stdscr, title, default, **kwargs): if title == "Bind host": return "127.0.0.1" if title == "Port": return "8080" return default def fake_menu(stdscr, title, options, **kwargs): return "vulkan" def fake_confirm(stdscr, question, **kwargs): confirm_questions.append(question) return False # decline the build with patch.object(make_server, "find_local_checkout", return_value=checkout), \ patch.object(tui, "checkbox_tree", side_effect=fake_tree), \ patch.object(tui, "line_edit", side_effect=fake_line_edit), \ patch.object(tui, "menu", side_effect=fake_menu), \ patch.object(tui, "confirm", side_effect=fake_confirm): settings = make_server._wizard(None, self._args(), make_server.build_parser()) self.assertIsNotNone(settings) self.assertEqual(settings["backend"], "vulkan") self.assertFalse(settings["build"]) # The build offer was shown (and declined); the old modify flow # skipped it entirely. self.assertTrue(any("not built for vulkan" in q for q in confirm_questions)) def test_bind_host_esc_returns_to_families_tree(self): # Esc on "Bind host" must fall back to the model-family tree, then # re-selecting proceeds through the rest of the wizard. checkout = self._checkout() catalog = make_server.load_model_catalog(checkout) supertonic = next(i for i, entry in enumerate(catalog) if entry["family"] == "supertonic") tree_calls = [] hosts = iter([make_server._GO_BACK, "127.0.0.1"]) def fake_tree(*args, **kwargs): tree_calls.append(1) return [(supertonic, "Supertonic-GGUF")] def fake_line_edit(stdscr, title, default, **kwargs): if title == "Bind host": return next(hosts) if title == "Port": return "8080" return default with patch.object(make_server, "find_local_checkout", return_value=checkout), \ patch.object(tui, "checkbox_tree", side_effect=fake_tree), \ patch.object(tui, "line_edit", side_effect=fake_line_edit), \ patch.object(tui, "menu", return_value="cuda"), \ patch.object(tui, "confirm", return_value=True): settings = make_server._wizard(None, self._args(), make_server.build_parser()) self.assertIsNotNone(settings) # The tree was re-shown after the host screen's Esc. self.assertEqual(len(tree_calls), 2) self.assertEqual(settings["host"], "127.0.0.1") self.assertEqual([m["id"] for m in settings["model_entries"]], ["supertonic"]) class UninstallTests(unittest.TestCase): """uninstall: stop the server and remove the checkout.""" def test_removes_checkout(self): with tempfile.TemporaryDirectory() as td: checkout = Path(td) / "audio.cpp" checkout.mkdir() with patch.object(make_server, "find_local_checkout", return_value=checkout), \ patch.object(make_server.servers, "stop") as mk_stop: rc = make_server.uninstall() self.assertEqual(rc, 0) self.assertFalse(checkout.exists()) mk_stop.assert_called_once_with("audiocpp") def test_no_checkout_is_a_noop(self): with patch.object(make_server, "find_local_checkout", return_value=None), \ patch.object(make_server.servers, "stop") as mk_stop: rc = make_server.uninstall() self.assertEqual(rc, 0) mk_stop.assert_called_once_with("audiocpp") if __name__ == "__main__": unittest.main() class SetupScreenTests(unittest.TestCase): """setup_screen: the wizard run on the hub's screen, setup tail via the in-TUI task view (two parallel lanes on a fresh install).""" def test_abort_returns_one_without_executing(self): with patch.object(make_server, "_wizard", return_value=None) as mk_wizard, \ patch.object(make_server, "_execute_lanes") as mk_lanes: rc = make_server.setup_screen(None) self.assertEqual(rc, 1) mk_wizard.assert_called_once() mk_lanes.assert_not_called() def test_success_runs_the_tail_in_the_task_view(self): settings = {"audiocpp_dir": Path("/x")} lanes = [make_server.taskview.TaskLane( "Build", [make_server.taskview.TaskStep("t", lambda emit, cancel: 0)])] with patch.object(make_server, "_wizard", return_value=settings), \ patch.object(make_server, "_execute_lanes", return_value=lanes) as mk_lanes, \ patch.object(make_server.taskview, "run_lanes", return_value=0) as mk_run: rc = make_server.setup_screen(None) self.assertEqual(rc, 0) mk_lanes.assert_called_once() self.assertIs(mk_lanes.call_args[0][0], settings) self.assertTrue(mk_lanes.call_args[1]["parallel"]) mk_run.assert_called_once() self.assertEqual(mk_run.call_args[0][2], lanes) class ExecuteLanesTests(unittest.TestCase): """_execute_lanes: two lanes (build + configure/download) and the flattened console order.""" def _settings(self, **overrides): settings = { "audiocpp_dir": Path("/x"), "backend": "cuda", "build": True, "download": True, "include_clone": False, "wav_dir": None, "plan": None, "sync_port": None, "sync_model_ids": None, "delete_unused": False, "unused_entries": [], "model_entries": [], "entry_ids": [], "install_guidance": [], "output_path": Path("/x/server.json"), "host": "127.0.0.1", "port": 8080, "lazy_load": True, } settings.update(overrides) return settings def test_two_lanes_when_building(self): args = make_server.build_parser().parse_args([]) lanes = make_server._execute_lanes(self._settings(), args) self.assertEqual([lane.title for lane in lanes], ["Build", "Configure & download"]) self.assertEqual([s.title for s in lanes[0].steps], ["Build audiocpp_server (cuda)"]) self.assertEqual([s.title for s in lanes[1].steps], ["Transcribe reference voices", "Write server.json & sync config", "Download models"]) def test_single_lane_when_not_building(self): args = make_server.build_parser().parse_args([]) lanes = make_server._execute_lanes( self._settings(build=False), args) self.assertEqual([lane.title for lane in lanes], ["Configure & download"]) def test_flattened_console_steps_keep_the_build_first(self): args = make_server.build_parser().parse_args([]) steps = make_server._execute_steps(self._settings(), args) self.assertEqual([s.title for s in steps], ["Build audiocpp_server (cuda)", "Transcribe reference voices", "Write server.json & sync config", "Download models"]) def test_download_step_prints_the_parallel_launch_hint(self): args = make_server.build_parser().parse_args([]) lanes = make_server._execute_lanes(self._settings(), args, parallel=True) install_step = lanes[1].steps[2] with patch.object(make_server, "_install_models") as mk_install, \ patch.object(make_server, "_print_launch_hint") as mk_hint: install_step.work(lambda line: None, threading.Event()) mk_hint.assert_called_once() self.assertTrue(mk_hint.call_args[1]["pending_build"]) def test_download_step_console_hint_is_not_pending(self): args = make_server.build_parser().parse_args([]) lanes = make_server._execute_lanes(self._settings(), args) install_step = lanes[1].steps[2] with patch.object(make_server, "_install_models") as mk_install, \ patch.object(make_server, "_print_launch_hint") as mk_hint: install_step.work(lambda line: None, threading.Event()) mk_hint.assert_called_once() self.assertFalse(mk_hint.call_args[1]["pending_build"]) class LaunchHintTests(unittest.TestCase): """_print_launch_hint: exact command vs. the pending-build message.""" def _capture(self, audiocpp_dir, output_path, pending_build=False): buf = io.StringIO() with redirect_stdout(buf), \ patch.object(make_server, "find_audiocpp_server_bin", return_value=None): make_server._print_launch_hint(audiocpp_dir, output_path, pending_build=pending_build) return buf.getvalue() def test_pending_build_names_the_post_build_command(self): out = self._capture(Path("/tmp/acpp"), Path("/tmp/acpp/server.json"), pending_build=True) self.assertIn("still building", out) self.assertNotIn("Build it first", out) self.assertIn("audiocpp_server --config /tmp/acpp/server.json", out) def test_missing_binary_gives_build_remediation(self): out = self._capture(Path("/tmp/acpp"), Path("/tmp/acpp/server.json")) self.assertIn("Build it first", out) self.assertNotIn("still building", out)