diff options
Diffstat (limited to 'tests/test_hub.py')
| -rw-r--r-- | tests/test_hub.py | 502 |
1 files changed, 0 insertions, 502 deletions
diff --git a/tests/test_hub.py b/tests/test_hub.py deleted file mode 100644 index b3a17e8..0000000 --- a/tests/test_hub.py +++ /dev/null @@ -1,502 +0,0 @@ -"""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() |