diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_backends.py | 91 | ||||
| -rw-r--r-- | tests/test_backends_audiocpp.py | 1062 | ||||
| -rw-r--r-- | tests/test_backends_faster.py (renamed from tests/test_make_faster_voices_json.py) | 67 | ||||
| -rw-r--r-- | tests/test_hub.py | 91 | ||||
| -rw-r--r-- | tests/test_make_audiocpp_server_json.py | 1743 | ||||
| -rw-r--r-- | tests/test_tui.py | 37 |
6 files changed, 1314 insertions, 1777 deletions
diff --git a/tests/test_backends.py b/tests/test_backends.py new file mode 100644 index 0000000..4017cd4 --- /dev/null +++ b/tests/test_backends.py @@ -0,0 +1,91 @@ +"""Tests for the backends package registry and detection aggregation.""" + +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from backends import REGISTRY, detect_all, get + + +class RegistryTests(unittest.TestCase): + def test_registry_has_the_three_backends(self): + keys = [info.key for info in REGISTRY] + self.assertEqual(keys, ["audiocpp", "qwen", "faster"]) + + def test_every_entry_has_detect_and_setup_tui(self): + for info in REGISTRY: + self.assertTrue(callable(info.detect), info.key) + self.assertTrue(callable(info.setup_tui), info.key) + self.assertIsInstance(info.modify_actions, list) + for action in info.modify_actions: + self.assertTrue(callable(action.run)) + + def test_get_returns_entry_by_key(self): + self.assertIs(get("audiocpp").key, "audiocpp") + self.assertIsNone(get("nonexistent")) + + +class DetectAllTests(unittest.TestCase): + def test_detect_all_returns_one_status_per_backend(self): + statuses = detect_all() + self.assertEqual([s.key for s in statuses], + ["audiocpp", "qwen", "faster"]) + for s in statuses: + self.assertIn(s.key, ("audiocpp", "qwen", "faster")) + # ready requires both installed and configured; on a clean + # machine none are ready. + if s.ready: + self.assertTrue(s.installed and s.configured) + + def test_audiocpp_status_when_cloned_built_configured(self): + with tempfile.TemporaryDirectory() as td: + root = Path(td) + checkout = root / "audio.cpp" + checkout.mkdir() + (checkout / "model_specs").mkdir() + (checkout / "build" / "linux-cuda-release" / "bin").mkdir( + parents=True) + (checkout / "build" / "linux-cuda-release" / "bin" + / "audiocpp_server").write_bytes(b"x") + (checkout / "server.json").write_text('{"models":[]}', + encoding="utf-8") + from backends import audiocpp + with patch.object(audiocpp, "find_local_checkout", + return_value=checkout): + status = audiocpp.detect() + self.assertTrue(status.installed) + self.assertTrue(status.configured) + self.assertTrue(status.ready) + self.assertIn("audiocpp_server", status.launch_hint) + + def test_qwen_status_reflects_install(self): + from backends import qwen + with patch.object(qwen, "_is_installed", return_value=True): + status = qwen.detect() + self.assertTrue(status.installed) + self.assertTrue(status.configured) + self.assertIn("qwen-tts-demo", status.launch_hint) + with patch.object(qwen, "_is_installed", return_value=False): + status = qwen.detect() + self.assertFalse(status.installed) + self.assertFalse(status.configured) + + def test_faster_status_reflects_install_clone_voices(self): + from backends import faster + with tempfile.TemporaryDirectory() as td: + checkout = Path(td) / "faster-qwen3-tts" + (checkout / "examples").mkdir(parents=True) + (checkout / "examples" / "openai_server.py").write_text("x") + (checkout / "voices.json").write_text('{"default":{}}', + encoding="utf-8") + with patch.object(faster, "_is_installed", return_value=True), \ + patch.object(faster, "_checkout", return_value=checkout): + status = faster.detect() + self.assertTrue(status.installed) + self.assertTrue(status.configured) + self.assertIn("openai_server.py", status.launch_hint) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_backends_audiocpp.py b/tests/test_backends_audiocpp.py new file mode 100644 index 0000000..9882ce1 --- /dev/null +++ b/tests/test_backends_audiocpp.py @@ -0,0 +1,1062 @@ +"""Tests for the audio.cpp backend setup module (backends/audiocpp.py).""" + +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 backends import audiocpp 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/<family>.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 CheckoutAutoSelectTests(unittest.TestCase): + """TUI browser auto-accept callback for an audio.cpp checkout.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.root = Path(self._td.name) + + def tearDown(self): + self._td.cleanup() + + def test_accepts_audio_cpp_containing_model_specs(self): + checkout = self.root / "audio.cpp" + checkout.mkdir() + (checkout / "model_specs").mkdir() + self.assertEqual(make_server._checkout_auto_select(checkout), + checkout) + + def test_rejects_audio_cpp_without_model_specs(self): + checkout = self.root / "audio.cpp" + checkout.mkdir() + self.assertIsNone(make_server._checkout_auto_select(checkout)) + + def test_rejects_other_name_even_with_model_specs(self): + other = self.root / "not-audiocpp" + other.mkdir() + (other / "model_specs").mkdir() + self.assertIsNone(make_server._checkout_auto_select(other)) + + def test_rejects_plain_directory(self): + plain = self.root / "somewhere" + plain.mkdir() + self.assertIsNone(make_server._checkout_auto_select(plain)) + + +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() + + +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 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") + + def tearDown(self): + self._td.cleanup() + + 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) + + +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, "") + + 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) + + +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.assertFalse(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) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_make_faster_voices_json.py b/tests/test_backends_faster.py index f26e645..641f6ee 100644 --- a/tests/test_make_faster_voices_json.py +++ b/tests/test_backends_faster.py @@ -1,4 +1,4 @@ -"""Tests for the faster-qwen3-tts voices.json generator tool.""" +"""Tests for the faster-qwen3-tts backend setup module (backends/faster.py).""" import json import sys @@ -7,7 +7,7 @@ import unittest from pathlib import Path from unittest.mock import patch -from tools import make_faster_voices_json as make_voices +from backends import faster as make_voices class FindWavFilesTests(unittest.TestCase): @@ -81,24 +81,35 @@ class BuildVoicesTests(unittest.TestCase): class MainTests(unittest.TestCase): + """The flag-only (non-TUI) path through main(), end to end.""" + def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.folder = Path(self._tmp.name) (self.folder / "narrator.wav").write_bytes(b"x") (self.folder / "alpha.wav").write_bytes(b"x") self.output = self.folder / "voices.json" + # Avoid touching the real converter/config.py and pip/git. + patcher = patch.object(make_voices.common, "update_config_value", + return_value=False) + patcher.start() + self.addCleanup(patcher.stop) + patcher = patch.object(make_voices, "_interactive", return_value=False) + patcher.start() + self.addCleanup(patcher.stop) def tearDown(self): self._tmp.cleanup() def _run(self, argv): - with patch.object(sys, "argv", ["make_voices.py"] + argv): + with patch.object(sys, "argv", ["backends/faster.py"] + argv), \ + patch.object(make_voices, "transcribe_reference_audio", + return_value="hello"): return make_voices.main() def test_writes_json_with_alphabetical_voice_order(self): - with patch.object(make_voices, "transcribe_reference_audio", - return_value="hello"): - exit_code = self._run([str(self.folder)]) + exit_code = self._run([str(self.folder), "--output", str(self.output), + "--skip-install", "--skip-clone"]) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) self.assertEqual(list(data), ["alpha", "narrator"]) @@ -107,59 +118,51 @@ class MainTests(unittest.TestCase): def test_custom_output_path(self): custom = Path(self._tmp.name) / "custom.json" - with patch.object(make_voices, "transcribe_reference_audio", - return_value="hello"): - self._run([str(self.folder), "--output", str(custom)]) + exit_code = self._run([str(self.folder), "--output", str(custom), + "--skip-install", "--skip-clone"]) + self.assertEqual(exit_code, 0) self.assertTrue(custom.exists()) self.assertFalse(self.output.exists()) def test_invalid_language_errors_before_work(self): with patch.object(make_voices, "transcribe_reference_audio") as mock_transcribe: with self.assertRaises(SystemExit) as ctx: - self._run([str(self.folder), "--language", "klingon"]) + self._run([str(self.folder), "--output", str(self.output), + "--language", "klingon", "--skip-install", + "--skip-clone"]) self.assertEqual(ctx.exception.code, 2) mock_transcribe.assert_not_called() def test_missing_input_dir_errors(self): with self.assertRaises(SystemExit) as ctx: - self._run([str(self.folder / "nope")]) + self._run([str(self.folder / "nope"), "--output", str(self.output), + "--skip-install", "--skip-clone"]) self.assertEqual(ctx.exception.code, 2) - def test_no_wav_files_errors(self): + def test_no_wav_files_returns_error(self): empty = Path(tempfile.mkdtemp()) try: - with self.assertRaises(SystemExit) as ctx: - self._run([str(empty)]) - self.assertEqual(ctx.exception.code, 2) + exit_code = self._run([str(empty), "--output", + str(empty / "voices.json"), + "--skip-install", "--skip-clone"]) + self.assertEqual(exit_code, 1) finally: empty.rmdir() def test_existing_output_declined_keeps_file(self): self.output.write_text('{"old": true}', encoding="utf-8") - with patch.object(make_voices, "transcribe_reference_audio") as mock_transcribe, \ - patch("builtins.input", return_value="n"): - exit_code = self._run([str(self.folder)]) + with patch.object(make_voices, "transcribe_reference_audio") as mock_transcribe: + exit_code = self._run([str(self.folder), "--output", str(self.output), + "--skip-install", "--skip-clone"]) self.assertEqual(exit_code, 1) mock_transcribe.assert_not_called() 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") - with patch.object(make_voices, "transcribe_reference_audio", - return_value="hello"), \ - patch("builtins.input", return_value="y"): - exit_code = self._run([str(self.folder)]) - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(list(data), ["alpha", "narrator"]) - def test_force_overwrites_without_prompt(self): self.output.write_text('{"old": true}', encoding="utf-8") - with patch.object(make_voices, "transcribe_reference_audio", - return_value="hello"), \ - patch("builtins.input", side_effect=AssertionError("prompted")): - exit_code = self._run([str(self.folder), "--force"]) + exit_code = self._run([str(self.folder), "--output", str(self.output), + "--force", "--skip-install", "--skip-clone"]) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) self.assertEqual(list(data), ["alpha", "narrator"]) diff --git a/tests/test_hub.py b/tests/test_hub.py new file mode 100644 index 0000000..5f6d992 --- /dev/null +++ b/tests/test_hub.py @@ -0,0 +1,91 @@ +"""Tests for the TUI hub (hub.py) menu and helpers. + +The hub drives the same curses widgets as tui.py, so these tests reuse the +fake curses/screen from test_tui to run the menu without a terminal. +""" + +import unittest +from pathlib import Path +from unittest.mock import patch + +import hub +import tui +from tests.test_tui import FakeCurses, FakeScreen + + +class HubHelperTests(unittest.TestCase): + """Pure helpers in hub.py (no curses).""" + + def test_is_float(self): + self.assertTrue(hub._is_float("1.0")) + self.assertTrue(hub._is_float("2")) + self.assertFalse(hub._is_float("abc")) + self.assertFalse(hub._is_float("")) + + def test_list_voices_from_dir(self): + with __import__("tempfile").TemporaryDirectory() as td: + d = Path(td) + (d / "Narrator.wav").write_bytes(b"x") + (d / "Alpha.WAV").write_bytes(b"x") + (d / "notes.txt").write_bytes(b"x") + voices = hub._list_voices(str(d)) + # Stems preserve case; sorting is case-insensitive. + self.assertEqual(voices, ["Alpha", "Narrator"]) + + def test_list_voices_missing_dir(self): + self.assertEqual(hub._list_voices("/no/such/dir"), []) + + def test_status_mark(self): + from backends import BackendStatus + ready = BackendStatus("k", "l", installed=True, configured=True) + half = BackendStatus("k", "l", installed=True, configured=False) + none = BackendStatus("k", "l", installed=False, configured=False) + self.assertEqual(hub._status_mark("k", [ready]), "ready") + self.assertEqual(hub._status_mark("k", [half]), "installed") + self.assertEqual(hub._status_mark("k", [none]), "not set up") + self.assertEqual(hub._status_mark("missing", []), "not set up") + + +class HubMenuTests(unittest.TestCase): + """Drive _hub_menu with a fake screen (no terminal).""" + + def setUp(self): + tui._THEME.clear() + self.curses = FakeCurses() + from unittest.mock import patch as _patch + self._patcher = _patch.dict("sys.modules", {"curses": self.curses}) + self._patcher.start() + self.addCleanup(self._patcher.stop) + self.addCleanup(tui._THEME.clear) + + def test_quit_returns_none(self): + # Main menu: move to "Quit" (4th option, index 3) and press Enter. + screen = FakeScreen(keys=[FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, + FakeCurses.KEY_DOWN, 10]) + with patch.object(hub, "detect_all", return_value=[]): + result = hub._hub_menu(screen) + self.assertIsNone(result) + + def test_convert_with_no_ready_backend_offers_setup(self): + # Convert -> "Set up a backend..." is the only entry -> Enter selects + # it -> setup menu lists 3 backends; press Esc to go back -> convert + # returns None -> main menu loops. Then quit (Down x3 + Enter). + from backends import BackendInfo, BackendStatus + none = BackendStatus("k", "l", installed=False, configured=False) + infos = [BackendInfo("audiocpp", "audio.cpp", lambda: none, + lambda: 0), + BackendInfo("qwen", "Qwen", lambda: none, lambda: 0), + BackendInfo("faster", "faster", lambda: none, lambda: 0)] + with patch.object(hub, "detect_all", return_value=[none, none, none]), \ + patch.object(hub, "REGISTRY", infos): + # Convert(Enter), setup-entry(Enter), Esc on setup menu, + # back at main menu -> Down x3 -> Enter (Quit). + screen = FakeScreen(keys=[10, 10, 27, + FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, + FakeCurses.KEY_DOWN, 10]) + result = hub._hub_menu(screen) + self.assertIsNone(result) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_make_audiocpp_server_json.py b/tests/test_make_audiocpp_server_json.py deleted file mode 100644 index 39d32a8..0000000 --- a/tests/test_make_audiocpp_server_json.py +++ /dev/null @@ -1,1743 +0,0 @@ -"""Tests for the audio.cpp server.json generator tool.""" - -import argparse -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 -from tools 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/<family>.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): - # Shallow search only: a wav dir two levels deep is not a candidate. - 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): - """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 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 CheckoutAutoSelectTests(unittest.TestCase): - """TUI browser auto-accept callback for an audio.cpp checkout.""" - - def setUp(self): - self._td = tempfile.TemporaryDirectory() - self.root = Path(self._td.name) - - def tearDown(self): - self._td.cleanup() - - def test_accepts_audio_cpp_containing_model_specs(self): - checkout = self.root / "audio.cpp" - checkout.mkdir() - (checkout / "model_specs").mkdir() - self.assertEqual(make_server._checkout_auto_select(checkout), - checkout) - - def test_rejects_audio_cpp_without_model_specs(self): - checkout = self.root / "audio.cpp" - checkout.mkdir() - self.assertIsNone(make_server._checkout_auto_select(checkout)) - - def test_rejects_other_name_even_with_model_specs(self): - other = self.root / "not-audiocpp" - other.mkdir() - (other / "model_specs").mkdir() - self.assertIsNone(make_server._checkout_auto_select(other)) - - def test_rejects_plain_directory(self): - plain = self.root / "somewhere" - plain.mkdir() - self.assertIsNone(make_server._checkout_auto_select(plain)) - - -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_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)) - # No family is marked "tested" anymore; all are treated equally. - 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): - # 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 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 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 (higgs_audio_tts and pocket_tts in alphabetical order). - with patch("builtins.input", side_effect=["1,3"]): - chosen = make_server.ask_families(self.catalog) - self.assertEqual(chosen, ["higgs_audio_tts", "pocket_tts"]) - - def test_space_separated_numbers(self): - with patch("builtins.input", side_effect=["2 4"]): - chosen = make_server.ask_families(self.catalog) - self.assertEqual(chosen, ["index_tts2", "qwen3_tts"]) - - 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, ["higgs_audio_tts", "index_tts2"]) - - 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, ["index_tts2"]) - - -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() - - -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 DesignPackageTests(unittest.TestCase): - """Voice-design package detection.""" - - def test_detects_voicedesign_in_id(self): - self.assertTrue(make_server.is_design_package( - {"id": "qwen3_tts_1_7b_voicedesign_q8_0"})) - - def test_detects_voicedesign_in_directory(self): - self.assertTrue(make_server.is_design_package( - {"target_directory": "Foo-VoiceDesign-GGUF"})) - - def test_detects_separated_voice_design(self): - self.assertTrue(make_server.is_design_package( - {"display_name": "Voice Design Q8_0"})) - - def test_ignores_other_packages(self): - self.assertFalse(make_server.is_design_package( - {"id": "higgs_audio_tts_4b_q8_0"})) - self.assertFalse(make_server.is_design_package({})) - - -class PackageDirOptionsTests(unittest.TestCase): - """Grouping a family's packages into distinct target directories.""" - - def test_groups_precisions_and_marks_recommended(self): - entry = { - "family": "qwen3_tts", - "packages": [ - {"id": "base_q8", "default": True, "format": "gguf", - "target_directory": "Base-GGUF"}, - {"id": "base_bf16", "format": "gguf", - "target_directory": "Base-GGUF"}, - {"id": "voicedesign_q8", "format": "gguf", - "target_directory": "VoiceDesign-GGUF"}, - ], - } - options = make_server.package_dir_options(entry) - self.assertEqual([o["target_directory"] for o in options], - ["Base-GGUF", "VoiceDesign-GGUF"]) - self.assertTrue(options[0]["recommended"]) - self.assertFalse(options[0]["design"]) - self.assertFalse(options[1]["recommended"]) - self.assertTrue(options[1]["design"]) - # The recommended precision inside the shared directory wins. - self.assertEqual(options[0]["install_id"], "base_q8") - - def test_recommended_comes_first_even_if_listed_later(self): - entry = { - "family": "demo_tts", - "packages": [ - {"id": "demo_other", "format": "gguf", - "target_directory": "Other-GGUF"}, - {"id": "demo_default", "default": True, "format": "gguf", - "target_directory": "Default-GGUF"}, - ], - } - options = make_server.package_dir_options(entry) - self.assertEqual([o["target_directory"] for o in options], - ["Default-GGUF", "Other-GGUF"]) - - -class _MainTestBase(unittest.TestCase): - """Shared fixtures for end-to-end main() tests.""" - - 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) - # Force the line-prompt flow regardless of the test terminal, so - # the builtins.input patches below are what actually answer the - # questions (the TUI path is exercised separately). - patcher = patch.object(make_server, "_tui_enabled", return_value=False) - patcher.start() - self.addCleanup(patcher.stop) - - def tearDown(self): - self._td.cleanup() - - 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() - - # Default single-family run inputs (no flags, port matches config): - # family, host, port, backend, lazy, model-id-sync. - def _defaults(self, sync="y"): - return ["", "", "", "", "", sync] - - -class MainTests(_MainTestBase): - """The default single-family flow and shared server settings.""" - - def _args(self, *extra): - return ["--wavs", str(self.folder), "--output", str(self.output), - "--audiocpp-dir", str(self.checkout)] + list(extra) - - def test_missing_wav_dir_prompted_errors(self): - # No --wavs and EOF at the prompt -> hard error. - buf = io.StringIO() - with patch.object(sys, "argv", - ["make_audiocpp_server_json.py", - "--output", str(self.output), - "--audiocpp-dir", str(self.checkout)]), \ - patch("builtins.input", side_effect=EOFError), \ - redirect_stdout(buf): - with self.assertRaises(SystemExit) as ctx: - make_server.main() - self.assertEqual(ctx.exception.code, 2) - - def test_missing_audiocpp_dir_errors(self): - with self.assertRaises(SystemExit) as ctx: - self._run(self._args("--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", - "--wavs", str(self.folder), - "--output", str(self.output)]), \ - patch("builtins.input", side_effect=EOFError), \ - redirect_stdout(buf): - with self.assertRaises(SystemExit) as ctx: - make_server.main() - self.assertEqual(ctx.exception.code, 2) - - def test_wav_prompt_defaults_to_detected_dir(self): - # No --wavs: the prompt default is the unique .wav directory detected - # across the checkouts; pressing Enter accepts it. - (self.folder / "narrator.wav").write_bytes(b"x") - argv = ["make_audiocpp_server_json.py", - "--output", str(self.output), - "--audiocpp-dir", str(self.checkout)] - # wav(Enter -> default), family, host, port, backend, lazy, sync(y) - inputs = ["", "", "", "", "", "", "y"] - with patch.object(sys, "argv", argv), \ - patch("builtins.input", side_effect=inputs), \ - patch.object(make_server, "detect_wav_dir", - return_value=self.folder), \ - patch.object(make_server, "transcribe_reference_audio", - side_effect=lambda path, model_name="base": "t"), \ - patch.object(make_server, "whisper_backend_available", - return_value="faster_whisper"): - exit_code = make_server.main() - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(data["voice_dir"], str(self.folder.resolve())) - - def test_default_run_hosts_recommended_entry(self): - exit_code = self._run(self._args(), inputs=self._defaults()) - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(data["host"], "127.0.0.1") - self.assertEqual(data["port"], make_server.config_port()) - self.assertEqual(data["backend"], "cuda") - # Single family -> one entry, lazy defaults to False. - self.assertFalse(data["lazy_load"]) - # The first family alphabetically is Higgs Audio v3 TTS 4B. - self.assertEqual([model["id"] for model in data["models"]], ["higgs"]) - self.assertEqual( - [model["path"] for model in data["models"]], - ["models/Higgs-Audio-v3-TTS-4B-GGUF"]) - self.assertEqual(data["models"][0]["task"], "tts") - # 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"]), 1) - - 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. - # family, host, port_sync(y), backend, lazy, sync(y) - inputs = ["", "", "y", "", "", "y"] - exit_code = self._run( - self._args("--port", "8080"), inputs=inputs) - self.assertEqual(exit_code, 0) - 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", "", "", "n"] - 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_no_final_confirm_prompt_writes_file(self): - # There is no final confirmation prompt anymore; the config is always - # written once the (single) overwrite check has been passed. - exit_code = self._run(self._args(), inputs=EOFError) - self.assertEqual(exit_code, 0) - self.assertTrue(self.output.exists()) - - def test_existing_output_declined_keeps_file(self): - self.output.write_text('{"old": true}', encoding="utf-8") - 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"]), 1) - - 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"]), 1) - - def test_flags_skip_prompts(self): - # --families qwen3_tts + server flags; port 9000 differs from config - # port 8080 -> the port sync prompt still fires. - exit_code = self._run( - self._args("--families", "qwen3_tts", - "--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_final_output_is_wrote_plus_install_commands(self): - # Two families -> two entries; the console output ends with the Wrote - # line and one full-path install command per model, nothing else. - code, out = self._run_capturing( - self._args("--families", "qwen3_tts,higgs_audio_tts"), - inputs=["", "", "", ""]) - self.assertEqual(code, 0) - self.assertIn(f"Wrote {self.output.resolve()} with 2 entries.", out) - manager = self.checkout / "tools" / "model_manager_v2.py" - self.assertIn(f"python {manager} install qwen3_tts_1_7b_base_q8_0", - out) - self.assertIn(f"python {manager} install higgs_audio_tts_4b_q8_0", out) - # The generated JSON and the old [INFO] notes are no longer echoed. - self.assertNotIn("[INFO]", out) - self.assertNotIn('"models"', out) - - def test_missing_wav_dir_flag_errors_with_message(self): - missing = self.root / "nope" - with self.assertRaises(SystemExit) as ctx, \ - patch("sys.stderr") as mock_stderr: - self._run(["--wavs", 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 _run_capturing(self, argv, inputs): - argv = ["make_audiocpp_server_json.py"] + argv - buf = io.StringIO() - with patch.object(sys, "argv", argv), \ - patch("builtins.input", side_effect=inputs), \ - patch.object(make_server, "transcribe_reference_audio"), \ - patch.object(make_server, "whisper_backend_available", - return_value="faster_whisper"), \ - redirect_stdout(buf): - code = make_server.main() - return code, buf.getvalue() - - def test_detected_backend_selected_by_default(self): - # A built backend in the checkout's build/ dir makes that backend the - # default; pressing Enter on the backend prompt accepts it. - build_dir = self.checkout / "build" / "linux-vulkan-release" / "bin" - build_dir.mkdir(parents=True) - (build_dir / "audiocpp_server").write_bytes(b"x") - # host, port, backend(Enter -> detected vulkan), lazy, sync(y) - inputs = ["", "", "", "", "", "y"] - code, out = self._run_capturing(self._args(), inputs=inputs) - self.assertEqual(code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(data["backend"], "vulkan") - - def test_download_accepted_runs_install_commands(self): - manager = self.checkout / "tools" / "model_manager_v2.py" - manager.parent.mkdir(parents=True, exist_ok=True) - manager.write_text("#!/usr/bin/env python3\n", encoding="utf-8") - argv = ["make_audiocpp_server_json.py"] + self._args( - "--families", "qwen3_tts") - with patch.object(sys, "argv", argv), \ - patch("builtins.input", side_effect=["", "", "", "", "y"]), \ - patch.object(make_server, "transcribe_reference_audio"), \ - patch.object(make_server, "whisper_backend_available", - return_value="faster_whisper"), \ - patch.object(make_server.subprocess, "run", - return_value=MagicMock(returncode=0)) as run: - code = make_server.main() - self.assertEqual(code, 0) - run.assert_called_once_with( - [sys.executable, str(manager), "install", - "qwen3_tts_1_7b_base_q8_0"], - cwd=str(self.checkout)) - - def test_all_packages_design_hosts_vdes_entry(self): - # --all-packages: pick the VoiceDesign package (menu 3) and accept the - # "design" default so it is hosted with task "vdes". - self.fake_config.write_text(FAKE_CONFIG_WITH_MODEL_IDS, - encoding="utf-8") - # packages(3=VoiceDesign), task(design default Enter), host, - # port, backend, lazy, sync(y) - inputs = ["3", "", "", "", "", "", "y"] - code, out = self._run_capturing( - self._args("--families", "qwen3_tts", "--all-packages"), - inputs=inputs) - self.assertEqual(code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(data["models"], [{ - "id": "qwen-design", - "family": "qwen3_tts", - "path": "models/Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF", - "task": "vdes", - "mode": "offline", - }]) - self.assertNotIn("voice_dir", data) - # Only the VoiceDesign package is installed (custom/base are not). - self.assertIn("install qwen3_tts_1_7b_voicedesign_q8_0", out) - self.assertNotIn("install qwen3_tts_1_7b_customvoice_q8_0", out) - self.assertNotIn("install qwen3_tts_1_7b_base_q8_0", out) - # Single-entry server: the converter ids are synced to the entry. - text = self.fake_config.read_text(encoding="utf-8") - self.assertIn('AUDIOCPP_MODEL_ID = "qwen-design"', text) - self.assertIn('AUDIOCPP_CLONE_MODEL_ID = "qwen-design"', text) - - def test_all_packages_non_design_package_gets_tts_no_prompt(self): - # CustomVoice (menu 2) is not a design package -> task "tts" with no - # task prompt. - # packages(2=CustomVoice), host, port, backend, lazy, sync(y) - inputs = ["2", "", "", "", "", "y"] - code, _ = self._run_capturing( - self._args("--families", "qwen3_tts", "--all-packages"), - inputs=inputs) - self.assertEqual(code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(data["models"], [{ - "id": "qwen", - "family": "qwen3_tts", - "path": "models/Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF", - "task": "tts", - "mode": "offline", - }]) - - def test_all_packages_both_tts_and_design(self): - # Pick Base (recommended) + VoiceDesign -> two entries; the design - # package prompts for its task. - # packages(1,3), task(design default Enter), host, port, backend, lazy - inputs = ["1,3", "", "", "", "", ""] - code, _ = self._run_capturing( - self._args("--families", "qwen3_tts", "--all-packages"), - inputs=inputs) - self.assertEqual(code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual([model["id"] for model in data["models"]], - ["qwen", "qwen-design"]) - self.assertEqual([model["task"] for model in data["models"]], - ["tts", "vdes"]) - - -class NonQwenFamilyMainTests(_MainTestBase): - """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 ["--wavs", str(self.folder), "--output", str(self.output), - "--audiocpp-dir", str(self.checkout), - "--families", family] + list(extra) - - def test_higgs_family_run(self): - (self.folder / "narrator.wav").write_bytes(b"x") - # Single family -> path comes from the catalog (no prompt); host, port, - # backend, lazy, model-id sync(y). - inputs = ["", "", "", "", "y"] - exit_code = self._run( - self._args("higgs_audio_tts"), inputs=inputs, - transcribe=lambda path, model_name="base": "a transcript") - 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 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") - # host, port, backend, lazy, sync(n) - inputs = ["", "", "", "", "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() - # host, port, backend, lazy, sync(y) - inputs = ["", "", "", "", "y"] - with patch.object(sys, "argv", - ["make_audiocpp_server_json.py", - "--wavs", 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 ["--wavs", str(self.folder), "--output", str(self.output), - "--audiocpp-dir", str(self.checkout)] + list(extra) - - def test_multiple_families_lazy_by_default_with_voice_dir(self): - (self.folder / "narrator.wav").write_bytes(b"x") - # --families selects qwen3_tts + higgs_audio_tts; each hosts its - # recommended package. host, port, backend, lazy(default True->Enter). - inputs = ["", "", "", ""] - exit_code = self._run( - self._args("--families", "qwen3_tts,higgs_audio_tts"), - inputs=inputs, - 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", "higgs"]) - # Two entries -> lazy defaults to True. - self.assertTrue(data["lazy_load"]) - self.assertEqual(data["voice_dir"], str(self.folder.resolve())) - higgs = data["models"][1] - self.assertEqual(higgs["path"], "models/Higgs-Audio-v3-TTS-4B-GGUF") - - def test_two_non_qwen_families_use_catalog_paths(self): - # Multiple families -> paths come from the catalog (no prompts). - # host, port, backend, lazy - inputs = ["", "", "", ""] - exit_code = self._run( - self._args("--families", "higgs_audio_tts,voxcpm2"), - inputs=inputs) - 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 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. - # host, port, backend, lazy, sync(n) - inputs = ["", "", "", "", "n"] - with patch.object(sys, "argv", - ["make_audiocpp_server_json.py", - "--wavs", 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 DefaultOutputTests(_MainTestBase): - """server.json defaults into the audio.cpp checkout unless declined.""" - - def test_default_output_written_into_checkout(self): - # No --output: server.json lands in the audio.cpp checkout. - argv = ["--wavs", str(self.folder), "--audiocpp-dir", str(self.checkout)] - exit_code = self._run(argv, inputs=self._defaults()) - self.assertEqual(exit_code, 0) - out = self.checkout / "server.json" - self.assertTrue(out.exists()) - data = json.loads(out.read_text(encoding="utf-8")) - self.assertEqual(len(data["models"]), 1) - - def test_declined_overwrite_falls_back_to_cwd(self): - # A pre-existing server.json in the checkout; declining the overwrite - # writes server.json into the current working directory instead. - checkout_out = self.checkout / "server.json" - checkout_out.write_text('{"old": true}', encoding="utf-8") - cwd = self.root / "run-cwd" - cwd.mkdir() - argv = ["--wavs", str(self.folder), "--audiocpp-dir", str(self.checkout)] - with patch.object(make_server.os, "getcwd", return_value=str(cwd)): - exit_code = self._run(argv, inputs=["n"] + self._defaults()) - self.assertEqual(exit_code, 0) - self.assertEqual(json.loads(checkout_out.read_text(encoding="utf-8")), - {"old": True}) - fallback = cwd / "server.json" - self.assertTrue(fallback.exists()) - data = json.loads(fallback.read_text(encoding="utf-8")) - self.assertEqual(len(data["models"]), 1) - - -class TranscriptWarningTests(_MainTestBase): - """Empty transcripts and a missing Whisper backend produce loud warnings.""" - - def _args(self, *extra): - return ["--wavs", 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") - inputs = self._defaults() - 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_install_warning(self): - (self.folder / "narrator.wav").write_bytes(b"x") - inputs = self._defaults() - code, out = self._run_capturing( - self._args(), inputs=inputs, - transcribe=lambda path, model_name="base": "a transcript", - whisper=None) - self.assertEqual(code, 0) - self.assertIn("Install whisper", out) - self.assertIn("faster_whisper", out) - - def test_all_transcripts_present_prints_no_end_warning(self): - (self.folder / "narrator.wav").write_bytes(b"x") - inputs = self._defaults() - 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) - - -class PromptTextReuseTests(_MainTestBase): - """Reusing an existing prompt_text and transcribing only new voices.""" - - def _args(self, *extra): - return ["--wavs", str(self.folder), "--output", str(self.output), - "--audiocpp-dir", str(self.checkout), - "--families", "higgs_audio_tts"] + list(extra) - - def _run_capturing(self, argv, inputs, transcribe): - argv = ["make_audiocpp_server_json.py"] + argv - buf = io.StringIO() - with patch.object(sys, "argv", argv), \ - patch("builtins.input", side_effect=inputs), \ - patch.object(make_server, "transcribe_reference_audio", - side_effect=transcribe), \ - patch.object(make_server, "whisper_backend_available", - return_value="faster_whisper"), \ - redirect_stdout(buf): - code = make_server.main() - return code, buf.getvalue() - - def _transcribe(self, called, text): - def transcribe(path, model_name="base"): - called.append(path) - return text - return transcribe - - def test_all_present_decline_keeps_file_and_skips_transcribe(self): - (self.folder / "narrator.wav").write_bytes(b"x") - prompt = self.folder / make_server.PROMPT_TEXT_FILENAME - prompt.write_text("narrator|An existing transcript.\n", - encoding="utf-8") - called = [] - # host, port, backend, lazy, re-transcribe(n), sync(y) - inputs = ["", "", "", "", "n", "y"] - code, out = self._run_capturing( - self._args(), inputs=inputs, - transcribe=self._transcribe(called, "Fresh.")) - self.assertEqual(code, 0) - self.assertEqual(called, []) - self.assertEqual(prompt.read_text(encoding="utf-8"), - "narrator|An existing transcript.\n") - self.assertIn("Kept existing", out) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(data["voice_dir"], str(self.folder.resolve())) - - def test_all_present_accept_retranscribes_and_overwrites(self): - (self.folder / "narrator.wav").write_bytes(b"x") - prompt = self.folder / make_server.PROMPT_TEXT_FILENAME - prompt.write_text("narrator|Old.\n", encoding="utf-8") - called = [] - # host, port, backend, lazy, re-transcribe(y), sync(y) - inputs = ["", "", "", "", "y", "y"] - code, _ = self._run_capturing( - self._args(), inputs=inputs, - transcribe=self._transcribe(called, "Fresh.")) - self.assertEqual(code, 0) - self.assertEqual(called, [str(self.folder / "narrator.wav")]) - self.assertIn("narrator|Fresh.", prompt.read_text(encoding="utf-8")) - - def test_new_voice_merges_preserving_hand_edits(self): - (self.folder / "existing.wav").write_bytes(b"x") - (self.folder / "new.wav").write_bytes(b"x") - prompt = self.folder / make_server.PROMPT_TEXT_FILENAME - prompt.write_text("existing|Hand edited transcript.\n", - encoding="utf-8") - called = [] - # host, port, backend, lazy, only-new(Enter -> y), sync(y) - inputs = ["", "", "", "", "", "y"] - code, _ = self._run_capturing( - self._args(), inputs=inputs, - transcribe=self._transcribe(called, "New transcript.")) - self.assertEqual(code, 0) - self.assertEqual(called, [str(self.folder / "new.wav")]) - text = prompt.read_text(encoding="utf-8") - self.assertIn("existing|Hand edited transcript.", text) - self.assertIn("new|New transcript.", text) - - def test_new_voice_decline_retranscribes_all(self): - (self.folder / "existing.wav").write_bytes(b"x") - (self.folder / "new.wav").write_bytes(b"x") - prompt = self.folder / make_server.PROMPT_TEXT_FILENAME - prompt.write_text("existing|Old.\n", encoding="utf-8") - called = [] - # host, port, backend, lazy, only-new(n), sync(y) - inputs = ["", "", "", "", "n", "y"] - code, _ = self._run_capturing( - self._args(), inputs=inputs, - transcribe=self._transcribe(called, "Fresh.")) - self.assertEqual(code, 0) - self.assertEqual(sorted(called), sorted([ - str(self.folder / "existing.wav"), str(self.folder / "new.wav")])) - self.assertIn("existing|Fresh.", prompt.read_text(encoding="utf-8")) - - def test_force_retranscribes_without_prompt(self): - (self.folder / "narrator.wav").write_bytes(b"x") - prompt = self.folder / make_server.PROMPT_TEXT_FILENAME - prompt.write_text("narrator|Old.\n", encoding="utf-8") - called = [] - # host, port, backend, lazy, sync(y); no re-transcribe prompt with force. - inputs = ["", "", "", "", "y"] - code, _ = self._run_capturing( - self._args("--force"), inputs=inputs, - transcribe=self._transcribe(called, "Fresh.")) - self.assertEqual(code, 0) - self.assertEqual(called, [str(self.folder / "narrator.wav")]) - self.assertIn("narrator|Fresh.", prompt.read_text(encoding="utf-8")) - - def test_empty_transcript_counts_as_missing(self): - (self.folder / "narrator.wav").write_bytes(b"x") - prompt = self.folder / make_server.PROMPT_TEXT_FILENAME - prompt.write_text("narrator|\n", encoding="utf-8") - called = [] - # Empty transcript is treated as missing -> the "only new voices" - # prompt fires (Enter -> y). - # host, port, backend, lazy, only-new(Enter), sync(y) - inputs = ["", "", "", "", "", "y"] - code, _ = self._run_capturing( - self._args(), inputs=inputs, - transcribe=self._transcribe(called, "Fresh.")) - self.assertEqual(code, 0) - self.assertEqual(called, [str(self.folder / "narrator.wav")]) - self.assertIn("narrator|Fresh.", prompt.read_text(encoding="utf-8")) - - -class WizardBackNavigationTests(unittest.TestCase): - """Esc-driven back navigation in the _wizard step state machine. - - The tui widget module is mocked so the steps can be driven directly; - Esc is simulated by having the mocked widgets return the _GO_BACK - sentinel (what the real widgets return for Esc via back_value). - """ - - def _settings(self, tmp, confirm_sequence): - checkout = _make_checkout(tmp) - manager = checkout / "tools" / "model_manager_v2.py" - manager.parent.mkdir(parents=True, exist_ok=True) - manager.write_text("#!/usr/bin/env python3\n", encoding="utf-8") - wav_dir = tmp / "wavs" - wav_dir.mkdir() - tui_mock = MagicMock() - tui_mock.menu.return_value = "cuda" - tui_mock.line_edit.side_effect = ["127.0.0.1", "8080"] - tui_mock.browse_directory.return_value = wav_dir - tui_mock.confirm.side_effect = confirm_sequence - args = argparse.Namespace( - audiocpp_dir=checkout, families="qwen3_tts", - all_packages=False, host=None, port=None, backend=None, - lazy_load=False, output=None, force=False, input_dir=None, - whisper_model="base") - with patch.object(make_server, "_load_tui", - return_value=tui_mock): - return make_server._wizard(None, args, None), tui_mock, checkout - - def test_esc_on_download_prompt_returns_to_previous_step(self): - tmp = tempfile.TemporaryDirectory() - self.addCleanup(tmp.cleanup) - # confirms: lazy(True), download Esc(_GO_BACK), download accept(True). - settings, tui_mock, _ = self._settings( - Path(tmp.name), [True, make_server._GO_BACK, True]) - self.assertIsNotNone(settings) - self.assertEqual(settings["backend"], "cuda") - downloads = [call[0][1] for call in tui_mock.confirm.call_args_list - if call[0][1] == "Automatically download the selected " - "models with model_manager_v2.py now?"] - self.assertEqual(len(downloads), 2, - "Esc on the download prompt must re-show it after " - "going back") - - def test_esc_back_to_wav_browser_rebrowses(self): - tmp = tempfile.TemporaryDirectory() - self.addCleanup(tmp.cleanup) - root = Path(tmp.name) - checkout = _make_checkout(root) - wav_dir = root / "wavs" - wav_dir.mkdir() - (wav_dir / "narrator.wav").write_bytes(b"x") - prompt = wav_dir / make_server.PROMPT_TEXT_FILENAME - prompt.write_text("narrator|already transcribed.\n", encoding="utf-8") - tui_mock = MagicMock() - tui_mock.menu.return_value = "cuda" - tui_mock.line_edit.side_effect = ["127.0.0.1", "8080"] - # 1st browse, then re-browse after Esc backs from the transcription - # plan, then browse again only if we re-reached step 4 once more. - tui_mock.browse_directory.side_effect = [wav_dir, wav_dir] - # lazy(True), transcription Esc(_GO_BACK), lazy(True again), - # transcription(True), download(True). - tui_mock.confirm.side_effect = [ - True, make_server._GO_BACK, True, True, True] - args = argparse.Namespace( - audiocpp_dir=checkout, families="qwen3_tts", - all_packages=False, host=None, port=None, backend=None, - lazy_load=False, output=None, force=False, input_dir=None, - whisper_model="base") - with patch.object(make_server, "_load_tui", return_value=tui_mock): - settings = make_server._wizard(None, args, None) - self.assertIsNotNone(settings) - self.assertEqual(tui_mock.browse_directory.call_count, 2, - "Esc on the transcription plan must re-open the " - "wav browser") - - def test_esc_on_first_step_aborts_wizard(self): - tmp = tempfile.TemporaryDirectory() - self.addCleanup(tmp.cleanup) - checkout = _make_checkout(Path(tmp.name)) - tui_mock = MagicMock() - tui_mock.browse_directory.side_effect = tui.WizardCancelled - args = argparse.Namespace( - audiocpp_dir=None, families="qwen3_tts", - all_packages=False, host=None, port=None, backend=None, - lazy_load=False, output=None, force=False, input_dir=None, - whisper_model="base") - with patch.object(make_server, "_load_tui", return_value=tui_mock), \ - self.assertRaises(tui.WizardCancelled): - make_server._wizard(None, args, None) - - def test_wav_browser_starts_in_detected_wav_dir(self): - tmp = tempfile.TemporaryDirectory() - self.addCleanup(tmp.cleanup) - root = Path(tmp.name) - checkout = _make_checkout(root) - # The only .wav directory across the checkout (and the real - # TTS_ROOT, which has none) is voices/ inside the checkout. - voices = checkout / "voices" - voices.mkdir() - (voices / "narrator.wav").write_bytes(b"x") - wav_dir = root / "wavs" - wav_dir.mkdir() - tui_mock = MagicMock() - tui_mock.menu.return_value = "cuda" - tui_mock.line_edit.side_effect = ["127.0.0.1", "8080"] - tui_mock.browse_directory.return_value = wav_dir - tui_mock.confirm.side_effect = [True, True] - args = argparse.Namespace( - audiocpp_dir=checkout, families="qwen3_tts", - all_packages=False, host=None, port=None, backend=None, - lazy_load=False, output=None, force=False, input_dir=None, - whisper_model="base") - with patch.object(make_server, "_load_tui", return_value=tui_mock): - make_server._wizard(None, args, None) - start = tui_mock.browse_directory.call_args[1].get("start") - self.assertEqual(start, voices.resolve()) - - -class ModeSelectionTests(unittest.TestCase): - """Choosing between the TUI wizard and the line prompts.""" - - def _args(self, notui=False): - return argparse.Namespace(notui=notui) - - def test_notui_flag_forces_prompt_mode(self): - # Even with a tty and an importable curses, --notui disables the TUI. - with patch.object(make_server, "_curses_importable", return_value=True), \ - patch.object(make_server.sys.stdin, "isatty", return_value=True), \ - patch.object(make_server.sys.stdout, "isatty", return_value=True): - self.assertFalse(make_server._tui_enabled(self._args(notui=True))) - - def test_non_tty_forces_prompt_mode(self): - with patch.object(make_server, "_curses_importable", return_value=True), \ - patch.object(make_server.sys.stdin, "isatty", return_value=False), \ - patch.object(make_server.sys.stdout, "isatty", return_value=True): - self.assertFalse(make_server._tui_enabled(self._args())) - - def test_tty_with_curses_uses_tui(self): - with patch.object(make_server, "_curses_importable", return_value=True), \ - patch.object(make_server.sys.stdin, "isatty", return_value=True), \ - patch.object(make_server.sys.stdout, "isatty", return_value=True): - self.assertTrue(make_server._tui_enabled(self._args())) - - def test_missing_curses_forces_prompt_mode(self): - with patch.object(make_server, "_curses_importable", return_value=False), \ - patch.object(make_server.sys.stdin, "isatty", return_value=True), \ - patch.object(make_server.sys.stdout, "isatty", return_value=True): - self.assertFalse(make_server._tui_enabled(self._args())) - - def test_curses_is_importable_on_this_platform(self): - # The TUI widget module imports without curses at module load time, - # but the wizard still needs the real curses package to run. - self.assertTrue(make_server._curses_importable()) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_tui.py b/tests/test_tui.py index 94fb638..ba6f99f 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -1,4 +1,4 @@ -"""Tests for the DOS-style curses TUI widgets in tools/tui.py. +"""Tests for the DOS-style curses TUI widgets in tui.py. The widget module imports curses lazily, so these tests swap the curses module for a small fake (patched into sys.modules) and drive @@ -14,7 +14,7 @@ import unittest from pathlib import Path from unittest.mock import patch -from tools import tui +import tui class FakeCurses: @@ -73,6 +73,9 @@ class FakeCurses: def curs_set(self, visibility): pass + def endwin(self): + pass + class FakeScreen: """Recording curses window; getch() replays scripted keys.""" @@ -105,6 +108,9 @@ class FakeScreen: def hline(self, y, x, ch, n, attr=0): pass + def redrawwin(self): + pass + def getch(self): if not self.keys: raise AssertionError("the script ran out of keys") @@ -469,5 +475,32 @@ class CheckboxTreeTests(TuiTestCase): back_value=marker) +class SuspendTests(TuiTestCase): + """tui.suspend leaves curses, runs code, then repaints.""" + + def test_suspend_runs_block_and_restores(self): + ran = [] + with tui.suspend(self.screen): + ran.append("inside") + self.assertEqual(ran, ["inside"]) + + def test_suspend_always_restores_on_exception(self): + class Boom(Exception): + pass + with self.assertRaises(Boom): + with tui.suspend(self.screen): + raise Boom() + + +class FlashTests(TuiTestCase): + """tui.flash shows a notice until any key is pressed.""" + + def test_notice_dismissed_by_any_key(self): + screen = FakeScreen(keys=[10]) + # Should return (None) after consuming one key; not raise. + tui.flash(screen, "a notice", kind="warn") + self.assertEqual(screen.keys, []) + + if __name__ == "__main__": unittest.main() |