diff options
Diffstat (limited to 'app/tests/test_backends.py')
| -rw-r--r-- | app/tests/test_backends.py | 405 |
1 files changed, 405 insertions, 0 deletions
diff --git a/app/tests/test_backends.py b/app/tests/test_backends.py index 5cf5633..1e1a3f9 100644 --- a/app/tests/test_backends.py +++ b/app/tests/test_backends.py @@ -15,6 +15,7 @@ from backends import ( get, invalidate_detect_cache, ) +from ui import tui class FormatLaunchHintTests(unittest.TestCase): @@ -50,6 +51,10 @@ class RegistryTests(unittest.TestCase): self.assertIs(get("audiocpp").key, "audiocpp") self.assertIsNone(get("nonexistent")) + def test_qwen_carries_the_per_model_configure_screen(self): + from backends import qwen + self.assertIs(get("qwen").configure_screen, qwen.models_screen) + class DetectAllTests(unittest.TestCase): def test_detect_all_returns_one_status_per_backend(self): @@ -313,6 +318,406 @@ class RemoteSuppressionTests(unittest.TestCase): self.assertEqual(status.remote_urls, {}) +class QwenModelCacheTests(unittest.TestCase): + """HF-cache awareness for the three demo repos (see backends.qwen).""" + + def test_repo_dirs_map_to_hf_cache_names(self): + from backends import qwen + with tempfile.TemporaryDirectory() as td, \ + patch.dict("os.environ", {"HF_HUB_CACHE": td}): + self.assertEqual( + qwen.model_repo_dir("CustomVoice"), + Path(td) / "models--Qwen--Qwen3-TTS-12Hz-1.7B-CustomVoice") + self.assertEqual( + qwen.model_repo_dir("Base"), + Path(td) / "models--Qwen--Qwen3-TTS-12Hz-1.7B-Base") + self.assertEqual( + qwen.model_repo_dir("VoiceDesign"), + Path(td) / "models--Qwen--Qwen3-TTS-12Hz-1.7B-VoiceDesign") + + def test_cache_dir_resolution_matches_huggingface_hub_precedence(self): + # HF_HUB_CACHE > HUGGINGFACE_HUB_CACHE > HF_HOME/hub > default. + # Listed vars are blanked so ambient env can't leak in. + from backends import qwen + with tempfile.TemporaryDirectory() as td: + base = Path(td) + with patch.dict("os.environ", {"HF_HUB_CACHE": str(base / "a"), + "HUGGINGFACE_HUB_CACHE": "", + "HF_HOME": ""}): + self.assertEqual(qwen._hf_cache_dir(), base / "a") + with patch.dict("os.environ", {"HF_HUB_CACHE": "", + "HUGGINGFACE_HUB_CACHE": + str(base / "b"), + "HF_HOME": ""}): + self.assertEqual(qwen._hf_cache_dir(), base / "b") + with patch.dict("os.environ", {"HF_HUB_CACHE": "", + "HUGGINGFACE_HUB_CACHE": "", + "HF_HOME": str(base / "c")}): + self.assertEqual(qwen._hf_cache_dir(), base / "c" / "hub") + with patch.dict("os.environ", {"HF_HUB_CACHE": "", + "HUGGINGFACE_HUB_CACHE": "", + "HF_HOME": ""}): + self.assertEqual(qwen._hf_cache_dir(), + Path.home() / ".cache" / "huggingface" + / "hub") + + def _seed_model(self, cache: Path, repo_id: str) -> Path: + """A fully-fetched-looking repo dir: refs/main + a snapshot file.""" + d = cache / ("models--" + repo_id.replace("/", "--")) + (d / "snapshots" / "abc123").mkdir(parents=True) + (d / "refs").mkdir() + (d / "refs" / "main").write_text("abc123\n", encoding="utf-8") + (d / "snapshots" / "abc123" / "config.json").write_bytes(b"x") + return d + + def test_installed_requires_refs_and_a_snapshot_file(self): + from backends import qwen + repo = qwen.MODEL_REPOS["CustomVoice"] + with tempfile.TemporaryDirectory() as td: + with patch.dict("os.environ", {"HF_HUB_CACHE": td}): + self.assertFalse(qwen.model_installed("CustomVoice")) + self.assertEqual(qwen.installed_models(), []) + self._seed_model(Path(td), repo) + self.assertTrue(qwen.model_installed("CustomVoice")) + self.assertEqual(qwen.installed_models(), ["CustomVoice"]) + + def test_partial_download_counts_as_not_installed(self): + # An interrupted fetch leaves blobs/ behind but no refs/main yet; + # resuming (Install or the next server start) takes over cleanly. + from backends import qwen + d = None + with tempfile.TemporaryDirectory() as td: + d = Path(td) + with patch.dict("os.environ", {"HF_HUB_CACHE": td}): + blob = (d / "models--Qwen--Qwen3-TTS-12Hz-1.7B-Base" + / "blobs") + blob.mkdir(parents=True) + (blob / "half.bin").write_bytes(b"x") + self.assertFalse(qwen.model_installed("Base")) + + +class QwenUninstallWeightsTests(unittest.TestCase): + """qwen.uninstall now also deletes every downloaded HF weight dir.""" + + def _seed_all(self, cache: Path): + from backends import qwen + for repo_id in qwen.MODEL_REPOS.values(): + d = cache / ("models--" + repo_id.replace("/", "--")) + (d / "snapshots" / "abc123").mkdir(parents=True) + (d / "snapshots" / "abc123" + / "model.safetensors").write_bytes(b"x") + (d / "refs").mkdir() + (d / "refs" / "main").write_text("abc123", encoding="utf-8") + + def test_uninstall_deletes_every_cached_model(self): + from backends import qwen + with tempfile.TemporaryDirectory() as td: + self._seed_all(Path(td)) + with patch.dict("os.environ", {"HF_HUB_CACHE": td}), \ + patch.object(qwen.servers, "pid_for", + return_value=None), \ + patch.object(qwen.common, "pip_uninstall", + return_value=0) as mk_pip: + rc = qwen.uninstall(emit="EMIT") + self.assertEqual(rc, 0) + mk_pip.assert_called_once_with([qwen.QWEN_PIP_PKG], emit="EMIT") + self.assertEqual(list(Path(td).iterdir()), []) + + def test_weights_deleted_even_when_pip_failed(self): + # The package is trivially re-installable; multi-GB snapshots are + # what actually cost disk. Deleting them is not conditional on pip. + from backends import qwen + with tempfile.TemporaryDirectory() as td: + self._seed_all(Path(td)) + with patch.dict("os.environ", {"HF_HUB_CACHE": td}), \ + patch.object(qwen.servers, "pid_for", + return_value=None), \ + patch.object(qwen.common, "pip_uninstall", + return_value=1): + rc = qwen.uninstall() + self.assertEqual(rc, 1) + self.assertEqual(list(Path(td).iterdir()), []) + + def test_cancel_after_pip_skips_weight_deletion(self): + import threading + + from backends import qwen + # Cancel fires mid-pip (the only moment the user can): everything + # through pip completes, but the weight-deletion phase never starts. + cancel = threading.Event() + + def pip_flips_cancel(*args, **kwargs): + cancel.set() + return 0 + + with tempfile.TemporaryDirectory() as td: + self._seed_all(Path(td)) + with patch.dict("os.environ", {"HF_HUB_CACHE": td}), \ + patch.object(qwen.servers, "pid_for", + return_value=None), \ + patch.object(qwen.common, "pip_uninstall", + side_effect=pip_flips_cancel) as mk_pip: + rc = qwen.uninstall(cancel=cancel) + self.assertEqual(rc, 130) + mk_pip.assert_called_once_with([qwen.QWEN_PIP_PKG], emit=None) + # Cancelled between phases: the weights stay untouched... + self.assertEqual(len(list(Path(td).iterdir())), 3) + + def test_only_qwen_repos_are_touched_in_the_shared_cache(self): + from backends import qwen + with tempfile.TemporaryDirectory() as td: + self._seed_all(Path(td)) + other = Path(td) / "models--Other--Repo" + other.mkdir() + (other / "weights.bin").write_bytes(b"x") + with patch.dict("os.environ", {"HF_HUB_CACHE": td}), \ + patch.object(qwen.servers, "pid_for", + return_value=None), \ + patch.object(qwen.common, "pip_uninstall", + return_value=0): + qwen.uninstall() + self.assertTrue(other.is_dir()) + + +class QwenUninstallModelTests(unittest.TestCase): + """Per-model uninstall: stop only a server serving THAT model.""" + + def test_stops_managed_server_only_when_it_serves_that_model(self): + from backends import qwen + cases = [("Base", True), ("CustomVoice", False), ("VoiceDesign", + False)] + for model, should_stop in cases: + with self.subTest(model=model): + with patch.object(qwen, "_managed_running_model", + return_value="Base"), \ + patch.object(qwen.servers, "stop") as mk_stop, \ + patch.object(qwen, "delete_model_weights") as mk_del: + rc = qwen.uninstall_model(model) + self.assertEqual((rc, mk_stop.called), + (0, should_stop)) + mk_del.assert_called_once_with([model]) + + # No managed server at all: nothing to stop either. + with patch.object(qwen, "_managed_running_model", + return_value=None), \ + patch.object(qwen.servers, "stop") as mk_stop: + qwen.uninstall_model("Base") + mk_stop.assert_not_called() + + def test_removes_only_that_models_cache_dir(self): + from backends import qwen + with tempfile.TemporaryDirectory() as td, \ + patch.dict("os.environ", {"HF_HUB_CACHE": td}), \ + patch.object(qwen, "_managed_running_model", + return_value=None): + kept = qwen.repo_dir(qwen.MODEL_REPOS["CustomVoice"]) + kept.mkdir(parents=True) + gone = qwen.repo_dir(qwen.MODEL_REPOS["VoiceDesign"]) + gone.mkdir(parents=True) + rc = qwen.uninstall_model("VoiceDesign") + self.assertEqual(rc, 0) + # Only the named model's directory is gone; every other cache + # entry (unrelated repos included) survives. + self.assertFalse(gone.exists()) + self.assertTrue(kept.is_dir()) + + def test_missing_weights_still_succeed(self): + # Idempotent removal, like apt purge on an already-clean system. + from backends import qwen + with patch.object(qwen, "_managed_running_model", + return_value=None): + self.assertEqual(qwen.uninstall_model("Base"), 0) + + def test_cancel_after_stop_skips_deletion(self): + import threading + + from backends import qwen + cancel = threading.Event() + cancel.set() + with patch.object(qwen, "_managed_running_model", + return_value="Base"), \ + patch.object(qwen.servers, "stop") as mk_stop, \ + patch.object(qwen, "delete_model_weights"): + rc = qwen.uninstall_model("Base", cancel=cancel) + self.assertEqual(rc, 130) + mk_stop.assert_called_once_with("qwen") + + +class QwenInstallModelTests(unittest.TestCase): + """install_model: venv hf CLI download of exactly one repo's weights.""" + + def test_hf_cli_prefers_hf_then_falls_back(self): + from backends import qwen + with tempfile.TemporaryDirectory() as td: + with patch.object(qwen.envs, "ENV_DIR", Path(td)): + self.assertIsNone(qwen._hf_download_prefix()) + cli = qwen.envs.env_script("huggingface-cli") + cli.parent.mkdir(parents=True) + cli.write_bytes(b"x") + self.assertEqual(qwen._hf_download_prefix(), [str(cli)]) + hf = qwen.envs.env_script("hf") + hf.write_bytes(b"x") + self.assertEqual(qwen._hf_download_prefix(), [str(hf)]) + + def test_download_runs_through_console_streaming(self): + from backends import qwen + with patch.object(qwen, "_hf_download_prefix", + return_value=["/venv/bin/hf"]), \ + patch.object(qwen.common, "run_console_subprocess", + return_value=0) as mk_run: + rc = qwen.install_model("VoiceDesign", emit="EMIT", + cancel="CANCEL") + self.assertEqual(rc, 0) + mk_run.assert_called_once_with( + ["/venv/bin/hf", "download", + "Qwen/Qwen3-TTS-12Hz-1.7B-VoiceDesign"], + emit="EMIT", cancel="CANCEL") + + def test_no_cli_is_a_failure_not_an_exception(self): + from backends import qwen + with patch.object(qwen, "_hf_download_prefix", return_value=None), \ + patch.object(qwen.common, "run_console_subprocess"): + rc = qwen.install_model("Base") + self.assertEqual(rc, 1) + + +class QwenModelsScreenTests(unittest.TestCase): + """models_screen: per-model menu driving task-view steps.""" + + def _screen(self, answers, *, installed=("CustomVoice",), + package=True, extra=()): + """Run models_screen with scripted menu answers; record calls. + + Returns ``(rc, menus, flashes, runs)``: menus holds one + (title, options, kwargs) per render, flashes every (text, kind), + and runs each task-view run's title while executing its first + step's work inline (so delegation to install/uninstall model + functions is observable). EXTRA holds additional patch context + managers entered around the whole run. + """ + import contextlib + + from backends import qwen + choices = list(answers) + menus = [] + flashes = [] + runs = [] + + def fake_menu(stdscr, title, options, **kwargs): + menus.append((title, options, kwargs)) + return choices.pop(0) + + def fake_flash(scr, text, kind="warn"): + flashes.append((text, kind)) + + def fake_run(scr, title, steps, **kwargs): + runs.append(title) + steps[0].work(None, None) + return 0 + + patches = [ + patch.object(qwen, "_is_installed", return_value=package), + patch.object(qwen, "installed_models", + return_value=list(installed)), + patch.object(qwen.tui, "menu", fake_menu), + patch.object(qwen.tui, "flash", fake_flash), + patch.object(qwen.taskview, "run_steps", fake_run), + *extra, + ] + with contextlib.ExitStack() as stack: + for ctx in patches: + stack.enter_context(ctx) + rc = qwen.models_screen(None) + return rc, menus, flashes, runs + + def test_options_reflect_disk_state_per_model(self): + rc, menus, _, _ = self._screen([tui.Wizard.BACK], + installed=("CustomVoice",)) + self.assertEqual(rc, 0) + title, options, kwargs = menus[0] + self.assertEqual(title, "Configure qwen-tts") + # One action per model, mirroring disk state; order follows + # MODEL_REPOS. The table repeats the state in color. + self.assertEqual(options, [ + ("Uninstall CustomVoice", ("uninstall", "CustomVoice")), + ("Install Base", ("install", "Base")), + ("Install VoiceDesign", ("install", "VoiceDesign")), + ]) + self.assertEqual(kwargs["table_title"], "Model state") + self.assertEqual(kwargs["table_rows"][0], + ("CustomVoice", "installed", "ok")) + + def test_install_action_runs_a_download_step_in_the_task_view(self): + from backends import qwen + requested = [] + + def capture(model, *, emit=None, cancel=None): + requested.append(model) + return 0 + + rc, _, flashes, runs = self._screen( + [("install", "Base"), tui.Wizard.BACK], installed=(), + extra=[patch.object(qwen, "install_model", + side_effect=capture)]) + self.assertEqual(rc, 0) + self.assertEqual(requested, ["Base"]) + self.assertEqual(len(runs), 1) + self.assertEqual(runs[0], "Download Qwen/Qwen3-TTS-12Hz-1.7B-Base") + self.assertEqual(flashes[-1], ("Base downloaded.", "ok")) + + def test_uninstall_action_stops_the_server_then_deletes_weights(self): + from backends import qwen + stopped = [] + with tempfile.TemporaryDirectory() as td: + gone = Path(td) / "models--Qwen--Qwen3-TTS-12Hz-1.7B-CustomVoice" + gone.mkdir(parents=True) + other = Path(td) / "models--Other--Repo" + other.mkdir(parents=True) + rc, _, flashes, _ = self._screen( + [("uninstall", "CustomVoice"), tui.Wizard.BACK], + extra=[ + patch.dict("os.environ", {"HF_HUB_CACHE": td}), + patch.object(qwen, "_managed_running_model", + return_value="CustomVoice"), + patch.object(qwen.servers, "stop", + side_effect=lambda name: + stopped.append(name)), + # delete_model_weights stays real: it must rm the exact + # directory below (inside a redirected HF_HUB_CACHE). + ]) + self.assertEqual(rc, 0) + self.assertEqual(stopped, ["qwen"]) + self.assertFalse(gone.exists()) + self.assertTrue(other.is_dir()) + self.assertEqual(flashes[-1], + ("CustomVoice weights removed.", "ok")) + + def test_install_without_package_flashes_guidance_instead(self): + rc, menus, flashes, runs = self._screen( + [("install", "Base"), tui.Wizard.BACK], installed=(), + package=False) + self.assertEqual(rc, 0) + # No task-view run, one guidance flash instead of a download. + self.assertEqual(runs, []) + guidance = ("Install the qwen-tts backend first " + "(Configure backends > Install Backend).") + self.assertEqual(flashes, [(guidance, "warn")]) + # While the backend is missing, Install is replaced by dimmed-out + # "(backend not installed)" placeholders — actions stay inert. + self.assertEqual([label for label, _ in menus[0][1]], + ["CustomVoice (backend not installed)", + "Base (backend not installed)", + "VoiceDesign (backend not installed)"]) + + def test_noop_placeholder_actions_change_nothing(self): + rc, _, flashes, runs = self._screen( + [("noop", "Base"), tui.Wizard.BACK], installed=(), + package=False) + self.assertEqual(rc, 0) + self.assertEqual(runs, []) + self.assertEqual(flashes, []) + + class DetectCacheTests(unittest.TestCase): """detect_all's short-TTL cache (menu renders re-probe only after it).""" |