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


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_the_three_backends(self):
        keys = [info.key for info in REGISTRY]
        self.assertEqual(keys, ["audiocpp", "qwen", "faster"])

    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_get_returns_entry_by_key(self):
        self.assertIs(get("audiocpp").key, "audiocpp")
        self.assertIsNone(get("nonexistent"))


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"])
        for s in statuses:
            self.assertIn(s.key, ("audiocpp", "qwen", "faster"))
            # 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_uses_one_spec_for_the_configured_model(self):
        # One demo server hosts one model on the single port: the spec's
        # argv launches config.QWEN_MODEL's repo, and its identity matches.
        from backends import qwen
        from backends.probe import (IDENTITY_QWEN_CLONE,
                                    IDENTITY_QWEN_CUSTOM,
                                    IDENTITY_QWEN_DESIGN)
        cases = {"CustomVoice": IDENTITY_QWEN_CUSTOM,
                 "Base": IDENTITY_QWEN_CLONE,
                 "VoiceDesign": IDENTITY_QWEN_DESIGN}
        for model, identity in cases.items():
            with self.subTest(model=model):
                with patch.object(qwen.config, "QWEN_MODEL", model), \
                        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)

    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 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.
        mk_pip.assert_called_once_with([qwen.QWEN_PIP_PKG], emit="EMIT")

    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)