diff options
Diffstat (limited to 'app/tests/test_backends.py')
| -rw-r--r-- | app/tests/test_backends.py | 178 |
1 files changed, 178 insertions, 0 deletions
diff --git a/app/tests/test_backends.py b/app/tests/test_backends.py new file mode 100644 index 0000000..c0e8d4a --- /dev/null +++ b/app/tests/test_backends.py @@ -0,0 +1,178 @@ +"""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. + 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) + 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) + + 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() |