"""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)