"""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, detect_all, get 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_and_setup_tui(self): for info in REGISTRY: self.assertTrue(callable(info.detect), info.key) self.assertTrue(callable(info.setup_tui), info.key) self.assertIsInstance(info.configure_actions, list) for action in info.configure_actions: self.assertTrue(callable(action.run)) 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_server_probe_succeeds(self): from backends import audiocpp with patch.object(audiocpp, "find_local_checkout", return_value=None), \ patch("backends.common.server_running", return_value=True): status = audiocpp.detect() # Not installed (no checkout) but an external server is up. self.assertFalse(status.installed) self.assertTrue(status.running) 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_port_is_up(self): # Either the CustomVoice port or the Base port counts as running, # and the status names which model answered. Probes: CustomVoice # (QWEN_API_URL) first, then Base (CLONE_API_URL). from backends import qwen with patch.object(qwen, "_is_installed", return_value=False), \ patch("backends.common.server_running", side_effect=[True, False]): status = qwen.detect() self.assertTrue(status.running) self.assertEqual(status.running_models, ["CustomVoice"]) with patch.object(qwen, "_is_installed", return_value=False), \ patch("backends.common.server_running", side_effect=[False, True]): status = qwen.detect() self.assertTrue(status.running) self.assertEqual(status.running_models, ["Base"]) def test_qwen_running_models_names_both_ports(self): # Both ports 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("backends.common.server_running", side_effect=[True, True]): status = qwen.detect() self.assertTrue(status.running) 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_server_probe_succeeds(self): from backends import faster with patch.object(faster, "_is_installed", return_value=False), \ patch.object(faster, "_is_cloned", return_value=False), \ patch("backends.common.server_running", return_value=True): status = faster.detect() self.assertTrue(status.running) 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("")) if __name__ == "__main__": unittest.main()