diff options
Diffstat (limited to 'app')
| -rw-r--r-- | app/backends/managed.py | 162 | ||||
| -rw-r--r-- | app/backends/qwen.py | 26 | ||||
| -rw-r--r-- | app/tests/test_audiobook_cli.py | 135 | ||||
| -rw-r--r-- | app/tests/test_backends.py | 2 | ||||
| -rw-r--r-- | app/tests/test_backends_managed.py | 283 | ||||
| -rw-r--r-- | app/ui/hub.py | 2 |
6 files changed, 600 insertions, 10 deletions
diff --git a/app/backends/managed.py b/app/backends/managed.py new file mode 100644 index 0000000..3b98aea --- /dev/null +++ b/app/backends/managed.py @@ -0,0 +1,162 @@ +"""Start and stop a managed TTS server around a CLI conversion run. + +The TUI owns the server lifecycle through its Generate flow (the hub's +autostart plan boots the server in the run view, and the stop-and-exit +toggle shuts it down afterwards). This module gives the CLI the same +capability: when ``audiobook.py`` runs without ``--api-url`` it calls +``ensure_running`` to boot the selected backend's managed server — the +instance installed through the TUI — converts against it, and stops it +again afterwards. Only a server this run started is ever stopped: one +found already answering at the backend's configured URL is used as-is and +left running when the run ends, whoever started it. + +The qwen backend hosts one model per process, so its "already running" +check is model-aware: a managed demo hosting another model than the run +needs is stopped and rebooted with the right one (the TUI's Generate form +does the same), while a foreign server with the wrong model refuses the +run with an actionable message instead of letting the conversion fail +against the wrong endpoints. The other backends host all their models in +one server process, so any server answering at the configured URL is +usable as-is. + +All server boot/stop console output comes from ``backends.servers`` (the +same lines the TUI's console tail prints); this module only adds the +decisions around them. Like ``servers`` it is presentation-agnostic +enough to run in the plain console — the CLI's only caller. +""" + +from dataclasses import dataclass +from typing import Optional + +from backends import ServerSpec, common, probe, servers +from converter.clients import BACKEND_QWEN + + +@dataclass +class ManagedServer: + """The server-lifecycle outcome of one conversion run's boot phase. + + SPEC is the server spec the conversion targets (its URL is the + endpoint the converter uses when no ``--api-url`` overrides it). + STARTED is True when this run spawned (or rebooted) the process — + only then does ``shutdown`` stop it; a server found already running + belongs to whoever launched it and is left alone. OK is False when + the boot failed or was refused: the caller must not convert, and + ``shutdown`` is a no-op (``servers.start`` owns any pid file left + behind, mirroring the TUI's behavior for a failed boot). + """ + + spec: ServerSpec + started: bool = False + ok: bool = True + + def shutdown(self) -> None: + """Stop the server when this run started it (no-op otherwise).""" + if self.started: + servers.stop(self.spec.name) + + +def ensure_running(backend: str, voice_mode: str) -> Optional[ManagedServer]: + """Make the backend's managed server ready for a conversion run. + + Resolves the server spec for BACKEND (qwen: the demo hosting the + model VOICE_MODE needs; the others: their single configured spec), + then starts it when its port is free — waiting out the boot and + streaming ``servers``' console progress — or reuses the server + already answering there (qwen: restarting a managed server that hosts + another model, refusing a foreign one). Returns the run's + ``ManagedServer`` (call ``shutdown`` when the conversion is over), or + None when the backend is not installed here and nothing can be + started: the caller proceeds unmanaged, since a foreign server at the + configured endpoint may still answer and otherwise the conversion + fails with the converter's own unreachable-server message. + + Raises KeyboardInterrupt when the boot poll is interrupted (after + stopping a server this call spawned, so nothing is left loading). + """ + from backends import detect as _registry_detect + + status = _registry_detect(backend) + if status is None or not status.servers or not status.installed: + label = status.label if status is not None else backend + print(f"[WARNING] {label} is not installed — cannot start a server " + "automatically; the conversion will use the configured " + "endpoint (see the TUI's Configure Backends to install it).") + return None + spec = _spec_for(status, backend, voice_mode) + return _boot(spec, backend, voice_mode) + + +def _spec_for(status, backend: str, voice_mode: str) -> ServerSpec: + """The server spec this run needs, from STATUS's detected servers.""" + if backend != BACKEND_QWEN: + return status.servers[0] + # qwen hosts one model per process: aim the spec at the model this + # run selected rather than the default one detect() reports. + from backends import qwen + return qwen.build_spec(qwen.model_for_voice_mode(voice_mode)) + + +def _boot(spec: ServerSpec, backend: str, voice_mode: str) -> ManagedServer: + """Start or reuse the server SPEC describes, per the run's needs.""" + wanted_model = None + if backend == BACKEND_QWEN: + from backends import qwen + wanted_model = qwen.model_for_voice_mode(voice_mode) + + if common.server_running(spec.url): + if wanted_model is None: + print(f"[INFO] using the {spec.name} server already running " + f"at {spec.url}") + return ManagedServer(spec) + running_model = qwen.model_for_identity( + probe.identify_server(spec.url)) + if running_model == wanted_model: + print(f"[INFO] using the {spec.name} server already running " + f"at {spec.url} (hosting {wanted_model})") + return ManagedServer(spec) + if not servers.alive(spec.name): + print(f"[ERROR] a server this tool did not start is running at " + f"{spec.url} hosting {running_model or 'an unknown'} — " + f"this run needs {wanted_model}. Stop that server first, " + "or adjust the voice flags to use the hosted model.") + return ManagedServer(spec, ok=False) + # Ours: stop it and boot the newly-selected model on the same + # port (the TUI's Generate form restarts a managed server the + # same way when the run's model selection changes). + print(f"[INFO] restarting the {spec.name} server to host " + f"{wanted_model}...") + servers.stop(spec.name) + return _start(spec) + + +def _start(spec: ServerSpec) -> ManagedServer: + """Spawn SPEC's server and wait for readiness (``servers.start``). + + STARTED is recorded only when ``servers.start`` actually spawned the + process — its "running" event (a server appeared under us between the + caller's port check and the spawn) marks the run as a reuser, so + ``shutdown`` never stops a server this run did not start. A failed + boot reports False and leaves any pid file in place, exactly like the + TUI's boot path. + """ + spawned = {"yes": True} + + def _progress(event: dict) -> None: + if event.get("kind") == "running": + spawned["yes"] = False + # Same-package reuse of the console printer: without it a custom + # progress callback would silence the CLI's boot output. + servers._console_progress(event) + + try: + ok = servers.start(spec, progress=_progress) + except KeyboardInterrupt: + # Ctrl-C while waiting out the boot: kill what we spawned so no + # half-booted server is left loading in the background. Without a + # pid file nothing was spawned (or it exited already) — skip the + # noisy "not started by this tool" notice. + if servers.pid_for(spec.name) is not None: + servers.stop(spec.name) + raise + return ManagedServer(spec, started=spawned["yes"] and ok, ok=ok) diff --git a/app/backends/qwen.py b/app/backends/qwen.py index a4cddb5..a94b3cc 100644 --- a/app/backends/qwen.py +++ b/app/backends/qwen.py @@ -44,7 +44,8 @@ from backends import ( setup, ) from converter import config -from converter.clients import QWEN3_TTS_SPEAKERS +from converter.clients import (QWEN3_TTS_SPEAKERS, VOICE_MODE_CLONE, + VOICE_MODE_CUSTOM, VOICE_MODE_DESIGN) from ui import taskview, tui QWEN_PIP_PKG = "qwen-tts" @@ -191,6 +192,18 @@ def model_for_identity(identity: Optional[str]) -> Optional[str]: return IDENTITY_TO_MODEL.get(identity) +def model_for_voice_mode(voice_mode: str) -> str: + """The MODEL_REPOS key a run with VOICE_MODE needs hosted. + + The mirror of ``model_for_identity`` for a planned (not yet running) + run: instructions design the voice (VoiceDesign), a reference .wav + clones (Base), otherwise built-in speakers (CustomVoice). + """ + return {VOICE_MODE_DESIGN: "VoiceDesign", + VOICE_MODE_CLONE: "Base", + VOICE_MODE_CUSTOM: "CustomVoice"}[voice_mode] + + def desired_identity(model: str) -> str: """The probe identity the model's demo answers as (used while booting).""" return { @@ -283,8 +296,13 @@ def build_parser() -> argparse.ArgumentParser: return parser -def _build_spec(model: str) -> ServerSpec: - """The single managed ServerSpec hosting MODEL on the configured port.""" +def build_spec(model: str) -> ServerSpec: + """The single managed ServerSpec hosting MODEL on the configured port. + + Public because callers outside this module (the hub's run preparation + and the CLI's managed-server bootstrap) need to boot exactly the model + their run selected, which can differ from the default one. + """ url = config.QWEN_API_URL return ServerSpec( "qwen", url, @@ -325,7 +343,7 @@ def detect() -> BackendStatus: "not installed — run setup to pip install qwen-tts") details.append(f"port: {_config_port(url, DEFAULT_PORT)}") details.append(f"default model: {model}") - specs = [_build_spec(model)] + specs = [build_spec(model)] managed = servers.manages(specs) # A locally-managed server names its running model via the probe of the # managed URL; a remotely-run demo names it via the remote-URL probe. diff --git a/app/tests/test_audiobook_cli.py b/app/tests/test_audiobook_cli.py index d947c5d..f13cf9e 100644 --- a/app/tests/test_audiobook_cli.py +++ b/app/tests/test_audiobook_cli.py @@ -1,12 +1,13 @@ -"""Tests for the audiobook.py CLI — single-book flags and arg validation. +"""Tests for the audiobook.py CLI — single-book flags, arg validation, and +the managed-server wiring. audiobook.py lives at the repo root (one level above app/), so the tests bootstrap the root onto sys.path to import it. main() runs with the managed-environment bootstrap stubbed (it would otherwise re-exec the process into envs/tts) and convert() mocked, asserting only argparse -behavior and what reaches convert(); convert()'s single-book wiring and -the pre-flight overrides are tested against the real functions with -temporary directories. +behavior and what reaches convert(); convert()'s single-book wiring, the +pre-flight overrides, and the manage_server lifecycle are tested against +the real functions with temporary directories. """ import contextlib @@ -250,6 +251,23 @@ class MainHappyPathTests(MainTestCase): self.assertIsNone(kwargs["input_file"]) self.assertIsNone(kwargs["output_file"]) + def test_manage_server_requested_without_api_url(self): + # Without --api-url the CLI asks convert() to manage the server + # lifecycle (convert() performs the actual boot/stop). + code, _, convert = self.run_main([]) + self.assertEqual(code, 0) + self.assertIs(convert.call_args.kwargs["manage_server"], True) + + def test_api_url_run_still_carries_the_manage_flag(self): + # The flag travels too; convert() itself skips management when an + # explicit api_url targets an external server. + code, _, convert = self.run_main( + ["--api-url", "10.20.30.40:8080"]) + self.assertEqual(code, 0) + self.assertEqual(convert.call_args.kwargs["api_url"], + "http://10.20.30.40:8080") + self.assertIs(convert.call_args.kwargs["manage_server"], True) + class ConvertWiringTests(unittest.TestCase): """convert() turns the single-book flags into the pre-flight overrides.""" @@ -314,6 +332,115 @@ class ConvertWiringTests(unittest.TestCase): self._convert(output_file=self.tmp / "dune.mp3") +class ManagedServerWiringTests(unittest.TestCase): + """convert(manage_server=True) boots and stops the server around the run. + + The lifecycle decisions live in backends.managed (tested there); these + pin convert()'s wiring: when the boot happens relative to pre-flight + and the conversion, that shutdown runs even on failure or Ctrl-C, and + that an external api_url or the hub's progress-callback path never + touch the server. + """ + + def setUp(self): + self.tmp = Path(tempfile.mkdtemp(prefix="audiobook_managed_")) + self.addCleanup(shutil.rmtree, self.tmp, True) + self.book = _make_book(self.tmp) + self._old_folders = (converter_mod.BOOKS_FOLDER, + converter_mod.AUDIOBOOKS_FOLDER) + self.addCleanup(self._restore_folders) + + def _restore_folders(self): + converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = \ + self._old_folders + + def _convert(self, *, server_ok=True, run_result=True, run_raises=None, + preflight_result=None, **kwargs): + """Run convert() with the managed-server and converter mocked. + + Returns (code, ensure mock, server mock, events, preflight mock); + EVENTS records the order of ensure_running / run / shutdown. + """ + kwargs.setdefault("backend", "audiocpp") + kwargs.setdefault("manage_server", True) + if preflight_result is None: + preflight_result = ([self.book], [(self.book, "dune")]) + preflight = MagicMock(return_value=preflight_result) + fake_instance = MagicMock() + events = [] + + def _run(): + events.append("run") + if run_raises is not None: + raise run_raises + return run_result + fake_instance.run.side_effect = _run + fake_class = MagicMock(return_value=fake_instance) + fake_class.preflight_overwrites = preflight + server = MagicMock() + server.ok = server_ok + server.shutdown.side_effect = lambda: events.append("shutdown") + ensure = MagicMock(return_value=server) + + def _ensure(backend, voice_mode): + events.append(("ensure", backend, voice_mode)) + return server + ensure.side_effect = _ensure + with patch.object(audiobook, "setup_logging"), \ + patch.object(audiobook, "setup_directories"), \ + patch.object(audiobook, "AudiobookConverter", fake_class), \ + patch("backends.managed.ensure_running", ensure): + code = audiobook.convert(**kwargs) + return code, ensure, server, events, preflight + + def test_server_boots_before_the_run_and_stops_after(self): + code, ensure, _, events, _ = self._convert() + self.assertEqual(code, 0) + self.assertEqual(events, [("ensure", "audiocpp", "custom_voice"), + "run", "shutdown"]) + + def test_qwen_run_needs_its_voice_mode_model(self): + _, ensure, _, events, _ = self._convert( + backend="qwen", clone="ref.wav") + self.assertEqual(events[0], ("ensure", "qwen", "voice_clone")) + + def test_not_ok_boot_stops_before_converting(self): + code, ensure, server, events, _ = self._convert(server_ok=False) + self.assertEqual(code, 1) + self.assertEqual(events, [("ensure", "audiocpp", "custom_voice"), + "shutdown"]) + + def test_shutdown_runs_when_the_conversion_fails(self): + code, _, _, events, _ = self._convert( + run_raises=RuntimeError("server unreachable")) + self.assertEqual(code, 1) + self.assertEqual(events, [("ensure", "audiocpp", "custom_voice"), + "run", "shutdown"]) + + def test_shutdown_runs_on_ctrl_c(self): + code, _, _, events, _ = self._convert(run_raises=KeyboardInterrupt) + self.assertEqual(code, 130) + self.assertEqual(events, [("ensure", "audiocpp", "custom_voice"), + "run", "shutdown"]) + + def test_no_management_for_an_explicit_api_url(self): + code, ensure, _, _, _ = self._convert( + api_url="http://10.20.30.40:8080") + self.assertEqual(code, 0) + ensure.assert_not_called() + + def test_no_management_for_the_hub_path(self): + # The run view boots/stops the server itself: manage_server False. + _, ensure, _, _, _ = self._convert(manage_server=False) + ensure.assert_not_called() + + def test_no_server_boot_when_nothing_to_convert(self): + code, ensure, _, _, preflight = self._convert( + preflight_result=([], [])) + self.assertEqual(code, 0) + ensure.assert_not_called() + + class PreflightOverrideTests(unittest.TestCase): """preflight_overwrites honors the explicit book list and output name.""" diff --git a/app/tests/test_backends.py b/app/tests/test_backends.py index 64fd483..7a39880 100644 --- a/app/tests/test_backends.py +++ b/app/tests/test_backends.py @@ -169,7 +169,7 @@ class DetectAllTests(unittest.TestCase): for model, wanted in (("Base", IDENTITY_QWEN_CLONE), ("VoiceDesign", IDENTITY_QWEN_DESIGN)): with self.subTest(model=model): - spec = qwen._build_spec(model) + spec = qwen.build_spec(model) self.assertEqual(spec.identity, wanted) self.assertIn(qwen.MODEL_REPOS[model], spec.argv) diff --git a/app/tests/test_backends_managed.py b/app/tests/test_backends_managed.py new file mode 100644 index 0000000..cd1f03a --- /dev/null +++ b/app/tests/test_backends_managed.py @@ -0,0 +1,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) diff --git a/app/ui/hub.py b/app/ui/hub.py index f63ff99..5c312cf 100644 --- a/app/ui/hub.py +++ b/app/ui/hub.py @@ -1852,7 +1852,7 @@ def _prepare_run_config(backend: str, kwargs: dict # run selected (same URL/port, matching probe identity), so an # autostart or model-switch restart boots exactly what the run # needs instead of the Start/Stop menu's default model. - spec = qwen_backend._build_spec(_qwen_wanted_model(kwargs)) + spec = qwen_backend.build_spec(_qwen_wanted_model(kwargs)) return runview.RunConfig( backend=backend, backend_label=label, kwargs=kwargs, book_files=book_files, planned=planned, |