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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
"""Tests for the backends package registry and detection aggregation."""
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from backends import REGISTRY, detect_all, get
class RegistryTests(unittest.TestCase):
def setUp(self):
# The registry is built lazily on first access (the backend modules
# pull in converter.tts and its deps, which are only available inside
# the managed venv). Trigger the build so these tests don't depend on
# another test class having called detect_all() first.
get("audiocpp")
def test_registry_has_the_three_backends(self):
keys = [info.key for info in REGISTRY]
self.assertEqual(keys, ["audiocpp", "qwen", "faster"])
def test_every_entry_has_detect_and_setup_tui(self):
for info in REGISTRY:
self.assertTrue(callable(info.detect), info.key)
self.assertTrue(callable(info.setup_tui), info.key)
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):
self.assertIs(get("audiocpp").key, "audiocpp")
self.assertIsNone(get("nonexistent"))
class DetectAllTests(unittest.TestCase):
def test_detect_all_returns_one_status_per_backend(self):
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:
self.assertIn(s.key, ("audiocpp", "qwen", "faster"))
# ready requires both installed and configured; on a clean
# 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:
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")
from backends import audiocpp
with patch.object(audiocpp, "find_local_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), \
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), \
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:
checkout = Path(td) / "faster-qwen3-tts"
(checkout / "examples").mkdir(parents=True)
(checkout / "examples" / "openai_server.py").write_text("x")
(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("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):
# Pick an unused port by opening + closing a socket, then probe it.
import socket
from backends import common
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()
|
