diff options
Diffstat (limited to 'app/tests')
| -rw-r--r-- | app/tests/test_backends_audiocpp.py | 268 | ||||
| -rw-r--r-- | app/tests/test_hub.py | 109 |
2 files changed, 370 insertions, 7 deletions
diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py index ada7954..08de777 100644 --- a/app/tests/test_backends_audiocpp.py +++ b/app/tests/test_backends_audiocpp.py @@ -834,6 +834,203 @@ class InstallModelsTests(unittest.TestCase): self.assertFalse(make_server.models._all_models_present( self.checkout, [{"path": "models/higgs"}])) + def _supporting_manager(self): + self.manager.write_text( + '#!/usr/bin/env python3\n' + 'parser.add_argument("--specs-dir", default="")\n' + 'parser.add_argument("--cancel-file", default="")\n', + encoding="utf-8") + + def _write_spec(self, name: str, packages: list) -> Path: + specs = self.checkout / "model_specs" + specs.mkdir(parents=True, exist_ok=True) + path = specs / name + path.write_text(json.dumps({"family": name[:-5], "packages": packages}), + encoding="utf-8") + return path + + def test_dot_strip_prefix_spec_installs_from_sanitized_copy(self): + self._supporting_manager() + original = self._write_spec("vietneu_tts.json", [{ + "id": "vietneu_tts_v3_turbo_q8_0", + "files": ["model.gguf"], + "strip_prefix": ".", + }]) + self._write_spec("other.json", [{ + "id": "other_q8_0", + "files": ["Other-GGUF/model.gguf"], + "strip_prefix": "Other-GGUF", + }]) + seen: dict = {} + + def record(argv, **kwargs): + specs_dir = Path(argv[3]) + seen["fixed"] = json.loads( + (specs_dir / "vietneu_tts.json").read_text(encoding="utf-8")) + seen["other"] = json.loads( + (specs_dir / "other.json").read_text(encoding="utf-8")) + return 0 + + buf = io.StringIO() + with redirect_stdout(buf), \ + patch.object(common, "run_console_subprocess", + side_effect=record) as run: + rc = make_server.models._install_models( + self.checkout, + [("VieNeu-TTS v3 Turbo GGUF", "vietneu_tts_v3_turbo_q8_0")], + download=True) + self.assertEqual(rc, 0) + argv = run.call_args[0][0] + self.assertEqual(argv[:2], [sys.executable, str(self.manager)]) + self.assertEqual(argv[2], "--specs-dir") + self.assertEqual(argv[4:], ["install", "vietneu_tts_v3_turbo_q8_0"]) + self.assertEqual(seen["fixed"]["packages"][0]["strip_prefix"], "") + self.assertEqual(seen["other"]["packages"][0]["strip_prefix"], + "Other-GGUF") + self.assertIn("sanitized copy", buf.getvalue()) + # The checkout's own specs are untouched and the temp copy is gone. + self.assertEqual(json.loads( + original.read_text(encoding="utf-8"))["packages"][0] + ["strip_prefix"], ".") + self.assertFalse(Path(argv[3]).exists()) + + def test_healthy_specs_do_not_add_specs_dir(self): + self._supporting_manager() + self._write_spec("ok.json", [{ + "id": "ok_q8_0", + "files": ["Ok-GGUF/model.gguf"], + "strip_prefix": "Ok-GGUF", + }]) + with patch.object(common, "run_console_subprocess", + return_value=0) as run: + make_server.models._install_models( + self.checkout, [("Ok", "ok_q8_0")], download=True) + self.assertEqual( + run.call_args[0][0], + [sys.executable, str(self.manager), "install", "ok_q8_0"]) + + def test_unknown_prefix_mismatch_left_for_warning_path(self): + # A real-directory prefix that matches no files cannot be repaired + # confidently; the install is left to fail with the manager's own + # error so the remaining downloads continue (warn-and-continue). + self._supporting_manager() + self._write_spec("broken.json", [{ + "id": "broken_q8_0", + "files": ["model.gguf"], + "strip_prefix": "Some-Dir", + }]) + with patch.object(common, "run_console_subprocess", + return_value=0) as run: + make_server.models._install_models( + self.checkout, [("Broken", "broken_q8_0")], download=True) + self.assertEqual( + run.call_args[0][0], + [sys.executable, str(self.manager), "install", "broken_q8_0"]) + + def test_specs_dir_unsupported_manager_leaves_argv_unchanged(self): + self._write_spec("vietneu_tts.json", [{ + "id": "vietneu_tts_v3_turbo_q8_0", + "files": ["model.gguf"], + "strip_prefix": ".", + }]) + with patch.object(common, "run_console_subprocess", + return_value=0) as run: + make_server.models._install_models( + self.checkout, + [("VieNeu-TTS v3 Turbo GGUF", "vietneu_tts_v3_turbo_q8_0")], + download=True) + self.assertEqual( + run.call_args[0][0], + [sys.executable, str(self.manager), "install", + "vietneu_tts_v3_turbo_q8_0"]) + + +class SanitizeModelSpecTests(unittest.TestCase): + """The dot strip_prefix repair and the --specs-dir staging copy.""" + + def test_dot_prefix_dropped_when_file_is_bare(self): + spec = {"packages": [{"files": ["model.gguf"], "strip_prefix": "."}]} + self.assertTrue(make_server.models._sanitize_model_spec(spec)) + self.assertEqual(spec["packages"][0]["strip_prefix"], "") + + def test_slash_dot_prefix_normalized_like_dot(self): + spec = {"packages": [{"files": ["model.gguf"], "strip_prefix": "./"}]} + self.assertTrue(make_server.models._sanitize_model_spec(spec)) + self.assertEqual(spec["packages"][0]["strip_prefix"], "") + + def test_dot_prefix_kept_when_files_carry_it(self): + spec = {"packages": [{"files": ["./model.gguf"], + "strip_prefix": "."}]} + self.assertFalse(make_server.models._sanitize_model_spec(spec)) + self.assertEqual(spec["packages"][0]["strip_prefix"], ".") + + def test_real_directory_prefix_untouched(self): + spec = {"packages": [{"files": ["model.gguf"], + "strip_prefix": "Kroko-ASR-GGUF"}]} + self.assertFalse(make_server.models._sanitize_model_spec(spec)) + self.assertEqual(spec["packages"][0]["strip_prefix"], + "Kroko-ASR-GGUF") + + def test_valid_prefix_untouched(self): + spec = {"packages": [{"files": ["Kroko-ASR-GGUF/model.gguf"], + "strip_prefix": "Kroko-ASR-GGUF"}]} + self.assertFalse(make_server.models._sanitize_model_spec(spec)) + + def test_missing_or_empty_files_untouched(self): + spec = {"packages": [{"strip_prefix": "."}, + {"files": [], "strip_prefix": "."}, + {"files": "model.gguf", "strip_prefix": "."}]} + self.assertFalse(make_server.models._sanitize_model_spec(spec)) + + def test_only_broken_packages_repaired(self): + spec = {"packages": [ + {"files": ["model.gguf"], "strip_prefix": "."}, + {"files": ["./model.gguf"], "strip_prefix": "."}, + ]} + self.assertTrue(make_server.models._sanitize_model_spec(spec)) + self.assertEqual([p["strip_prefix"] for p in spec["packages"]], + ["", "."]) + + 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 test_prepare_specs_dir_none_without_specs(self): + self.assertIsNone( + make_server.models._prepare_specs_dir(self.checkout)) + + def test_prepare_specs_dir_none_when_healthy(self): + specs = self.checkout / "model_specs" + specs.mkdir() + (specs / "ok.json").write_text(json.dumps( + {"packages": [{"files": ["Ok-GGUF/m.gguf"], + "strip_prefix": "Ok-GGUF"}]}), encoding="utf-8") + self.assertIsNone( + make_server.models._prepare_specs_dir(self.checkout)) + + def test_prepare_specs_dir_writes_all_specs_and_repairs(self): + specs = self.checkout / "model_specs" + specs.mkdir() + (specs / "broken.json").write_text(json.dumps( + {"packages": [{"files": ["model.gguf"], + "strip_prefix": "."}]}), encoding="utf-8") + (specs / "plain.json").write_text("not json", encoding="utf-8") + staging = make_server.models._prepare_specs_dir(self.checkout) + try: + self.assertIsNotNone(staging) + repaired = json.loads( + (staging / "broken.json").read_text(encoding="utf-8")) + self.assertEqual(repaired["packages"][0]["strip_prefix"], "") + self.assertEqual( + (staging / "plain.json").read_text(encoding="utf-8"), + "not json") + finally: + shutil.rmtree(staging, ignore_errors=True) + class ConfigFormTranscriptionToggleTests(unittest.TestCase): """The combined form's Voice transcripts row: one fixed two-way toggle. @@ -3467,10 +3664,18 @@ class PrebuiltUpdateRoutingTests(unittest.TestCase): marker.write_text(json.dumps({"tag": tag, "asset": "x"}), encoding="utf-8") + def _marker(self, checkout: Path) -> Path: + return make_server.prebuilt.marker_path(checkout, "cpu", + platform="darwin") + def _run(self, checkout: Path, *, marker: Optional[dict], - release: Optional[dict], tag: Optional[str] = None): + release: Optional[dict], tag: Optional[str] = None, + install_rc: int = 0, build_rc: int = 0): # TAG is what the quota-free redirect resolution reports; RELEASE # is the API fallback (only consulted when the redirect fails). + # INSTALL_RC is what the prebuilt re-download reports: 0 (or a + # cancellation) returns directly, a failure falls back to the + # source-build route, whose BUILD_RC decides the final exit code. with patch("sys.platform", "darwin"), \ patch.object(make_server.build, "find_local_checkout", return_value=checkout), \ @@ -3485,16 +3690,18 @@ class PrebuiltUpdateRoutingTests(unittest.TestCase): patch.object(make_server.prebuilt, "fetch_latest_release", return_value=release) as mk_fetch, \ patch.object(make_server.prebuilt, "install_prebuilt", - return_value=0) as mk_install, \ + return_value=install_rc) as mk_install, \ patch.object(common, "git_update", - return_value=0) as mk_git: + return_value=0) as mk_git, \ + patch.object(make_server.build, "build_audiocpp", + return_value=build_rc) as mk_build: rc = make_server.build.update() - return rc, mk_install, mk_git, mk_tag, mk_fetch + return rc, mk_install, mk_git, mk_tag, mk_fetch, mk_build def test_newer_release_triggers_a_redownload(self): checkout = self._checkout() self._mark_prebuilt(checkout, "v0.6.0") - rc, mk_install, mk_git, _mk_tag, mk_fetch = self._run( + rc, mk_install, mk_git, _mk_tag, mk_fetch, mk_build = self._run( checkout, marker={"tag": "v0.6.0", "asset": "x"}, release=_release([], tag="v0.7.0"), tag="v0.7.0") self.assertEqual(rc, 0) @@ -3502,22 +3709,24 @@ class PrebuiltUpdateRoutingTests(unittest.TestCase): mk_install.assert_called_once_with(checkout, "cpu", emit=None, cancel=None) mk_fetch.assert_not_called() + mk_build.assert_not_called() def test_current_release_is_a_noop(self): checkout = self._checkout() - rc, mk_install, mk_git, _mk_tag, mk_fetch = self._run( + rc, mk_install, mk_git, _mk_tag, mk_fetch, mk_build = self._run( checkout, marker={"tag": "v0.7.0", "asset": "x"}, release=None, tag="v0.7.0") self.assertEqual(rc, 0) mk_install.assert_not_called() mk_git.assert_not_called() + mk_build.assert_not_called() # The redirect already answered: the API (rate-limited easily) # must not be touched for an "already current" check. mk_fetch.assert_not_called() def test_unreachable_github_keeps_the_install(self): checkout = self._checkout() - rc, mk_install, mk_git, mk_tag, mk_fetch = self._run( + rc, mk_install, mk_git, mk_tag, mk_fetch, mk_build = self._run( checkout, marker={"tag": "v0.7.0", "asset": "x"}, release=None, tag=None) self.assertEqual(rc, 0) @@ -3525,6 +3734,51 @@ class PrebuiltUpdateRoutingTests(unittest.TestCase): mk_fetch.assert_called_once() # the API fallback tried too mk_install.assert_not_called() mk_git.assert_not_called() + mk_build.assert_not_called() + + def test_failed_redownload_falls_back_to_a_source_build(self): + checkout = self._checkout() + self._mark_prebuilt(checkout, "v0.6.0") + rc, mk_install, mk_git, _mk_tag, _mk_fetch, mk_build = self._run( + checkout, marker={"tag": "v0.6.0", "asset": "x"}, + release=_release([], tag="v0.7.0"), tag="v0.7.0", install_rc=1) + self.assertEqual(rc, 0) + mk_install.assert_called_once_with(checkout, "cpu", emit=None, + cancel=None) + mk_git.assert_called_once() + mk_build.assert_called_once_with(checkout, "cpu", emit=None, + cancel=None) + # The fallback source build replaces the prebuilt install: the + # marker must go, or the next update re-downloads over the + # freshly built binary. + self.assertFalse(self._marker(checkout).exists()) + + def test_failed_redownload_keeps_the_marker_when_the_build_fails(self): + # A failed fallback leaves the previous prebuilt binary in place + # (install_prebuilt downloads before touching it), so the marker + # stays truthful and the next update retries the re-download. + checkout = self._checkout() + self._mark_prebuilt(checkout, "v0.6.0") + rc, _mk_install, mk_git, _mk_tag, _mk_fetch, mk_build = self._run( + checkout, marker={"tag": "v0.6.0", "asset": "x"}, + release=_release([], tag="v0.7.0"), tag="v0.7.0", install_rc=1, + build_rc=1) + self.assertEqual(rc, 1) + mk_git.assert_called_once() + mk_build.assert_called_once() + self.assertTrue(self._marker(checkout).exists()) + + def test_cancelled_redownload_does_not_fall_back(self): + checkout = self._checkout() + self._mark_prebuilt(checkout, "v0.6.0") + rc, _mk_install, mk_git, _mk_tag, _mk_fetch, mk_build = self._run( + checkout, marker={"tag": "v0.6.0", "asset": "x"}, + release=_release([], tag="v0.7.0"), tag="v0.7.0", + install_rc=130) + self.assertEqual(rc, 130) + mk_git.assert_not_called() + mk_build.assert_not_called() + self.assertTrue(self._marker(checkout).exists()) def test_source_builds_still_route_through_git(self): checkout = self._checkout() diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py index 39b8c5a..de7e4ea 100644 --- a/app/tests/test_hub.py +++ b/app/tests/test_hub.py @@ -1031,6 +1031,115 @@ class ConvertFlowTests(unittest.TestCase): instr = self._field("instructions") self.assertTrue(instr["visible"](fields)) + def test_audiocpp_model_switch_keeps_the_picked_voice(self): + # Switching models whose voice list is unchanged (two clone + # entries sharing one server's voices) keeps the picked voice + # instead of snapping back to the list's first entry. + self._patch_remote( + [{"id": "alpha", "family": "higgs_audio_tts", "task": "tts"}, + {"id": "beta", "family": "qwen3_tts", "task": "tts"}], + voices=["narrator", "second"]) + self._answer_form(backend="audiocpp-remote", model_id="alpha", + audiocpp_voice="second", instructions="") + self._convert(None, [self._remote("audiocpp", "audio.cpp")]) + fields = self.tui.forms_seen[0][1] + model_field = self._field("model_id") + voice_field = self._field("audiocpp_voice") + # The form opens on the list's first voice; the user picks another. + self.assertEqual(voice_field["value"], "narrator") + voice_field["value"] = "second" + model_field["value"] = "beta" + model_field["on_change"](fields) + self.assertEqual(voice_field["value"], "second") + model_field["value"] = "alpha" + model_field["on_change"](fields) + self.assertEqual(voice_field["value"], "second") + + def test_audiocpp_local_model_switch_keeps_the_picked_voice(self): + # The managed entry's voice list is shared by every model in + # server.json, so switching models keeps the picked voice. + with tempfile.TemporaryDirectory() as td: + root = Path(td) + (root / "server.json").write_text(json.dumps({ + "models": [{"id": "qwen-1_7b", "family": "qwen3_tts", + "task": "tts"}, + {"id": "qwen-0_6b", "family": "qwen3_tts", + "task": "tts"}], + "voice_dir": str(root), + }), encoding="utf-8") + (root / "Narrator.wav").write_bytes(b"x") + (root / "Second.wav").write_bytes(b"x") + with patch.object(hub.audiocpp_backend, "find_local_checkout", + return_value=root): + self._answer_form(backend="audiocpp", model_id="qwen-1_7b", + audiocpp_voice="Second", instructions="") + self._convert(None, + [self._ready("audiocpp", "audio.cpp")]) + fields = self.tui.forms_seen[0][1] + model_field = self._field("model_id") + voice_field = self._field("audiocpp_voice") + self.assertEqual(voice_field["value"], "Narrator") + voice_field["value"] = "Second" + model_field["value"] = "qwen-0_6b" + model_field["on_change"](fields) + self.assertEqual(voice_field["value"], "Second") + + def test_audiocpp_model_switch_resets_when_the_pick_is_gone(self): + # A remote server may host different voices per model: switching + # to a model whose list no longer offers the pick falls back to + # that model's first voice (and re-points again on the way back). + models = patch.object( + hub.audiocpp_backend, "fetch_server_models", + lambda url: [{"id": "alpha", "family": "higgs_audio_tts", + "task": "tts"}, + {"id": "beta", "family": "qwen3_tts", + "task": "tts"}]) + voices = patch.object( + hub.audiocpp_backend, "fetch_server_voices", + lambda url, model_id: {"alpha": ["narrator", "second"], + "beta": ["other"]}[model_id]) + with models, voices: + self._answer_form(backend="audiocpp-remote", model_id="alpha", + audiocpp_voice="second", instructions="") + self._convert(None, [self._remote("audiocpp", "audio.cpp")]) + fields = self.tui.forms_seen[0][1] + model_field = self._field("model_id") + voice_field = self._field("audiocpp_voice") + model_field["value"] = "beta" + model_field["on_change"](fields) + self.assertEqual(voice_field["value"], "other") + model_field["value"] = "alpha" + model_field["on_change"](fields) + self.assertEqual(voice_field["value"], "narrator") + + def test_audiocpp_model_switch_resets_across_capabilities(self): + # Keep-the-pick only applies within one voice list: a clone pick + # never survives a move to a built-in-speaker entry (and vice + # versa), and a design entry clears the voice again. + self._patch_remote( + [{"id": "clone", "family": "higgs_audio_tts", "task": "tts"}, + {"id": "Qwen3-TTS-CustomVoice-GGUF", "family": "qwen3_tts", + "task": "tts"}, + {"id": "design", "family": "qwen3_tts", "task": "vdes"}], + voices=["narrator", "second"]) + self._answer_form(backend="audiocpp-remote", model_id="clone", + audiocpp_voice="second", instructions="") + self._convert(None, [self._remote("audiocpp", "audio.cpp")]) + fields = self.tui.forms_seen[0][1] + model_field = self._field("model_id") + voice_field = self._field("audiocpp_voice") + voice_field["value"] = "second" + model_field["value"] = "Qwen3-TTS-CustomVoice-GGUF" + model_field["on_change"](fields) + self.assertEqual(voice_field["value"], hub.QWEN3_TTS_SPEAKERS[0]) + voice_field["value"] = "Ryan" + model_field["value"] = "clone" + model_field["on_change"](fields) + self.assertEqual(voice_field["value"], "narrator") + model_field["value"] = "design" + model_field["on_change"](fields) + self.assertIsNone(voice_field["value"]) + def test_audiocpp_remote_without_voices_refuses_generate_with_hint(self): # A clone-capable entry (e.g. Qwen Base) whose server lists no # voices at all: the Voice picker stays visible but is empty — |