"""Tests for the backends package registry and detection aggregation.""" import tempfile import unittest from pathlib import Path from unittest.mock import patch import backends from backends import ( REGISTRY, BackendStatus, ServerSpec, detect_all, format_launch_hint, get, invalidate_detect_cache, ) from ui import tui class FormatLaunchHintTests(unittest.TestCase): def test_plain_specs_join_argv(self): specs = [ServerSpec("a", "http://x", ["cmd", "--flag"])] self.assertEqual(format_launch_hint(specs), "cmd --flag") def test_cwd_prefixes_the_command(self): specs = [ServerSpec("a", "http://x", ["cmd"], cwd=Path("/opt/audio.cpp"))] self.assertEqual(format_launch_hint(specs), "cd /opt/audio.cpp && cmd") class RegistryTests(unittest.TestCase): def setUp(self): # The registry is built lazily on first access (the backend modules # pull in converter.clients and its deps, which are only available inside # the managed venv). Trigger the build so these tests don't depend on # another test class having called detect_all() first. get("audiocpp") def test_registry_has_every_backend(self): keys = [info.key for info in REGISTRY] self.assertEqual(keys, ["audiocpp", "qwen", "faster", "sglomni"]) def test_every_entry_has_detect_setup_and_uninstall(self): for info in REGISTRY: self.assertTrue(callable(info.detect), info.key) self.assertTrue(callable(info.setup_screen), info.key) self.assertTrue(callable(info.uninstall), info.key) def test_every_entry_has_an_update_action(self): for info in REGISTRY: self.assertTrue(callable(info.update), info.key) def test_get_returns_entry_by_key(self): 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): with patch("backends.common.server_running", return_value=False): statuses = detect_all() self.assertEqual([s.key for s in statuses], ["audiocpp", "qwen", "faster", "sglomni"]) for s in statuses: self.assertIn(s.key, ("audiocpp", "qwen", "faster", "sglomni")) # ready requires both installed and configured; on a clean # machine none are ready. if s.ready: self.assertTrue(s.installed and s.configured) # running is always probed; patched False here so a dev machine # running a real server can't flake the test. self.assertFalse(s.running) 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.build, "find_local_checkout", return_value=checkout), \ patch("backends.common.server_running", return_value=False): status = audiocpp.detect() self.assertTrue(status.installed) self.assertTrue(status.configured) self.assertTrue(status.ready) self.assertFalse(status.running) self.assertIn("audiocpp_server", status.launch_hint) def test_audiocpp_running_when_remote_server_identified(self): from backends import audiocpp with patch.object(audiocpp.build, "find_local_checkout", return_value=None), \ patch.object(audiocpp.status.probe, "identify_server", return_value="audiocpp"): status = audiocpp.detect() # Not installed (no checkout) but a remote server answers. self.assertFalse(status.installed) self.assertTrue(status.running) self.assertTrue(status.remote) self.assertIn("audiocpp", status.remote_urls) def test_qwen_status_reflects_install(self): from backends import qwen with patch.object(qwen, "_is_installed", return_value=True), \ patch("backends.common.server_running", return_value=False): status = qwen.detect() self.assertTrue(status.installed) self.assertTrue(status.configured) self.assertFalse(status.running) self.assertIn("qwen-tts-demo", status.launch_hint) with patch.object(qwen, "_is_installed", return_value=False), \ patch("backends.common.server_running", return_value=False): status = qwen.detect() self.assertFalse(status.installed) self.assertFalse(status.configured) def test_qwen_running_when_remote_url_is_up(self): # The single remote URL answering as any of the three demos counts # as running, and the status names which model answered. from backends import qwen for identity, model in (("qwen-custom", "CustomVoice"), ("qwen-clone", "Base"), ("qwen-design", "VoiceDesign")): with self.subTest(identity=identity): with patch.object(qwen, "_is_installed", return_value=False), \ patch.object(qwen.probe, "identify_server", return_value=identity): status = qwen.detect() self.assertTrue(status.running) self.assertTrue(status.remote) self.assertEqual(status.remote_models, [model]) self.assertEqual(status.running_models, [model]) def test_qwen_detect_builds_one_spec_for_the_default_model(self): # One demo server hosts one model on the single port: the detect() # spec launches the default model's repo (CustomVoice), and its # identity matches; runs wanting another model boot their own spec. from backends import qwen from backends.probe import (IDENTITY_QWEN_CLONE, IDENTITY_QWEN_CUSTOM, IDENTITY_QWEN_DESIGN) identity = IDENTITY_QWEN_CUSTOM model = qwen.DEFAULT_MODEL with patch.object(qwen, "_is_installed", return_value=True), \ patch("backends.common.server_running", return_value=False): status = qwen.detect() self.assertEqual([spec.name for spec in status.servers], ["qwen"]) spec = status.servers[0] self.assertEqual(spec.identity, identity) self.assertIn(qwen.MODEL_REPOS[model], spec.argv) self.assertIn(qwen.MODEL_REPOS[model], status.launch_hint) # An explicit per-run spec can target any of the three models. for model, wanted in (("Base", IDENTITY_QWEN_CLONE), ("VoiceDesign", IDENTITY_QWEN_DESIGN)): with self.subTest(model=model): spec = qwen.build_spec(model) self.assertEqual(spec.identity, wanted) self.assertIn(qwen.MODEL_REPOS[model], spec.argv) def test_qwen_detect_marks_our_server_as_managed(self): from backends import qwen from backends import servers as servers_mod with tempfile.TemporaryDirectory() as td: (Path(td) / "qwen-server.pid").write_text( "4242", encoding="utf-8") with patch.object(qwen, "_is_installed", return_value=False), \ patch("backends.common.server_running", return_value=False), \ patch.object(servers_mod, "LOG_DIR", Path(td)), \ patch.object(servers_mod, "_pid_alive", return_value=True): status = qwen.detect() self.assertTrue(status.managed) # Without a live pid file the same server counts as remote. with tempfile.TemporaryDirectory() as td, \ patch.object(qwen, "_is_installed", return_value=False), \ patch("backends.common.server_running", return_value=False), \ patch.object(servers_mod, "LOG_DIR", Path(td)): status = qwen.detect() self.assertFalse(status.managed) 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), \ patch("backends.common.server_running", return_value=False): status = faster.detect() self.assertTrue(status.installed) self.assertTrue(status.configured) self.assertFalse(status.running) self.assertIn("openai_server.py", status.launch_hint) def test_faster_running_when_remote_server_identified(self): from backends import faster with patch.object(faster, "_is_installed", return_value=False), \ patch.object(faster, "_is_cloned", return_value=False), \ patch.object(faster.probe, "identify_server", return_value="faster"): status = faster.detect() self.assertTrue(status.running) self.assertTrue(status.remote) self.assertIn("faster", status.remote_urls) class ServerRunningTests(unittest.TestCase): """backends.common.server_running: TCP probe against a real socket.""" def test_true_for_open_port(self): import socket from backends import common server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.bind(("127.0.0.1", 0)) server.listen(1) host, port = server.getsockname() url = f"http://127.0.0.1:{port}" try: self.assertTrue(common.server_running(url)) finally: server.close() def test_false_for_closed_port(self): # Pick an unused port by opening + closing a socket, then probe it. import socket from backends import common s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.bind(("127.0.0.1", 0)) _, port = s.getsockname() s.close() self.assertFalse(common.server_running(f"http://127.0.0.1:{port}")) def test_false_for_invalid_url(self): from backends import common self.assertFalse(common.server_running("not a url")) self.assertFalse(common.server_running("")) class RemoteUrlTests(unittest.TestCase): """backends.common.normalize_remote_url: host:port / URL -> http(s)://.""" def test_bare_host_port_gets_http_scheme(self): from backends import common self.assertEqual(common.normalize_remote_url("10.0.0.5:8080"), "http://10.0.0.5:8080") def test_full_url_preserved(self): from backends import common self.assertEqual(common.normalize_remote_url( "https://10.0.0.5:8443/path"), "https://10.0.0.5:8443/path") def test_empty_means_disabled(self): from backends import common self.assertEqual(common.normalize_remote_url(""), "") self.assertEqual(common.normalize_remote_url(" "), "") def test_whitespace_stripped(self): from backends import common self.assertEqual(common.normalize_remote_url(" 10.0.0.5:8080 "), "http://10.0.0.5:8080") def test_invalid_rejected(self): from backends import common for value in ("http://", "not a url", "10.0.0.5:notaport", "://"): with self.assertRaises(ValueError, msg=value): common.normalize_remote_url(value) class RemoteSuppressionTests(unittest.TestCase): """A server this tool started must not also be reported as remote.""" def test_audiocpp_own_server_suppresses_remote(self): from backends import audiocpp from backends import servers as servers_mod 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") (Path(td) / "audiocpp-server.pid").write_text( "4242", encoding="utf-8") with patch.object(audiocpp.build, "find_local_checkout", return_value=checkout), \ patch.object(servers_mod, "LOG_DIR", Path(td)), \ patch.object(servers_mod, "_pid_alive", return_value=True), \ patch.object(audiocpp.status.probe, "identify_server", return_value="audiocpp"): status = audiocpp.detect() self.assertTrue(status.managed) self.assertTrue(status.running) self.assertFalse(status.remote) 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", env_dir=qwen.QWEN_ENV) 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, env_dir=qwen.QWEN_ENV) # 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: # The CLI is looked up in the qwen backend's own venv. with patch.object(qwen, "QWEN_ENV", Path(td)): self.assertIsNone(qwen._hf_download_prefix()) cli = qwen.envs.env_script("huggingface-cli", qwen.QWEN_ENV) cli.parent.mkdir(parents=True) cli.write_bytes(b"x") self.assertEqual(qwen._hf_download_prefix(), [str(cli)]) hf = qwen.envs.env_script("hf", qwen.QWEN_ENV) 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") # Every row is a (name, status, kind) triple: tui.menu reads # row[2] for the status color, so a short row crashes the menu # with "tuple index out of range". self.assertEqual(kwargs["table_rows"], [ ("CustomVoice", "installed", "ok"), ("Base", "not installed", "warn"), ("VoiceDesign", "not installed", "warn"), ]) 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).""" def setUp(self): get("audiocpp") # build the lazy registry before patching its entries invalidate_detect_cache() self.addCleanup(invalidate_detect_cache) self.probes = [] self.patches = [] for info in REGISTRY: def fake_detect(key=info.key): self.probes.append(key) return BackendStatus(key, key, installed=False, configured=False) self.patches.append(patch.object(info, "detect", side_effect=fake_detect)) for p in self.patches: p.start() self.addCleanup(p.stop) def test_repeated_calls_within_the_ttl_probe_once(self): first = detect_all() second = detect_all() self.assertEqual(first, second) self.assertEqual(sorted(self.probes), sorted(i.key for i in REGISTRY)) self.assertEqual(len(self.probes), len(REGISTRY)) def test_refresh_bypasses_the_cache(self): detect_all() detect_all(refresh=True) self.assertEqual(len(self.probes), 2 * len(REGISTRY)) def test_invalidate_forces_the_next_call_to_reprobe(self): detect_all() invalidate_detect_cache() detect_all() self.assertEqual(len(self.probes), 2 * len(REGISTRY)) def test_expiry_after_the_ttl_reprobes(self): with patch.object(backends, "DETECT_TTL_SECONDS", 0.0): detect_all() detect_all() self.assertEqual(len(self.probes), 2 * len(REGISTRY)) if __name__ == "__main__": unittest.main() class QwenSetupScreenTests(unittest.TestCase): """qwen.setup_screen: a question-free setup on the hub's screen. The qwen wizard asks nothing (ports live in Settings, the speaker is chosen on Generate), so it cannot be aborted: an already-installed package is a silent no-op, everything else runs in the task view. """ def test_already_installed_is_a_silent_noop(self): from backends import qwen settings = {"do_install": False} with patch.object(qwen, "_wizard", return_value=settings) as mk_wizard, \ patch.object(qwen.taskview, "run_steps") as mk_run: rc = qwen.setup_screen(None) self.assertEqual(rc, 0) mk_wizard.assert_called_once() mk_run.assert_not_called() def test_missing_package_runs_the_tail_in_the_task_view(self): from backends import qwen settings = {"do_install": True} steps = [qwen.taskview.TaskStep("t", lambda emit, cancel: 0)] with patch.object(qwen, "_wizard", return_value=settings), \ patch.object(qwen, "_execute_steps", return_value=steps) as mk_steps, \ patch.object(qwen.taskview, "run_steps", return_value=0) as mk_run: rc = qwen.setup_screen(None) self.assertEqual(rc, 0) mk_steps.assert_called_once() self.assertIs(mk_steps.call_args[0][0], settings) mk_run.assert_called_once() self.assertEqual(mk_run.call_args[0][2], steps) def test_wizard_has_no_port_or_speaker_settings(self): # The screens for CustomVoice/Base ports and the built-in speaker # are gone; settings only carry whether to pip install. from backends import qwen args = qwen.build_parser().parse_args([]) with patch.object(qwen, "_is_installed", return_value=False): settings = qwen._wizard(None, args) self.assertEqual(settings, {"do_install": True}) with patch.object(qwen, "_is_installed", return_value=True): settings = qwen._wizard(None, args) self.assertEqual(settings, {"do_install": False}) class QwenUninstallTests(unittest.TestCase): """qwen.uninstall: stop the single server, then pip-uninstall the package.""" def test_stops_servers_and_pips(self): from backends import qwen # A pid file exists for the managed server, so stop runs. with patch.object(qwen.servers, "pid_for", return_value=1234), \ patch.object(qwen.servers, "stop") as mk_stop, \ patch.object(qwen.common, "pip_uninstall", return_value=0) as mk_pip: rc = qwen.uninstall(emit="EMIT") self.assertEqual(rc, 0) self.assertEqual([c.args[0] for c in mk_stop.call_args_list], ["qwen"]) # The task view's emit is forwarded so pip never touches the terminal, # and the package comes out of the qwen backend's own venv. mk_pip.assert_called_once_with([qwen.QWEN_PIP_PKG], emit="EMIT", env_dir=qwen.QWEN_ENV) def test_skips_stop_when_no_server_was_started(self): # No pid files: stop() is not called (no "not started by this # tool" noise during an uninstall). from backends import qwen with patch.object(qwen.servers, "pid_for", return_value=None), \ patch.object(qwen.servers, "stop") as mk_stop, \ patch.object(qwen.common, "pip_uninstall", return_value=0): rc = qwen.uninstall() self.assertEqual(rc, 0) mk_stop.assert_not_called() def test_cancel_before_pip_skips_uninstall(self): import threading from backends import qwen cancel = threading.Event() cancel.set() with patch.object(qwen.servers, "pid_for", return_value=1234), \ patch.object(qwen.servers, "stop") as mk_stop, \ patch.object(qwen.common, "pip_uninstall") as mk_pip: rc = qwen.uninstall(cancel=cancel) self.assertEqual(rc, 130) self.assertEqual(mk_stop.call_count, 1) mk_pip.assert_not_called() def test_pip_failure_propagates_the_exit_code(self): from backends import qwen with patch.object(qwen.servers, "pid_for", return_value=1234), \ patch.object(qwen.servers, "stop"), \ patch.object(qwen.common, "pip_uninstall", return_value=1): rc = qwen.uninstall() self.assertEqual(rc, 1) class QwenUpdateTests(unittest.TestCase): """qwen.update: stop the single server, then pip install -U the package.""" def test_stops_server_and_pip_upgrades_into_the_qwen_env(self): from backends import qwen with patch.object(qwen.servers, "pid_for", return_value=1234), \ patch.object(qwen.servers, "stop") as mk_stop, \ patch.object(qwen.common, "pip_install", return_value=0) as mk_pip: rc = qwen.update(emit="EMIT") self.assertEqual(rc, 0) self.assertEqual([c.args[0] for c in mk_stop.call_args_list], ["qwen"]) # The task view's emit is forwarded, the install is an upgrade, # and the package lands in the qwen backend's own venv. mk_pip.assert_called_once_with([qwen.QWEN_PIP_PKG], emit="EMIT", cancel=None, env_dir=qwen.QWEN_ENV, upgrade=True) def test_skips_stop_when_no_server_was_started(self): from backends import qwen with patch.object(qwen.servers, "pid_for", return_value=None), \ patch.object(qwen.servers, "stop") as mk_stop, \ patch.object(qwen.common, "pip_install", return_value=0): rc = qwen.update() self.assertEqual(rc, 0) mk_stop.assert_not_called() def test_cancel_before_pip_skips_the_upgrade(self): import threading from backends import qwen cancel = threading.Event() cancel.set() with patch.object(qwen.servers, "pid_for", return_value=1234), \ patch.object(qwen.servers, "stop") as mk_stop, \ patch.object(qwen.common, "pip_install") as mk_pip: rc = qwen.update(cancel=cancel) self.assertEqual(rc, 130) self.assertEqual(mk_stop.call_count, 1) mk_pip.assert_not_called() def test_pip_failure_propagates_the_exit_code(self): from backends import qwen with patch.object(qwen.servers, "pid_for", return_value=None), \ patch.object(qwen.servers, "stop"), \ patch.object(qwen.common, "pip_install", return_value=1): rc = qwen.update() self.assertEqual(rc, 1)