"""Tests for the TUI hub (ui/hub.py) menu and helpers. The hub drives the same curses widgets as ui/tui.py, so these tests reuse the fake curses/screen from test_tui to run the menu without a terminal. """ import contextlib import io import json import tempfile import unittest from pathlib import Path from unittest.mock import patch from backends import BackendInfo, BackendStatus, ServerSpec from tests.test_tui import FakeCurses, FakeScreen from ui import hub, tui class _ScriptedTUI: """Stand-in for the tui widget module: answers each menu/line_edit/ confirm/form call from a scripted answer list and records every prompt.""" def __init__(self): self.script = [] self.prompts = [] self.options_seen = [] self.flashes = [] self.form_script = [] # values dicts / None / sentinel for tui.form self.forms_seen = [] # (title, fields, kwargs) for tui.form def _next(self, prompt, options=None): self.prompts.append(prompt) if options is not None: self.options_seen.append(options) return self.script.pop(0) def menu(self, stdscr, title, options, **kwargs): return self._next(title, options) def line_edit(self, stdscr, title, default, **kwargs): self.prompts.append(f"{title} [default: {default!r}]") return self.script.pop(0) def confirm(self, stdscr, question, **kwargs): return self._next(question) def form(self, stdscr, title, fields, **kwargs): self.forms_seen.append((title, fields, kwargs)) return self.form_script.pop(0) def flash(self, stdscr, text, kind="warn"): self.flashes.append(text) class HubHelperTests(unittest.TestCase): """Pure helpers in hub.py (no curses).""" def test_is_float(self): self.assertTrue(hub._is_float("1.0")) self.assertTrue(hub._is_float("2")) self.assertFalse(hub._is_float("abc")) self.assertFalse(hub._is_float("")) def test_list_voices_from_dir(self): with __import__("tempfile").TemporaryDirectory() as td: d = Path(td) (d / "Narrator.wav").write_bytes(b"x") (d / "Alpha.WAV").write_bytes(b"x") (d / "notes.txt").write_bytes(b"x") voices = hub._list_voices(str(d)) # Stems preserve case; sorting is case-insensitive. self.assertEqual(voices, ["Alpha", "Narrator"]) def test_list_voices_missing_dir(self): self.assertEqual(hub._list_voices("/no/such/dir"), []) def test_status_mark(self): from backends import BackendStatus local = BackendStatus("k", "l", installed=True, configured=True, running=True, managed=True) remote = BackendStatus("k", "l", installed=True, configured=True, running=True, remote=True) both = BackendStatus("k", "l", installed=True, configured=True, running=True, managed=True, remote=True) models = BackendStatus("k", "l", installed=True, configured=True, running=True, managed=True, running_models=["Base", "CustomVoice"]) remote_models = BackendStatus("k", "l", installed=True, configured=True, running=True, remote=True, running_models=["Base"]) installed = BackendStatus("k", "l", installed=True, configured=False) none = BackendStatus("k", "l", installed=False, configured=False) downloaded = BackendStatus("k", "l", installed=False, configured=False, partial="downloaded (not built)") built_unconfigured = BackendStatus("k", "l", installed=True, configured=False, partial="built (not configured)") # running beats installed (a server is up even if not configured); # only a backend that is neither installed nor running is dimmed. self.assertEqual(hub._status_mark(local), ("running [local]", "ok", "body")) # A server this tool did not start, found at the remote URL. self.assertEqual(hub._status_mark(remote), ("running [remote]", "ok", "body")) self.assertEqual(hub._status_mark(both), ("running [local, remote]", "ok", "body")) # Multi-model backends name the models that answered. self.assertEqual(hub._status_mark(models), ("running [local] (Base, CustomVoice)", "ok", "body")) self.assertEqual(hub._status_mark(remote_models), ("running [remote] (Base)", "ok", "body")) self.assertEqual(hub._status_mark(installed), ("installed", "warn", "body")) self.assertEqual(hub._status_mark(none), ("unavailable", "err", "dim")) self.assertEqual(hub._status_mark(None), ("unavailable", "err", "dim")) # Part-way states: amber text; the name is dimmed while the backend # is still unusable (not installed), bright once it is built. self.assertEqual(hub._status_mark(downloaded), ("downloaded (not built)", "warn", "dim")) self.assertEqual(hub._status_mark(built_unconfigured), ("built (not configured)", "warn", "body")) class HubMenuTests(unittest.TestCase): """Drive the hub's screen stack with a fake screen (no terminal).""" def setUp(self): tui._THEME.clear() self.curses = FakeCurses() from unittest.mock import patch as _patch self._patcher = _patch.dict("sys.modules", {"curses": self.curses}) self._patcher.start() self.addCleanup(self._patcher.stop) self.addCleanup(tui._THEME.clear) def _none_status(self, key="k", label="l"): from backends import BackendStatus return BackendStatus(key, label, installed=False, configured=False) def test_quit_returns_none_when_no_backend(self): # No backends installed/running: menu is [Configure backends, # Settings, Quit]. Quit is the 3rd option (Down twice) then Enter. screen = FakeScreen(keys=[FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, 10]) with patch.object(hub, "detect_all", return_value=[]): result = hub._Hub(screen).run() self.assertIsNone(result) def test_run_prints_post_tui_notices_after_session(self): # The TUI runs in curses, so setup steps queue notices for the # console; hub.run must print them once the session ends. def fake_app(stdscr): hub.common.record_post_tui_notice( "[ERROR] audio.cpp build failed (exit code 2).\n" " Build log: /tmp/audiocpp_build_20260101_000000.log") hub.common.record_post_tui_notice("second notice") def fake_wrapper(func, *args, **kwargs): func(None) return 0 buffer = io.StringIO() self.curses.wrapper = fake_wrapper with patch.object(hub, "_app", fake_app), \ contextlib.redirect_stdout(buffer): rc = hub.run() self.assertEqual(rc, 0) out = buffer.getvalue() self.assertIn("[ERROR] audio.cpp build failed (exit code 2).", out) self.assertIn("Build log: /tmp/audiocpp_build_20260101_000000.log", out) self.assertIn("second notice", out) self.assertEqual(hub.common.drain_post_tui_notices(), []) def test_menu_has_only_configure_settings_and_quit_without_backends(self): # Capture the options handed to tui.menu: with nothing installed or # running, Convert/Server must be absent. captured = {} def fake_menu(stdscr, title, options, **kwargs): captured["options"] = options return "quit" screen = FakeScreen() with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "detect_all", return_value=[]): hub._Hub(screen).run() labels = [label for label, _ in captured["options"]] self.assertEqual(labels, ["Configure backends", "Settings", "Quit"]) def test_menu_has_all_five_when_one_installed(self): captured = {} def fake_menu(stdscr, title, options, **kwargs): captured["options"] = options captured["rows"] = kwargs.get("table_rows") return "quit" screen = FakeScreen() st = self._none_status("qwen", "qwen-tts") st.installed = True with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "detect_all", return_value=[st]): hub._Hub(screen).run() labels = [label for label, _ in captured["options"]] self.assertEqual( labels, ["Convert books", "Configure backends", "Start/Stop Backend Servers", "Settings", "Quit"]) # The status table is passed through, one row per backend. self.assertEqual(captured["rows"], [("qwen-tts", "installed", "warn", "body")]) def test_table_dims_name_when_not_installed_and_not_running(self): captured = {} def fake_menu(stdscr, title, options, **kwargs): captured["rows"] = kwargs.get("table_rows") return "quit" screen = FakeScreen() dead = self._none_status("audiocpp", "audio.cpp") external = self._none_status("qwen", "qwen-tts") external.running = True external.remote = True with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "detect_all", return_value=[dead, external]): hub._Hub(screen).run() # Unusable backend: dim name. Running-but-not-installed stays # bright and is tagged remote (found at its remote URL). self.assertEqual( captured["rows"], [("audio.cpp", "unavailable", "err", "dim"), ("qwen-tts", "running [remote]", "ok", "body")]) def test_menu_hides_server_when_only_running(self): # Running but not installed (an external server) still unlocks # Convert — but Start/Stop needs the backend on this machine. captured = {} def fake_menu(stdscr, title, options, **kwargs): captured["options"] = options return "quit" screen = FakeScreen() st = self._none_status("qwen", "qwen-tts") st.running = True with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "detect_all", return_value=[st]): hub._Hub(screen).run() labels = [label for label, _ in captured["options"]] self.assertEqual( labels, ["Convert books", "Configure backends", "Settings", "Quit"]) def test_ffmpeg_warning_shown_when_missing(self): # ffmpeg not on PATH → a red notice is passed above the table. captured = {} def fake_menu(stdscr, title, options, **kwargs): captured["notice_lines"] = kwargs.get("notice_lines") return "quit" screen = FakeScreen() with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "detect_all", return_value=[]), \ patch.object(hub.shutil, "which", return_value=None): hub._Hub(screen).run() self.assertEqual(captured["notice_lines"], [("Warning: ffmpeg not installed!", "err")]) def test_ffmpeg_warning_hidden_when_installed(self): # ffmpeg on PATH → no notice is passed at all. captured = {} def fake_menu(stdscr, title, options, **kwargs): captured["notice_lines"] = kwargs.get("notice_lines") return "quit" screen = FakeScreen() with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "detect_all", return_value=[]), \ patch.object(hub.shutil, "which", return_value="/usr/bin/ffmpeg"): hub._Hub(screen).run() self.assertIsNone(captured["notice_lines"]) def test_convert_with_no_available_backend_flashes(self): # Installed-but-not-ready backends → Convert is offered, but the # convert flow has nothing to list: it flashes a hint (no "Configure # backends" detour anymore) and returns to the main menu. Then # quit: 5 main-menu options, Quit is the 5th (Down x4). from backends import BackendStatus statuses = [BackendStatus("audiocpp", "audio.cpp", installed=True, configured=False), BackendStatus("qwen", "qwen-tts", installed=True, configured=False), BackendStatus("faster", "faster", installed=True, configured=False)] flashed = [] def fake_flash(stdscr, text, kind="warn"): flashed.append(text) with patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "flash", fake_flash): # Convert(Enter) → flash → main menu; Down x4 -> Quit, Enter. screen = FakeScreen(keys=[10, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, 10]) result = hub._Hub(screen).run() self.assertIsNone(result) self.assertEqual(len(flashed), 1) self.assertIn("No backend is ready", flashed[0]) class SubmenuStatusTableTests(unittest.TestCase): """First picker screen of every flow repeats the backend status table. Entries themselves stay clean: the configure-backends menu lists flat actions, and the Start/Stop menu offers only installed backends' servers, showing running/stopped inline instead of the table. """ def _capture_menu(self, captured): def fake_menu(stdscr, title, options, **kwargs): captured["title"] = title captured["options"] = options captured.update(kwargs) return tui.Wizard.BACK # Esc: back out immediately return fake_menu def _labels(self, options): """Option labels, skipping MENU_SEPARATOR divider rows.""" return [opt[0] for opt in options if opt is not tui.MENU_SEPARATOR] def _capture_form(self, captured): def fake_form(stdscr, title, fields, **kwargs): captured["title"] = title captured["fields"] = fields captured.update(kwargs) return tui.Wizard.BACK # Cancel: back out immediately return fake_form def test_configure_backends_menu_lists_actions_and_status_table(self): captured = {} infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0), BackendInfo("faster", "faster-qwen3-tts", lambda: None, lambda: 0)] statuses = [ BackendStatus("qwen", "qwen-tts", installed=True, configured=True), BackendStatus("faster", "faster-qwen3-tts", installed=False, configured=False, running=True, remote=True), ] with patch.object(hub, "REGISTRY", infos), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.shutil, "which", return_value="/usr/bin/ffmpeg"): result = hub._Hub(None).screen_configure() self.assertIs(result, tui.Wizard.BACK) # Configure (qwen installed), Install (faster uninstalled), then # Uninstall (qwen installed); no audio.cpp means no model actions. self.assertEqual([label for label, _ in captured["options"]], ["Configure qwen-tts", "Install Backend", "Uninstall Backend"]) # ...the shared status table carries the states instead. self.assertEqual(captured["table_title"], "Backend status") self.assertEqual( captured["table_rows"], [("qwen-tts", "installed", "warn", "body"), ("faster-qwen3-tts", "running [remote]", "ok", "body")]) self.assertIsNone(captured["notice_lines"]) def test_configure_backends_menu_install_only_when_nothing_installed(self): captured = {} infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0)] statuses = [BackendStatus("qwen", "qwen-tts", installed=False, configured=False)] with patch.object(hub, "REGISTRY", infos), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.shutil, "which", return_value="/x"): result = hub._Hub(None).screen_configure() self.assertIs(result, tui.Wizard.BACK) # Nothing installed: only the install entry is offered. self.assertEqual([label for label, _ in captured["options"]], ["Install Backend"]) def test_configure_backends_menu_audiocpp_model_actions(self): captured = {} infos = [BackendInfo("audiocpp", "audio.cpp", lambda: None, lambda: 0)] statuses = [BackendStatus("audiocpp", "audio.cpp", installed=True, configured=True)] with tempfile.TemporaryDirectory() as td: checkout = Path(td) (checkout / "server.json").write_text(json.dumps({ "models": [{"id": "present", "path": "models/present"}, {"id": "absent", "path": "models/absent"}], }), encoding="utf-8") (checkout / "models" / "present").mkdir(parents=True) (checkout / "models" / "present" / "m.gguf").write_bytes(b"x") binary = checkout / "build" / "linux-cuda-release" / "bin" binary.mkdir(parents=True) (binary / "audiocpp_server").write_bytes(b"x") with patch.object(hub, "REGISTRY", infos), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=checkout), \ patch.object(hub.shutil, "which", return_value="/x"): result = hub._Hub(None).screen_configure() self.assertIs(result, tui.Wizard.BACK) labels = self._labels(captured["options"]) # The missing-model download heads the menu as the recommended next # step (yellow suffix), separated from the rest by a blank line; # Configure + Uninstall follow. The backend is built, so no "Build" # action is offered. Deleting unused models now lives inside the # "Configure audio.cpp" wizard, not here. self.assertEqual( labels, ["Download Missing Models (audio.cpp)", "Configure audio.cpp", "Uninstall Backend"]) self.assertEqual(captured["options"][0], ("Download Missing Models (audio.cpp)", "download_models", ("[recommended]", "warn"))) self.assertIs(captured["options"][1], tui.MENU_SEPARATOR) def test_configure_backends_menu_offers_build_when_not_built(self): captured = {} infos = [BackendInfo("audiocpp", "audio.cpp", lambda: None, lambda: 0)] # installed=False (not built), but configured (server.json exists). statuses = [BackendStatus("audiocpp", "audio.cpp", installed=False, configured=True)] with tempfile.TemporaryDirectory() as td: checkout = Path(td) (checkout / "server.json").write_text(json.dumps({ "models": [{"id": "absent", "path": "models/absent"}], }), encoding="utf-8") with patch.object(hub, "REGISTRY", infos), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=checkout), \ patch.object(hub.shutil, "which", return_value="/x"): result = hub._Hub(None).screen_configure() self.assertIs(result, tui.Wizard.BACK) labels = self._labels(captured["options"]) # Not built → the Build action heads the menu as the recommended # next step (yellow suffix, blank separator below); Uninstall follows # (a downloaded checkout is removable). A downloaded-but-unbuilt # checkout is NOT installable, so no "Install Backend" entry, and the # model download stays hidden until the binary exists — Build and # Download never coexist. Configure needs an installed (built) # backend. self.assertEqual(labels, ["Build audio.cpp server", "Uninstall Backend"]) self.assertEqual(captured["options"][0], ("Build audio.cpp server", "build_audiocpp", ("[recommended]", "warn"))) self.assertIs(captured["options"][1], tui.MENU_SEPARATOR) def test_configure_backends_menu_omits_build_when_built(self): captured = {} infos = [BackendInfo("audiocpp", "audio.cpp", lambda: None, lambda: 0)] statuses = [BackendStatus("audiocpp", "audio.cpp", installed=True, configured=True)] with tempfile.TemporaryDirectory() as td: checkout = Path(td) (checkout / "server.json").write_text(json.dumps({"models": []}), encoding="utf-8") binary = checkout / "build" / "linux-cuda-release" / "bin" binary.mkdir(parents=True) (binary / "audiocpp_server").write_bytes(b"x") with patch.object(hub, "REGISTRY", infos), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=checkout), \ patch.object(hub.shutil, "which", return_value="/x"): result = hub._Hub(None).screen_configure() self.assertIs(result, tui.Wizard.BACK) labels = self._labels(captured["options"]) self.assertNotIn("Build audio.cpp server", labels) def test_configure_backends_menu_configure_only_when_built_unconfigured(self): captured = {} infos = [BackendInfo("audiocpp", "audio.cpp", lambda: None, lambda: 0)] # Built but no server.json: only Configure (the next step) plus # Uninstall — no Build, no Download, no Install entry. statuses = [BackendStatus("audiocpp", "audio.cpp", installed=True, configured=False)] with tempfile.TemporaryDirectory() as td: checkout = Path(td) binary = checkout / "build" / "linux-cuda-release" / "bin" binary.mkdir(parents=True) (binary / "audiocpp_server").write_bytes(b"x") with patch.object(hub, "REGISTRY", infos), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=checkout), \ patch.object(hub.shutil, "which", return_value="/x"): result = hub._Hub(None).screen_configure() self.assertIs(result, tui.Wizard.BACK) self.assertEqual(self._labels(captured["options"]), ["Configure audio.cpp", "Uninstall Backend"]) def test_convert_menu_builds_one_form_with_backend_field(self): captured = {} st = BackendStatus("qwen", "qwen-tts", installed=True, configured=True) with patch.object(hub.tui, "form", self._capture_form(captured)), \ patch.object(hub, "detect_all", return_value=[st]), \ patch.object(hub.shutil, "which", return_value="/x"): result = hub._Hub(None).screen_convert() self.assertIs(result, tui.Wizard.BACK) self.assertEqual(captured["title"], "Convert books") # One form, no picker menu: the first field is the Backend picker, # and only convertible backends are offered in it. self.assertEqual(captured["fields"][0]["key"], "backend") self.assertEqual(captured["fields"][0]["choices"], [("qwen-tts", "qwen")]) self.assertEqual(captured["buttons"], ("Generate!", "Cancel")) self.assertTrue(captured["start_on_buttons"]) def test_convert_with_nothing_ready_flashes_instead_of_menu(self): # Installed-but-unconfigured → nothing convertible: a hint flash # replaces the old fallback menu entirely. flashed = [] menus = [] def fake_menu(*args, **kwargs): menus.append((args, kwargs)) return tui.Wizard.BACK def fake_flash(stdscr, text, kind="warn"): flashed.append(text) st = BackendStatus("qwen", "qwen-tts", installed=True, configured=False) with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub.tui, "flash", fake_flash), \ patch.object(hub, "detect_all", return_value=[st]): result = hub._Hub(None).screen_convert() self.assertIs(result, tui.Wizard.BACK) self.assertEqual(menus, []) self.assertIn("No backend is ready", flashed[0]) def test_configure_backends_menu_shows_status_table(self): captured = {} infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0)] statuses = [BackendStatus("qwen", "qwen-tts", installed=True, configured=True)] with patch.object(hub, "REGISTRY", infos), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.shutil, "which", return_value="/x"): result = hub._Hub(None).screen_configure() self.assertIs(result, tui.Wizard.BACK) self.assertEqual(captured["table_title"], "Backend status") self.assertEqual( captured["table_rows"], [("qwen-tts", "installed", "warn", "body")]) def test_server_menu_lists_only_installed_backends(self): captured = {} installed = BackendStatus("audiocpp", "audio.cpp", installed=True, configured=True, servers=[ServerSpec("audiocpp", "http://127.0.0.1:8080", [])]) remote = BackendStatus("qwen", "qwen-tts", installed=False, configured=False, running=True) gone = BackendStatus("faster", "faster-qwen3-tts", installed=False, configured=False) with patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.common, "server_running", return_value=False), \ patch.object(hub, "detect_all", return_value=[installed, remote, gone]), \ patch.object(hub.shutil, "which", return_value="/x"): result = hub._Hub(None).screen_server() self.assertIs(result, tui.Wizard.BACK) # Only the installed backend's server is offered; a running external # server (remote) can't be stopped from here and must not appear. self.assertEqual([opt[0] for opt in captured["options"]], ["audio.cpp"]) # The running/stopped state lives in the status table above the # menu (not on the entries, whose colors the selection bar covers). self.assertEqual(captured["table_title"], "Server status") self.assertEqual(captured["table_rows"], [("audio.cpp", "stopped", "err", "body")]) def test_server_menu_flashes_when_nothing_installed(self): flashed = [] def fake_flash(stdscr, text, kind="warn"): flashed.append(text) remote = BackendStatus("qwen", "qwen-tts", installed=False, configured=False, running=True) with patch.object(hub.tui, "flash", fake_flash), \ patch.object(hub, "detect_all", return_value=[remote]): result = hub._Hub(None).screen_server() self.assertIs(result, tui.Wizard.BACK) self.assertEqual(len(flashed), 1) self.assertIn("No backend is installed", flashed[0]) def test_submenu_repeats_ffmpeg_warning(self): captured = {} infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0)] statuses = [BackendStatus("qwen", "qwen-tts", installed=True, configured=True)] with patch.object(hub, "REGISTRY", infos), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.shutil, "which", return_value=None): hub._Hub(None).screen_configure() self.assertEqual(captured["notice_lines"], [("Warning: ffmpeg not installed!", "err")]) class ConvertFlowTests(unittest.TestCase): """_convert_form / screen_convert: one form whose first field is the Backend picker, followed by that backend's options (local config or live remote queries).""" def setUp(self): self.tui = _ScriptedTUI() for name in ("menu", "line_edit", "confirm", "form", "flash"): patcher = patch.object(hub.tui, name, getattr(self.tui, name)) patcher.start() self.addCleanup(patcher.stop) def _form_values(self, **overrides): """A fully-populated form result, with sensible defaults.""" values = {"output_format": "m4b", "speed": "1.0", "single_file": False, "debug": False, "stop_and_exit": True} values.update(overrides) return values def _answer_form(self, **overrides): self.tui.form_script.append(self._form_values(**overrides)) def _field(self, key, form_index=-1): _, fields, _ = self.tui.forms_seen[form_index] return next(f for f in fields if f["key"] == key) def _ready(self, key, label): """A backend status that is ready to convert with.""" return BackendStatus(key, label, installed=True, configured=True) def _convert(self, stdscr, statuses, run_result=None): """Run the convert flow with STATUSES, returning the command tuple. ``_run_conversion`` is stubbed so the accepted command is captured instead of launching the run view (and reporting RUN_RESULT — True when the user answered "stop the server and exit"); None is returned when the flow aborts before reaching a conversion (nothing ready, a flash). The screen's navigation result lands on ``self.nav``. """ captured = {} def fake_run_conversion(self_, backend, kwargs): captured["backend"] = backend captured["kwargs"] = kwargs return run_result with patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub._Hub, "_run_conversion", fake_run_conversion): self.nav = hub._Hub(None).screen_convert() if "backend" not in captured: return None return ("convert", captured["backend"], captured["kwargs"]) # ------------------------------------------------------------------ # audio.cpp: remote server (no local checkout / server.json) # ------------------------------------------------------------------ def _patch_remote(self, models, voices=None): """Fetch helpers return MODELS/VOICES for a remote audio.cpp server.""" fetched_models = patch.object(hub.audiocpp_backend, "fetch_server_models", lambda url: models) fetched_voices = patch.object(hub.audiocpp_backend, "fetch_server_voices", lambda url, model_id: voices) for patcher in (fetched_models, fetched_voices): patcher.start() self.addCleanup(patcher.stop) def _remote(self, key, label, spec_name=None, url=None, remote_urls=None, remote_models=None): """A running remote backend status (not installed on this machine).""" if remote_urls is None: remote_urls = {spec_name or key: url or f"http://{key}.local:8080"} return BackendStatus(key, label, installed=False, configured=False, running=True, remote=True, remote_urls=remote_urls, remote_models=list(remote_models or [])) def test_audiocpp_remote_builds_one_form(self): self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="narrator", instructions="", speed="1.5") cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertEqual(cmd[0], "convert") self.assertEqual(cmd[1], hub.BACKEND_AUDIOCPP) kwargs = cmd[2] self.assertEqual(kwargs["model_id"], "higgs") self.assertEqual(kwargs["voice"], "narrator") self.assertIsNone(kwargs["instructions"]) self.assertEqual(kwargs["api_url"], "http://audiocpp.local:8080") self.assertEqual(kwargs["output_format"], "m4b") self.assertEqual(kwargs["speed"], 1.5) self.assertFalse(kwargs["single_file"]) self.assertFalse(kwargs["debug"]) # One form, not a cascade of menus/editors. self.assertEqual(len(self.tui.forms_seen), 1) title, fields, form_kwargs = self.tui.forms_seen[0] self.assertEqual(title, "Convert books") self.assertEqual([f["key"] for f in fields], ["backend", "model_id", "audiocpp_voice", "instructions", "output_format", "speed", "single_file", "debug", "stop_and_exit"]) self.assertEqual(form_kwargs["buttons"], ("Generate!", "Cancel")) self.assertTrue(form_kwargs["start_on_buttons"]) # The stop-and-exit toggle ships on by default. stop_field = self._field("stop_and_exit") self.assertEqual(stop_field["kind"], "bool") self.assertTrue(stop_field["value"]) self.assertTrue(cmd[2]["stop_and_exit"]) # The backend field offers the remote entry under a [remote] label. self.assertEqual(fields[0]["choices"], [("audio.cpp [remote]", "audiocpp-remote")]) # The model menu was fed from the live query (label, id). self.assertEqual(self._field("model_id")["choices"], [("higgs (higgs_audio_tts, clone)", "higgs")]) def test_audiocpp_customvoice_entry_lists_builtin_speakers(self): # A CustomVoice entry populates the Voice menu with the Qwen3-TTS # built-in speakers and maps the pick to --voice (which the client # resolves as speaker mode). self._patch_remote( [{"id": "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF", "family": "qwen3_tts", "task": "tts"}]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self._answer_form( backend="audiocpp-remote", model_id="Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF", audiocpp_voice="Ryan", instructions="") cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) # The picked speaker is passed as --voice; no separate speaker kwarg. self.assertEqual(cmd[2]["voice"], "Ryan") self.assertNotIn("speaker", cmd[2]) fields = self.tui.forms_seen[0][1] voice_field = self._field("audiocpp_voice") self.assertEqual(voice_field["choices"](fields), [(s, s) for s in hub.QWEN3_TTS_SPEAKERS]) # CustomVoice reads a style instruction, so the field stays visible. instr = self._field("instructions") self.assertTrue(instr["visible"](fields)) self.assertIsNone(instr["validate"]("")) def test_audiocpp_qwen3_tts_base_entry_lists_clone_voices(self): # A Base entry populates the Voice menu with the server's clone # voices only (no built-in speakers) and maps the pick to --voice. self._patch_remote( [{"id": "Qwen3-TTS-12Hz-1.7B-Base-GGUF", "family": "qwen3_tts", "task": "tts"}], voices=["narrator"]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self._answer_form( backend="audiocpp-remote", model_id="Qwen3-TTS-12Hz-1.7B-Base-GGUF", audiocpp_voice="narrator", instructions="") cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertEqual(cmd[2]["voice"], "narrator") self.assertNotIn("speaker", cmd[2]) self.assertIsNone(cmd[2]["instructions"]) fields = self.tui.forms_seen[0][1] voice_field = self._field("audiocpp_voice") self.assertEqual(voice_field["choices"](fields), [("narrator", "narrator")]) # Base (clone) ignores instructions, so the field is hidden. instr = self._field("instructions") self.assertFalse(instr["visible"](fields)) def test_audiocpp_remote_missing_family_is_clone_capable(self): # A missing family is unknown — not guessed as qwen3_tts — so the # entry is clone-only: it needs a --voice rather than offering a # built-in speaker. self._patch_remote([{"id": "legacy", "family": "", "task": ""}], voices=[]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self._answer_form(backend="audiocpp-remote", model_id="legacy", audiocpp_voice="", instructions="") cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNotNone(cmd) self.assertIsNone(cmd[2]["voice"]) self.assertNotIn("speaker", cmd[2]) voice_field = self._field("audiocpp_voice") # Clone-only: an empty voice is refused (no built-in speaker option). self.assertIsNotNone(voice_field["validate"]("")) def test_audiocpp_vdes_hides_voice_and_requires_instructions(self): self._patch_remote( [{"id": "design", "family": "qwen3_tts", "task": "vdes"}]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self._answer_form(backend="audiocpp-remote", model_id="design", audiocpp_voice=None, instructions="A warm British narrator") cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNone(cmd[2]["voice"]) self.assertEqual(cmd[2]["instructions"], "A warm British narrator") fields = self.tui.forms_seen[0][1] voice_field = self._field("audiocpp_voice") self.assertFalse(voice_field["visible"](fields)) instr = self._field("instructions") self.assertTrue(instr["visible"](fields)) self.assertIsNotNone(instr["validate"]("")) self.assertIsNone(instr["validate"]("describe me")) def test_audiocpp_clone_drops_stale_instructions(self): # A Base/clone entry ignores instructions: even if the form held a # leftover value, the mapper must not send it to the model. self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="narrator", instructions="stale description") cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNone(cmd[2]["instructions"]) def test_audiocpp_required_voice_validates(self): # A non-qwen3_tts family needs a --voice; a blank value refuses. self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="narrator", instructions="") self._convert(None, [self._remote("audiocpp", "audio.cpp")]) voice_field = self._field("audiocpp_voice") self.assertIsNotNone(voice_field["validate"]("")) self.assertIsNone(voice_field["validate"]("narrator")) def test_audiocpp_remote_unreachable_models_flash_and_abort(self): self._patch_remote(None) # endpoint did not answer valid JSON cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNone(cmd) self.assertIn("Could not list models", self.tui.flashes[0]) def test_audiocpp_remote_empty_models_flash_and_abort(self): self._patch_remote([]) cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNone(cmd) self.assertIn("hosts no model entries", self.tui.flashes[0]) def test_audiocpp_remote_no_server_voices_leaves_voice_blank(self): # No voices listed for a required-voice model: the form still opens # with an empty Voice field (Generate-time validation reports it). self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=[]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="", instructions="") cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNotNone(cmd) self.assertIsNone(cmd[2]["voice"]) fields = self.tui.forms_seen[0][1] voice_field = self._field("audiocpp_voice") self.assertEqual(voice_field["choices"](fields), []) # ------------------------------------------------------------------ # audio.cpp: local managed setup keeps reading its server.json # ------------------------------------------------------------------ def test_audiocpp_local_still_reads_server_json(self): queried = [] def must_not_query(url): queried.append(url) raise AssertionError("live query on the local path") with tempfile.TemporaryDirectory() as td: root = Path(td) (root / "server.json").write_text(json.dumps({ "models": [{"id": "qwen", "family": "qwen3_tts", "task": "tts"}], "voice_dir": str(root), }), encoding="utf-8") (root / "Narrator.wav").write_bytes(b"x") with patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=root), \ patch.object(hub.audiocpp_backend, "fetch_server_models", must_not_query), \ patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self._answer_form(backend="audiocpp", model_id="qwen", audiocpp_voice="Narrator", instructions="") cmd = self._convert(None, [self._ready("audiocpp", "audio.cpp")]) self.assertEqual(queried, []) self.assertIsNotNone(cmd) self.assertEqual(cmd[2]["model_id"], "qwen") self.assertEqual(cmd[2]["voice"], "Narrator") def test_managed_and_remote_both_offered(self): # A ready managed audio.cpp (server.json) AND a running remote # audio.cpp: both entries appear. The managed entry reads server.json # (no api_url), the remote entry live-queries (api_url set). self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) with tempfile.TemporaryDirectory() as td: root = Path(td) (root / "server.json").write_text(json.dumps({ "models": [{"id": "qwen", "family": "qwen3_tts", "task": "tts"}], }), encoding="utf-8") with patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=root), \ patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self._answer_form(backend="audiocpp", model_id="qwen", audiocpp_voice="", instructions="") cmd = self._convert(None, [ self._ready("audiocpp", "audio.cpp"), self._remote("audiocpp", "audio.cpp")]) self.assertEqual(cmd[0], "convert") self.assertEqual(cmd[1], hub.BACKEND_AUDIOCPP) # Managed entry: no api_url override (uses the configured local URL). self.assertNotIn("api_url", cmd[2]) self.assertEqual(cmd[2]["model_id"], "qwen") fields = self.tui.forms_seen[0][1] self.assertEqual(fields[0]["choices"], [("audio.cpp", "audiocpp"), ("audio.cpp [remote]", "audiocpp-remote")]) def test_audiocpp_remote_mapper_adds_api_url(self): self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="narrator", instructions="") cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp", url="http://10.0.0.5:8080")]) self.assertEqual(cmd[2]["api_url"], "http://10.0.0.5:8080") # ------------------------------------------------------------------ # common fields: output format, speed, single-file, debug # ------------------------------------------------------------------ def test_common_fields_hide_combine_for_m4b(self): fields = hub._common_fields() fmt = next(f for f in fields if f["key"] == "output_format") single = next(f for f in fields if f["key"] == "single_file") fmt["value"] = "m4b" self.assertFalse(single["visible"](fields)) fmt["value"] = "mp3" self.assertTrue(single["visible"](fields)) # ------------------------------------------------------------------ # qwen: speaker or clone # ------------------------------------------------------------------ def test_qwen_builds_speaker_and_clone_form(self): with patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["Vivian", "Serena"]), \ patch.object(hub.config, "SPEAKER", "Vivian"): self._answer_form(backend="qwen", mode="custom", speaker="Serena", clone="") with patch.object(hub.common, "update_config_value") as mk_update: cmd = self._convert(None, [self._ready("qwen", "qwen-tts")]) speaker_in_memory = hub.config.SPEAKER self.assertEqual(cmd[0], "convert") self.assertEqual(cmd[1], hub.BACKEND_QWEN) self.assertIsNone(cmd[2]["clone"]) # The speaker choice is persisted for future runs too. mk_update.assert_called_once_with("SPEAKER", "Serena") self.assertEqual(speaker_in_memory, "Serena") fields = self.tui.forms_seen[0][1] self.assertEqual([f["key"] for f in fields], ["backend", "mode", "speaker", "clone", "output_format", "speed", "single_file", "debug", "stop_and_exit"]) mode_field = self._field("mode") self.assertEqual(mode_field["choices"], [("Built-in speaker", "custom"), ("Clone from a .wav file", "clone")]) speaker_field = self._field("speaker") clone_field = self._field("clone") # Speaker shows in custom mode; the .wav path shows in clone mode. self.assertTrue(speaker_field["visible"](fields)) self.assertFalse(clone_field["visible"](fields)) mode_field["value"] = "clone" self.assertFalse(speaker_field["visible"](fields)) self.assertTrue(clone_field["visible"](fields)) def test_qwen_clone_mode_passes_path_and_keeps_speaker(self): with patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["Vivian"]), \ patch.object(hub.config, "SPEAKER", "Vivian"): self._answer_form(backend="qwen", mode="clone", speaker="Vivian", clone="/tmp/ref.wav") with patch.object(hub.common, "update_config_value") as mk_update: cmd = self._convert(None, [self._ready("qwen", "qwen-tts")]) self.assertEqual(cmd[2]["clone"], "/tmp/ref.wav") # Clone mode does not touch the global speaker. mk_update.assert_not_called() # ------------------------------------------------------------------ # faster: remote server (no local voices.json) # ------------------------------------------------------------------ def test_faster_remote_uses_a_text_voice_field(self): with tempfile.TemporaryDirectory() as td: with patch.object(hub.faster_backend, "_checkout", return_value=Path(td)): self._answer_form(backend="faster", faster_voice="obama") cmd = self._convert( None, [self._ready("faster", "faster-qwen3-tts")]) self.assertEqual(cmd[0], "convert") self.assertEqual(cmd[1], "faster") self.assertEqual(cmd[2]["voice"], "obama") self.assertEqual(self._field("faster_voice")["kind"], "text") def test_faster_local_still_lists_voices_json(self): with tempfile.TemporaryDirectory() as td: checkout = Path(td) (checkout / "voices.json").write_text( json.dumps({"default": {}, "obama": {}}), encoding="utf-8") with patch.object(hub.faster_backend, "_checkout", return_value=checkout): self._answer_form(backend="faster", faster_voice="obama") cmd = self._convert( None, [self._ready("faster", "faster-qwen3-tts")]) self.assertEqual(cmd[2]["voice"], "obama") voice_field = self._field("faster_voice") self.assertEqual(voice_field["kind"], "choice") self.assertEqual(voice_field["choices"], [("default", "default"), ("obama", "obama")]) def test_faster_remote_uses_text_voice_and_api_url(self): st = self._remote("faster", "faster-qwen3-tts", url="http://10.0.0.5:8000") self._answer_form(backend="faster-remote", faster_voice="obama") cmd = self._convert(None, [st]) self.assertEqual(cmd[0], "convert") self.assertEqual(cmd[1], "faster") self.assertEqual(cmd[2]["voice"], "obama") self.assertEqual(cmd[2]["api_url"], "http://10.0.0.5:8000") self.assertEqual(self._field("faster_voice")["kind"], "text") def test_stop_and_exit_answer_quits_the_tui(self): # Yes on the run view's stop-and-exit prompt ends the whole TUI: # screen_convert returns None instead of Wizard.BACK. st = self._remote("faster", "faster-qwen3-tts", url="http://10.0.0.5:8000") self._answer_form(backend="faster-remote", faster_voice="obama") cmd = self._convert(None, [st], run_result=True) self.assertIsNotNone(cmd) self.assertIsNone(self.nav) def test_normal_finish_returns_to_the_menu(self): st = self._remote("faster", "faster-qwen3-tts", url="http://10.0.0.5:8000") self._answer_form(backend="faster-remote", faster_voice="obama") cmd = self._convert(None, [st]) self.assertIsNotNone(cmd) self.assertIs(self.nav, hub.tui.Wizard.BACK) def test_qwen_remote_limited_modes_and_api_url(self): # A remote qwen with only the Base (clone) demo answering: the form # offers only clone mode and targets the clone remote URL. st = self._remote( "qwen", "qwen-tts", remote_urls={"qwen-clone": "http://10.0.0.5:7861"}, remote_models=["Base"]) with patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["Vivian"]), \ patch.object(hub.config, "SPEAKER", "Vivian"): self._answer_form(backend="qwen-remote", mode="clone", speaker="Vivian", clone="/tmp/ref.wav") cmd = self._convert(None, [st]) self.assertEqual(cmd[1], hub.BACKEND_QWEN) self.assertEqual(cmd[2]["clone"], "/tmp/ref.wav") self.assertEqual(cmd[2]["api_url"], "http://10.0.0.5:7861") fields = self.tui.forms_seen[0][1] self.assertEqual(self._field("mode")["choices"], [("Clone from a .wav file", "clone")]) # ------------------------------------------------------------------ # multiple backends: the Backend picker gates which options show # ------------------------------------------------------------------ def test_multiple_backends_gate_options_on_backend_value(self): # Two ready backends: the form leads with a Backend picker and the # per-backend fields are hidden/shown by its value. with tempfile.TemporaryDirectory() as td: root = Path(td) (root / "server.json").write_text(json.dumps({ "models": [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], }), encoding="utf-8") with patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=root), \ patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self._answer_form(backend="qwen", mode="custom", speaker="Vivian", clone="") cmd = self._convert(None, [ self._ready("audiocpp", "audio.cpp"), self._ready("qwen", "qwen-tts")]) self.assertEqual(cmd[0], "convert") self.assertEqual(cmd[1], hub.BACKEND_QWEN) fields = self.tui.forms_seen[0][1] self.assertEqual(fields[0]["key"], "backend") self.assertEqual(fields[0]["choices"], [("audio.cpp", "audiocpp"), ("qwen-tts", "qwen")]) self.assertEqual( [f["key"] for f in fields], ["backend", "model_id", "audiocpp_voice", "instructions", "mode", "speaker", "clone", "output_format", "speed", "single_file", "debug", "stop_and_exit"]) # The form opens on the configured default (audio.cpp): its fields # show, the other backend's hide. (Instructions is hidden too: the # default higgs entry is clone-only, which ignores instructions.) for key in ("model_id", "audiocpp_voice"): self.assertTrue(self._field(key)["visible"](fields)) self.assertFalse(self._field("instructions")["visible"](fields)) for key in ("mode", "speaker", "clone"): self.assertFalse(self._field(key)["visible"](fields)) # Picking qwen in the Backend field swaps which options show. fields[0]["value"] = "qwen" self.assertTrue(self._field("mode")["visible"](fields)) self.assertTrue(self._field("speaker")["visible"](fields)) self.assertFalse(self._field("clone")["visible"](fields)) # qwen's clone mode hides the speaker and shows the .wav path. self._field("mode")["value"] = "clone" self.assertFalse(self._field("speaker")["visible"](fields)) self.assertTrue(self._field("clone")["visible"](fields)) for key in ("model_id", "audiocpp_voice", "instructions"): self.assertFalse(self._field(key)["visible"](fields)) # And back to audio.cpp. fields[0]["value"] = "audiocpp" for key in ("model_id", "audiocpp_voice"): self.assertTrue(self._field(key)["visible"](fields)) self.assertFalse(self._field("instructions")["visible"](fields)) for key in ("mode", "speaker", "clone"): self.assertFalse(self._field(key)["visible"](fields)) class SelectSpecTests(unittest.TestCase): """_select_spec: mode-aware server selection (qwen has two servers).""" def _qwen_status(self): return BackendStatus( "qwen", "qwen-tts", installed=True, configured=True, servers=[ServerSpec("qwen-custom", "http://127.0.0.1:7860", []), ServerSpec("qwen-clone", "http://127.0.0.1:7861", [])]) def test_qwen_custom_mode(self): spec = hub._select_spec(self._qwen_status(), {"clone": None}) self.assertEqual(spec.name, "qwen-custom") def test_qwen_clone_mode(self): spec = hub._select_spec(self._qwen_status(), {"clone": "ref.wav"}) self.assertEqual(spec.name, "qwen-clone") def test_audiocpp_returns_single_spec(self): st = BackendStatus("audiocpp", "audio.cpp", installed=True, configured=True, servers=[ServerSpec("audiocpp", "http://x", [])]) spec = hub._select_spec(st, {}) self.assertEqual(spec.name, "audiocpp") def test_none_when_no_servers(self): st = BackendStatus("qwen", "qwen-tts", installed=False, configured=False) self.assertIsNone(hub._select_spec(st, {})) class PrepareRunConfigTests(unittest.TestCase): """_prepare_run_config: the run view's inputs from the accepted form.""" def _spec(self, name="qwen-custom", url="http://127.0.0.1:7860"): return ServerSpec(name, url, ["x"]) def test_remote_targets_the_api_url(self): with patch.object(hub, "detect_all", return_value=[]) as mk_detect: cfg = hub._prepare_run_config( "audiocpp", {"api_url": "http://10.0.0.5:8080"}) self.assertEqual(cfg.server_url, "http://10.0.0.5:8080") self.assertIsNone(cfg.autostart_spec) self.assertEqual(cfg.server_identity, "audiocpp") self.assertIn("remote", cfg.backend_label) # The remote path never re-detects or touches managed-instance state. mk_detect.assert_not_called() def test_autostart_sets_the_spec_and_pops_the_flag(self): spec = self._spec() kwargs = {"autostart": "qwen-custom"} with patch.object(hub, "detect_all", return_value=[]), \ patch.object(hub, "_find_spec", return_value=spec): cfg = hub._prepare_run_config("qwen", kwargs) self.assertIs(cfg.autostart_spec, spec) self.assertEqual(cfg.server_name, "qwen-custom") self.assertNotIn("autostart", kwargs) def test_stop_and_exit_travels_on_the_config_not_the_kwargs(self): # The run-view toggle is not a converter kwarg: it moves onto the # config (and defaults to off when the form did not send it). with tempfile.TemporaryDirectory() as tmp: with patch.object(hub, "LOGS_FOLDER", Path(tmp)), \ patch.object(hub, "detect_all", return_value=[]): cfg = hub._prepare_run_config( "audiocpp", {"stop_and_exit": True}) self.assertTrue(cfg.stop_and_exit) cfg = hub._prepare_run_config("audiocpp", {}) self.assertFalse(cfg.stop_and_exit) def test_remote_config_carries_stop_and_exit(self): cfg = hub._prepare_run_config( "audiocpp", {"api_url": "http://10.0.0.5:8080", "stop_and_exit": True}) self.assertTrue(cfg.stop_and_exit) def test_autostart_with_missing_spec_continues_with_notice(self): kwargs = {"autostart": "gone"} with patch.object(hub, "detect_all", return_value=[]), \ patch.object(hub, "_find_spec", return_value=None): cfg = hub._prepare_run_config("qwen", kwargs) self.assertIsNone(cfg.autostart_spec) self.assertIn("gone", cfg.notice) def test_managed_conversion_flags_foreign_server_on_port(self): spec = ServerSpec("audiocpp", "http://127.0.0.1:8080", ["x"]) status = BackendStatus("audiocpp", "audio.cpp", installed=True, configured=True, servers=[spec]) with patch.object(hub, "detect_all", return_value=[status]), \ patch("backends.common.server_running", return_value=True), \ patch.object(hub.servers, "alive", return_value=False): cfg = hub._prepare_run_config("audiocpp", {}) self.assertEqual(cfg.server_url, spec.url) self.assertIsNone(cfg.autostart_spec) self.assertIn("did not start", cfg.notice) def test_managed_not_running_sets_url_without_autostart(self): spec = self._spec() status = BackendStatus("qwen", "qwen-tts", installed=True, configured=True, servers=[spec]) with patch.object(hub, "detect_all", return_value=[status]), \ patch("backends.common.server_running", return_value=False): cfg = hub._prepare_run_config("qwen", {"clone": None}) self.assertEqual(cfg.server_url, spec.url) self.assertIsNone(cfg.autostart_spec) def test_book_files_travel_on_the_config_not_the_kwargs(self): # _preflight stashes the plan in the form kwargs; keeping it there # collided with convert()'s named book_files/planned parameters. spec = self._spec() status = BackendStatus("qwen", "qwen-tts", installed=True, configured=True, servers=[spec]) kwargs = {"book_files": ["b.txt"], "planned": ["b.txt"]} with tempfile.TemporaryDirectory() as tmp, \ patch.object(hub, "LOGS_FOLDER", Path(tmp)), \ patch.object(hub, "detect_all", return_value=[status]), \ patch("backends.common.server_running", return_value=False): cfg = hub._prepare_run_config("qwen", kwargs) self.assertNotIn("book_files", kwargs) self.assertNotIn("planned", kwargs) self.assertEqual(cfg.book_files, ["b.txt"]) self.assertEqual(cfg.planned, ["b.txt"]) def test_the_dated_log_file_exists_once_a_run_is_prepared(self): # The run view advertises this file on failures; create it up front # so a crash before setup_logging still points somewhere real. with tempfile.TemporaryDirectory() as tmp: with patch.object(hub, "LOGS_FOLDER", Path(tmp)), \ patch.object(hub, "detect_all", return_value=[]): cfg = hub._prepare_run_config( "audiocpp", {"api_url": "http://10.0.0.5:8080"}) self.assertTrue(Path(cfg.log_path).exists()) class PreflightTests(unittest.TestCase): """_preflight: overwrite prompts run in the TUI, plan stashed in kwargs.""" def _cmd(self): return ("convert", "qwen", {"clone": None, "output_format": "mp3"}) def test_books_stashed_on_kwargs(self): stdscr = object() cmd = self._cmd() with patch.object(hub.AudiobookConverter, "preflight_overwrites", return_value=(["book.txt"], [("book.txt", "x")])) \ as mk_pre: self.assertTrue(hub._preflight(stdscr, cmd)) self.assertEqual(cmd[2]["book_files"], ["book.txt"]) self.assertEqual(cmd[2]["planned"], [("book.txt", "x")]) # The confirm callback passed to preflight is a TUI yes/no. confirm = mk_pre.call_args.kwargs["confirm"] with patch.object(hub.tui, "confirm", return_value=True) as mk_confirm: self.assertTrue(confirm("overwrite?", True)) mk_confirm.assert_called_once() def test_nothing_to_convert_flashes_and_returns_false(self): stdscr = object() with patch.object(hub.AudiobookConverter, "preflight_overwrites", return_value=([], [])), \ patch.object(hub.tui, "flash") as mk_flash: self.assertFalse(hub._preflight(stdscr, self._cmd())) mk_flash.assert_called_once() def test_all_skipped_flashes_and_returns_false(self): stdscr = object() with patch.object(hub.AudiobookConverter, "preflight_overwrites", return_value=(["book.txt"], [])), \ patch.object(hub.tui, "flash") as mk_flash: self.assertFalse(hub._preflight(stdscr, self._cmd())) mk_flash.assert_called_once() def test_confirm_esc_raises_back_to_form(self): # Esc on an overwrite confirm backs out to the form (one screen), # not "No" — which could dump the user on the main menu. stdscr = object() with patch.object(hub.AudiobookConverter, "preflight_overwrites", return_value=(["book.txt"], [("book.txt", "x")])) \ as mk_pre: hub._preflight(stdscr, self._cmd()) confirm = mk_pre.call_args.kwargs["confirm"] with patch.object(hub.tui, "confirm", return_value=hub._CANCEL): with self.assertRaises(hub._BackToForm): confirm("overwrite?", True) class DispatchConversionTests(unittest.TestCase): """_Hub._run_conversion: builds the config and runs the run view.""" def test_runs_run_view_on_the_hub_screen(self): timeouts = [] class Screen: def timeout(self, ms): timeouts.append(ms) class FakeView: run_result = None def __init__(self, scr, config): self.config = config self.scr = scr def run(self): return type(self).run_result screen = Screen() with patch.object(hub, "_prepare_run_config", return_value=hub.runview.RunConfig( backend="qwen", backend_label="qwen-tts", kwargs={}, book_files=[], planned=[])) as mk_cfg, \ patch.object(hub.runview, "RunView", FakeView): result = hub._Hub(screen)._run_conversion("qwen", {}) mk_cfg.assert_called_once() # The run view leaves a timed getch behind; it is reset so the hub # menus block for keys again. self.assertEqual(timeouts, [-1]) # A normal finish lands back on the menu (no quit signal). self.assertFalse(result) def test_stop_and_exit_answer_propagates_from_the_view(self): class Screen: def timeout(self, ms): pass class FakeView: def __init__(self, scr, config): self.config = config self.scr = scr def run(self): return True with patch.object(hub, "_prepare_run_config", return_value=hub.runview.RunConfig( backend="qwen", backend_label="qwen-tts", kwargs={}, book_files=[], planned=[])), \ patch.object(hub.runview, "RunView", FakeView): result = hub._Hub(Screen())._run_conversion("qwen", {}) self.assertTrue(result) def test_no_run_config_skips_the_view(self): with patch.object(hub, "_prepare_run_config", return_value=None) as mk_cfg, \ patch.object(hub.runview, "RunView") as mk_view: hub._Hub(None)._run_conversion("qwen", {}) mk_cfg.assert_called_once() mk_view.assert_not_called() class AddAutostartTests(unittest.TestCase): """_add_autostart: always starts the server when it isn't running.""" def _status(self): spec = ServerSpec("qwen-custom", "http://127.0.0.1:7860", ["x"]) return BackendStatus("qwen", "qwen-tts", installed=True, configured=True, running=False, servers=[spec]) def test_sets_autostart_when_not_running(self): cmd = ("convert", "qwen", {"clone": None}) with patch.object(hub, "detect_all", return_value=[self._status()]), \ patch("backends.common.server_running", return_value=False): hub._add_autostart(cmd, [self._status()]) self.assertEqual(cmd[2]["autostart"], "qwen-custom") def test_no_autostart_when_server_already_running(self): cmd = ("convert", "qwen", {"clone": None}) with patch.object(hub, "detect_all", return_value=[self._status()]), \ patch("backends.common.server_running", return_value=True): hub._add_autostart(cmd, [self._status()]) self.assertNotIn("autostart", cmd[2]) def test_no_autostart_for_remote_conversion(self): # A remote conversion (api_url set) never autostarts: the server is # external to this tool, so there is nothing to start/stop here. cmd = ("convert", "audiocpp", {"api_url": "http://10.0.0.5:8080"}) hub._add_autostart(cmd, []) self.assertNotIn("autostart", cmd[2]) class SettingsTests(unittest.TestCase): """Settings menu: field collection, validation, config.py writing.""" def test_write_config_preserves_comments_and_other_lines(self): import tempfile with tempfile.TemporaryDirectory() as td: path = Path(td) / "config.py" path.write_text( "# Default output options\n" 'AUDIO_FORMAT = "m4b"\n' 'AUDIO_BITRATE = "128k"\n' 'LANGUAGE = "English"\n' "\n" "CHUNK_SIZE = 250 # words per request\n", encoding="utf-8") with patch.object(hub.config, "__file__", str(path)): hub._write_config({"AUDIO_FORMAT": "mp3", "AUDIO_BITRATE": "192k", "LANGUAGE": "Japanese", "CHUNK_SIZE": 300}) text = path.read_text(encoding="utf-8") self.assertEqual( text, "# Default output options\n" 'AUDIO_FORMAT = "mp3"\n' 'AUDIO_BITRATE = "192k"\n' 'LANGUAGE = "Japanese"\n' "\n" "CHUNK_SIZE = 300 # words per request\n") def test_write_config_missing_key_raises(self): import tempfile with tempfile.TemporaryDirectory() as td: path = Path(td) / "config.py" path.write_text("X = 1\n", encoding="utf-8") with patch.object(hub.config, "__file__", str(path)): with self.assertRaises(ValueError): hub._write_config({"AUDIO_FORMAT": "mp3"}) def test_apply_settings_writes_and_reloads_in_memory(self): written = {} def fake_write(updates): written.update(updates) original = {name: getattr(hub.config, name) for name in ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", "CHUNK_SIZE", "AUDIOCPP_UNLOAD_MODELS", "QWEN_API_URL", "CLONE_API_URL", "FASTER_API_URL", "AUDIOCPP_API_URL", "QWEN_REMOTE_URL", "CLONE_REMOTE_URL", "FASTER_REMOTE_URL", "AUDIOCPP_REMOTE_URL")} self.addCleanup(lambda: [setattr(hub.config, name, value) for name, value in original.items()]) values = {"audio_format": "ogg", "audio_bitrate": " 192k ", "language": "en", "chunk_size": "300", "unload_models": True, "qwen_custom_port": "7862", "qwen_clone_port": "7863", "faster_port": "8001", "audiocpp_port": "8081", "audiocpp_remote_url": "10.0.0.5:8080", "faster_remote_url": "http://10.0.0.6:8000", "qwen_custom_remote_url": "", "qwen_clone_remote_url": ""} with patch.object(hub, "_write_config", fake_write), \ patch.object(hub, "_sync_audiocpp_server_port"): hub._apply_settings(values) # Values are trimmed and language normalized to a display name; # remote URLs are normalized to full http(s) URLs (empty = off). self.assertEqual(written, {"AUDIO_FORMAT": "ogg", "AUDIO_BITRATE": "192k", "LANGUAGE": "English", "CHUNK_SIZE": 300, "AUDIOCPP_UNLOAD_MODELS": True, "QWEN_API_URL": "http://127.0.0.1:7862", "CLONE_API_URL": "http://127.0.0.1:7863", "FASTER_API_URL": "http://127.0.0.1:8001", "AUDIOCPP_API_URL": "http://127.0.0.1:8081", "QWEN_REMOTE_URL": "", "CLONE_REMOTE_URL": "", "FASTER_REMOTE_URL": "http://10.0.0.6:8000", "AUDIOCPP_REMOTE_URL": "http://10.0.0.5:8080"}) # In-memory config is reloaded so this session sees the change. self.assertEqual(hub.config.AUDIO_FORMAT, "ogg") self.assertEqual(hub.config.AUDIO_BITRATE, "192k") self.assertEqual(hub.config.LANGUAGE, "English") self.assertEqual(hub.config.CHUNK_SIZE, 300) self.assertEqual(hub.config.AUDIOCPP_UNLOAD_MODELS, True) self.assertEqual(hub.config.QWEN_API_URL, "http://127.0.0.1:7862") self.assertEqual(hub.config.FASTER_API_URL, "http://127.0.0.1:8001") self.assertEqual(hub.config.AUDIOCPP_REMOTE_URL, "http://10.0.0.5:8080") def test_apply_settings_rejects_bad_values(self): original = {name: getattr(hub.config, name) for name in ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", "CHUNK_SIZE", "AUDIOCPP_UNLOAD_MODELS", "QWEN_API_URL", "CLONE_API_URL", "FASTER_API_URL", "AUDIOCPP_API_URL", "QWEN_REMOTE_URL", "CLONE_REMOTE_URL", "FASTER_REMOTE_URL", "AUDIOCPP_REMOTE_URL")} self.addCleanup(lambda: [setattr(hub.config, name, value) for name, value in original.items()]) base = {"audio_format": "m4b", "audio_bitrate": "128k", "language": "English", "chunk_size": "250", "unload_models": True, "qwen_custom_port": "7860", "qwen_clone_port": "7861", "faster_port": "8000", "audiocpp_port": "8080"} with patch.object(hub, "_write_config") as mk_write: with self.assertRaises(ValueError): hub._apply_settings({**base, "language": "Klingon"}) with self.assertRaises(ValueError): hub._apply_settings({**base, "chunk_size": "0"}) with self.assertRaises(ValueError): hub._apply_settings({**base, "audiocpp_port": "70000"}) with self.assertRaises(ValueError): hub._apply_settings({**base, "audiocpp_remote_url": "not a url"}) mk_write.assert_not_called() def test_field_validators(self): self.assertIsNone(hub._validate_bitrate("128k")) self.assertIsNotNone(hub._validate_bitrate(" ")) self.assertIsNone(hub._validate_language("English")) self.assertIsNone(hub._validate_language("en")) self.assertIsNotNone(hub._validate_language("Klingon")) self.assertIsNone(hub._validate_chunk_size("250")) self.assertIsNotNone(hub._validate_chunk_size("abc")) self.assertIsNotNone(hub._validate_chunk_size("0")) self.assertIsNone(hub._validate_port("8080")) self.assertIsNone(hub._validate_port("1")) self.assertIsNone(hub._validate_port("65535")) self.assertIsNotNone(hub._validate_port("0")) self.assertIsNotNone(hub._validate_port("70000")) self.assertIsNotNone(hub._validate_port("abc")) def test_settings_menu_builds_form_and_saves(self): captured = {} def fake_form(stdscr, title, fields, back_value=None): captured["fields"] = fields return {"audio_format": "ogg", "audio_bitrate": "192k", "language": "English", "chunk_size": "300", "unload_models": True, "qwen_custom_port": "7860", "qwen_clone_port": "7861", "faster_port": "8000", "audiocpp_port": "8080"} applied = [] def fake_apply(values): applied.append(values) def fake_flash(stdscr, text, kind="warn"): captured["flash"] = (text, kind) with patch.object(hub.tui, "form", fake_form), \ patch.object(hub, "_apply_settings", fake_apply), \ patch.object(hub.tui, "flash", fake_flash): hub._Hub(None).screen_settings() self.assertEqual([f["key"] for f in captured["fields"]], ["audio_format", "audio_bitrate", "language", "chunk_size", "unload_models", "audiocpp_port", "faster_port", "qwen_custom_port", "qwen_clone_port", "audiocpp_remote_url", "faster_remote_url", "qwen_custom_remote_url", "qwen_clone_remote_url"]) kinds = {f["key"]: f["kind"] for f in captured["fields"]} self.assertEqual(kinds["audio_format"], "choice") self.assertEqual(kinds["audio_bitrate"], "text") self.assertEqual(kinds["audiocpp_port"], "text") self.assertEqual(kinds["unload_models"], "bool") self.assertEqual(kinds["audiocpp_remote_url"], "text") labels = {f["key"]: f["label"] for f in captured["fields"]} self.assertEqual(labels["qwen_clone_port"], "qwen-tts Base port") self.assertEqual(labels["audiocpp_remote_url"], "audio.cpp remote URL") self.assertNotIn("(clone)", " ".join(labels.values())) # The ports section note hangs off the first port field, the remote # section note off the first remote URL field. notes = {f["key"]: f.get("note") for f in captured["fields"]} self.assertTrue(notes["audiocpp_port"]) self.assertTrue(notes["audiocpp_remote_url"]) self.assertIsNone(notes["audio_format"]) self.assertIsNone(notes["qwen_custom_port"]) self.assertEqual(applied, [{"audio_format": "ogg", "audio_bitrate": "192k", "language": "English", "chunk_size": "300", "unload_models": True, "qwen_custom_port": "7860", "qwen_clone_port": "7861", "faster_port": "8000", "audiocpp_port": "8080"}]) self.assertEqual(captured["flash"], ("Settings saved.", "ok")) def test_settings_menu_cancel_does_not_apply(self): def fake_form(stdscr, title, fields, back_value=None): return back_value # user pressed Cancel applied = [] def fake_apply(values): applied.append(values) with patch.object(hub.tui, "form", fake_form), \ patch.object(hub, "_apply_settings", fake_apply): hub._Hub(None).screen_settings() self.assertEqual(applied, []) def test_settings_menu_writes_config_end_to_end(self): import tempfile tui._THEME.clear() self.addCleanup(tui._THEME.clear) curses = FakeCurses() patcher = patch.dict("sys.modules", {"curses": curses}) patcher.start() self.addCleanup(patcher.stop) original = {name: getattr(hub.config, name) for name in ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", "CHUNK_SIZE", "AUDIOCPP_UNLOAD_MODELS", "QWEN_API_URL", "CLONE_API_URL", "FASTER_API_URL", "AUDIOCPP_API_URL", "QWEN_REMOTE_URL", "CLONE_REMOTE_URL", "FASTER_REMOTE_URL", "AUDIOCPP_REMOTE_URL")} self.addCleanup(lambda: [setattr(hub.config, name, value) for name, value in original.items()]) with tempfile.TemporaryDirectory() as td: path = Path(td) / "config.py" path.write_text( "# Default output options\n" 'AUDIO_FORMAT = "m4b"\n' 'AUDIO_BITRATE = "128k"\n' 'LANGUAGE = "English"\n' "\n" "CHUNK_SIZE = 250\n" "AUDIOCPP_UNLOAD_MODELS = True\n" 'QWEN_API_URL = "http://127.0.0.1:7860"\n' 'CLONE_API_URL = "http://127.0.0.1:7861"\n' 'FASTER_API_URL = "http://127.0.0.1:8000"\n' 'AUDIOCPP_API_URL = "http://127.0.0.1:8080"\n' 'QWEN_REMOTE_URL = "http://127.0.0.1:7860"\n' 'CLONE_REMOTE_URL = "http://127.0.0.1:7861"\n' 'FASTER_REMOTE_URL = "http://127.0.0.1:8000"\n' 'AUDIOCPP_REMOTE_URL = "http://127.0.0.1:8080"\n', encoding="utf-8") with patch.object(hub.config, "__file__", str(path)): # Down to Chunk size, Enter -> editor, Ctrl-U + '300', # Enter; Tab -> Save, Enter; a key dismisses the flash. screen = FakeScreen(keys=[ FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, 10, 21, ord("3"), ord("0"), ord("0"), 10, 9, 10, 10]) hub._Hub(screen).screen_settings() text = path.read_text(encoding="utf-8") self.assertIn('AUDIO_FORMAT = "m4b"', text) self.assertIn("CHUNK_SIZE = 300", text) # The running session also picked up the change in-memory. self.assertEqual(hub.config.CHUNK_SIZE, 300) def test_settings_menu_updates_backend_ports(self): import tempfile original = {name: getattr(hub.config, name) for name in ("QWEN_API_URL", "CLONE_API_URL", "FASTER_API_URL", "AUDIOCPP_API_URL")} self.addCleanup(lambda: [setattr(hub.config, name, value) for name, value in original.items()]) with tempfile.TemporaryDirectory() as td: path = Path(td) / "config.py" path.write_text( 'QWEN_API_URL = "http://127.0.0.1:7860"\n' 'CLONE_API_URL = "http://127.0.0.1:7861"\n' 'FASTER_API_URL = "http://127.0.0.1:8000"\n' 'AUDIOCPP_API_URL = "http://127.0.0.1:8080"\n', encoding="utf-8") with patch.object(hub.config, "__file__", str(path)), \ patch.object(hub, "_sync_audiocpp_server_port"): hub._write_config({ "QWEN_API_URL": "http://127.0.0.1:7862", "CLONE_API_URL": "http://127.0.0.1:7863", "FASTER_API_URL": "http://127.0.0.1:8001", "AUDIOCPP_API_URL": "http://127.0.0.1:8081", }) text = path.read_text(encoding="utf-8") self.assertIn('QWEN_API_URL = "http://127.0.0.1:7862"', text) self.assertIn('CLONE_API_URL = "http://127.0.0.1:7863"', text) self.assertIn('FASTER_API_URL = "http://127.0.0.1:8001"', text) self.assertIn('AUDIOCPP_API_URL = "http://127.0.0.1:8081"', text) class AudiocppServerConfigTests(unittest.TestCase): """update_server_config_port: rewriting the checkout's server.json.""" def _make_checkout(self, td, port=8080): from backends import audiocpp as audiocpp_backend import json checkout = Path(td) / "audio.cpp" checkout.mkdir() server_json = checkout / "server.json" server_json.write_text( json.dumps({"host": "127.0.0.1", "port": port, "models": [{"id": "qwen"}]}, indent=2), encoding="utf-8") return audiocpp_backend, checkout, server_json def test_rewrites_existing_server_json_port(self): import tempfile import json with tempfile.TemporaryDirectory() as td: mod, checkout, server_json = self._make_checkout(td, port=8080) with patch.object(mod, "find_local_checkout", return_value=checkout): self.assertTrue(mod.update_server_config_port(9090)) data = json.loads(server_json.read_text(encoding="utf-8")) self.assertEqual(data["port"], 9090) # Other keys are preserved. self.assertEqual(data["host"], "127.0.0.1") self.assertEqual(data["models"], [{"id": "qwen"}]) def test_noop_when_port_unchanged(self): import tempfile with tempfile.TemporaryDirectory() as td: mod, checkout, server_json = self._make_checkout(td, port=8080) before = server_json.read_text(encoding="utf-8") with patch.object(mod, "find_local_checkout", return_value=checkout): self.assertTrue(mod.update_server_config_port(8080)) self.assertEqual(server_json.read_text(encoding="utf-8"), before) def test_false_when_no_checkout(self): from backends import audiocpp as audiocpp_backend with patch.object(audiocpp_backend, "find_local_checkout", return_value=None): self.assertFalse(audiocpp_backend.update_server_config_port(9090)) class ConfigureBackendsDispatchTests(unittest.TestCase): """The configure-backends screens dispatch their backend actions.""" def test_setup_screen_runs_the_backend_wizard_and_goes_back(self): info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0) with patch.object(info, "setup_screen") as mk_setup: result = hub._Hub(None).screen_setup(info)() mk_setup.assert_called_once_with(None) self.assertIs(result, tui.Wizard.BACK) def test_setup_screen_flashes_on_crash_and_goes_back(self): info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0) flashes = [] def boom(scr): raise RuntimeError("kaboom") with patch.object(info, "setup_screen", boom), \ patch.object(hub.tui, "flash", lambda scr, text, kind="warn": flashes.append(text)): result = hub._Hub(None).screen_setup(info)() self.assertIs(result, tui.Wizard.BACK) self.assertEqual(flashes, ["kaboom"]) def test_screen_install_runs_setup_inline_then_goes_back(self): # No stack frame for the setup: BACK pops past the picker straight # to the Configure menu instead of re-showing "Install Backend". info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0) with patch.object(hub._Hub, "_pick_backend", return_value=info), \ patch.object(info, "setup_screen") as mk_setup: result = hub._Hub(None).screen_install() mk_setup.assert_called_once_with(None) self.assertIs(result, tui.Wizard.BACK) def test_screen_install_esc_on_picker_goes_back_without_setup(self): info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0) with patch.object(hub._Hub, "_pick_backend", return_value=None), \ patch.object(info, "setup_screen") as mk_setup: result = hub._Hub(None).screen_install() mk_setup.assert_not_called() self.assertIs(result, tui.Wizard.BACK) def test_screen_install_flashes_on_crash_and_still_goes_back(self): info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0) flashes = [] def boom(scr): raise RuntimeError("kaboom") with patch.object(hub._Hub, "_pick_backend", return_value=info), \ patch.object(info, "setup_screen", boom), \ patch.object(hub.tui, "flash", lambda scr, text, kind="warn": flashes.append(text)): result = hub._Hub(None).screen_install() self.assertIs(result, tui.Wizard.BACK) self.assertEqual(flashes, ["kaboom"]) def test_screen_uninstall_runs_in_task_view_and_goes_back(self): info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0) with patch.object(hub._Hub, "_pick_backend", return_value=info), \ patch.object(hub.tui, "confirm", return_value=True), \ patch.object(hub.taskview, "run_steps", return_value=0) as mk_run, \ patch.object(info, "uninstall") as mk_uninstall, \ patch.object(hub.tui, "flash"): result = hub._Hub(None).screen_uninstall() # The uninstall runs as one task-view step on the session (no # suspend); executing the step forwards emit/cancel to uninstall. mk_run.assert_called_once() self.assertEqual(mk_run.call_args[0][0], None) steps = mk_run.call_args[0][2] self.assertEqual([step.title for step in steps], ["Uninstall qwen-tts"]) self.assertFalse(mk_run.call_args.kwargs["wait_on_finish"]) emit = lambda line: None steps[0].work(emit, None) mk_uninstall.assert_called_once_with(emit=emit, cancel=None) self.assertIs(result, tui.Wizard.BACK) def test_screen_uninstall_success_flashes_ok(self): flashes = [] def fake_flash(scr, text, kind="warn"): flashes.append((text, kind)) info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0) with patch.object(hub._Hub, "_pick_backend", return_value=info), \ patch.object(hub.tui, "confirm", return_value=True), \ patch.object(hub.taskview, "run_steps", return_value=0), \ patch.object(info, "uninstall"), \ patch.object(hub.tui, "flash", fake_flash): result = hub._Hub(None).screen_uninstall() self.assertIs(result, tui.Wizard.BACK) self.assertEqual(flashes[-1], ("qwen-tts uninstalled.", "ok")) def test_screen_uninstall_failure_flashes_error(self): flashes = [] def fake_flash(scr, text, kind="warn"): flashes.append((text, kind)) info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0) with patch.object(hub._Hub, "_pick_backend", return_value=info), \ patch.object(hub.tui, "confirm", return_value=True), \ patch.object(hub.taskview, "run_steps", return_value=1), \ patch.object(info, "uninstall"), \ patch.object(hub.tui, "flash", fake_flash): result = hub._Hub(None).screen_uninstall() self.assertIs(result, tui.Wizard.BACK) self.assertEqual(flashes[-1][1], "err") self.assertIn("Could not fully uninstall", flashes[-1][0]) def test_screen_uninstall_confirm_declined_skips_uninstall(self): info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0) with patch.object(hub._Hub, "_pick_backend", return_value=info), \ patch.object(hub.tui, "confirm", return_value=False) \ as mk_confirm, \ patch.object(hub.taskview, "run_steps") as mk_run, \ patch.object(info, "uninstall") as mk_uninstall: result = hub._Hub(None).screen_uninstall() mk_run.assert_not_called() mk_uninstall.assert_not_called() self.assertIs(result, tui.Wizard.BACK) # The confirm names the backend and is Esc-able (cancel_value set). self.assertIn("Uninstall qwen-tts?", mk_confirm.call_args[0][1]) def test_screen_uninstall_esc_on_confirm_backs_out(self): info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0) with patch.object(hub._Hub, "_pick_backend", return_value=info), \ patch.object(hub.tui, "confirm", return_value=tui.Wizard.BACK), \ patch.object(hub.taskview, "run_steps") as mk_run, \ patch.object(info, "uninstall") as mk_uninstall: result = hub._Hub(None).screen_uninstall() mk_run.assert_not_called() mk_uninstall.assert_not_called() self.assertIs(result, tui.Wizard.BACK) def _capture_flashes(self): flashes = [] def fake_flash(stdscr, text, kind="warn"): flashes.append((text, kind)) return patch.object(hub.tui, "flash", fake_flash), flashes def test_download_models_action_flashes_hand_install_guidance(self): with tempfile.TemporaryDirectory() as td: checkout = Path(td) (checkout / "server.json").write_text(json.dumps({"models": []}), encoding="utf-8") missing = [{"id": "qwen", "rel": "models/Qwen3-TTS-12Hz-0.6B-Base-GGUF"}] patch_flash, flashes = self._capture_flashes() with patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=checkout), \ patch.object(hub.audiocpp_backend, "missing_model_entries", return_value=missing), \ patch.object(hub.audiocpp_backend, "missing_model_install_guidance", return_value=[]), \ patch.object(hub.audiocpp_backend, "hand_install_guidance", return_value="do it by hand") as mk_hand, \ patch_flash: hub._download_models_action(None) self.assertEqual(flashes, [("do it by hand", "err")]) mk_hand.assert_called_once() def test_download_models_action_flashes_ok_when_nothing_missing(self): with tempfile.TemporaryDirectory() as td: checkout = Path(td) (checkout / "server.json").write_text(json.dumps({"models": []}), encoding="utf-8") patch_flash, flashes = self._capture_flashes() with patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=checkout), \ patch.object(hub.audiocpp_backend, "missing_model_entries", return_value=[]), \ patch_flash: hub._download_models_action(None) self.assertEqual(len(flashes), 1) self.assertEqual(flashes[0][1], "ok") def test_download_models_action_runs_in_task_view_and_installs(self): with tempfile.TemporaryDirectory() as td: checkout = Path(td) (checkout / "server.json").write_text(json.dumps({"models": []}), encoding="utf-8") missing = [{"id": "qwen", "rel": "models/q"}] guidance = [("qwen", "qwen3_tts_0_6b_base_q8_0")] patch_flash, flashes = self._capture_flashes() with patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=checkout), \ patch.object(hub.audiocpp_backend, "missing_model_entries", return_value=missing), \ patch.object(hub.audiocpp_backend, "missing_model_install_guidance", return_value=guidance), \ patch.object(hub.taskview, "run_steps", return_value=0) as mk_run, \ patch.object(hub.audiocpp_backend, "install_models") as mk_install, \ patch_flash: hub._download_models_action(None) # The downloads run in the TUI task view (one step), not via # suspend; executing the step forwards emit/cancel to # install_models. mk_run.assert_called_once() self.assertEqual(mk_run.call_args[0][0], None) steps = mk_run.call_args[0][2] self.assertEqual([step.title for step in steps], ["Download missing models"]) emit = lambda line: None steps[0].work(emit, None) mk_install.assert_called_once_with( checkout, guidance, emit=emit, cancel=None) self.assertEqual(len(flashes), 1) self.assertEqual(flashes[0][1], "ok") def test_download_models_action_flashes_error_when_no_checkout(self): patch_flash, flashes = self._capture_flashes() with patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=None), patch_flash: hub._download_models_action(None) self.assertEqual(len(flashes), 1) self.assertEqual(flashes[0][1], "err") def test_pick_backend_install_lists_uninstalled_only(self): captured = {} def fake_menu(stdscr, title, options, **kwargs): captured["options"] = options return tui.Wizard.BACK infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0), BackendInfo("faster", "faster-qwen3-tts", lambda: None, lambda: 0)] statuses = [BackendStatus("qwen", "qwen-tts", installed=True, configured=True), BackendStatus("faster", "faster-qwen3-tts", installed=False, configured=False)] with patch.object(hub, "REGISTRY", infos), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", fake_menu): result = hub._Hub(None)._pick_backend(installed_only=False) self.assertIsNone(result) self.assertEqual([label for label, _ in captured["options"]], ["faster-qwen3-tts"]) def test_pick_backend_install_skips_downloaded_not_built_audiocpp(self): captured = {} def fake_menu(stdscr, title, options, **kwargs): captured["options"] = options return tui.Wizard.BACK infos = [BackendInfo("audiocpp", "audio.cpp", lambda: None, lambda: 0), BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0)] statuses = [BackendStatus("audiocpp", "audio.cpp", installed=False, configured=False), BackendStatus("qwen", "qwen-tts", installed=False, configured=False)] # audio.cpp has a checkout (downloaded but not built): its next step # is the Build action, so it must not reappear in the Install picker. with patch.object(hub, "REGISTRY", infos), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=Path("/tmp/audiocpp")): result = hub._Hub(None)._pick_backend(installed_only=False) self.assertIsNone(result) self.assertEqual([label for label, _ in captured["options"]], ["qwen-tts"]) def test_pick_backend_install_lists_audiocpp_without_checkout(self): captured = {} def fake_menu(stdscr, title, options, **kwargs): captured["options"] = options return tui.Wizard.BACK infos = [BackendInfo("audiocpp", "audio.cpp", lambda: None, lambda: 0)] statuses = [BackendStatus("audiocpp", "audio.cpp", installed=False, configured=False)] with patch.object(hub, "REGISTRY", infos), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=None): result = hub._Hub(None)._pick_backend(installed_only=False) self.assertIsNone(result) self.assertEqual([label for label, _ in captured["options"]], ["audio.cpp"]) def test_pick_backend_uninstall_lists_installed_only(self): captured = {} def fake_menu(stdscr, title, options, **kwargs): captured["options"] = options return "qwen" infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0), BackendInfo("faster", "faster-qwen3-tts", lambda: None, lambda: 0)] statuses = [BackendStatus("qwen", "qwen-tts", installed=True, configured=True), BackendStatus("faster", "faster-qwen3-tts", installed=False, configured=False)] with patch.object(hub, "REGISTRY", infos), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "get", return_value=infos[0]) as mk_get: result = hub._Hub(None)._pick_backend(installed_only=True) self.assertEqual(result, infos[0]) mk_get.assert_called_once_with("qwen") self.assertEqual([label for label, _ in captured["options"]], ["qwen-tts"]) def test_pick_backend_uninstall_lists_downloaded_not_built_audiocpp(self): captured = {} def fake_menu(stdscr, title, options, **kwargs): captured["options"] = options return tui.Wizard.BACK infos = [BackendInfo("audiocpp", "audio.cpp", lambda: None, lambda: 0)] # Downloaded but not built (installed=False): still removable, so the # uninstall picker must list it (its checkout lives on disk). statuses = [BackendStatus("audiocpp", "audio.cpp", installed=False, configured=False)] with patch.object(hub, "REGISTRY", infos), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=Path("/tmp/audiocpp")): result = hub._Hub(None)._pick_backend(installed_only=True) self.assertIsNone(result) self.assertEqual([label for label, _ in captured["options"]], ["audio.cpp"]) class HubNavigationTests(unittest.TestCase): """Esc (and q) steps back exactly one screen across the whole hub.""" def setUp(self): tui._THEME.clear() self.addCleanup(tui._THEME.clear) def _info(self, key="audiocpp", label="audio.cpp"): return BackendInfo(key, label, lambda: None, lambda: 0) def _status(self, key="audiocpp", label="audio.cpp"): return BackendStatus(key, label, installed=True, configured=True) def _drive(self, script, statuses, registry): """Run the hub, feeding SCRIPT (one value per menu) to tui.menu. Records the title of every menu shown, in order. ``tui.Wizard.BACK`` in the script simulates Esc on that menu. """ titles = [] def menu(stdscr, title, options, **kwargs): titles.append(title) return script.pop(0) def get(key): return next((i for i in registry if i.key == key), None) with patch.object(hub, "REGISTRY", registry), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub, "get", side_effect=get), \ patch.object(hub.tui, "menu", menu), \ patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=None): hub._Hub(None).run() return titles def test_esc_on_wizard_first_screen_returns_to_configure(self): # The reported bug: Esc on the audio.cpp "Select TTS model # families" tree (the wizard's first screen) must land back on # "Configure backends", not the main menu. info = self._info() with patch.object(info, "setup_screen", return_value=1): titles = self._drive( ["configure_backends", ("configure", "audiocpp"), tui.Wizard.BACK, tui.Wizard.BACK], [self._status()], [info]) self.assertEqual( titles, ["tts-audiobook-generator", "Configure backends", "Configure backends", "tts-audiobook-generator"]) def test_esc_on_install_picker_returns_to_configure(self): registry = [self._info("audiocpp", "audio.cpp"), self._info("qwen", "qwen-tts")] statuses = [self._status("audiocpp", "audio.cpp"), BackendStatus("qwen", "qwen-tts", installed=False, configured=False)] titles = self._drive( ["configure_backends", "install", tui.Wizard.BACK, tui.Wizard.BACK, tui.Wizard.BACK], statuses, registry) self.assertEqual( titles, ["tts-audiobook-generator", "Configure backends", "Install Backend", "Configure backends", "tts-audiobook-generator"]) def test_esc_on_server_action_returns_one_screen_at_a_time(self): specs = [ServerSpec("qwen-custom", "http://127.0.0.1:7860", []), ServerSpec("qwen-clone", "http://127.0.0.1:7861", [])] status = BackendStatus("qwen", "qwen-tts", installed=True, configured=True, servers=specs) registry = [self._info("qwen", "qwen-tts")] titles = [] script = ["server", specs[0], tui.Wizard.BACK, tui.Wizard.BACK] def menu(stdscr, title, options, **kwargs): titles.append(title) return script.pop(0) with patch.object(hub, "REGISTRY", registry), \ patch.object(hub, "detect_all", return_value=[status]), \ patch.object(hub.tui, "menu", menu), \ patch.object(hub.common, "server_running", return_value=False), \ patch.object(hub.taskview, "run_steps", return_value=0), \ patch.object(hub.tui, "flash", lambda *a, **k: None), \ patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=None): hub._Hub(None).run() # Selecting a server toggles it directly (no action sub-menu), then # Esc steps back one screen at a time to the server list and main. self.assertEqual( titles, ["tts-audiobook-generator", "Start / Stop a server", "Start / Stop a server", "tts-audiobook-generator"]) def test_esc_on_main_menu_quits(self): titles = self._drive([tui.Wizard.BACK], [], []) self.assertEqual(titles, ["tts-audiobook-generator"]) if __name__ == "__main__": unittest.main()