"""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 converter.clients import audiocpp as audiocpp_client 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", "ok", "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, Help, Quit]. Quit is the 4th option (Down x3) then # Enter. screen = FakeScreen(keys=[FakeCurses.KEY_DOWN, 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", "Help", "Quit"]) def test_menu_lists_convert_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, ["Generate Audiobooks", "Configure Backends", "Settings", "Help", "Quit"]) # The status table is passed through, one row per backend. self.assertEqual(captured["rows"], [("qwen-tts", "installed", "ok", "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_server_menu_hidden_from_main_when_only_running(self): # Running but not installed (an external server) still unlocks # Convert — but Start/Stop needs the backend on this machine, and # lives in Configure Backends. 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, ["Generate Audiobooks", "Configure Backends", "Settings", "Help", "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]) def test_help_opens_viewer_and_backs_out(self): # Selecting Help opens the text viewer with the quick-start text # (real folder paths); closing it lands back on the main menu. calls = [] def fake_viewer(stdscr, title, lines, **kwargs): calls.append((title, list(lines), kwargs)) return kwargs.get("back_value") with patch.object(hub, "detect_all", return_value=[]), \ patch.object(hub.tui, "text_viewer", fake_viewer): # Help is the 3rd main-menu option (Down x2), then Enter; # back on the main menu Quit is the 4th (Down x3), Enter. screen = FakeScreen(keys=[FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, 10, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, 10]) result = hub._Hub(screen).run() self.assertIsNone(result) self.assertEqual(len(calls), 1) title, items, kwargs = calls[0] self.assertEqual(title, "Help") self.assertIs(kwargs.get("back_value"), tui.Wizard.BACK) # Flatten the viewer rows: (segments, indent) pairs join their # texts; plain strings (the blank lines) pass through. text = "\n".join( "".join(part for part, _ in item[0]) if isinstance(item, tuple) else item for item in items) self.assertIn("1. Put your ebooks (epub, txt, or pdf) here:", text) self.assertIn("2. Put any .wavs of voices to clone here:", text) self.assertIn("3. If no backend is installed, go to Configure " "Backends > Install Backend and install audio.cpp.", text) self.assertIn("4. Select TTS models to install. If you're unsure, " "try these qwen3-tts models:", text) self.assertIn("qwen3_tts_1_7b_base_q8_0", text) self.assertIn("qwen3_tts_1_7b_customvoice_q8_0", text) self.assertIn("5. Go to Generate Audiobooks.", text) # The "no need to manually start/stop" sentence sits on its own # indented line below the step-5 paragraph (one leading space # beyond the 2-space indent, lining up with the step text). self.assertIn("stop it.\n There is no need to manually " "start/stop servers.", text) self.assertIn("6. Generated audiobooks (m4b, mp3, etc.) will " "output here:", text) # The three folder paths are their own indented, white-bold # ("input") rows; the numbered steps start at the margin and # the indented rows carry a leading space of their own, so # they line up with the step text after the "N. " prefixes. # They resolve live from the converter module, so a Settings # change this session is reflected. self.assertIn(([(" " + str(hub.converter_mod.BOOKS_FOLDER), "input")], 1), items) self.assertIn(([(" " + str(hub.common.VOICES_DIR), "input")], 1), items) self.assertIn(([(" " + str(hub.converter_mod.AUDIOBOOKS_FOLDER), "input")], 1), items) self.assertEqual(items[0][1], 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) # Start/Stop (qwen installed), then Install (faster uninstalled), # then Uninstall; no audio.cpp means no model actions. qwen has no # Configure entry (its wizard asks nothing to configure). self.assertEqual([label for label, _ in captured["options"]], ["Start/Stop Backend Servers", "Install Backend", "Uninstall Backend"]) # ...the shared status table carries the states instead (no title). self.assertIsNone(captured.get("table_title")) self.assertEqual( captured["table_rows"], [("qwen-tts", "installed", "ok", "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_lists_update_between_install_uninstall(self): captured = {} infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0, update=lambda **kw: 0), BackendInfo("faster", "faster-qwen3-tts", lambda: None, lambda: 0, update=lambda **kw: 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", self._capture_menu(captured)), \ patch.object(hub.shutil, "which", return_value="/x"): result = hub._Hub(None).screen_configure() self.assertIs(result, tui.Wizard.BACK) # Update sits between Install and Uninstall; it is offered once for # the whole set of installed backends (faster has nothing on disk # and so contributes nothing). Start/Stop heads the tail actions # (qwen is installed). self.assertEqual([label for label, _ in captured["options"]], ["Start/Stop Backend Servers", "Install Backend", "Update Backends", "Uninstall 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; # Start/Stop + 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", "Start/Stop Backend Servers", "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 # Start/Stop and 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", "Start/Stop Backend Servers", "Uninstall Backend"]) def test_configure_menu_offers_qwens_per_model_manager(self): # qwen ships a dedicated configure screen (per-model weight # installs): once installed, "Configure qwen-tts" appears in the # flat action list. captured = {} infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0, uninstall=lambda **kwargs: 0, configure_screen=lambda scr: 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="/usr/bin/ffmpeg"): result = hub._Hub(None).screen_configure() self.assertIs(result, tui.Wizard.BACK) self.assertEqual(self._labels(captured["options"]), ["Configure qwen-tts", "Start/Stop Backend Servers", "Uninstall Backend"]) self.assertEqual(captured["table_rows"], [("qwen-tts", "installed", "ok", "body")]) def test_selecting_qwen_runs_its_configure_screen_not_setup(self): ran = [] invalidated = [] def configure(scr): ran.append("configure") return 0 def setup(scr): ran.append("setup") return 0 info = BackendInfo("qwen", "qwen-tts", lambda: None, setup, uninstall=lambda **kwargs: 0, configure_screen=configure) def first_pick(stdscr, title, options, **kwargs): return ("configure", "qwen") statuses = [BackendStatus("qwen", "qwen-tts", installed=True, configured=True)] with patch.object(hub, "REGISTRY", [info]), \ patch.object(hub, "get", return_value=info), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", first_pick), \ patch.object(hub, "invalidate_detect_cache", side_effect=lambda: invalidated.append(True)), \ patch.object(hub.shutil, "which", return_value="/x"): leaf = hub._Hub(None).screen_configure() # The selection pushes the backend's configure screen as one # leaf of the wizard stack (which invalidates the status # cache when it finishes). self.assertIs(leaf(), tui.Wizard.BACK) self.assertEqual(ran, ["configure"]) self.assertEqual(invalidated, [True]) def test_bare_qwen_without_configure_screen_still_has_no_entry(self): # Without a dedicated configure screen, plain qwen stays excluded # from Configure Backends (its wizard asks nothing to configure). captured = {} infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0, uninstall=lambda **kwargs: 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(self._labels(captured["options"]), ["Start/Stop Backend Servers", "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"], "Generate Audiobooks") # 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.assertIsNone(captured.get("table_title")) self.assertEqual( captured["table_rows"], [("qwen-tts", "installed", "ok", "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.assertIsNone(captured.get("table_title")) 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) # Family voice policies are resolved from the local audio.cpp # checkout's model_specs, which a fresh clone does not have (the # checkout is downloaded by setup): seed the client's spec cache # with the classifications these tests rely on, so they stay # hermetic. Unknown families keep the clone-only default. spec_cache = audiocpp_client._FAMILY_SPECS spec_cache.clear() spec_cache.update({ "higgs_audio_tts": {"tasks": ["tts", "clone"]}, "supertonic": {"tasks": ["tts"]}, }) self.addCleanup(spec_cache.clear) # Keys shared by every backend entry; a "-remote" backend's other # option keys are namespaced under "." in the form dict # (mirroring hub.py), so _form_values maps them automatically. _COMMON_KEYS = frozenset(("backend", "single_file")) def _form_values(self, **overrides): """A fully-populated form result, with sensible defaults. Output format, language, speed, debug, and stop-and-exit are no longer form fields: they live in config.py (Settings) and reach the run kwargs via _common_kwargs(). """ values = {"single_file": False} values.update(overrides) backend = values.get("backend") or "" if backend.endswith("-remote"): prefix = f"{backend}." values = {(prefix + key if key not in self._COMMON_KEYS else key): value for key, value in values.items()} 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 or f["key"].endswith("." + 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"]) # Pin the config settings the common kwargs are built from, so # this test does not depend on the user's saved config.py values. with patch.object(hub.config, "STOP_SERVER_AND_EXIT", True), \ patch.object(hub.config, "AUDIO_FORMAT", "m4b"), \ patch.object(hub.config, "LANGUAGE", "English"), \ patch.object(hub.config, "SPEED", 1.25), \ patch.object(hub.config, "DEBUG", False): self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="narrator", instructions="") 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") # The output settings come from config (the Settings menu), not # the form. self.assertEqual(kwargs["output_format"], "m4b") self.assertEqual(kwargs["language"], "English") self.assertEqual(kwargs["speed"], 1.25) self.assertFalse(kwargs["single_file"]) self.assertFalse(kwargs["debug"]) self.assertTrue(kwargs["stop_and_exit"]) # 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, "Generate Audiobooks") self.assertEqual([f["key"] for f in fields], ["backend", "audiocpp-remote.model_id", "audiocpp-remote.audiocpp_voice", "audiocpp-remote.instructions", "audiocpp-remote.request_options", "single_file"]) self.assertEqual(form_kwargs["buttons"], ("Generate!", "Cancel")) self.assertTrue(form_kwargs["start_on_buttons"]) # 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); ids are # padded so the capability columns line up across entries. A mixed # tts+clone family reads as the "tts" and "clone" columns. self.assertEqual(self._field("model_id")["choices"], [("higgs tts clone", "higgs")]) def test_model_menu_lines_the_type_column_up(self): # Ids are padded to the widest id, and every capability word sits # in its own fixed column (tts | clone | design): the picker reads # as a table where "tts", "clone" (and "design") line up. self._patch_remote( [{"id": "short", "family": "higgs_audio_tts", "task": "tts"}, {"id": "a-much-longer-model-id", "family": "qwen3_tts", "task": "tts"}]) self._answer_form(backend="audiocpp-remote", model_id="short", audiocpp_voice="", instructions="") cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNotNone(cmd) choices = self._field("model_id")["choices"] # "a-much-longer-model-id" is 22 columns wide; both capabilities # start at the same offsets: "tts" at 24, "clone" at 29. The # clone-only qwen3_tts entry leaves the tts column blank. self.assertEqual(choices[0], ("short".ljust(22) + " tts clone", "short")) self.assertEqual(choices[1], ("a-much-longer-model-id".ljust(22) + " clone", "a-much-longer-model-id")) self.assertEqual({label.index("clone") for label, _ in choices}, {29}) self.assertEqual({label.index("tts") for label, _ in choices if "tts" in label}, {24}) # A plain qwen3_tts entry (no CustomVoice in the id) is clone-only. self.assertEqual(choices[1][1], "a-much-longer-model-id") self.assertTrue(choices[1][0].endswith("clone")) self.assertNotIn("design", choices[1][0]) def test_model_menu_shows_design_for_design_capable_families(self): # A family whose spec advertises a design task designs on the # regular entry from the Instructions text: the model menu grows # a "design" column entry behind tts/clone. spec_cache = audiocpp_client._FAMILY_SPECS spec_cache["omnivoice"] = {"tasks": ["tts", "clone", "design"]} self.addCleanup(spec_cache.pop, "omnivoice", None) self._patch_remote( [{"id": "omnivoice", "family": "omnivoice", "task": "tts"}], voices=["narrator"]) self._answer_form(backend="audiocpp-remote", model_id="omnivoice", audiocpp_voice="", instructions="") cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNotNone(cmd) self.assertEqual(self._field("model_id")["choices"], [("omnivoice tts clone design", "omnivoice")]) def test_model_menu_shows_design_only_on_vdes_entries(self): # A task-"vdes" entry is a design model: its row shows only the # design column (no tts, no clone), aligned with other rows. self._patch_remote( [{"id": "moss_voicegen", "family": "moss_voicegen", "task": "vdes"}]) self._answer_form(backend="audiocpp-remote", model_id="moss_voicegen", audiocpp_voice=None, instructions="a warm narrator") cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNotNone(cmd) self.assertEqual( self._field("model_id")["choices"], [("moss_voicegen design", "moss_voicegen")]) def test_model_menu_keeps_design_off_qwen3_tts_nondesign_entries(self): # Qwen3-TTS designs only through its separate VoiceDesign entry: # the Base row stays "clone" while the vdes row shows "design", # both words on the same columns. self._patch_remote( [{"id": "Qwen3-TTS-12Hz-1.7B-Base-GGUF", "family": "qwen3_tts", "task": "tts"}, {"id": "Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF", "family": "qwen3_tts", "task": "vdes"}]) self._answer_form( backend="audiocpp-remote", model_id="Qwen3-TTS-12Hz-1.7B-Base-GGUF", audiocpp_voice="", instructions="") cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNotNone(cmd) choices = self._field("model_id")["choices"] base = "Qwen3-TTS-12Hz-1.7B-Base-GGUF" design = "Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF" labels = {value: label for label, value in choices} self.assertEqual(labels, {base: base.ljust(len(design)) + " clone", design: design.ljust(len(design)) + " design"}) # "clone" and "design" each sit on one shared column. self.assertEqual({label.index("clone") for label, _ in choices if "clone" in label}, {label.index("design") - 7 for label, _ in choices if "design" in label}) 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"}]) 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"]) 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")]) # Instructions are optional on clone entries too (a style/delivery # instruction, or the voice itself on families that read one). instr = self._field("instructions") self.assertTrue(instr["visible"](fields)) def test_audiocpp_model_switch_keeps_the_picked_voice(self): # Switching models whose voice list is unchanged (two clone-only # entries sharing one server's voices) keeps the picked voice # instead of snapping back to the list's first entry. self._patch_remote( [{"id": "alpha", "family": "chatterbox", "task": "clon"}, {"id": "beta", "family": "qwen3_tts", "task": "tts"}], voices=["narrator", "second"]) self._answer_form(backend="audiocpp-remote", model_id="alpha", audiocpp_voice="second", instructions="") self._convert(None, [self._remote("audiocpp", "audio.cpp")]) fields = self.tui.forms_seen[0][1] model_field = self._field("model_id") voice_field = self._field("audiocpp_voice") # The form opens on the list's first voice; the user picks another. self.assertEqual(voice_field["value"], "narrator") voice_field["value"] = "second" model_field["value"] = "beta" model_field["on_change"](fields) self.assertEqual(voice_field["value"], "second") model_field["value"] = "alpha" model_field["on_change"](fields) self.assertEqual(voice_field["value"], "second") def test_audiocpp_local_model_switch_keeps_the_picked_voice(self): # The managed entry's voice list is shared by every model in # server.json, so switching models keeps the picked voice. with tempfile.TemporaryDirectory() as td: root = Path(td) (root / "server.json").write_text(json.dumps({ "models": [{"id": "qwen-1_7b", "family": "qwen3_tts", "task": "tts"}, {"id": "qwen-0_6b", "family": "qwen3_tts", "task": "tts"}], "voice_dir": str(root), }), encoding="utf-8") (root / "Narrator.wav").write_bytes(b"x") (root / "Second.wav").write_bytes(b"x") with patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=root): self._answer_form(backend="audiocpp", model_id="qwen-1_7b", audiocpp_voice="Second", instructions="") self._convert(None, [self._ready("audiocpp", "audio.cpp")]) fields = self.tui.forms_seen[0][1] model_field = self._field("model_id") voice_field = self._field("audiocpp_voice") self.assertEqual(voice_field["value"], "Narrator") voice_field["value"] = "Second" model_field["value"] = "qwen-0_6b" model_field["on_change"](fields) self.assertEqual(voice_field["value"], "Second") def test_audiocpp_model_switch_resets_when_the_pick_is_gone(self): # A remote server may host different voices per model: switching # to a model whose list no longer offers the pick falls back to # that model's first voice (and re-points again on the way back). models = patch.object( hub.audiocpp_backend, "fetch_server_models", lambda url: [{"id": "alpha", "family": "chatterbox", "task": "clon"}, {"id": "beta", "family": "qwen3_tts", "task": "tts"}]) voices = patch.object( hub.audiocpp_backend, "fetch_server_voices", lambda url, model_id: {"alpha": ["narrator", "second"], "beta": ["other"]}[model_id]) with models, voices: self._answer_form(backend="audiocpp-remote", model_id="alpha", audiocpp_voice="second", instructions="") self._convert(None, [self._remote("audiocpp", "audio.cpp")]) fields = self.tui.forms_seen[0][1] model_field = self._field("model_id") voice_field = self._field("audiocpp_voice") model_field["value"] = "beta" model_field["on_change"](fields) self.assertEqual(voice_field["value"], "other") model_field["value"] = "alpha" model_field["on_change"](fields) self.assertEqual(voice_field["value"], "narrator") def test_audiocpp_model_switch_resets_across_capabilities(self): # Keep-the-pick only applies within one voice list: a clone pick # never survives a move to a built-in-speaker entry (and vice # versa), and a design entry clears the voice again. self._patch_remote( [{"id": "clone", "family": "chatterbox", "task": "clon"}, {"id": "Qwen3-TTS-CustomVoice-GGUF", "family": "qwen3_tts", "task": "tts"}, {"id": "design", "family": "qwen3_tts", "task": "vdes"}], voices=["narrator", "second"]) self._answer_form(backend="audiocpp-remote", model_id="clone", audiocpp_voice="second", instructions="") self._convert(None, [self._remote("audiocpp", "audio.cpp")]) fields = self.tui.forms_seen[0][1] model_field = self._field("model_id") voice_field = self._field("audiocpp_voice") voice_field["value"] = "second" model_field["value"] = "Qwen3-TTS-CustomVoice-GGUF" model_field["on_change"](fields) self.assertEqual(voice_field["value"], hub.QWEN3_TTS_SPEAKERS[0]) voice_field["value"] = "Ryan" model_field["value"] = "clone" model_field["on_change"](fields) self.assertEqual(voice_field["value"], "narrator") model_field["value"] = "design" model_field["on_change"](fields) self.assertIsNone(voice_field["value"]) def test_audiocpp_remote_without_voices_refuses_generate_with_hint(self): # A clone-capable entry (e.g. Qwen Base) whose server lists no # voices at all: the Voice picker stays visible but is empty — # opening it flashes where to configure voices instead of # crashing — and Generate! refuses with the same hint. self._patch_remote( [{"id": "Qwen3-TTS-12Hz-1.7B-Base-GGUF", "family": "qwen3_tts", "task": "tts"}], voices=[]) self._answer_form( backend="audiocpp-remote", model_id="Qwen3-TTS-12Hz-1.7B-Base-GGUF", audiocpp_voice="", instructions="") self._convert(None, [self._remote("audiocpp", "audio.cpp")]) fields = self.tui.forms_seen[0][1] voice_field = self._field("audiocpp_voice") self.assertTrue(voice_field["visible"](fields)) self.assertEqual(voice_field["choices"](fields), []) error = voice_field["validate"]("") self.assertIsNotNone(error) self.assertIn(".wav", error) self.assertIn("server", error) def test_audiocpp_local_without_voice_dir_points_at_configure(self): # The managed entry's server.json has no voice_dir: clone-capable # models get an empty Voice picker whose hint sends the user to # Configure Backends instead of crashing on menu(). 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): self._answer_form(backend="audiocpp", model_id="qwen", audiocpp_voice="", instructions="") self._convert(None, [self._ready("audiocpp", "audio.cpp")]) fields = self.tui.forms_seen[0][1] voice_field = self._field("audiocpp_voice") self.assertTrue(voice_field["visible"](fields)) self.assertEqual(voice_field["choices"](fields), []) error = voice_field["validate"]("") self.assertIsNotNone(error) self.assertIn(".wav", error) self.assertIn("Configure Backends", error) 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=[]) 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"}]) 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_forwards_instructions(self): # Instructions are optional on clone entries: the mapper forwards a # submitted instruction (style/delivery control, or the voice itself # on families that condition synthesis on instructions alone). self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) 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.assertEqual(cmd[2]["instructions"], "stale description") def test_audiocpp_required_voice_validates(self): # A clone-only family (Chatterbox) needs a --voice; a blank value # refuses. A mixed tts+clone family (higgs_audio_tts) accepts the # blank pick — it means plain TTS without a reference. self._patch_remote( [{"id": "chatterbox", "family": "chatterbox", "task": "clon"}, {"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) self._answer_form(backend="audiocpp-remote", model_id="chatterbox", 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")) fields = self.tui.forms_seen[0][1] model_field = self._field("model_id") voice_field["value"] = "narrator" model_field["value"] = "higgs" model_field["on_change"](fields) # Mixed family: the blank (no clone) pick is valid. self.assertIsNone(voice_field["validate"]("")) def test_audiocpp_builtin_speaker_entry_labels_the_field_built_in(self): # On a CustomVoice entry the Voice field is labelled "Built-in # voice" — the pick is one of the model's speakers, not a clone ref. self._patch_remote( [{"id": "Qwen3-TTS-CustomVoice-GGUF", "family": "qwen3_tts", "task": "tts"}]) self._answer_form( backend="audiocpp-remote", model_id="Qwen3-TTS-CustomVoice-GGUF", audiocpp_voice="Vivian", instructions="") self._convert(None, [self._remote("audiocpp", "audio.cpp")]) fields = self.tui.forms_seen[0][1] label = self._field("audiocpp_voice")["label"] self.assertEqual(label(fields), "Built-in voice") def test_audiocpp_clone_entry_labels_the_field_voice_to_clone(self): # Any non-speaker entry clones a server-side preset: "Voice to clone". self._patch_remote( [{"id": "Qwen3-TTS-Base-GGUF", "family": "qwen3_tts", "task": "tts"}], voices=["narrator"]) self._answer_form(backend="audiocpp-remote", model_id="Qwen3-TTS-Base-GGUF", audiocpp_voice="narrator", instructions="") self._convert(None, [self._remote("audiocpp", "audio.cpp")]) fields = self.tui.forms_seen[0][1] label = self._field("audiocpp_voice")["label"] self.assertEqual(label(fields), "Voice to clone") def test_audiocpp_clone_with_instructions_accepts_an_empty_voice(self): # An Instructions text substitutes for the voice on clone-only # families: blank Voice passes validation when instructions are # present, and is still refused without one. self._patch_remote( [{"id": "chatterbox", "family": "chatterbox", "task": "clon"}], voices=["narrator"]) self._answer_form(backend="audiocpp-remote", model_id="chatterbox", audiocpp_voice="", instructions="") self._convert(None, [self._remote("audiocpp", "audio.cpp")]) fields = self.tui.forms_seen[0][1] voice = self._field("audiocpp_voice") instr = self._field("instructions") instr["value"] = "an elderly narrator" self.assertIsNone(voice["validate"]("")) instr["value"] = "" self.assertIsNotNone(voice["validate"]("")) def test_audiocpp_no_voices_with_instructions_still_converts(self): # A clone-only entry whose server lists no voices is refused by # default — but an instruction provides the voice instead. self._patch_remote( [{"id": "chatterbox", "family": "chatterbox", "task": "clon"}], voices=[]) self._answer_form(backend="audiocpp-remote", model_id="chatterbox", audiocpp_voice="", instructions="") cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) fields = self.tui.forms_seen[0][1] voice = self._field("audiocpp_voice") instr = self._field("instructions") instr["value"] = "" # No voices and no instruction: the usual refusal hint. self.assertIsNotNone(voice["validate"]("")) # An instruction provides the voice instead. instr["value"] = "designed narrator" self.assertIsNone(voice["validate"]("")) def test_audiocpp_pure_tts_entry_hides_the_voice_menu(self): # Pure-TTS families (spec tasks without "clone") synthesize with # no voice at all: the Voice menu is hidden entirely, the model # menu reads "tts", and Generate! sends no voice. self._patch_remote( [{"id": "supertonic", "family": "supertonic", "task": "tts"}]) self._answer_form(backend="audiocpp-remote", model_id="supertonic", audiocpp_voice=None, 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.assertFalse(voice_field["visible"](fields)) self.assertEqual(self._field("model_id")["choices"], [("supertonic tts", "supertonic")]) def test_audiocpp_mixed_family_offers_a_built_in_blank_pick(self): # Mixed tts+clone families lead the Voice menu with a blank # " (no clone)" pick meaning plain TTS (the model's own # default voice, no reference cloned), and it is the default. self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) 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.assertTrue(voice_field["visible"](fields)) self.assertEqual(voice_field["choices"](fields), [(" (no clone)", ""), ("narrator", "narrator")]) # A kept clone pick survives a mixed-family switch; blank is valid. voice_field["value"] = "narrator" self.assertIsNone(voice_field["validate"]("narrator")) self.assertIsNone(voice_field["validate"]("")) def test_audiocpp_request_options_map_to_kwargs(self): self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) self._answer_form( backend="audiocpp-remote", model_id="higgs", audiocpp_voice="narrator", instructions="", request_options="emotion=neutral, speed=1.1") cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertEqual(cmd[2]["request_options"], {"emotion": "neutral", "speed": "1.1"}) def test_audiocpp_request_options_validate_and_recover_from_garbage(self): self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="narrator", instructions="", request_options="oops") cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) options_field = self._field("request_options") self.assertIsNone(options_field["validate"]("emotion=neutral")) self.assertIsNotNone(options_field["validate"]("oops")) # The scripted form bypasses validation, so a garbage submit falls # back to no options instead of crashing the mapper. self.assertEqual(cmd[2]["request_options"], {}) # ------------------------------------------------------------------ # audio.cpp: Request options gated by model_specs option support # ------------------------------------------------------------------ def _specs_checkout(self, families_with_options=()): """A fake checkout whose specs mark FAMILIES_WITH_OPTIONS supportive.""" root = Path(self.enterContext(tempfile.TemporaryDirectory())) specs = root / "model_specs" specs.mkdir() for family in ("higgs_audio_tts", "qwen3_tts"): request = ([{"id": "temperature"}] if family in families_with_options else []) spec = {"family": family, "display_name": family.title(), "packages": [{"id": f"{family}_q8_0", "default": True, "format": "gguf", "target_directory": f"{family}-GGUF"}]} if request: spec["options"] = {"request": request} (specs / f"{family}.json").write_text(json.dumps(spec), encoding="utf-8") return root def test_options_field_visible_when_spec_proves_support(self): self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) with patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=self._specs_checkout(("higgs_audio_tts",))): self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="narrator", instructions="") cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) fields = self.tui.forms_seen[0][1] self.assertTrue(self._field("request_options")["visible"](fields)) self.assertIsNotNone(cmd) def test_options_field_hidden_when_spec_lacks_the_family(self): self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) # A checkout exists but only qwen3_tts declares request options: # higgs is provably unsupported -> hidden. with patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=self._specs_checkout(("qwen3_tts",))): self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="narrator", instructions="") self._convert(None, [self._remote("audiocpp", "audio.cpp")]) fields = self.tui.forms_seen[0][1] self.assertFalse(self._field("request_options")["visible"](fields)) def test_options_field_hidden_without_a_local_checkout(self): # Unknown support (no specs anywhere) hides the field — strict. self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) with patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=None): self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="narrator", instructions="") self._convert(None, [self._remote("audiocpp", "audio.cpp")]) fields = self.tui.forms_seen[0][1] self.assertFalse(self._field("request_options")["visible"](fields)) def test_instructions_help_is_short_and_shared(self): # One compact static help text for every capability: two lines, # naming style instructions, partial clone-model support, and an # example. (Design entries enforce their requirement by validation.) self._patch_remote([ {"id": "design", "family": "qwen3_tts", "task": "vdes"}, {"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}, ], voices=["narrator"]) self._answer_form(backend="audiocpp-remote", model_id="design", audiocpp_voice=None, instructions="A warm British narrator") self._convert(None, [self._remote("audiocpp", "audio.cpp")]) instr = self._field("instructions") self.assertEqual(instr["help"], [ "TTS style instructions. Supported by some clone models. Example:", '"Speak in a calm, soothing, and happy tone."', ]) def test_options_help_is_two_lines_with_examples(self): self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) with patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=self._specs_checkout(("qwen3_tts", "higgs_audio_tts"))): self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="narrator", instructions="") self._convert(None, [self._remote("audiocpp", "audio.cpp")]) options = self._field("request_options") fields = self.tui.forms_seen[0][1] self.assertTrue(options["visible"](fields)) self.assertEqual(len(options["help"]), 2) help_text = "\n".join(options["help"]) self.assertIn("KEY=VALUE", help_text) self.assertIn("emotion=neutral", help_text) def test_language_passes_through_from_config(self): # The Settings Language setting travels on the run kwargs as-is; # the converter normalizes it (short codes included). with patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["Vivian"]), \ patch.object(hub.config, "LANGUAGE", "en"): self._answer_form(backend="qwen", mode="custom", speaker="Vivian", clone="") cmd = self._convert(None, [self._ready("qwen", "qwen-tts")]) self.assertEqual(cmd[2]["language"], "en") 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": "chatterbox", "family": "chatterbox", "task": "clon"}], voices=[]) self._answer_form(backend="audiocpp-remote", model_id="chatterbox", 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): 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_audiocpp_local_rehosts_clone_only_entries(self): # server.json written before clone-only hosting existed carries # task "tts" for Chatterbox: opening the form re-hosts it with # task "clon" on disk and flags the run so the autostart plan # restarts the managed server with the corrected config. with tempfile.TemporaryDirectory() as td: root = Path(td) server_json = root / "server.json" server_json.write_text(json.dumps({ "models": [{"id": "Chatterbox-GGUF", "family": "chatterbox", "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): self._answer_form(backend="audiocpp", model_id="Chatterbox-GGUF", audiocpp_voice="Narrator", instructions="") cmd = self._convert(None, [self._ready("audiocpp", "audio.cpp")]) self.assertIsNotNone(cmd) self.assertTrue(cmd[2]["audiocpp_rehost"]) # The repair was persisted: the entry is hosted with "clon". data = json.loads(server_json.read_text(encoding="utf-8")) self.assertEqual(data["models"][0]["task"], "clon") 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): 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")]) # The two entries' fields are namespaced, so both carry their own # values and picking one never leaks the other's into the run. keys = [f["key"] for f in fields] self.assertIn("model_id", keys) self.assertIn("audiocpp-remote.model_id", keys) def test_managed_and_remote_entries_do_not_overwrite_each_other(self): # Regression: the form returns one flat {key: value} dict. When # managed and remote entries shared field keys, the remote entry's # hidden defaults silently won over the user's edits on whichever # entry was selected. self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) statuses = [self._ready("audiocpp", "audio.cpp"), self._remote("audiocpp", "audio.cpp")] common = {"single_file": False} 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): # Managed selected: its picks must survive next to the # remote entry's same-shaped fields. self.tui.form_script.append({ "backend": "audiocpp", "model_id": "qwen", "audiocpp_voice": "", "instructions": "", "audiocpp-remote.model_id": "higgs", "audiocpp-remote.audiocpp_voice": "narrator", **common}) managed_cmd = self._convert(None, statuses) self.assertIsNotNone(managed_cmd) self.assertEqual(managed_cmd[2]["model_id"], "qwen") # The remote entry's hidden "narrator" voice must not leak # into the managed run (the old duplicate-key behavior). self.assertIsNone(managed_cmd[2]["voice"]) # Remote selected: its picks win instead. self.tui.form_script.append({ "backend": "audiocpp-remote", "audiocpp-remote.model_id": "higgs", "audiocpp-remote.audiocpp_voice": "narrator", "model_id": "qwen", **common}) remote_cmd = self._convert(None, statuses) self.assertIsNotNone(remote_cmd) self.assertEqual(remote_cmd[2]["model_id"], "higgs") self.assertEqual(remote_cmd[2]["voice"], "narrator") def test_audiocpp_remote_mapper_adds_api_url(self): self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) 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: the per-run combine-all-chapters toggle # ------------------------------------------------------------------ def test_common_fields_hide_combine_for_m4b(self): # The toggle's visibility tracks the configured output format # (the Settings menu's Audio format), not a form field. fields = hub._common_fields() single = next(f for f in fields if f["key"] == "single_file") with patch.object(hub.config, "AUDIO_FORMAT", "m4b"): self.assertFalse(single["visible"](fields)) with patch.object(hub.config, "AUDIO_FORMAT", "mp3"): self.assertTrue(single["visible"](fields)) # ------------------------------------------------------------------ # qwen: model picker (Base / CustomVoice / VoiceDesign) # ------------------------------------------------------------------ def test_qwen_builds_speaker_and_clone_form(self): with patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["Vivian", "Serena"]): self._answer_form(backend="qwen", mode="custom", speaker="Serena", clone="") cmd = self._convert(None, [self._ready("qwen", "qwen-tts")]) self.assertEqual(cmd[0], "convert") self.assertEqual(cmd[1], hub.BACKEND_QWEN) self.assertIsNone(cmd[2]["clone"]) self.assertIsNone(cmd[2].get("instructions")) # The picked speaker travels with the run; nothing is persisted. self.assertEqual(cmd[2]["voice"], "Serena") fields = self.tui.forms_seen[0][1] self.assertEqual([f["key"] for f in fields], ["backend", "mode", "speaker", "clone_dir", "clone", "qwen_instructions", "single_file"]) mode_field = self._field("mode") # Model names are padded to the widest ("CustomVoice"/"VoiceDesign" # are 11 columns) plus a two-space gutter, so every (purpose) opens # on the same column and the picker reads as a two-column table. self.assertEqual(mode_field["choices"], [("CustomVoice".ljust(11) + " (built-in voices)", "custom"), ("Base".ljust(11) + " (voice cloning)", "clone"), ("VoiceDesign".ljust(11) + " (design)", "design")]) self.assertEqual({label.index("(") for label, _ in mode_field["choices"]}, {13}) speaker_field = self._field("speaker") clone_dir_field = self._field("clone_dir") # The .wav directory browser alerts on the .wavs it lists. self.assertIs(clone_dir_field["info"], hub.common.wav_dir_info) self.assertIs(clone_dir_field["preview"], hub.common.wav_dir_preview) clone_field = self._field("clone") design_field = self._field("qwen_instructions") # Speaker shows in custom mode; the .wav directory browser and # picker in clone mode and the instruction in design mode. self.assertTrue(speaker_field["visible"](fields)) self.assertFalse(clone_dir_field["visible"](fields)) self.assertFalse(clone_field["visible"](fields)) self.assertFalse(design_field["visible"](fields)) mode_field["value"] = "clone" self.assertFalse(speaker_field["visible"](fields)) self.assertTrue(clone_dir_field["visible"](fields)) self.assertTrue(clone_field["visible"](fields)) mode_field["value"] = "design" self.assertFalse(speaker_field["visible"](fields)) self.assertFalse(clone_dir_field["visible"](fields)) self.assertFalse(clone_field["visible"](fields)) self.assertTrue(design_field["visible"](fields)) def test_qwen_design_mode_passes_instructions(self): with patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["Vivian"]): self._answer_form(backend="qwen", mode="design", qwen_instructions="A warm narrator") 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"], None) self.assertEqual(cmd[2]["instructions"], "A warm narrator") # Per-run choices are not persisted to the config file. mk_update.assert_not_called() def test_qwen_clone_mode_passes_path(self): with patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["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") mk_update.assert_not_called() def test_qwen_custom_mode_passes_the_speaker(self): with patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["Vivian"]): self._answer_form(backend="qwen", mode="custom", speaker="Vivian") with patch.object(hub.common, "update_config_value") as mk_update: cmd = self._convert(None, [self._ready("qwen", "qwen-tts")]) self.assertIsNotNone(cmd) self.assertEqual(cmd[2]["voice"], "Vivian") mk_update.assert_not_called() def test_qwen_form_defaults_to_the_first_model(self): # No persisted default: the Model picker opens on CustomVoice. with patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["Vivian"]), \ patch.object(hub.common, "update_config_value"): self._answer_form(backend="qwen", mode="design", qwen_instructions="A warm narrator") self._convert(None, [self._ready("qwen", "qwen-tts")]) self.assertEqual(self._field("mode")["value"], "custom") def test_qwen_clone_dir_defaults_to_the_project_voices(self): # The Clone .wav directory is the shared directory widget, seeded # with the project's ./voices; the picker starts on its first .wav # (alphabetically), ignoring non-.wav files. with tempfile.TemporaryDirectory() as td: root = Path(td) (root / "narrator.wav").write_bytes(b"") (root / "alice.wav").write_bytes(b"") (root / "notes.txt").write_text("", encoding="utf-8") expected_choices = [ ("alice.wav", str(root / "alice.wav")), ("narrator.wav", str(root / "narrator.wav")), ] with patch.object(hub.common, "VOICES_DIR", root), \ patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["Vivian"]), \ patch.object(hub.common, "update_config_value"): self._answer_form(backend="qwen", mode="custom", speaker="Vivian") cmd = self._convert(None, [self._ready("qwen", "qwen-tts")]) fields = self.tui.forms_seen[0][1] clone_dir_field = self._field("clone_dir") clone_field = self._field("clone") self.assertIsNone(cmd[2]["clone"]) # custom mode: no clone self.assertEqual(clone_dir_field["kind"], "dir") self.assertEqual(clone_dir_field["value"], root) self.assertEqual(clone_field["kind"], "choice") self.assertEqual(clone_field["choices"](fields), expected_choices) self.assertEqual(clone_field["value"], expected_choices[0][1]) def test_qwen_clone_picker_resets_when_the_directory_changes(self): # Changing the directory browser re-points the picker at the new # directory's first .wav; an empty directory clears the pick. with tempfile.TemporaryDirectory() as td, \ tempfile.TemporaryDirectory() as other, \ tempfile.TemporaryDirectory() as nowhere: root, other = Path(td), Path(other) (root / "one.wav").write_bytes(b"") (other / "beta.wav").write_bytes(b"") (other / "alpha.wav").write_bytes(b"") with patch.object(hub.common, "VOICES_DIR", root), \ patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["Vivian"]), \ patch.object(hub.common, "update_config_value"): self._answer_form(backend="qwen", mode="clone", clone="") self._convert(None, [self._ready("qwen", "qwen-tts")]) fields = self.tui.forms_seen[0][1] clone_dir_field = self._field("clone_dir") clone_field = self._field("clone") # The picker seeds itself with the default directory's first # .wav, and follows the directory browser from there. self.assertEqual(clone_field["value"], str(root / "one.wav")) clone_dir_field["value"] = other clone_dir_field["on_change"](fields) self.assertEqual(clone_field["value"], str(other / "alpha.wav")) clone_dir_field["value"] = Path(nowhere) / "no-such-dir" clone_dir_field["on_change"](fields) self.assertEqual(clone_field["value"], "") def test_qwen_clone_picker_refuses_generate_without_wavs(self): # No .wav files in the directory: the picker stays empty, opening # it flashes the hint, and Generate! is refused with the same # message naming the directory. with tempfile.TemporaryDirectory() as td: empty = Path(td) with patch.object(hub.common, "VOICES_DIR", empty), \ patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["Vivian"]), \ patch.object(hub.common, "update_config_value"): self._answer_form(backend="qwen", mode="clone", clone="") cmd = self._convert(None, [self._ready("qwen", "qwen-tts")]) self.assertIsNone(cmd[2]["clone"]) fields = self.tui.forms_seen[0][1] clone_field = self._field("clone") self.assertEqual(clone_field["choices"](fields), []) message = clone_field["on_empty_choices"](fields) self.assertIn(str(empty), message) self.assertIn("No .wav files", message) self.assertEqual(clone_field["validate"](""), message) self.assertIsNone(clone_field["validate"](str(empty / "x.wav"))) # ------------------------------------------------------------------ # 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") # faster voices are server-side clone references. self.assertEqual(self._field("faster_voice")["label"], "Voice to clone") 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_settings_default_feeds_the_stop_and_exit_toggle(self): # The Settings menu's "Stop server and exit" value decides the # run's stop-and-exit behavior (it travels on the run kwargs). st = self._remote("faster", "faster-qwen3-tts", url="http://10.0.0.5:8000") with patch.object(hub.config, "STOP_SERVER_AND_EXIT", False): self._answer_form(backend="faster-remote", faster_voice="obama") cmd = self._convert(None, [st]) self.assertIsNotNone(cmd) self.assertFalse(cmd[2]["stop_and_exit"]) def test_qwen_remote_limited_modes_and_api_url(self): # A remote qwen with only the Base (clone) demo answering: the form # offers only the Base model and targets the single remote URL. st = self._remote( "qwen", "qwen-tts", remote_urls={"qwen": "http://10.0.0.5:7861"}, remote_models=["Base"]) with patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["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") self.assertEqual(self._field("mode")["choices"], [("Base".ljust(11) + " (voice cloning)", "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): 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", "request_options", "mode", "speaker", "clone_dir", "clone", "qwen_instructions", "single_file"]) # The form opens on the configured default (audio.cpp): its fields # show, the other backend's hide. Instructions shows too (optional # style/delivery control even on the clone-only higgs entry), while # Request options stays hidden — higgs has no option-supporting # spec on this machine's checkout, so its support is unknown. for key in ("model_id", "audiocpp_voice", "instructions"): self.assertTrue(self._field(key)["visible"](fields)) self.assertFalse(self._field("request_options")["visible"](fields)) for key in ("mode", "speaker", "clone_dir", "clone", "qwen_instructions"): 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_dir")["visible"](fields)) self.assertFalse(self._field("clone")["visible"](fields)) # qwen's clone mode hides the speaker and shows the .wav directory # browser and the picker of the .wavs inside it. self._field("mode")["value"] = "clone" self.assertFalse(self._field("speaker")["visible"](fields)) self.assertTrue(self._field("clone_dir")["visible"](fields)) self.assertTrue(self._field("clone")["visible"](fields)) for key in ("model_id", "audiocpp_voice", "instructions", "request_options"): 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)) for key in ("mode", "speaker", "clone_dir", "clone", "qwen_instructions"): self.assertFalse(self._field(key)["visible"](fields)) class SelectSpecTests(unittest.TestCase): """_select_spec: single-server selection (qwen hosts one model at a time).""" def _qwen_status(self): return BackendStatus( "qwen", "qwen-tts", installed=True, configured=True, servers=[ServerSpec("qwen", "http://127.0.0.1:7860", [])]) def test_qwen_returns_the_single_spec(self): spec = hub._select_spec(self._qwen_status(), {"clone": None}) self.assertEqual(spec.name, "qwen") spec = hub._select_spec(self._qwen_status(), {"clone": "ref.wav"}) self.assertEqual(spec.name, "qwen") 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", 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"} with patch.object(hub, "detect_all", return_value=[]), \ patch.object(hub, "_find_spec", return_value=spec): cfg = hub._prepare_run_config("qwen", kwargs) # The qwen spec is rebuilt for the model this run selected, so an # autostart boots exactly what the conversion needs. self.assertEqual(cfg.autostart_spec.name, "qwen") self.assertEqual(cfg.autostart_spec.identity, "qwen-custom") self.assertEqual(cfg.autostart_spec.url, spec.url) self.assertFalse(cfg.restart_first) self.assertEqual(cfg.server_name, "qwen") self.assertNotIn("autostart", kwargs) def test_restart_first_stops_and_boots_before_converting(self): # A running managed server hosting another model than this run # selected: the recorded spec boots again after a stop. spec = self._spec() status = BackendStatus("qwen", "qwen-tts", installed=True, configured=True, servers=[spec]) kwargs = {"restart_server": "qwen", "clone": "/tmp/ref.wav"} with patch.object(hub, "detect_all", return_value=[status]), \ patch("backends.common.server_running", return_value=True), \ patch.object(hub.servers, "alive", return_value=True): cfg = hub._prepare_run_config("qwen", kwargs) self.assertEqual(cfg.autostart_spec.name, spec.name) # The restart spec hosts the Base model (the run clones a voice). self.assertEqual(cfg.autostart_spec.identity, "qwen-clone") self.assertTrue(cfg.restart_first) self.assertNotIn("restart_server", kwargs) self.assertEqual(cfg.server_url, spec.url) def test_rehost_flag_is_popped_and_reported_in_the_notice(self): # The convert form's config repair travels as "audiocpp_rehost": # popped from the converter kwargs and surfaced as the run notice. spec = self._spec("audiocpp", "http://127.0.0.1:8080") status = BackendStatus("audiocpp", "audio.cpp", installed=True, configured=True, servers=[spec]) kwargs = {"restart_server": "audiocpp", "audiocpp_rehost": True} with patch.object(hub, "detect_all", return_value=[status]), \ patch("backends.common.server_running", return_value=True), \ patch.object(hub.servers, "alive", return_value=True): cfg = hub._prepare_run_config("audiocpp", kwargs) self.assertTrue(cfg.restart_first) self.assertNotIn("audiocpp_rehost", kwargs) self.assertIn("clon", cfg.notice) self.assertIn("restarted", cfg.notice) def test_rehost_notice_without_restart_when_server_was_down(self): # The autostart path boots the fixed server.json anyway, so the # notice only reports the re-hosting. kwargs = {"audiocpp_rehost": True} with patch.object(hub, "detect_all", return_value=[]): cfg = hub._prepare_run_config("audiocpp", kwargs) self.assertFalse(cfg.restart_first) self.assertNotIn("audiocpp_rehost", kwargs) self.assertIn("clon", cfg.notice) self.assertNotIn("restarted", cfg.notice) 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", "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): self.assertIsNone(hub._add_autostart(cmd, [self._status()])) self.assertEqual(cmd[2]["autostart"], "qwen") def test_no_autostart_when_server_already_running_same_model(self): cmd = ("convert", "qwen", {"clone": None}) with patch.object(hub, "detect_all", return_value=[self._status()]), \ patch("backends.common.server_running", return_value=True), \ patch.object(hub.backend_probe, "identify_server", return_value="qwen-custom"): self.assertIsNone(hub._add_autostart(cmd, [self._status()])) self.assertNotIn("autostart", cmd[2]) def test_running_server_hosting_another_model_is_restarted(self): # The managed qwen server hosts CustomVoice but the run selected # the Base (clone) model: restart (stop + boot) before converting. cmd = ("convert", "qwen", {"clone": "/tmp/ref.wav"}) with patch.object(hub, "detect_all", return_value=[self._status()]), \ patch("backends.common.server_running", return_value=True), \ patch.object(hub.servers, "alive", return_value=True), \ patch.object(hub.backend_probe, "identify_server", return_value="qwen-custom"): self.assertIsNone(hub._add_autostart(cmd, [self._status()])) self.assertEqual(cmd[2]["restart_server"], "qwen") def test_foreign_server_with_wrong_model_refuses_the_run(self): cmd = ("convert", "qwen", {"clone": "/tmp/ref.wav"}) with patch.object(hub, "detect_all", return_value=[self._status()]), \ patch("backends.common.server_running", return_value=True), \ patch.object(hub.servers, "alive", return_value=False), \ patch.object(hub.backend_probe, "identify_server", return_value="qwen-custom"): message = hub._add_autostart(cmd, [self._status()]) self.assertIsNotNone(message) self.assertIn("CustomVoice", message) self.assertIn("Base", message) self.assertNotIn("autostart", cmd[2]) self.assertNotIn("restart_server", 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, []) def _audiocpp_status(self): spec = ServerSpec("audiocpp", "http://127.0.0.1:8080", ["x"]) return BackendStatus("audiocpp", "audio.cpp", installed=True, configured=True, running=True, servers=[spec]) def test_rehosted_config_restarts_the_managed_audiocpp_server(self): # The convert form re-hosted clone-only families with task "clon" # in server.json: the running managed server still hosts the stale # tasks, so it is stopped and rebooted before converting. cmd = ("convert", "audiocpp", {"audiocpp_rehost": True}) with patch.object(hub, "detect_all", return_value=[]), \ patch("backends.common.server_running", return_value=True), \ patch.object(hub.servers, "alive", return_value=True): self.assertIsNone(hub._add_autostart(cmd, [self._audiocpp_status()])) self.assertEqual(cmd[2]["restart_server"], "audiocpp") self.assertNotIn("autostart", cmd[2]) def test_rehosted_config_with_foreign_server_refuses_the_run(self): cmd = ("convert", "audiocpp", {"audiocpp_rehost": True}) with patch.object(hub, "detect_all", return_value=[]), \ patch("backends.common.server_running", return_value=True), \ patch.object(hub.servers, "alive", return_value=False): message = hub._add_autostart(cmd, [self._audiocpp_status()]) self.assertIsNotNone(message) self.assertIn("stop it first", message) self.assertNotIn("restart_server", cmd[2]) def test_rehosted_config_autostarts_when_server_is_down(self): # Server not running: the plain autostart path boots it with the # corrected server.json — no restart needed. cmd = ("convert", "audiocpp", {"audiocpp_rehost": True}) with patch.object(hub, "detect_all", return_value=[]), \ patch("backends.common.server_running", return_value=False): self.assertIsNone(hub._add_autostart(cmd, [self._audiocpp_status()])) self.assertEqual(cmd[2]["autostart"], "audiocpp") self.assertNotIn("restart_server", cmd[2]) class SettingsTests(unittest.TestCase): """Settings menu: field collection, validation, config.py writing.""" # Keys _apply_settings persists; every test that triggers a real or # fake config write restores these afterwards. _SETTING_KEYS = ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", "CHUNK_SIZE", "INPUT_DIR", "OUTPUT_DIR", "SPEED", "DEBUG", "STOP_SERVER_AND_EXIT", "AUDIOCPP_UNLOAD_MODELS", "QWEN_API_URL", "FASTER_API_URL", "AUDIOCPP_API_URL", "QWEN_REMOTE_URL", "FASTER_REMOTE_URL", "AUDIOCPP_REMOTE_URL") def _snapshot_settings(self): original = {name: getattr(hub.config, name) for name in self._SETTING_KEYS} self.addCleanup(lambda: [setattr(hub.config, name, value) for name, value in original.items()]) def test_update_config_value_preserves_comments_and_other_lines(self): import tempfile self._snapshot_settings() 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") for key, value in (("AUDIO_FORMAT", "mp3"), ("AUDIO_BITRATE", "192k"), ("LANGUAGE", "Japanese"), ("CHUNK_SIZE", 300)): self.assertTrue(hub.common.update_config_value( key, value, config_path=path)) 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") # The imported module mirrors the file immediately. self.assertEqual(hub.config.AUDIO_FORMAT, "mp3") self.assertEqual(hub.config.CHUNK_SIZE, 300) def test_update_config_value_missing_key_returns_false(self): import tempfile self._snapshot_settings() with tempfile.TemporaryDirectory() as td: path = Path(td) / "config.py" path.write_text("X = 1\n", encoding="utf-8") self.assertFalse(hub.common.update_config_value( "AUDIO_FORMAT", "mp3", config_path=path)) def test_apply_settings_writes_and_reloads_in_memory(self): written = {} def fake_update(key, value, config_path=None): written[key] = value setattr(hub.config, key, value) return True self._snapshot_settings() # _apply_settings re-derives the converter module's folder # globals; restore them afterwards. original_folders = (hub.converter_mod.BOOKS_FOLDER, hub.converter_mod.AUDIOBOOKS_FOLDER) self.addCleanup(setattr, hub.converter_mod, "BOOKS_FOLDER", original_folders[0]) self.addCleanup(setattr, hub.converter_mod, "AUDIOBOOKS_FOLDER", original_folders[1]) values = {"audio_format": "ogg", "audio_bitrate": " 192k ", "language": "en", "chunk_size": "300", "input_dir": " /books ", "output_dir": "/audiobooks", "speed": "1.5", "debug": True, "stop_and_exit": False, "unload_models": True, "qwen_port": "7862", "faster_port": "8001", "audiocpp_port": "8081", "audiocpp_remote_url": "10.0.0.5:8080", "faster_remote_url": "http://10.0.0.6:8000", "qwen_remote_url": ""} with patch.object(hub.common, "update_config_value", fake_update), \ 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, "INPUT_DIR": "/books", "OUTPUT_DIR": "/audiobooks", "SPEED": 1.5, "DEBUG": True, "STOP_SERVER_AND_EXIT": False, "AUDIOCPP_UNLOAD_MODELS": True, "QWEN_API_URL": "http://127.0.0.1:7862", "FASTER_API_URL": "http://127.0.0.1:8001", "AUDIOCPP_API_URL": "http://127.0.0.1:8081", "QWEN_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, # and the converter module's folder globals follow the directories. 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.INPUT_DIR, "/books") self.assertEqual(hub.config.OUTPUT_DIR, "/audiobooks") self.assertEqual(hub.config.SPEED, 1.5) self.assertEqual(hub.config.DEBUG, True) self.assertEqual(hub.config.STOP_SERVER_AND_EXIT, False) 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") self.assertEqual(hub.converter_mod.BOOKS_FOLDER, Path("/books")) self.assertEqual(hub.converter_mod.AUDIOBOOKS_FOLDER, Path("/audiobooks")) def test_apply_settings_rejects_bad_values(self): self._snapshot_settings() base = {"audio_format": "m4b", "audio_bitrate": "128k", "language": "English", "chunk_size": "250", "input_dir": "./input", "output_dir": "./output", "speed": "1.0", "debug": False, "stop_and_exit": True, "unload_models": True, "qwen_port": "7860", "faster_port": "8000", "audiocpp_port": "8080"} with patch.object(hub.common, "update_config_value") as mk_update: 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, "speed": "0"}) with self.assertRaises(ValueError): hub._apply_settings({**base, "speed": "fast"}) with self.assertRaises(ValueError): hub._apply_settings({**base, "input_dir": " "}) with self.assertRaises(ValueError): hub._apply_settings({**base, "output_dir": ""}) 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_update.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_speed("1.0")) self.assertIsNone(hub._validate_speed("1.5")) self.assertIsNone(hub._validate_speed(" 2 ")) self.assertIsNotNone(hub._validate_speed("0")) self.assertIsNotNone(hub._validate_speed("-1")) self.assertIsNotNone(hub._validate_speed("fast")) self.assertIsNotNone(hub._validate_speed("")) self.assertIsNone(hub._validate_dir("./input")) self.assertIsNone(hub._validate_dir(Path("/books"))) self.assertIsNotNone(hub._validate_dir("")) self.assertIsNotNone(hub._validate_dir(" ")) 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_language_fields_are_pickers_with_edit_hint(self): # The Settings Language field is a static picker over the # audio.cpp-menu languages, with a dim hint inside its edit # dialog. Common languages lead. expected_choices = ["English", "Spanish", "Chinese", "French", "German", "Italian", "Portuguese", "Japanese", "Korean", "Russian", "Arabic", "Hindi", "Vietnamese", "Auto"] hint = hub._LANGUAGE_EDIT_HINT self.assertEqual(hint, ["Check model documentation for supported " "languages."]) fields = hub._settings_fields() field = next(f for f in fields if f["key"] == "language") self.assertEqual(field["label"], "Language") self.assertEqual(field["kind"], "choice") self.assertEqual(field["choices"], expected_choices) self.assertEqual(field["help"], hint) self.assertEqual(field["value"], hub.config.LANGUAGE) def test_directory_fields_are_browsers(self): # The Input/Output Directory settings use the DOS-style directory # browser, seeded with the configured folder resolved against the # project root. fields = hub._settings_fields() for key, config_name in (("input_dir", "INPUT_DIR"), ("output_dir", "OUTPUT_DIR")): field = next(f for f in fields if f["key"] == key) self.assertEqual(field["kind"], "dir") self.assertTrue(field["label"].endswith("Directory")) self.assertEqual(field["value"], hub.converter_mod.resolve_dir( getattr(hub.config, config_name), key.removesuffix("_dir"))) self.assertIsNone(field["validate"](str(field["value"]))) self.assertIsNotNone(field["validate"](" ")) with patch.object(hub.config, "INPUT_DIR", "books"): fields = hub._settings_fields() field = next(f for f in fields if f["key"] == "input_dir") self.assertEqual( field["value"], hub.converter_mod.BASE_DIR / "books") 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", "input_dir": "/books", "output_dir": "/audiobooks", "speed": "1.0", "debug": False, "stop_and_exit": True, "unload_models": True, "qwen_port": "7860", "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", "input_dir", "output_dir", "speed", "debug", "stop_and_exit", "unload_models", "audiocpp_port", "faster_port", "qwen_port", "audiocpp_remote_url", "faster_remote_url", "qwen_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["input_dir"], "dir") self.assertEqual(kinds["output_dir"], "dir") self.assertEqual(kinds["speed"], "text") self.assertEqual(kinds["debug"], "bool") self.assertEqual(kinds["audiocpp_port"], "text") self.assertEqual(kinds["stop_and_exit"], "bool") 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_port"], "qwen-tts port") self.assertEqual(labels["audiocpp_remote_url"], "audio.cpp remote URL") self.assertEqual(labels["input_dir"], "Input Directory") self.assertEqual(labels["output_dir"], "Output Directory") self.assertNotIn("(clone)", " ".join(labels.values())) # The language setting is a picker labelled "Language". self.assertEqual(labels["language"], "Language") self.assertEqual(kinds["language"], "choice") # 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.assertIsNone(notes["input_dir"]) self.assertIsNone(notes["output_dir"]) self.assertIsNone(notes["debug"]) self.assertTrue(notes["stop_and_exit"]) self.assertTrue(notes["unload_models"]) self.assertTrue(notes["audiocpp_port"]) self.assertTrue(notes["audiocpp_remote_url"]) self.assertIsNone(notes["audio_format"]) self.assertIsNone(notes["speed"]) self.assertIsNone(notes["qwen_port"]) self.assertEqual(applied, [{"audio_format": "ogg", "audio_bitrate": "192k", "language": "English", "chunk_size": "300", "input_dir": "/books", "output_dir": "/audiobooks", "speed": "1.0", "debug": False, "stop_and_exit": True, "unload_models": True, "qwen_port": "7860", "faster_port": "8000", "audiocpp_port": "8080"}]) # Saving is silent: no confirmation flash either way. self.assertNotIn("flash", captured) def test_settings_menu_cancel_does_not_apply(self): # An actual edit triggers the save prompt; "no" discards it. def fake_form(stdscr, title, fields, back_value=None): next(f for f in fields if f["key"] == "chunk_size")["value"] = "300" return back_value # user pressed Cancel / q / Esc applied = [] def fake_apply(values): applied.append(values) with patch.object(hub.tui, "form", fake_form), \ patch.object(hub.tui, "confirm_yn_cancel", return_value="no") as mk_prompt, \ patch.object(hub, "_apply_settings", fake_apply): hub._Hub(None).screen_settings() mk_prompt.assert_called_once_with(None, "Save settings?") self.assertEqual(applied, []) def test_settings_menu_exit_without_changes_skips_prompt(self): # Leaving with untouched fields never asks about saving. def fake_form(stdscr, title, fields, back_value=None): return back_value # user pressed Cancel / q / Esc applied = [] with patch.object(hub.tui, "form", fake_form), \ patch.object(hub.tui, "confirm_yn_cancel") as mk_prompt, \ patch.object(hub, "_apply_settings", lambda values: applied.append(values)): hub._Hub(None).screen_settings() mk_prompt.assert_not_called() self.assertEqual(applied, []) def test_settings_menu_reverted_edit_skips_the_prompt(self): # Typing a value and typing it back leaves nothing to save. def fake_form(stdscr, title, fields, back_value=None): field = next(f for f in fields if f["key"] == "chunk_size") untouched = field["value"] field["value"] = "300" field["value"] = untouched return back_value applied = [] with patch.object(hub.tui, "form", fake_form), \ patch.object(hub.tui, "confirm_yn_cancel") as mk_prompt, \ patch.object(hub, "_apply_settings", lambda values: applied.append(values)): hub._Hub(None).screen_settings() mk_prompt.assert_not_called() self.assertEqual(applied, []) def test_settings_menu_whitespace_edit_skips_the_prompt(self): # Surrounding whitespace alone is not a change: _apply_settings # trims text values, so saving would be a no-op. def fake_form(stdscr, title, fields, back_value=None): field = next(f for f in fields if f["key"] == "language") field["value"] = " " + field["value"] + " " return back_value with patch.object(hub.tui, "form", fake_form), \ patch.object(hub.tui, "confirm_yn_cancel") as mk_prompt, \ patch.object(hub, "_apply_settings", lambda values: None): hub._Hub(None).screen_settings() mk_prompt.assert_not_called() def test_settings_menu_exit_yes_applies_the_edited_fields(self): # Leaving via Esc and answering Yes applies a values dict built # from the (edited) field list. def fake_form(stdscr, title, fields, back_value=None): next(f for f in fields if f["key"] == "chunk_size")["value"] = \ "300" return back_value # leave without the Save button applied = [] with patch.object(hub.tui, "form", fake_form), \ patch.object(hub.tui, "confirm_yn_cancel", return_value="yes"), \ patch.object(hub, "_apply_settings", lambda values: applied.append(values)): hub._Hub(None).screen_settings() self.assertEqual(len(applied), 1) self.assertEqual(applied[0]["chunk_size"], "300") # Every settings field's value travels on the dict. self.assertIn("stop_and_exit", applied[0]) self.assertIn("audiocpp_port", applied[0]) def test_settings_menu_exit_cancel_reopens_the_form(self): # "Cancel" on the save prompt returns to the form with edits kept; # leaving through Save afterwards applies once. seen = [] fields_seen = [] def fake_form(stdscr, title, fields, back_value=None): seen.append(title) fields_seen.append(fields) if len(seen) == 1: next(f for f in fields if f["key"] == "chunk_size")["value"] = "300" return back_value # first exit: Esc return {"audio_format": "m4b", "audio_bitrate": "128k", "language": "English", "chunk_size": "250", "input_dir": "./input", "output_dir": "./output", "speed": "1.0", "debug": False, "stop_and_exit": True, "unload_models": True, "qwen_port": "7860", "faster_port": "8000", "audiocpp_port": "8080"} applied = [] with patch.object(hub.tui, "form", fake_form), \ patch.object(hub.tui, "confirm_yn_cancel", side_effect=["cancel"]), \ patch.object(hub, "_apply_settings", lambda values: applied.append(values)): hub._Hub(None).screen_settings() # The form reopened with the same field objects (edits intact). self.assertEqual(len(seen), 2) self.assertIs(fields_seen[0], fields_seen[1]) self.assertEqual( next(f for f in fields_seen[1] if f["key"] == "chunk_size")["value"], "300") self.assertEqual(len(applied), 1) 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", "INPUT_DIR", "OUTPUT_DIR", "SPEED", "DEBUG", "STOP_SERVER_AND_EXIT", "AUDIOCPP_UNLOAD_MODELS", "QWEN_API_URL", "FASTER_API_URL", "AUDIOCPP_API_URL", "QWEN_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" 'INPUT_DIR = "./input"\n' 'OUTPUT_DIR = "./output"\n' "SPEED = 1.0\n" "DEBUG = False\n" "STOP_SERVER_AND_EXIT = True\n" "AUDIOCPP_UNLOAD_MODELS = True\n" 'QWEN_API_URL = "http://127.0.0.1:7860"\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' '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.common, "CONFIG_PATH", path), \ patch.object(hub, "_sync_audiocpp_server_port"): # 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 settings-only fields are written back unchanged. self.assertIn('INPUT_DIR = "', text) self.assertIn('OUTPUT_DIR = "', text) self.assertIn("SPEED = 1.0", text) self.assertIn("DEBUG = False", 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", "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' 'FASTER_API_URL = "http://127.0.0.1:8000"\n' 'AUDIOCPP_API_URL = "http://127.0.0.1:8080"\n', encoding="utf-8") self._snapshot_settings() for key, value in ( ("QWEN_API_URL", "http://127.0.0.1:7862"), ("FASTER_API_URL", "http://127.0.0.1:8001"), ("AUDIOCPP_API_URL", "http://127.0.0.1:8081")): hub.common.update_config_value(key, value, config_path=path) text = path.read_text(encoding="utf-8") self.assertIn('QWEN_API_URL = "http://127.0.0.1:7862"', 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.build, "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.build, "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.build, "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"]) def emit(line): pass 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"]) def emit(line): pass steps[0].work(emit, None) mk_install.assert_called_once_with( checkout, guidance, emit=emit, cancel=None) self.assertEqual(flashes, []) 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_update_backends_action_runs_one_step_per_installed_backend(self): calls = [] def make_update(name): def update(*, emit=None, cancel=None): calls.append((name, emit, cancel)) return 0 return update # faster is installed in statuses but has no update action → one # step fewer; audiocpp's on-disk check routes through the checkout. infos = [BackendInfo("audiocpp", "audio.cpp", lambda: None, lambda: 0, update=make_update("audiocpp")), BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0, update=make_update("qwen")), BackendInfo("faster", "faster-qwen3-tts", lambda: None, lambda: 0)] statuses = [BackendStatus("audiocpp", "audio.cpp", installed=True, configured=True), BackendStatus("qwen", "qwen-tts", installed=True, configured=True), BackendStatus("faster", "faster-qwen3-tts", installed=True, configured=True)] patch_flash, flashes = self._capture_flashes() with patch.object(hub, "REGISTRY", infos), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=Path("/co")), \ patch.object(hub.taskview, "run_steps", return_value=0) as mk_run, \ patch_flash: hub._update_backends_action(None) # One task-view run titled "Update Backends", one step per # updatable backend in registry order; executing a step # forwards emit/cancel to that backend's update. mk_run.assert_called_once() self.assertEqual(mk_run.call_args[0][0], None) self.assertEqual(mk_run.call_args[0][1], "Update Backends") steps = mk_run.call_args[0][2] self.assertEqual([step.title for step in steps], ["Update audio.cpp", "Update qwen-tts"]) def emit(line): pass steps[1].work(emit, "CANCEL") self.assertEqual(calls, [("qwen", emit, "CANCEL")]) self.assertEqual(flashes, []) def test_update_backends_action_flashes_error_when_something_failed(self): infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0, update=lambda **kw: 0)] statuses = [BackendStatus("qwen", "qwen-tts", installed=True, configured=True)] patch_flash, flashes = self._capture_flashes() with patch.object(hub, "REGISTRY", infos), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.taskview, "run_steps", return_value=1), \ patch_flash: hub._update_backends_action(None) self.assertEqual(flashes[-1][1], "err") self.assertIn("did not complete", flashes[-1][0]) def test_update_backends_action_flashes_warn_when_cancelled(self): infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0, update=lambda **kw: 0)] statuses = [BackendStatus("qwen", "qwen-tts", installed=True, configured=True)] patch_flash, flashes = self._capture_flashes() with patch.object(hub, "REGISTRY", infos), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.taskview, "run_steps", return_value=130), \ patch_flash: hub._update_backends_action(None) self.assertEqual(flashes[-1][1], "warn") self.assertIn("cancelled", flashes[-1][0]) def test_update_backends_action_without_targets_flashes_a_hint(self): # An installed backend without an update action (and nothing else # installed): the entry never shows, but a direct call still # explains itself instead of running an empty task view. infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0)] statuses = [BackendStatus("qwen", "qwen-tts", installed=True, configured=True)] patch_flash, flashes = self._capture_flashes() with patch.object(hub, "REGISTRY", infos), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.taskview, "run_steps") as mk_run, \ patch_flash: hub._update_backends_action(None) mk_run.assert_not_called() self.assertEqual(flashes, [("No installed backend supports updating.", "warn")]) def test_selecting_update_runs_the_action_and_reshows_the_menu(self): titles = [] def fake_menu(stdscr, title, options, **kwargs): titles.append(title) return "update" if len(titles) == 1 else tui.Wizard.BACK ran = [] invalidated = [] info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0, update=lambda **kw: 0) statuses = [BackendStatus("qwen", "qwen-tts", installed=True, configured=True)] with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub, "REGISTRY", [info]), \ patch.object(hub, "_update_backends_action", side_effect=lambda scr: ran.append("update")), \ patch.object(hub, "invalidate_detect_cache", side_effect=lambda: invalidated.append(True)), \ patch.object(hub.shutil, "which", return_value="/x"): result = hub._Hub(None).screen_configure() self.assertIs(result, tui.Wizard.BACK) self.assertEqual(ran, ["update"]) self.assertEqual(invalidated, [True]) # An inline action: the same menu re-shows (second title) with a # freshly detected status table. self.assertEqual(titles, ["Configure Backends", "Configure Backends"]) 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 = ["configure_backends", "server", specs[0], tui.Wizard.BACK, 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: server list → Configure # Backends → main menu. self.assertEqual( titles, ["tts-audiobook-generator", "Configure Backends", "Start / Stop A Server", "Start / Stop A Server", "Configure Backends", "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()