"""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 json import tempfile import unittest from pathlib import Path from unittest.mock import patch from backends import BackendInfo, BackendStatus, ServerSpec from tests.test_tui import FakeCurses, FakeScreen from ui import hub, tui class _ScriptedTUI: """Stand-in for the tui widget module: answers each menu/line_edit/ confirm call from a scripted answer list and records every prompt.""" def __init__(self): self.script = [] self.prompts = [] self.options_seen = [] self.flashes = [] 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 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 running = BackendStatus("k", "l", installed=True, configured=True, running=True, managed=True) remote = BackendStatus("k", "l", installed=True, configured=True, running=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, running_models=["Base"]) installed = BackendStatus("k", "l", installed=True, configured=False) none = BackendStatus("k", "l", installed=False, configured=False) # 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(running), ("running", "ok", "body")) # A server without a live recorded pid was started externally. self.assertEqual(hub._status_mark(remote), ("running [remote]", "ok", "body")) # Multi-model backends name the models that answered. self.assertEqual(hub._status_mark(models), ("running (Base, CustomVoice)", "ok", "body")) self.assertEqual(hub._status_mark(remote_models), ("running [remote] (Base)", "ok", "body")) self.assertEqual(hub._status_mark(installed), ("installed", "warn", "body")) self.assertEqual(hub._status_mark(none), ("unavailable", "err", "dim")) self.assertEqual(hub._status_mark(None), ("unavailable", "err", "dim")) class HubMenuTests(unittest.TestCase): """Drive _hub_menu 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 [Set up, Settings, Quit]. # Quit is the 3rd option (Down twice) then Enter. screen = FakeScreen(keys=[FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, 10]) with patch.object(hub, "detect_all", return_value=[]): result = hub._hub_menu(screen) self.assertIsNone(result) def test_menu_has_only_setup_settings_and_quit_without_backends(self): # Capture the options handed to tui.menu: with nothing installed or # running, Convert/Configure 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_menu(screen) labels = [label for label, _ in captured["options"]] self.assertEqual(labels, ["Set up a backend", "Settings", "Quit"]) def test_menu_has_all_six_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_menu(screen) labels = [label for label, _ in captured["options"]] self.assertEqual( labels, ["Convert books", "Set up a backend", "Configure a backend", "Start/Stop Backend Servers", "Settings", "Quit"]) # The status table is passed through, one row per backend. self.assertEqual(captured["rows"], [("qwen-tts", "installed", "warn", "body")]) def test_table_dims_name_when_not_installed_and_not_running(self): captured = {} def fake_menu(stdscr, title, options, **kwargs): captured["rows"] = kwargs.get("table_rows") return "quit" screen = FakeScreen() dead = self._none_status("audiocpp", "audio.cpp") external = self._none_status("qwen", "qwen-tts") external.running = True with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "detect_all", return_value=[dead, external]): hub._hub_menu(screen) # Unusable backend: dim name. Running-but-not-installed stays # bright and is tagged remote (no pid file → not started by us). self.assertEqual( captured["rows"], [("audio.cpp", "unavailable", "err", "dim"), ("qwen-tts", "running [remote]", "ok", "body")]) def test_menu_hides_configure_and_server_when_only_running(self): # Running but not installed (an external server) still unlocks # Convert — but Configure/Server need the backend on this machine. captured = {} def fake_menu(stdscr, title, options, **kwargs): captured["options"] = options return "quit" screen = FakeScreen() st = self._none_status("qwen", "qwen-tts") st.running = True with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "detect_all", return_value=[st]): hub._hub_menu(screen) labels = [label for label, _ in captured["options"]] self.assertEqual( labels, ["Convert books", "Set up a backend", "Settings", "Quit"]) def test_ffmpeg_warning_shown_when_missing(self): # ffmpeg not on PATH → a red notice is passed above the table. captured = {} def fake_menu(stdscr, title, options, **kwargs): captured["notice_lines"] = kwargs.get("notice_lines") return "quit" screen = FakeScreen() with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "detect_all", return_value=[]), \ patch.object(hub.shutil, "which", return_value=None): hub._hub_menu(screen) 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_menu(screen) 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 "Set up # a backend" detour anymore) and returns to the main menu. Then # quit: 6 main-menu options, Quit is the 6th (Down x5). 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 x5 -> Quit, Enter. screen = FakeScreen(keys=[10, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, 10]) result = hub._hub_menu(screen) self.assertIsNone(result) self.assertEqual(len(flashed), 1) self.assertIn("No backend is ready", flashed[0]) class SubmenuStatusTableTests(unittest.TestCase): """First picker screen of every flow repeats the backend status table. Entries themselves stay clean: setup lists bare labels, and the Start/Stop menu offers only installed backends. """ def _capture_menu(self, captured): def fake_menu(stdscr, title, options, **kwargs): captured["title"] = title captured["options"] = options captured.update(kwargs) return hub._GO_BACK # Esc: back out immediately return fake_menu def test_setup_menu_lists_bare_labels_and_status_table(self): captured = {} infos = [BackendInfo("audiocpp", "audio.cpp", lambda: None, lambda: 0), BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0)] statuses = [ BackendStatus("audiocpp", "audio.cpp", installed=True, configured=True), BackendStatus("qwen", "qwen-tts", installed=False, configured=False, running=True), ] with patch.object(hub, "REGISTRY", infos), \ patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.shutil, "which", return_value="/usr/bin/ffmpeg"): result = hub._setup_menu(None, statuses) self.assertIsNone(result) # No inline "(running)"-style suffix on the entries anymore... self.assertEqual([label for label, _ in captured["options"]], ["audio.cpp", "qwen-tts"]) # ...the shared status table carries the states instead. self.assertEqual(captured["table_title"], "Backend status") self.assertEqual( captured["table_rows"], [("audio.cpp", "installed", "warn", "body"), ("qwen-tts", "running [remote]", "ok", "body")]) self.assertIsNone(captured["notice_lines"]) def test_convert_menu_shows_status_table(self): captured = {} st = BackendStatus("qwen", "qwen-tts", installed=True, configured=True) with patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.shutil, "which", return_value="/x"): result = hub._convert_menu(None, [st]) self.assertIsNone(result) self.assertEqual(captured["title"], "Convert books with...") # Only convertible backends are listed — no "Set up a backend" # detour inside the Convert flow. self.assertEqual([label for label, _ in captured["options"]], ["qwen-tts"]) self.assertEqual( captured["table_rows"], [("qwen-tts", "installed", "warn", "body")]) 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 hub._GO_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): result = hub._convert_menu(None, [st]) self.assertIsNone(result) self.assertEqual(menus, []) self.assertIn("No backend is ready", flashed[0]) def test_configure_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.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.shutil, "which", return_value="/x"): result = hub._configure_menu(None, statuses) self.assertIsNone(result) self.assertEqual(captured["table_title"], "Backend status") self.assertEqual( captured["table_rows"], [("qwen-tts", "installed", "warn", "body")]) def test_server_menu_lists_only_installed_backends(self): captured = {} installed = BackendStatus("audiocpp", "audio.cpp", installed=True, configured=True) 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.shutil, "which", return_value="/x"): result = hub._server_menu(None, [installed, remote, gone]) self.assertIsNone(result) # Only the installed backend is offered; a running external server # (remote) can't be stopped from here and must not appear. self.assertEqual([label for label, _ in captured["options"]], ["audio.cpp"]) # The status table still shows all three, states included. self.assertEqual([row[0] for row in captured["table_rows"]], ["audio.cpp", "qwen-tts", "faster-qwen3-tts"]) 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): result = hub._server_menu(None, [remote]) self.assertIsNone(result) 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.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.shutil, "which", return_value=None): hub._setup_menu(None, statuses) self.assertEqual(captured["notice_lines"], [("Warning: ffmpeg not installed!", "err")]) class ConvertFlowTests(unittest.TestCase): """_convert_audiocpp / _convert_faster: local-config menus vs. live queries against a running remote server.""" def setUp(self): self.tui = _ScriptedTUI() for name in ("menu", "line_edit", "confirm", "flash"): patcher = patch.object(hub.tui, name, getattr(self.tui, name)) patcher.start() self.addCleanup(patcher.stop) def _answer_common_options(self): # Output format, speed, single-file, chunk, debug. self.tui.script += ["m4b", "1.5", False, False, False] # ------------------------------------------------------------------ # audio.cpp: remote server (no local checkout / server.json) # ------------------------------------------------------------------ def _patch_remote(self, models, voices=None): """No local checkout; fetch helpers return MODELS/VOICES.""" checkout = patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=None) 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 (checkout, fetched_models, fetched_voices): patcher.start() self.addCleanup(patcher.stop) def test_audiocpp_remote_queries_live_models_and_voices(self): self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self.tui.script += ["higgs", "narrator", ""] self._answer_common_options() cmd = hub._convert_audiocpp(None, []) 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"]) # The model menu was fed from the live query. self.assertEqual(self.tui.options_seen[0], [("higgs (higgs_audio_tts, tts)", "higgs")]) def test_audiocpp_remote_qwen3_tts_offers_builtin_speaker_first(self): self._patch_remote( [{"id": "qwen", "family": "qwen3_tts", "task": "tts"}], voices=["narrator"]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self.tui.script += ["qwen", None, ""] self._answer_common_options() cmd = hub._convert_audiocpp(None, []) self.assertIsNone(cmd[2]["voice"]) self.assertEqual(self.tui.options_seen[1], [("(built-in speaker)", None), ("narrator", "narrator")]) def test_audiocpp_remote_missing_family_treated_as_qwen3_tts(self): # Legacy servers omit family/task; the converter defaults them to # qwen3_tts/tts and so must the menus (voice optional). self._patch_remote([{"id": "legacy", "family": "", "task": ""}], voices=[]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): # Empty server voices: no Voice menu, built-in speaker implied. self.tui.script += ["legacy", ""] self._answer_common_options() cmd = hub._convert_audiocpp(None, []) self.assertIsNotNone(cmd) self.assertIsNone(cmd[2]["voice"]) def test_audiocpp_remote_vdes_needs_instructions_not_voice(self): self._patch_remote( [{"id": "design", "family": "qwen3_tts", "task": "vdes"}]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self.tui.script += ["design", "A warm British narrator"] self._answer_common_options() cmd = hub._convert_audiocpp(None, []) self.assertIsNone(cmd[2]["voice"]) self.assertEqual(cmd[2]["instructions"], "A warm British narrator") # No voice prompt happened at all. self.assertNotIn("Voice", [p for p in self.tui.prompts]) def test_audiocpp_remote_unreachable_models_flash_and_abort(self): self._patch_remote(None) # endpoint did not answer valid JSON cmd = hub._convert_audiocpp(None, []) 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 = hub._convert_audiocpp(None, []) self.assertIsNone(cmd) self.assertIn("hosts no model entries", self.tui.flashes[0]) def test_audiocpp_remote_no_server_voices_for_clone_model_aborts(self): self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=[]) self.tui.script += ["higgs"] cmd = hub._convert_audiocpp(None, []) self.assertIsNone(cmd) self.assertIn("lists none", self.tui.flashes[0]) def test_audiocpp_remote_failed_voices_query_aborts(self): self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=None) self.tui.script += ["higgs"] cmd = hub._convert_audiocpp(None, []) self.assertIsNone(cmd) self.assertIn("Could not list voices", self.tui.flashes[0]) # ------------------------------------------------------------------ # audio.cpp: local managed setup keeps reading its server.json # ------------------------------------------------------------------ def test_audiocpp_local_still_reads_server_json(self): queried = [] def must_not_query(url): queried.append(url) raise AssertionError("live query on the local path") with tempfile.TemporaryDirectory() as td: root = Path(td) (root / "server.json").write_text(json.dumps({ "models": [{"id": "qwen", "family": "qwen3_tts", "task": "tts"}], "voice_dir": str(root), }), encoding="utf-8") (root / "Narrator.wav").write_bytes(b"x") with patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=root), \ patch.object(hub.audiocpp_backend, "fetch_server_models", must_not_query), \ patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self.tui.script += ["qwen", "Narrator", ""] self._answer_common_options() cmd = hub._convert_audiocpp(None, []) self.assertEqual(queried, []) self.assertIsNotNone(cmd) self.assertEqual(cmd[2]["model_id"], "qwen") self.assertEqual(cmd[2]["voice"], "Narrator") # ------------------------------------------------------------------ # faster: remote server (no local voices.json) # ------------------------------------------------------------------ def test_faster_remote_prompts_for_a_voice_name(self): with tempfile.TemporaryDirectory() as td: with patch.object(hub.faster_backend, "_checkout", return_value=Path(td)): self.tui.script += ["obama"] self._answer_common_options() cmd = hub._convert_faster(None) self.assertEqual(cmd[0], "convert") self.assertEqual(cmd[1], "faster") self.assertEqual(cmd[2]["voice"], "obama") self.assertIn("Server-side voice", self.tui.prompts[0]) 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.tui.script += ["obama"] self._answer_common_options() cmd = hub._convert_faster(None) self.assertEqual(cmd[2]["voice"], "obama") # The voice came from a menu over voices.json, not a text field. self.assertIn("Select the voice to clone", self.tui.prompts[0]) class SelectSpecTests(unittest.TestCase): """_select_spec: mode-aware server selection (qwen has two servers).""" def _qwen_status(self): return BackendStatus( "qwen", "qwen-tts", installed=True, configured=True, servers=[ServerSpec("qwen-custom", "http://127.0.0.1:7860", []), ServerSpec("qwen-clone", "http://127.0.0.1:7861", [])]) def test_qwen_custom_mode(self): spec = hub._select_spec(self._qwen_status(), {"clone": None}) self.assertEqual(spec.name, "qwen-custom") def test_qwen_clone_mode(self): spec = hub._select_spec(self._qwen_status(), {"clone": "ref.wav"}) self.assertEqual(spec.name, "qwen-clone") def test_audiocpp_returns_single_spec(self): st = BackendStatus("audiocpp", "audio.cpp", installed=True, configured=True, servers=[ServerSpec("audiocpp", "http://x", [])]) spec = hub._select_spec(st, {}) self.assertEqual(spec.name, "audiocpp") def test_none_when_no_servers(self): st = BackendStatus("qwen", "qwen-tts", installed=False, configured=False) self.assertIsNone(hub._select_spec(st, {})) class RunConversionTests(unittest.TestCase): """_run_conversion: autostart, hint-when-manual, and stop-after.""" def test_autostart_starts_server_then_converts(self): spec = ServerSpec("qwen-custom", "http://127.0.0.1:7860", ["x"]) status = BackendStatus("qwen", "qwen-tts", installed=True, configured=True, running=False, servers=[spec]) kwargs = {"autostart": "qwen-custom"} with patch.object(hub, "detect_all", return_value=[status]), \ patch.object(hub, "_find_spec", return_value=spec), \ patch.object(hub.servers, "start", return_value=True) as mk_start, \ patch.object(hub.audiobook, "convert", return_value=0) as mk_conv, \ patch("builtins.input", return_value="n") as mk_input, \ patch.object(hub.servers, "stop") as mk_stop: hub._run_conversion("qwen", kwargs) mk_start.assert_called_once_with(spec) mk_conv.assert_called_once() # User declined stopping → stop not called. mk_stop.assert_not_called() def test_autostart_stop_when_user_says_yes(self): spec = ServerSpec("qwen-custom", "http://127.0.0.1:7860", ["x"]) status = BackendStatus("qwen", "qwen-tts", installed=True, configured=True, running=False, servers=[spec]) kwargs = {"autostart": "qwen-custom"} with patch.object(hub, "detect_all", return_value=[status]), \ patch.object(hub, "_find_spec", return_value=spec), \ patch.object(hub.servers, "start", return_value=True), \ patch.object(hub.audiobook, "convert", return_value=0), \ patch("builtins.input", return_value="y"), \ patch.object(hub.servers, "stop") as mk_stop: hub._run_conversion("qwen", kwargs) mk_stop.assert_called_once_with("qwen-custom") def test_autostart_aborts_when_server_fails(self): spec = ServerSpec("qwen-custom", "http://127.0.0.1:7860", ["x"]) status = BackendStatus("qwen", "qwen-tts", installed=True, configured=True, running=False, launch_hint="hint cmd", servers=[spec]) kwargs = {"autostart": "qwen-custom"} with patch.object(hub, "detect_all", return_value=[status]), \ patch.object(hub, "_find_spec", return_value=spec), \ patch.object(hub.servers, "start", return_value=False), \ patch.object(hub.audiobook, "convert") as mk_conv, \ patch.object(hub.servers, "stop") as mk_stop: hub._run_conversion("qwen", kwargs) mk_conv.assert_not_called() mk_stop.assert_not_called() def test_no_autostart_prints_hint_when_not_running(self): status = BackendStatus("qwen", "qwen-tts", installed=True, configured=True, running=False, launch_hint="the-hint") with patch.object(hub, "detect_all", return_value=[status]), \ patch.object(hub.audiobook, "convert", return_value=0) as mk_conv: hub._run_conversion("qwen", {}) mk_conv.assert_called_once() class AddAutostartTests(unittest.TestCase): """_add_autostart: offers to start the server when it isn't running.""" def setUp(self): tui._THEME.clear() self.curses = FakeCurses() self._patcher = patch.dict("sys.modules", {"curses": self.curses}) self._patcher.start() self.addCleanup(self._patcher.stop) self.addCleanup(tui._THEME.clear) def _status(self): spec = ServerSpec("qwen-custom", "http://127.0.0.1:7860", ["x"]) return BackendStatus("qwen", "qwen-tts", installed=True, configured=True, running=False, servers=[spec]) def test_sets_autostart_when_user_confirms(self): screen = FakeScreen(keys=[10]) # Enter = Yes cmd = ("convert", "qwen", {"clone": None}) with patch.object(hub, "detect_all", return_value=[self._status()]), \ patch("backends.common.server_running", return_value=False): hub._add_autostart(screen, cmd, [self._status()]) self.assertEqual(cmd[2]["autostart"], "qwen-custom") def test_no_autostart_when_server_already_running(self): screen = FakeScreen(keys=[10]) cmd = ("convert", "qwen", {"clone": None}) with patch.object(hub, "detect_all", return_value=[self._status()]), \ patch("backends.common.server_running", return_value=True): hub._add_autostart(screen, cmd, [self._status()]) self.assertNotIn("autostart", cmd[2]) class SettingsTests(unittest.TestCase): """Settings menu: field collection, validation, config.py writing.""" def test_write_config_preserves_comments_and_other_lines(self): import tempfile with tempfile.TemporaryDirectory() as td: path = Path(td) / "config.py" path.write_text( "# Default output options\n" 'AUDIO_FORMAT = "m4b"\n' 'AUDIO_BITRATE = "128k"\n' 'LANGUAGE = "English"\n' "\n" "CHUNK_SIZE = 250 # words per request\n", encoding="utf-8") with patch.object(hub.config, "__file__", str(path)): hub._write_config({"AUDIO_FORMAT": "mp3", "AUDIO_BITRATE": "192k", "LANGUAGE": "Japanese", "CHUNK_SIZE": 300}) text = path.read_text(encoding="utf-8") self.assertEqual( text, "# Default output options\n" 'AUDIO_FORMAT = "mp3"\n' 'AUDIO_BITRATE = "192k"\n' 'LANGUAGE = "Japanese"\n' "\n" "CHUNK_SIZE = 300 # words per request\n") def test_write_config_missing_key_raises(self): import tempfile with tempfile.TemporaryDirectory() as td: path = Path(td) / "config.py" path.write_text("X = 1\n", encoding="utf-8") with patch.object(hub.config, "__file__", str(path)): with self.assertRaises(ValueError): hub._write_config({"AUDIO_FORMAT": "mp3"}) def test_apply_settings_writes_and_reloads_in_memory(self): written = {} def fake_write(updates): written.update(updates) original = {name: getattr(hub.config, name) for name in ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", "CHUNK_SIZE", "QWEN_API_URL", "CLONE_API_URL", "FASTER_API_URL", "AUDIOCPP_API_URL")} self.addCleanup(lambda: [setattr(hub.config, name, value) for name, value in original.items()]) values = {"audio_format": "ogg", "audio_bitrate": " 192k ", "language": "en", "chunk_size": "300", "qwen_custom_port": "7862", "qwen_clone_port": "7863", "faster_port": "8001", "audiocpp_port": "8081"} with patch.object(hub, "_write_config", fake_write), \ patch.object(hub, "_sync_audiocpp_server_port"): hub._apply_settings(values) # Values are trimmed and language normalized to a display name. self.assertEqual(written, {"AUDIO_FORMAT": "ogg", "AUDIO_BITRATE": "192k", "LANGUAGE": "English", "CHUNK_SIZE": 300, "QWEN_API_URL": "http://127.0.0.1:7862", "CLONE_API_URL": "http://127.0.0.1:7863", "FASTER_API_URL": "http://127.0.0.1:8001", "AUDIOCPP_API_URL": "http://127.0.0.1:8081"}) # In-memory config is reloaded so this session sees the change. self.assertEqual(hub.config.AUDIO_FORMAT, "ogg") self.assertEqual(hub.config.AUDIO_BITRATE, "192k") self.assertEqual(hub.config.LANGUAGE, "English") self.assertEqual(hub.config.CHUNK_SIZE, 300) self.assertEqual(hub.config.QWEN_API_URL, "http://127.0.0.1:7862") self.assertEqual(hub.config.FASTER_API_URL, "http://127.0.0.1:8001") def test_apply_settings_rejects_bad_values(self): original = {name: getattr(hub.config, name) for name in ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", "CHUNK_SIZE", "QWEN_API_URL", "CLONE_API_URL", "FASTER_API_URL", "AUDIOCPP_API_URL")} self.addCleanup(lambda: [setattr(hub.config, name, value) for name, value in original.items()]) base = {"audio_format": "m4b", "audio_bitrate": "128k", "language": "English", "chunk_size": "250", "qwen_custom_port": "7860", "qwen_clone_port": "7861", "faster_port": "8000", "audiocpp_port": "8080"} with patch.object(hub, "_write_config") as mk_write: with self.assertRaises(ValueError): hub._apply_settings({**base, "language": "Klingon"}) with self.assertRaises(ValueError): hub._apply_settings({**base, "chunk_size": "0"}) with self.assertRaises(ValueError): hub._apply_settings({**base, "audiocpp_port": "70000"}) mk_write.assert_not_called() def test_field_validators(self): self.assertIsNone(hub._validate_bitrate("128k")) self.assertIsNotNone(hub._validate_bitrate(" ")) self.assertIsNone(hub._validate_language("English")) self.assertIsNone(hub._validate_language("en")) self.assertIsNotNone(hub._validate_language("Klingon")) self.assertIsNone(hub._validate_chunk_size("250")) self.assertIsNotNone(hub._validate_chunk_size("abc")) self.assertIsNotNone(hub._validate_chunk_size("0")) self.assertIsNone(hub._validate_port("8080")) self.assertIsNone(hub._validate_port("1")) self.assertIsNone(hub._validate_port("65535")) self.assertIsNotNone(hub._validate_port("0")) self.assertIsNotNone(hub._validate_port("70000")) self.assertIsNotNone(hub._validate_port("abc")) def test_settings_menu_builds_form_and_saves(self): captured = {} def fake_form(stdscr, title, fields, back_value=None): captured["fields"] = fields return {"audio_format": "ogg", "audio_bitrate": "192k", "language": "English", "chunk_size": "300", "qwen_custom_port": "7860", "qwen_clone_port": "7861", "faster_port": "8000", "audiocpp_port": "8080"} applied = [] def fake_apply(values): applied.append(values) def fake_flash(stdscr, text, kind="warn"): captured["flash"] = (text, kind) with patch.object(hub.tui, "form", fake_form), \ patch.object(hub, "_apply_settings", fake_apply), \ patch.object(hub.tui, "flash", fake_flash): hub._settings_menu(None) self.assertEqual([f["key"] for f in captured["fields"]], ["audio_format", "audio_bitrate", "language", "chunk_size", "audiocpp_port", "faster_port", "qwen_custom_port", "qwen_clone_port"]) kinds = {f["key"]: f["kind"] for f in captured["fields"]} self.assertEqual(kinds["audio_format"], "choice") self.assertEqual(kinds["audio_bitrate"], "text") self.assertEqual(kinds["audiocpp_port"], "text") labels = {f["key"]: f["label"] for f in captured["fields"]} self.assertEqual(labels["qwen_clone_port"], "qwen-tts Base port") self.assertNotIn("(clone)", " ".join(labels.values())) # The ports section note hangs off the first port field so it # renders between the output settings and the ports. notes = {f["key"]: f.get("note") for f in captured["fields"]} self.assertTrue(notes["audiocpp_port"]) self.assertIsNone(notes["audio_format"]) self.assertIsNone(notes["qwen_custom_port"]) self.assertEqual(applied, [{"audio_format": "ogg", "audio_bitrate": "192k", "language": "English", "chunk_size": "300", "qwen_custom_port": "7860", "qwen_clone_port": "7861", "faster_port": "8000", "audiocpp_port": "8080"}]) self.assertEqual(captured["flash"], ("Settings saved.", "ok")) def test_settings_menu_cancel_does_not_apply(self): def fake_form(stdscr, title, fields, back_value=None): return back_value # user pressed Cancel applied = [] def fake_apply(values): applied.append(values) with patch.object(hub.tui, "form", fake_form), \ patch.object(hub, "_apply_settings", fake_apply): hub._settings_menu(None) self.assertEqual(applied, []) def test_settings_menu_writes_config_end_to_end(self): import tempfile tui._THEME.clear() self.addCleanup(tui._THEME.clear) curses = FakeCurses() patcher = patch.dict("sys.modules", {"curses": curses}) patcher.start() self.addCleanup(patcher.stop) original = {name: getattr(hub.config, name) for name in ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", "CHUNK_SIZE", "QWEN_API_URL", "CLONE_API_URL", "FASTER_API_URL", "AUDIOCPP_API_URL")} self.addCleanup(lambda: [setattr(hub.config, name, value) for name, value in original.items()]) with tempfile.TemporaryDirectory() as td: path = Path(td) / "config.py" path.write_text( "# Default output options\n" 'AUDIO_FORMAT = "m4b"\n' 'AUDIO_BITRATE = "128k"\n' 'LANGUAGE = "English"\n' "\n" "CHUNK_SIZE = 250\n" 'QWEN_API_URL = "http://127.0.0.1:7860"\n' 'CLONE_API_URL = "http://127.0.0.1:7861"\n' 'FASTER_API_URL = "http://127.0.0.1:8000"\n' 'AUDIOCPP_API_URL = "http://127.0.0.1:8080"\n', encoding="utf-8") with patch.object(hub.config, "__file__", str(path)): # Down to Chunk size, Enter -> editor, Ctrl-U + '300', # Enter; Tab -> Save, Enter; a key dismisses the flash. screen = FakeScreen(keys=[ FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, 10, 21, ord("3"), ord("0"), ord("0"), 10, 9, 10, 10]) hub._settings_menu(screen) text = path.read_text(encoding="utf-8") self.assertIn('AUDIO_FORMAT = "m4b"', text) self.assertIn("CHUNK_SIZE = 300", text) # The running session also picked up the change in-memory. self.assertEqual(hub.config.CHUNK_SIZE, 300) def test_settings_menu_updates_backend_ports(self): import tempfile original = {name: getattr(hub.config, name) for name in ("QWEN_API_URL", "CLONE_API_URL", "FASTER_API_URL", "AUDIOCPP_API_URL")} self.addCleanup(lambda: [setattr(hub.config, name, value) for name, value in original.items()]) with tempfile.TemporaryDirectory() as td: path = Path(td) / "config.py" path.write_text( 'QWEN_API_URL = "http://127.0.0.1:7860"\n' 'CLONE_API_URL = "http://127.0.0.1:7861"\n' 'FASTER_API_URL = "http://127.0.0.1:8000"\n' 'AUDIOCPP_API_URL = "http://127.0.0.1:8080"\n', encoding="utf-8") with patch.object(hub.config, "__file__", str(path)), \ patch.object(hub, "_sync_audiocpp_server_port"): hub._write_config({ "QWEN_API_URL": "http://127.0.0.1:7862", "CLONE_API_URL": "http://127.0.0.1:7863", "FASTER_API_URL": "http://127.0.0.1:8001", "AUDIOCPP_API_URL": "http://127.0.0.1:8081", }) text = path.read_text(encoding="utf-8") self.assertIn('QWEN_API_URL = "http://127.0.0.1:7862"', text) self.assertIn('CLONE_API_URL = "http://127.0.0.1:7863"', text) self.assertIn('FASTER_API_URL = "http://127.0.0.1:8001"', text) self.assertIn('AUDIOCPP_API_URL = "http://127.0.0.1:8081"', text) class AudiocppServerConfigTests(unittest.TestCase): """update_server_config_port: rewriting the checkout's server.json.""" def _make_checkout(self, td, port=8080): from backends import audiocpp as audiocpp_backend import json checkout = Path(td) / "audio.cpp" checkout.mkdir() server_json = checkout / "server.json" server_json.write_text( json.dumps({"host": "127.0.0.1", "port": port, "models": [{"id": "qwen"}]}, indent=2), encoding="utf-8") return audiocpp_backend, checkout, server_json def test_rewrites_existing_server_json_port(self): import tempfile import json with tempfile.TemporaryDirectory() as td: mod, checkout, server_json = self._make_checkout(td, port=8080) with patch.object(mod, "find_local_checkout", return_value=checkout): self.assertTrue(mod.update_server_config_port(9090)) data = json.loads(server_json.read_text(encoding="utf-8")) self.assertEqual(data["port"], 9090) # Other keys are preserved. self.assertEqual(data["host"], "127.0.0.1") self.assertEqual(data["models"], [{"id": "qwen"}]) def test_noop_when_port_unchanged(self): import tempfile with tempfile.TemporaryDirectory() as td: mod, checkout, server_json = self._make_checkout(td, port=8080) before = server_json.read_text(encoding="utf-8") with patch.object(mod, "find_local_checkout", return_value=checkout): self.assertTrue(mod.update_server_config_port(8080)) self.assertEqual(server_json.read_text(encoding="utf-8"), before) def test_false_when_no_checkout(self): from backends import audiocpp as audiocpp_backend with patch.object(audiocpp_backend, "find_local_checkout", return_value=None): self.assertFalse(audiocpp_backend.update_server_config_port(9090)) if __name__ == "__main__": unittest.main()