feat: manage qwen-tts model installs manually, delete model(s) when uninstalled

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)."""