diff options
Diffstat (limited to 'app/tests/test_backends_audiocpp.py')
| -rw-r--r-- | app/tests/test_backends_audiocpp.py | 300 |
1 files changed, 222 insertions, 78 deletions
diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py index 59bd039..8cd7981 100644 --- a/app/tests/test_backends_audiocpp.py +++ b/app/tests/test_backends_audiocpp.py @@ -1,6 +1,5 @@ """Tests for the audio.cpp backend setup module (backends/audiocpp.py).""" -import contextlib import io import json import sys @@ -662,41 +661,105 @@ class InstallModelsTests(unittest.TestCase): 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.models._decide_download(self.checkout, [], confirm)) - confirm.assert_not_called() + def _entry_paths(self): + return [{"path": "models/higgs"}, {"path": "models/qwen"}] - def test_decide_download_asks_when_manager_present(self): - confirm = MagicMock(return_value=True) - self.assertTrue(make_server.models._decide_download(self.checkout, [], confirm)) - confirm.assert_called_once() + def test_installed_model_prints_no_command_for_it(self): + # Mixed selection: qwen is on disk, higgs is not. The print path + # reports the installed one without a python command, explains + # that setup downloads automatically, then lists the rest. + (self.checkout / "models" / "qwen").mkdir(parents=True) + (self.checkout / "models" / "qwen" / "f.bin").write_bytes(b"x") + buf = io.StringIO() + with redirect_stdout(buf), \ + patch.object(common, + "run_console_subprocess") as run: + make_server.models._install_models( + self.checkout, + [("Higgs Audio v3 TTS 4B", "higgs_audio_tts_4b_q8_0"), + ("Qwen3-TTS", "qwen3_tts_1_7b_base_q8_0")], + download=False, + model_entries=self._entry_paths()) + out = buf.getvalue() + self.assertIn("[OK] Qwen3-TTS is already installed.", out) + self.assertIn("downloaded automatically", out) + self.assertIn("python {} install higgs_audio_tts_4b_q8_0".format( + self.manager), out) + self.assertNotIn("install qwen3_tts_1_7b_base_q8_0", out) + run.assert_not_called() - def test_decide_download_defaults_to_yes(self): - confirm = MagicMock(return_value=True) - make_server.models._decide_download(self.checkout, [], confirm) - self.assertIs(confirm.call_args[0][1], True) + def test_all_models_present_prints_no_commands(self): + for name in ("higgs", "qwen"): + target = self.checkout / "models" / name + target.mkdir(parents=True) + (target / "f.bin").write_bytes(b"x") + buf = io.StringIO() + with redirect_stdout(buf), \ + patch.object(common, + "run_console_subprocess") as run: + rc = make_server.models._install_models( + self.checkout, + [("Higgs Audio v3 TTS 4B", "higgs_audio_tts_4b_q8_0"), + ("Qwen3-TTS", "qwen3_tts_1_7b_base_q8_0")], + download=False, + model_entries=self._entry_paths()) + out = buf.getvalue() + self.assertEqual(rc, 0) + self.assertIn("All selected models are already installed.", out) + self.assertNotIn("model_manager_v2.py install", out) + run.assert_not_called() - def test_decide_download_skips_prompt_when_all_models_present(self): + def test_download_skips_installed_models(self): + (self.checkout / "models" / "qwen").mkdir(parents=True) + (self.checkout / "models" / "qwen" / "f.bin").write_bytes(b"x") + with redirect_stdout(io.StringIO()), \ + patch.object(common, + "run_console_subprocess", + return_value=0) as run: + make_server.models._install_models( + self.checkout, + [("Higgs Audio v3 TTS 4B", "higgs_audio_tts_4b_q8_0"), + ("Qwen3-TTS", "qwen3_tts_1_7b_base_q8_0")], + download=True, + model_entries=self._entry_paths()) + self.assertEqual(run.call_count, 1) + self.assertEqual(run.call_args[0][0][3], "higgs_audio_tts_4b_q8_0") + + def test_entries_without_guidance_do_not_filter(self): + # A length mismatch means no filtering is possible: every model + # is treated as missing (the pre-change behavior). + buf = io.StringIO() + with redirect_stdout(buf): + make_server.models._install_models( + self.checkout, self.guidance, download=False, + model_entries=[{"path": "models/qwen"}]) + out = buf.getvalue() + self.assertIn("higgs_audio_tts_4b_q8_0", out) + self.assertIn("qwen3_tts_1_7b_base_q8_0", out) + + def test_download_applicable_false_without_manager(self): + self.manager.unlink() + self.assertFalse( + make_server.models.download_applicable(self.checkout, [])) + + def test_download_applicable_when_manager_present(self): + self.assertTrue( + make_server.models.download_applicable(self.checkout, [])) + + def test_download_applicable_skipped_when_all_models_present(self): target = self.checkout / "models" / "higgs" target.mkdir(parents=True) (target / "model.gguf").write_bytes(b"x") - confirm = MagicMock() - self.assertFalse(make_server.models._decide_download( - self.checkout, [{"path": "models/higgs"}], confirm)) - confirm.assert_not_called() + self.assertFalse(make_server.models.download_applicable( + self.checkout, [{"path": "models/higgs"}])) - def test_decide_download_prompts_when_a_model_is_missing(self): + def test_download_applicable_when_a_model_is_missing(self): target = self.checkout / "models" / "higgs" target.mkdir(parents=True) (target / "model.gguf").write_bytes(b"x") - confirm = MagicMock(return_value=True) - self.assertTrue(make_server.models._decide_download( + self.assertTrue(make_server.models.download_applicable( self.checkout, - [{"path": "models/higgs"}, {"path": "models/absent"}], - confirm)) - confirm.assert_called_once() + [{"path": "models/higgs"}, {"path": "models/absent"}])) def test_all_models_present_true_when_all_paths_hold_files(self): target = self.checkout / "models" / "higgs" @@ -729,6 +792,33 @@ class InstallModelsTests(unittest.TestCase): self.checkout, [{"path": "models/higgs"}])) +class TranscriptionChoicesTests(unittest.TestCase): + """_transcription_choices: the renamed voice-transcripts options.""" + + def test_fresh_directory_offers_the_renamed_all(self): + choices, default = make_server.wizard._transcription_choices( + [], {}, prompt_exists=False) + self.assertEqual(default, "all") + self.assertEqual(choices, [("Re-transcribe all", "all")]) + + def test_existing_transcripts_offer_new_only_and_all(self): + wavs = [Path("/x/narrator.wav"), Path("/x/new.wav")] + choices, default = make_server.wizard._transcription_choices( + wavs, {"narrator": "old transcript"}, prompt_exists=True) + self.assertEqual(default, "missing") + self.assertEqual([label for label, _mode in choices], + ["Only transcribe new voices", "Re-transcribe all"]) + + def test_complete_transcripts_offer_keep_and_all(self): + wavs = [Path("/x/narrator.wav")] + choices, default = make_server.wizard._transcription_choices( + wavs, {"narrator": "old transcript"}, prompt_exists=True) + self.assertEqual(default, "keep") + self.assertEqual([label for label, _mode in choices], + ["Keep the existing transcripts", + "Re-transcribe all"]) + + class TranscribeWavDirTests(unittest.TestCase): def setUp(self): self._td = tempfile.TemporaryDirectory() @@ -1150,27 +1240,28 @@ class NonInteractiveMainTests(unittest.TestCase): ["Higgs-Audio-v3-TTS-4B-GGUF"]) self.assertNotIn("voice_dir", data) - def test_port_sync_accepted_updates_config(self): + def test_port_comes_from_config_and_leaves_config_alone(self): + # Ports are not a wizard question anymore: server.json always + # records the port in AUDIOCPP_API_URL (edited in Settings), and + # app/converter/config.py itself is never rewritten by setup. 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", + self._args("--families", "higgs_audio_tts", "--no-sync-model-ids")) self.assertEqual(exit_code, 0) - self.assertIn('"http://127.0.0.1:8080"', + self.assertIn('"http://127.0.0.1:9999"', self.fake_config.read_text(encoding="utf-8")) data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(data["port"], 8080) + self.assertEqual(data["port"], 9999) - 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_host_port_sync_flags_removed(self): + # No bind-host or port questions anywhere: 127.0.0.1 is fixed and + # the port follows Settings, so their flags are gone. + parser = make_server.wizard.build_parser() + for flag in ("--host", "--port", "--no-sync-port"): + with self.assertRaises(SystemExit): + parser.parse_args([flag, "x"]) def test_model_id_sync_accepted_updates_config(self): self.fake_config.write_text(FAKE_CONFIG_WITH_MODEL_IDS, @@ -1767,9 +1858,9 @@ class WizardNavigationTests(unittest.TestCase): def test_modify_flow_offers_build_when_not_built(self): # A server.json recording "vulkan" exists, but nothing is built: the - # wizard must still reach the backend menu (pre-selecting vulkan) and - # offer the build — instead of silently skipping it because the - # existing server.json already records a backend. + # combined config form must still ask the backend (pre-selecting + # vulkan) and offer the build — instead of silently skipping it + # because the existing server.json already records a backend. checkout = self._checkout() (checkout / "server.json").write_text( json.dumps({"models": [], "backend": "vulkan"}), @@ -1777,79 +1868,132 @@ class WizardNavigationTests(unittest.TestCase): catalog = make_server.catalog.load_model_catalog(checkout) supertonic = next(i for i, entry in enumerate(catalog) if entry["family"] == "supertonic") - confirm_questions = [] def fake_tree(*args, **kwargs): return [(supertonic, "Supertonic-GGUF")] - def fake_line_edit(stdscr, title, default, **kwargs): - if title == "Bind host": - return "127.0.0.1" - if title == "Port": - return "8080" - return default + captured = {} - def fake_menu(stdscr, title, options, **kwargs): - return "vulkan" - - def fake_confirm(stdscr, question, **kwargs): - confirm_questions.append(question) - return False # decline the build + def fake_form(stdscr, title, fields, **kwargs): + captured["title"] = title + captured["keys"] = [f["key"] for f in fields] + by_key = {f["key"]: f for f in fields} + return {f["key"]: f["value"] for f in fields} | { + "backend": by_key["backend"]["value"], + "build": False, # decline the build + } with patch.object(make_server.build, "find_local_checkout", return_value=checkout), \ patch.object(tui, "checkbox_tree", side_effect=fake_tree), \ - patch.object(tui, "line_edit", side_effect=fake_line_edit), \ - patch.object(tui, "menu", side_effect=fake_menu), \ - patch.object(tui, "confirm", side_effect=fake_confirm): + patch.object(tui, "form", side_effect=fake_form): settings = make_server.wizard._wizard(None, self._args(), make_server.wizard.build_parser()) self.assertIsNotNone(settings) self.assertEqual(settings["backend"], "vulkan") self.assertFalse(settings["build"]) - # The build offer was shown (and declined); the old modify flow - # skipped it entirely. - self.assertTrue(any("not built for vulkan" in q - for q in confirm_questions)) - - def test_bind_host_esc_returns_to_families_tree(self): - # Esc on "Bind host" must fall back to the model-family tree, then - # re-selecting proceeds through the rest of the wizard. + # The config screen is one combined form (not one question per + # screen) that includes both the backend pick and the build offer. + self.assertEqual(captured["title"], "Configure audio.cpp") + self.assertIn("backend", captured["keys"]) + self.assertIn("build", captured["keys"]) + + def test_esc_on_config_form_returns_to_families_tree(self): + # Esc on the combined config form must fall back to the model-family + # tree; re-selecting then proceeds through the rest of the wizard. checkout = self._checkout() catalog = make_server.catalog.load_model_catalog(checkout) supertonic = next(i for i, entry in enumerate(catalog) if entry["family"] == "supertonic") tree_calls = [] - hosts = iter([make_server.wizard._GO_BACK, "127.0.0.1"]) + form_calls = [] def fake_tree(*args, **kwargs): tree_calls.append(1) return [(supertonic, "Supertonic-GGUF")] - def fake_line_edit(stdscr, title, default, **kwargs): - if title == "Bind host": - return next(hosts) - if title == "Port": - return "8080" - return default + def fake_form(stdscr, title, fields, **kwargs): + form_calls.append(title) + if len(form_calls) == 1: + return tui.Wizard.BACK # Esc on the config form + return {f["key"]: f["value"] for f in fields} with patch.object(make_server.build, "find_local_checkout", return_value=checkout), \ patch.object(tui, "checkbox_tree", side_effect=fake_tree), \ - patch.object(tui, "line_edit", - side_effect=fake_line_edit), \ - patch.object(tui, "menu", return_value="cuda"), \ - patch.object(tui, "confirm", return_value=True): + patch.object(tui, "form", + side_effect=fake_form): settings = make_server.wizard._wizard(None, self._args(), make_server.wizard.build_parser()) self.assertIsNotNone(settings) - # The tree was re-shown after the host screen's Esc. + # The tree was re-shown after the form's Esc. self.assertEqual(len(tree_calls), 2) - self.assertEqual(settings["host"], "127.0.0.1") + self.assertEqual(form_calls, + ["Configure audio.cpp", "Configure audio.cpp"]) self.assertEqual([m["id"] for m in settings["model_entries"]], ["Supertonic-GGUF"]) + def test_combined_form_defaults_and_fixed_host_port(self): + # One screen collects everything: the form value defaults produce a + # complete settings dict whose host/port never came from questions. + checkout = self._checkout() + catalog = make_server.catalog.load_model_catalog(checkout) + supertonic = next(i for i, entry in enumerate(catalog) + if entry["family"] == "supertonic") + + def fake_tree(*args, **kwargs): + return [(supertonic, "Supertonic-GGUF")] + + def fake_form(stdscr, title, fields, **kwargs): + return {f["key"]: f["value"] for f in fields} + + with patch.object(make_server.build, "find_local_checkout", + return_value=checkout), \ + patch.object(tui, "checkbox_tree", side_effect=fake_tree), \ + patch.object(tui, "form", side_effect=fake_form): + settings = make_server.wizard._wizard(None, self._args(), + make_server.wizard.build_parser()) + self.assertIsNotNone(settings) + self.assertEqual(settings["host"], "127.0.0.1") + self.assertEqual(settings["port"], + make_server.configsync.config_port()) + self.assertEqual(settings["backend"], "cuda") # default choice + self.assertTrue(settings["build"]) # not built yet → offered (default Yes) + self.assertFalse(settings["download"]) # no manager script here + self.assertTrue(settings["sync_model_ids"]) + + def test_build_offer_hidden_when_backend_already_built(self): + # A checkout with a built binary for the chosen backend must not + # show (or honor) a build offer. + checkout = self._checkout() + catalog = make_server.catalog.load_model_catalog(checkout) + supertonic = next(i for i, entry in enumerate(catalog) + if entry["family"] == "supertonic") + binary = checkout / "build" / "linux-cuda-release" / "bin" \ + / "audiocpp_server" + binary.parent.mkdir(parents=True) + binary.write_bytes(b"x") + + def fake_tree(*args, **kwargs): + return [(supertonic, "Supertonic-GGUF")] + + def fake_form(stdscr, title, fields, **kwargs): + keys = [f["key"] for f in fields] + self.assertNotIn("build", keys) + self.assertNotIn("backend", keys) + return {f["key"]: f["value"] for f in fields} + + with patch.object(make_server.build, "find_local_checkout", + return_value=checkout), \ + patch.object(tui, "checkbox_tree", side_effect=fake_tree), \ + patch.object(tui, "form", side_effect=fake_form): + settings = make_server.wizard._wizard(None, self._args(), + make_server.wizard.build_parser()) + self.assertIsNotNone(settings) + self.assertFalse(settings["build"]) + self.assertEqual(settings["backend"], "cuda") + class UninstallTests(unittest.TestCase): """uninstall: stop the server and remove the checkout.""" |