From 194c63e4d11e6de9792a736a7b99788f1db78741 Mon Sep 17 00:00:00 2001 From: historia Date: Mon, 24 Aug 2026 00:41:52 -0400 Subject: feat: running process detection, menu gating --- tests/test_backends.py | 92 ++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 85 insertions(+), 7 deletions(-) (limited to 'tests/test_backends.py') diff --git a/tests/test_backends.py b/tests/test_backends.py index 4017cd4..8ee1be8 100644 --- a/tests/test_backends.py +++ b/tests/test_backends.py @@ -17,8 +17,8 @@ class RegistryTests(unittest.TestCase): for info in REGISTRY: self.assertTrue(callable(info.detect), info.key) self.assertTrue(callable(info.setup_tui), info.key) - self.assertIsInstance(info.modify_actions, list) - for action in info.modify_actions: + 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): @@ -28,7 +28,8 @@ class RegistryTests(unittest.TestCase): class DetectAllTests(unittest.TestCase): def test_detect_all_returns_one_status_per_backend(self): - statuses = detect_all() + 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: @@ -37,6 +38,9 @@ class DetectAllTests(unittest.TestCase): # 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: @@ -52,25 +56,55 @@ class DetectAllTests(unittest.TestCase): encoding="utf-8") from backends import audiocpp with patch.object(audiocpp, "find_local_checkout", - return_value=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): + 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): + 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: @@ -80,12 +114,56 @@ class DetectAllTests(unittest.TestCase): (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.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): + from backends import common + # Pick an unused port by opening + closing a socket, then probe it. + import socket + 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() -- cgit v1.2.3