From 1ff9a635bd9b033b631a6b525891b7eb44e189d3 Mon Sep 17 00:00:00 2001 From: historia Date: Mon, 24 Aug 2026 16:08:33 -0400 Subject: feat: clearer split between local (managed) and remote URLs and server status --- app/tests/test_backends_probe.py | 104 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 104 insertions(+) create mode 100644 app/tests/test_backends_probe.py (limited to 'app/tests/test_backends_probe.py') diff --git a/app/tests/test_backends_probe.py b/app/tests/test_backends_probe.py new file mode 100644 index 0000000..08f9fd9 --- /dev/null +++ b/app/tests/test_backends_probe.py @@ -0,0 +1,104 @@ +"""Tests for backends.probe: identifying which backend answers at a URL.""" + +import json +import unittest +from unittest.mock import patch + +from backends import probe + + +class _FakeResponse: + def __init__(self, payload): + self._payload = payload + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def read(self): + return json.dumps(self._payload).encode("utf-8") + + +class IdentifyServerTests(unittest.TestCase): + def _patch_http(self, routes): + """routes: URL path -> JSON payload dict (missing = HTTP error).""" + import urllib.parse + + def fake_urlopen(url, timeout=None): + path = urllib.parse.urlsplit(url).path + payload = routes.get(path) + if payload is None: + raise OSError("HTTP 404") + return _FakeResponse(payload) + + return patch.object(probe.urllib.request, "urlopen", + side_effect=fake_urlopen) + + def test_audiocpp_identified(self): + routes = {"/health": {"status": "ok"}, + "/v1/models": {"data": [{"id": "qwen", "family": "qwen3_tts", + "task": "tts"}]}} + with patch.object(probe.common, "server_running", return_value=True), \ + self._patch_http(routes): + self.assertEqual(probe.identify_server("http://127.0.0.1:8080"), + "audiocpp") + + def test_faster_identified(self): + with patch.object(probe.common, "server_running", return_value=True), \ + self._patch_http({"/health": {"model_loaded": True}}): + self.assertEqual(probe.identify_server("http://127.0.0.1:8000"), + "faster") + + def test_qwen_custom_and_clone_identified(self): + with patch.object(probe.common, "server_running", return_value=True), \ + self._patch_http({"/info": {"named_endpoints": + {"/run_instruct": {}}}}): + self.assertEqual(probe.identify_server("http://x:7860"), + "qwen-custom") + with patch.object(probe.common, "server_running", return_value=True), \ + self._patch_http({"/info": {"named_endpoints": + {"/run_voice_clone": {}}}}): + self.assertEqual(probe.identify_server("http://x:7861"), + "qwen-clone") + + def test_unreachable_returns_none(self): + with patch.object(probe.common, "server_running", return_value=False): + self.assertIsNone(probe.identify_server("http://127.0.0.1:8080")) + + def test_health_ok_without_models_catalog_is_not_audiocpp(self): + # A service that answers {"status": "ok"} but not /v1/models is not + # recognized as audio.cpp. + with patch.object(probe.common, "server_running", return_value=True), \ + self._patch_http({"/health": {"status": "ok"}}): + self.assertIsNone(probe.identify_server("http://x")) + + def test_empty_url_returns_none(self): + self.assertIsNone(probe.identify_server("")) + self.assertIsNone(probe.identify_server(None)) + + +class SameEndpointTests(unittest.TestCase): + def test_same_host_port(self): + self.assertTrue(probe.same_endpoint("http://127.0.0.1:8080", + "http://127.0.0.1:8080/")) + + def test_scheme_ignored(self): + self.assertTrue(probe.same_endpoint("http://h:8080", "https://h:8080")) + + def test_different_port(self): + self.assertFalse(probe.same_endpoint("http://127.0.0.1:8080", + "http://127.0.0.1:8000")) + + def test_different_host(self): + self.assertFalse(probe.same_endpoint("http://127.0.0.1:8080", + "http://10.0.0.5:8080")) + + def test_empty_url(self): + self.assertFalse(probe.same_endpoint("", "http://h:8080")) + self.assertFalse(probe.same_endpoint("http://h:8080", "")) + + +if __name__ == "__main__": + unittest.main() -- cgit v1.2.3