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

from backends import REGISTRY, ServerSpec, detect_all, format_launch_hint, get


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.tts 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_tui), 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, "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, "find_local_checkout",
                          return_value=None), \
                patch.object(audiocpp.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_either_remote_url_is_up(self):
        # Either the CustomVoice or the Base remote URL answering counts as
        # running, and the status names which model answered. Probes: Base
        # (CLONE_REMOTE_URL) first, then CustomVoice (QWEN_REMOTE_URL).
        from backends import qwen
        with patch.object(qwen, "_is_installed", return_value=False), \
                patch.object(qwen.probe, "identify_server",
                             side_effect=[None, "qwen-custom"]):
            status = qwen.detect()
        self.assertTrue(status.running)
        self.assertTrue(status.remote)
        self.assertEqual(status.remote_models, ["CustomVoice"])
        self.assertEqual(status.running_models, ["CustomVoice"])
        with patch.object(qwen, "_is_installed", return_value=False), \
                patch.object(qwen.probe, "identify_server",
                             side_effect=["qwen-clone", None]):
            status = qwen.detect()
        self.assertTrue(status.running)
        self.assertEqual(status.remote_models, ["Base"])
        self.assertEqual(status.running_models, ["Base"])

    def test_qwen_running_models_names_both_ports(self):
        # Both remote URLs up → both models, Base first (the hub renders
        # "running (Base, CustomVoice)").
        from backends import qwen
        with patch.object(qwen, "_is_installed", return_value=False), \
                patch.object(qwen.probe, "identify_server",
                             side_effect=["qwen-clone", "qwen-custom"]):
            status = qwen.detect()
        self.assertTrue(status.running)
        self.assertEqual(status.remote_models, ["Base", "CustomVoice"])
        self.assertEqual(status.running_models, ["Base", "CustomVoice"])

    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-custom-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, "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.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, {})


if __name__ == "__main__":
    unittest.main()