1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
|
"""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_qwen_design_identified(self):
with patch.object(probe.common, "server_running", return_value=True), \
self._patch_http({"/info": {"named_endpoints":
{"/run_voice_design": {}}}}):
self.assertEqual(probe.identify_server("http://x:7860"),
"qwen-design")
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()
|
