From f00249db9d1ea051d29aa1bcca869fc4b88e83eb Mon Sep 17 00:00:00 2001 From: historia Date: Mon, 24 Aug 2026 02:59:26 -0400 Subject: refactor: add app directory, dir structure change --- app/tests/test_hub.py | 502 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 502 insertions(+) create mode 100644 app/tests/test_hub.py (limited to 'app/tests/test_hub.py') diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py new file mode 100644 index 0000000..b3a17e8 --- /dev/null +++ b/app/tests/test_hub.py @@ -0,0 +1,502 @@ +"""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() -- cgit v1.2.3