"""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 unittest from pathlib import Path from unittest.mock import patch from backends import BackendStatus, ServerSpec from tests.test_tui import FakeCurses, FakeScreen from ui import hub, tui 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) 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")) 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...", "Server...", "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. self.assertEqual( captured["rows"], [("audio.cpp", "unavailable", "err", "dim"), ("qwen-tts", "running", "ok", "body")]) def test_menu_has_all_six_when_one_running_only(self): # Running but not installed (an external server) still unlocks the # Convert/Configure/Server entries. 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...", "Configure a backend...", "Server...", "Settings...", "Quit"]) def test_convert_with_no_available_backend_offers_setup(self): # One installed-but-not-ready backend → Convert is offered. The # convert menu lists no available backend, so only "Set up a # backend..." is shown; Enter selects it → setup menu lists 3 # backends; Esc goes back → convert returns None → main menu loops. # Then quit: main menu now has 5 options, Quit is the 5th (Down x4). from backends import BackendInfo, BackendStatus none = BackendStatus("k", "l", installed=True, configured=False) infos = [BackendInfo("audiocpp", "audio.cpp", lambda: none, lambda: 0), BackendInfo("qwen", "qwen-tts", lambda: none, lambda: 0), BackendInfo("faster", "faster", lambda: none, lambda: 0)] # installed=True so the main menu shows Convert; but ready/running # is False so the convert menu's available list is empty. statuses = [BackendStatus("audiocpp", "audio.cpp", installed=True, configured=False), BackendStatus("qwen", "qwen-tts", installed=True, configured=False), BackendStatus("faster", "faster", installed=True, configured=False)] with patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub, "REGISTRY", infos): # Convert(Enter), setup-entry(Enter), Esc on setup menu, # back at main menu -> Down x5 -> Enter (Quit; Settings sits # just before it). screen = FakeScreen(keys=[10, 10, 27, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, 10]) result = hub._hub_menu(screen) self.assertIsNone(result) 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")} 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"} with patch.object(hub, "_write_config", fake_write): 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}) # 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) def test_apply_settings_rejects_bad_values(self): original = {name: getattr(hub.config, name) for name in ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", "CHUNK_SIZE")} self.addCleanup(lambda: [setattr(hub.config, name, value) for name, value in original.items()]) with patch.object(hub, "_write_config") as mk_write: with self.assertRaises(ValueError): hub._apply_settings({"audio_format": "m4b", "audio_bitrate": "128k", "language": "Klingon", "chunk_size": "250"}) with self.assertRaises(ValueError): hub._apply_settings({"audio_format": "m4b", "audio_bitrate": "128k", "language": "English", "chunk_size": "0"}) 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")) 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"} 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"]) kinds = {f["key"]: f["kind"] for f in captured["fields"]} self.assertEqual(kinds["audio_format"], "choice") self.assertEqual(kinds["audio_bitrate"], "text") self.assertEqual(applied, [{"audio_format": "ogg", "audio_bitrate": "192k", "language": "English", "chunk_size": "300"}]) 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")} 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", 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) if __name__ == "__main__": unittest.main()