path: root/app/tests/test_backends.py

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