diff options
Diffstat (limited to 'app/tests/test_backends.py')
| -rw-r--r-- | app/tests/test_backends.py | 105 |
1 files changed, 89 insertions, 16 deletions
diff --git a/app/tests/test_backends.py b/app/tests/test_backends.py index ac4c8ef..2cb5a95 100644 --- a/app/tests/test_backends.py +++ b/app/tests/test_backends.py @@ -73,15 +73,18 @@ class DetectAllTests(unittest.TestCase): self.assertFalse(status.running) self.assertIn("audiocpp_server", status.launch_hint) - def test_audiocpp_running_when_server_probe_succeeds(self): + def test_audiocpp_running_when_remote_server_identified(self): from backends import audiocpp with patch.object(audiocpp, "find_local_checkout", return_value=None), \ - patch("backends.common.server_running", return_value=True): + patch.object(audiocpp.probe, "identify_server", + return_value="audiocpp"): status = audiocpp.detect() - # Not installed (no checkout) but an external server is up. + # Not installed (no checkout) but a remote server answers. self.assertFalse(status.installed) self.assertTrue(status.running) + self.assertTrue(status.remote) + self.assertIn("audiocpp", status.remote_urls) def test_qwen_status_reflects_install(self): from backends import qwen @@ -98,33 +101,37 @@ class DetectAllTests(unittest.TestCase): 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). + def test_qwen_running_when_either_remote_url_is_up(self): + # Either the CustomVoice or the Base remote URL answering counts as + # running, and the status names which model answered. Probes: Base + # (CLONE_REMOTE_URL) first, then CustomVoice (QWEN_REMOTE_URL). from backends import qwen with patch.object(qwen, "_is_installed", return_value=False), \ - patch("backends.common.server_running", - side_effect=[True, False]): + patch.object(qwen.probe, "identify_server", + side_effect=[None, "qwen-custom"]): status = qwen.detect() self.assertTrue(status.running) + self.assertTrue(status.remote) + self.assertEqual(status.remote_models, ["CustomVoice"]) self.assertEqual(status.running_models, ["CustomVoice"]) with patch.object(qwen, "_is_installed", return_value=False), \ - patch("backends.common.server_running", - side_effect=[False, True]): + patch.object(qwen.probe, "identify_server", + side_effect=["qwen-clone", None]): status = qwen.detect() self.assertTrue(status.running) + self.assertEqual(status.remote_models, ["Base"]) 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 + # Both remote URLs 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]): + patch.object(qwen.probe, "identify_server", + side_effect=["qwen-clone", "qwen-custom"]): status = qwen.detect() self.assertTrue(status.running) + self.assertEqual(status.remote_models, ["Base", "CustomVoice"]) self.assertEqual(status.running_models, ["Base", "CustomVoice"]) def test_qwen_detect_marks_our_server_as_managed(self): @@ -169,13 +176,16 @@ class DetectAllTests(unittest.TestCase): self.assertFalse(status.running) self.assertIn("openai_server.py", status.launch_hint) - def test_faster_running_when_server_probe_succeeds(self): + def test_faster_running_when_remote_server_identified(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): + patch.object(faster.probe, "identify_server", + return_value="faster"): status = faster.detect() self.assertTrue(status.running) + self.assertTrue(status.remote) + self.assertIn("faster", status.remote_urls) class ServerRunningTests(unittest.TestCase): @@ -212,5 +222,68 @@ class ServerRunningTests(unittest.TestCase): self.assertFalse(common.server_running("")) +class RemoteUrlTests(unittest.TestCase): + """backends.common.normalize_remote_url: host:port / URL -> http(s)://.""" + + def test_bare_host_port_gets_http_scheme(self): + from backends import common + self.assertEqual(common.normalize_remote_url("10.0.0.5:8080"), + "http://10.0.0.5:8080") + + def test_full_url_preserved(self): + from backends import common + self.assertEqual(common.normalize_remote_url( + "https://10.0.0.5:8443/path"), "https://10.0.0.5:8443/path") + + def test_empty_means_disabled(self): + from backends import common + self.assertEqual(common.normalize_remote_url(""), "") + self.assertEqual(common.normalize_remote_url(" "), "") + + def test_whitespace_stripped(self): + from backends import common + self.assertEqual(common.normalize_remote_url(" 10.0.0.5:8080 "), + "http://10.0.0.5:8080") + + def test_invalid_rejected(self): + from backends import common + for value in ("http://", "not a url", "10.0.0.5:notaport", "://"): + with self.assertRaises(ValueError, msg=value): + common.normalize_remote_url(value) + + +class RemoteSuppressionTests(unittest.TestCase): + """A server this tool started must not also be reported as remote.""" + + def test_audiocpp_own_server_suppresses_remote(self): + from backends import audiocpp + from backends import servers as servers_mod + 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") + (Path(td) / "audiocpp-server.pid").write_text( + "4242", encoding="utf-8") + with patch.object(audiocpp, "find_local_checkout", + return_value=checkout), \ + patch.object(servers_mod, "LOG_DIR", Path(td)), \ + patch.object(servers_mod, "_pid_alive", + return_value=True), \ + patch.object(audiocpp.probe, "identify_server", + return_value="audiocpp"): + status = audiocpp.detect() + self.assertTrue(status.managed) + self.assertTrue(status.running) + self.assertFalse(status.remote) + self.assertEqual(status.remote_urls, {}) + + if __name__ == "__main__": unittest.main() |