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
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
|
"""Tests for the CLI's managed-server bootstrap (backends/managed.py).
The CLI (without --api-url) calls ``ensure_running`` before converting and
``ManagedServer.shutdown`` after; both are tested against mocked
``servers``/``probe``/registry-detect so no process is ever spawned. The
qwen paths exercise the real ``build_spec``/``model_for_voice_mode``
mapping (pure config reads) to pin the model-per-voice-mode contract.
"""
import io
import unittest
from contextlib import redirect_stdout
from types import SimpleNamespace
from unittest.mock import ANY, patch
from backends import ServerSpec, managed, servers
from backends.managed import ManagedServer, ensure_running
from backends.probe import (IDENTITY_AUDIOCPP, IDENTITY_QWEN_CLONE,
IDENTITY_QWEN_CUSTOM, IDENTITY_QWEN_DESIGN)
from converter.clients import (BACKEND_AUDIOCPP, BACKEND_QWEN,
VOICE_MODE_CLONE, VOICE_MODE_CUSTOM,
VOICE_MODE_DESIGN)
def _spec(name="audiocpp", url="http://127.0.0.1:8080", identity=None):
return ServerSpec(name, url, ["/bin/fake_server"], identity=identity)
def _status(specs, installed=True, label="audio.cpp"):
"""A minimal stand-in for the registry's BackendStatus."""
return SimpleNamespace(servers=specs, installed=installed, label=label)
class ShutdownTests(unittest.TestCase):
"""ManagedServer.shutdown stops only what the run started."""
def test_noop_when_not_started(self):
with patch.object(servers, "stop") as mk_stop:
ManagedServer(_spec()).shutdown()
mk_stop.assert_not_called()
def test_stops_the_spec_when_started(self):
spec = _spec()
with patch.object(servers, "stop") as mk_stop:
ManagedServer(spec, started=True).shutdown()
mk_stop.assert_called_once_with(spec.name)
class EnsureRunningTests(unittest.TestCase):
"""ensure_running resolves the spec and boots or reuses the server."""
def setUp(self):
self.spec = _spec(identity=IDENTITY_AUDIOCPP)
def _run(self, backend=BACKEND_AUDIOCPP, voice_mode="custom_voice"):
out = io.StringIO()
with redirect_stdout(out):
result = ensure_running(backend, voice_mode)
return result, out.getvalue()
def test_not_installed_returns_none_with_warning(self):
with patch("backends.detect",
return_value=_status([self.spec], installed=False)), \
patch.object(servers, "start") as mk_start:
result, output = self._run()
self.assertIsNone(result)
self.assertIn("not installed", output)
mk_start.assert_not_called()
def test_unknown_backend_returns_none_with_warning(self):
with patch("backends.detect", return_value=None), \
patch.object(servers, "start") as mk_start:
result, output = self._run(backend="gone")
self.assertIsNone(result)
self.assertIn("not installed", output)
mk_start.assert_not_called()
def test_no_specs_returns_none_with_warning(self):
with patch("backends.detect",
return_value=_status([], installed=True)), \
patch.object(servers, "start") as mk_start:
result, _ = self._run()
self.assertIsNone(result)
mk_start.assert_not_called()
def test_starts_when_port_free(self):
with patch("backends.detect",
return_value=_status([self.spec])), \
patch("backends.common.server_running",
return_value=False), \
patch.object(servers, "start", return_value=True) as mk_start:
result, _ = self._run()
mk_start.assert_called_once_with(self.spec, progress=ANY)
self.assertTrue(result.ok)
self.assertTrue(result.started)
self.assertEqual(result.spec, self.spec)
def test_start_failure_reports_not_ok_and_never_stops(self):
with patch("backends.detect",
return_value=_status([self.spec])), \
patch("backends.common.server_running",
return_value=False), \
patch.object(servers, "start", return_value=False), \
patch.object(servers, "stop") as mk_stop:
result, _ = self._run()
result.shutdown()
self.assertFalse(result.ok)
self.assertFalse(result.started)
mk_stop.assert_not_called()
def test_running_server_is_reused_and_left_running(self):
with patch("backends.detect",
return_value=_status([self.spec])), \
patch("backends.common.server_running",
return_value=True), \
patch.object(servers, "start") as mk_start:
result, _ = self._run()
result.shutdown()
mk_start.assert_not_called()
self.assertTrue(result.ok)
self.assertFalse(result.started)
def test_running_event_marks_the_run_as_a_reuser(self):
"""A server appearing under us between the port check and the spawn.
``servers.start`` reports "running" instead of spawning in that
race; the run must then not stop the server at shutdown.
"""
seen = []
def fake_start(spec, progress=None, cancel=None):
event = {"kind": "running", "name": spec.name, "url": spec.url}
seen.append(event)
if progress is not None:
progress(event)
return True
with patch("backends.detect",
return_value=_status([self.spec])), \
patch("backends.common.server_running",
return_value=False), \
patch.object(servers, "start", side_effect=fake_start), \
patch.object(servers, "_console_progress") as mk_print, \
patch.object(servers, "stop") as mk_stop:
result, _ = self._run()
result.shutdown()
self.assertTrue(result.ok)
self.assertFalse(result.started)
# The console still hears the event (delegated printer).
mk_print.assert_called_once_with(seen[0])
mk_stop.assert_not_called()
def test_keyboard_interrupt_stops_the_spawned_server(self):
with patch("backends.detect",
return_value=_status([self.spec])), \
patch("backends.common.server_running",
return_value=False), \
patch.object(servers, "start",
side_effect=KeyboardInterrupt), \
patch.object(servers, "pid_for", return_value=4242), \
patch.object(servers, "stop") as mk_stop, \
self.assertRaises(KeyboardInterrupt):
self._run()
mk_stop.assert_called_once_with(self.spec.name)
def test_keyboard_interrupt_without_spawn_skips_the_stop(self):
with patch("backends.detect",
return_value=_status([self.spec])), \
patch("backends.common.server_running",
return_value=False), \
patch.object(servers, "start",
side_effect=KeyboardInterrupt), \
patch.object(servers, "pid_for", return_value=None), \
patch.object(servers, "stop") as mk_stop, \
self.assertRaises(KeyboardInterrupt):
self._run()
mk_stop.assert_not_called()
class QwenEnsureRunningTests(unittest.TestCase):
"""qwen hosts one model per server: the running-model check is aware."""
def _detect_qwen(self):
from backends import qwen
return _status([qwen.build_spec("CustomVoice")],
installed=True, label="qwen-tts")
def _run(self, voice_mode, backend=BACKEND_QWEN):
out = io.StringIO()
with redirect_stdout(out):
result = ensure_running(backend, voice_mode)
return result, out.getvalue()
def test_spec_aims_at_the_model_the_voice_mode_needs(self):
from backends import qwen
cases = [(VOICE_MODE_CLONE, "Base", IDENTITY_QWEN_CLONE),
(VOICE_MODE_CUSTOM, "CustomVoice", IDENTITY_QWEN_CUSTOM),
(VOICE_MODE_DESIGN, "VoiceDesign", IDENTITY_QWEN_DESIGN)]
for voice_mode, model, identity in cases:
with self.subTest(voice_mode=voice_mode):
with patch("backends.detect", return_value=self._detect_qwen()), \
patch("backends.common.server_running",
return_value=False), \
patch.object(servers, "start",
return_value=True) as mk_start:
result, _ = self._run(voice_mode)
spec = mk_start.call_args.args[0]
self.assertEqual(spec.identity, identity)
self.assertIn(qwen.MODEL_REPOS[model], spec.argv)
self.assertTrue(result.started)
def test_running_server_hosting_the_wanted_model_is_reused(self):
with patch("backends.detect", return_value=self._detect_qwen()), \
patch("backends.common.server_running", return_value=True), \
patch("backends.probe.identify_server",
return_value=IDENTITY_QWEN_CLONE), \
patch.object(servers, "start") as mk_start, \
patch.object(servers, "stop") as mk_stop:
result, _ = self._run(VOICE_MODE_CLONE)
result.shutdown()
mk_start.assert_not_called()
mk_stop.assert_not_called()
self.assertTrue(result.ok)
self.assertFalse(result.started)
def test_managed_server_hosting_another_model_is_rebooted(self):
with patch("backends.detect", return_value=self._detect_qwen()), \
patch("backends.common.server_running", return_value=True), \
patch("backends.probe.identify_server",
return_value=IDENTITY_QWEN_CUSTOM), \
patch.object(servers, "alive", return_value=True), \
patch.object(servers, "start", return_value=True) as mk_start, \
patch.object(servers, "stop") as mk_stop:
result, output = self._run(VOICE_MODE_CLONE)
self.assertIn("restarting", output)
mk_stop.assert_called_once_with("qwen")
spec = mk_start.call_args.args[0]
self.assertEqual(spec.identity, IDENTITY_QWEN_CLONE)
self.assertTrue(result.started)
# The rebooted server is this run's: shutdown stops it again.
with patch.object(servers, "stop") as mk_stop:
result.shutdown()
mk_stop.assert_called_once_with("qwen")
def test_foreign_server_hosting_another_model_refuses_the_run(self):
with patch("backends.detect", return_value=self._detect_qwen()), \
patch("backends.common.server_running", return_value=True), \
patch("backends.probe.identify_server",
return_value=IDENTITY_QWEN_CUSTOM), \
patch.object(servers, "alive", return_value=False), \
patch.object(servers, "start") as mk_start, \
patch.object(servers, "stop") as mk_stop:
result, output = self._run(VOICE_MODE_CLONE)
self.assertFalse(result.ok)
self.assertIn("this run needs Base", output)
mk_start.assert_not_called()
mk_stop.assert_not_called()
def test_unidentified_running_server_counts_as_unknown(self):
"""A port squatting service that probes as nothing known refuses."""
with patch("backends.detect", return_value=self._detect_qwen()), \
patch("backends.common.server_running", return_value=True), \
patch("backends.probe.identify_server", return_value=None), \
patch.object(servers, "alive", return_value=False), \
patch.object(servers, "start") as mk_start:
result, output = self._run(VOICE_MODE_CLONE)
self.assertFalse(result.ok)
self.assertIn("an unknown", output)
mk_start.assert_not_called()
class ManagedModuleSmokeTests(unittest.TestCase):
"""Import-surface sanity for the module the CLI wires in."""
def test_managed_server_defaults(self):
server = ManagedServer(_spec())
self.assertFalse(server.started)
self.assertTrue(server.ok)
def test_module_uses_the_servers_module_singletons(self):
# managed delegates to the same servers module the TUI uses, so
# pid/log files and console output stay identical.
self.assertIs(managed.servers, servers)
|
