feat: running process detection, menu gating

1 files changed, 85 insertions, 7 deletions

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()