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