"""Tests for the run view (ui/runview.py) — the conversion status screen. The view is driven the same way as the other TUI widgets: the fake curses module and recording screen from test_tui stand in for a terminal, the worker/monitor threads are stubbed, and events are fed through the view's own queue to exercise state transitions, rendering, and the Esc/q cancel → stop-server flow. """ import os import sys import tempfile import types import unittest from queue import Empty from unittest.mock import patch from tests.test_tui import FakeCurses, FakeScreen from ui import runview def _config(**overrides): kwargs = dict(backend="audiocpp", backend_label="audio.cpp", kwargs={}, book_files=["book.txt"], planned=["book.txt"], server_name="audiocpp", server_url="http://127.0.0.1:8080", server_identity="audiocpp") kwargs.update(overrides) return runview.RunConfig(**kwargs) class _FakeTui: """Stand-in for the curses module (installed into sys.modules).""" def setUp(self): self.curses = FakeCurses() patcher = patch.dict(sys.modules, {"curses": self.curses}) patcher.start() self.addCleanup(patcher.stop) runview.tui._THEME.clear() self.addCleanup(runview.tui._THEME.clear) def make_view(self, keys=(), width=80, height=24, **cfg): screen = FakeScreen(keys=keys, width=width, height=height) # Patch the thread targets at the class level BEFORE construction so # __init__'s Thread(target=self._worker_main) binds the stub. with patch.object(runview.RunView, "_worker_main", lambda self: None), \ patch.object(runview.RunView, "_monitor_main", lambda self: None): view = runview.RunView(screen, _config(**cfg), clock=lambda: 1000.0) return view, screen class FormatTests(_FakeTui, unittest.TestCase): def test_format_elapsed(self): self.assertEqual(runview._format_elapsed(0), "0:00") self.assertEqual(runview._format_elapsed(65), "1:05") self.assertEqual(runview._format_elapsed(3661), "1:01:01") def test_fit_truncates_with_tilde(self): self.assertEqual(runview._fit("hello", 3), "he~") self.assertEqual(runview._fit("hi", 10), "hi") def test_fit_truncates_wide_characters_by_display_columns(self): # Japanese characters fill two terminal cells each, so fitting # by character count would let lines overflow their pane. self.assertEqual(runview._fit("你好", 4), "你好") self.assertEqual(runview._fit("你好你好", 5), "你好~") self.assertEqual(runview._fit("こんにちは", 3), "こ~") def test_wrap_wraps_on_word_boundaries(self): self.assertEqual(runview._wrap("aaaa bbbb cccc dddd", 12), ["aaaa bbbb", "cccc dddd"]) def test_wrap_counts_wide_characters_as_two_columns(self): self.assertEqual(runview._wrap("日本語 テスト", 10), ["日本語", "テスト"]) class StateTransitionTests(_FakeTui, unittest.TestCase): def test_boot_flow_starting_to_ready(self): view, _ = self.make_view() view.handle_event({"kind": "starting", "name": "audiocpp", "pid": 1, "log_path": "/tmp/x.log"}) self.assertEqual(view.phase, "boot") self.assertEqual(view.server, "starting") self.assertTrue(view.started_server) view.handle_event({"kind": "ready", "name": "audiocpp", "url": "http://x"}) self.assertEqual(view.server, "ready") def test_chunk_progress_updates(self): view, _ = self.make_view() view.handle_event({"kind": "book", "index": 1, "total": 2, "name": "book.txt"}) self.assertEqual(view.phase, "convert") view.handle_event({"kind": "chunks", "total": 10}) view.handle_event({"kind": "chunk_done", "chunk": 4, "total": 10}) self.assertEqual(view.chunk_done, 4) self.assertEqual(view.chunk_total, 10) def test_done_all_books_is_terminal(self): view, _ = self.make_view() view.handle_event({"kind": "book", "index": 1, "total": 1, "name": "b"}) view.handle_event({"kind": "book_done", "name": "b", "ok": True}) view.handle_event({"kind": "done", "ok": 1, "total": 1}) self.assertEqual(view.phase, "done") def test_chunk_failure_leads_to_error(self): view, _ = self.make_view() view.handle_event({"kind": "book", "index": 1, "total": 1, "name": "b"}) view.handle_event({"kind": "chunk_failed", "chunk": 3, "total": 5}) view.handle_event({"kind": "book_done", "name": "b", "ok": False}) view.handle_event({"kind": "done", "ok": 0, "total": 1}) self.assertEqual(view.phase, "error") self.assertTrue(view.error_message) def test_server_exit_during_boot_is_error(self): view, _ = self.make_view() view.handle_event({"kind": "starting", "name": "audiocpp"}) view.handle_event({"kind": "exited", "name": "audiocpp", "returncode": 1, "log_tail": ["boom"]}) self.assertEqual(view.phase, "error") self.assertEqual(view.server, "error") self.assertEqual(view.log_tail, ["boom"]) def test_server_down_during_convert(self): view, _ = self.make_view() view.handle_event({"kind": "book", "index": 1, "total": 1, "name": "b"}) view.handle_event({"kind": "server_down"}) self.assertEqual(view.server, "down") def test_server_stopped_event_marks_server_stopped(self): view, _ = self.make_view() view.server = "stopping" view.handle_event({"kind": "server_stopped"}) self.assertEqual(view.server, "stopped") class LogAppenderTests(_FakeTui, unittest.TestCase): """_LogAppender: stray console output survives in the run's log file.""" def _appender(self): tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) path = os.path.join(tmp.name, "audiobook_20260828.log") return path, runview._LogAppender(path) def test_write_splits_lines_into_the_file(self): path, appender = self._appender() appender.write("one\ntwo\n") appender.write("three") appender.flush() with open(path, encoding="utf-8") as handle: lines = handle.read().splitlines() self.assertEqual(len(lines), 3) self.assertTrue(lines[0].endswith(" - one")) self.assertTrue(lines[2].endswith(" - three")) def test_blank_lines_and_empty_path_are_skipped(self): path, appender = self._appender() appender.write("\n\n") appender.flush() appender.write("x") appender.flush() empty = runview._LogAppender("") empty.write("ignored\n") # must not raise with open(path, encoding="utf-8") as handle: self.assertEqual(len(handle.read().splitlines()), 1) def test_unwritable_path_never_raises(self): appender = runview._LogAppender("/nonexistent-dir-zz/log.log") appender.write("boom\n") # must not raise appender.flush() def test_worker_stdout_is_mirrored_to_the_run_log(self): tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) log_path = os.path.join(tmp.name, "audiobook_20260828.log") view, _ = self.make_view(log_path=log_path) fake = types.ModuleType("audiobook") def convert(**_kwargs): print("[INFO] stray console output") fake.convert = convert with patch.dict(sys.modules, {"audiobook": fake}): view._worker_main() with open(log_path, encoding="utf-8") as handle: text = handle.read() self.assertIn("[INFO] stray console output", text) # The worker signed off normally through the event queue. kinds = [] while True: try: kinds.append(view._queue.get_nowait()["kind"]) except Empty: break self.assertEqual(kinds, ["worker_exit"]) class RenderTests(_FakeTui, unittest.TestCase): def _strings(self, screen): return " ".join(text for _, _, text, _ in screen.strings) def test_boot_screen_shows_server_and_status(self): view, screen = self.make_view() view.handle_event({"kind": "starting", "name": "audiocpp", "pid": 1, "log_path": "/tmp/x.log"}) view.render() text = self._strings(screen) self.assertIn("Server", text) self.assertIn("audio.cpp", text) self.assertIn("Status", text) self.assertIn("starting", text) self.assertIn("Esc or q: cancel", text) def test_summary_screen_after_done(self): view, screen = self.make_view() view.handle_event({"kind": "book", "index": 1, "total": 1, "name": "book.txt"}) view.handle_event({"kind": "book_done", "name": "book.txt", "ok": True}) view.handle_event({"kind": "done", "ok": 1, "total": 1}) view.render() text = self._strings(screen) self.assertIn("completed", text) self.assertIn("book.txt", text) self.assertIn("press any key", text) def test_stopping_status_shows_elapsed(self): view, screen = self.make_view() view.handle_event({"kind": "book", "index": 1, "total": 1, "name": "book.txt"}) view.handle_event({"kind": "book_done", "name": "book.txt", "ok": True}) view.handle_event({"kind": "done", "ok": 1, "total": 1}) view.server = "stopping" view.stop_started = 999.0 view.render() text = self._strings(screen) self.assertIn("stopping", text) self.assertIn("(1s)", text) # The stop wait ignores keys, so the footer must not promise that # pressing one returns to the menu. self.assertNotIn("press any key", text) self.assertIn("stopping the audiocpp server...", text) def test_error_screen_shows_detail_and_log(self): view, screen = self.make_view(log_path="/tmp/audiobook.log") view.handle_event({"kind": "book", "index": 1, "total": 1, "name": "b"}) view.handle_event({"kind": "chunk_failed", "chunk": 1, "total": 3}) view.handle_event({"kind": "book_done", "name": "b", "ok": False}) view.handle_event({"kind": "done", "ok": 0, "total": 1}) view.render() text = self._strings(screen) self.assertIn("failed", text) self.assertIn("/tmp/audiobook.log", text) class RunLoopTests(_FakeTui, unittest.TestCase): def test_terminal_screen_key_returns(self): view, screen = self.make_view(keys=[ord("x")]) view._queue.put({"kind": "done", "ok": 1, "total": 1}) view.run() # Returned to the menu without touching the server stop prompt # (started_server is False). self.assertEqual(view.phase, "done") def test_run_leaves_the_screen_blocking_again(self): # The timed redraw cadence must not leak into the hub: blocking # input is restored so later dialogs (tui.flash) wait for keys. view, screen = self.make_view(keys=[ord("x")]) view._queue.put({"kind": "done", "ok": 1, "total": 1}) view.run() self.assertEqual(screen.timeouts[-1], -1) def test_esc_cancels_and_confirms_stop_server(self): confirm_answers = [True, True] # cancel? yes; stop server? yes with patch.object(runview.tui, "confirm", side_effect=confirm_answers), \ patch.object(runview.servers, "alive", return_value=True), \ patch.object(runview.servers, "stop") as mk_stop: view, screen = self.make_view(keys=[27, ord("x")], autostart_spec="SPEC") view.started_server = True view.run() mk_stop.assert_called_once_with("audiocpp") def test_esc_decline_cancel_keeps_running(self): # First Esc: "cancel?" answered No → the run continues; a second # key then exits via a terminal state the test feeds. with patch.object(runview.tui, "confirm", side_effect=[False]): view, screen = self.make_view(keys=[27]) # Feeds a done event after the declined cancel so run() can exit. view._queue.put({"kind": "done", "ok": 1, "total": 1}) screen.keys.append(ord("x")) view.run() self.assertEqual(view.phase, "done") def test_stop_server_not_asked_when_dead(self): with patch.object(runview.tui, "confirm", return_value=True), \ patch.object(runview.servers, "alive", return_value=False), \ patch.object(runview.servers, "stop") as mk_stop: view, screen = self.make_view(keys=[27, ord("x")], autostart_spec="SPEC") view.started_server = True view.run() mk_stop.assert_not_called() def test_finished_run_toggle_off_waits_for_key_then_menu(self): # Toggle off: a key on the summary screen returns to the menu; no # prompt is asked and the server keeps running. with patch.object(runview.tui, "confirm") as mk_confirm, \ patch.object(runview.servers, "alive", return_value=True), \ patch.object(runview.servers, "stop") as mk_stop: view, screen = self.make_view(keys=[ord("x")], autostart_spec="SPEC") view.started_server = True view._queue.put({"kind": "done", "ok": 1, "total": 1}) result = view.run() self.assertFalse(result) mk_confirm.assert_not_called() mk_stop.assert_not_called() def test_stop_and_exit_stops_server_and_quits_without_keys(self): # Toggle on: the moment the run ends the server is stopped and the # view reports quit — no key press and no prompt anywhere. with patch.object(runview.servers, "alive", return_value=True), \ patch.object(runview.servers, "stop") as mk_stop, \ patch.object(runview.common, "record_post_tui_notice") as mk_notice: view, screen = self.make_view([], autostart_spec="SPEC", stop_and_exit=True) view.started_server = True view._queue.put({"kind": "book", "index": 1, "total": 1, "name": "book.txt"}) view._queue.put({"kind": "book_done", "name": "book.txt", "ok": True, "files": ["book_test_michael.mp3"]}) view._queue.put({"kind": "done", "ok": 1, "total": 1}) result = view.run() self.assertTrue(result) mk_stop.assert_called_once_with("audiocpp") text = mk_notice.call_args[0][0] self.assertIn("Output directory:", text) self.assertIn("[OK] book.txt: book_test_michael.mp3", text) self.assertIn("1 of 1 book(s) generated successfully", text) self.assertIn("Elapsed time:", text) def test_stop_and_exit_leaves_external_servers_alone(self): # A server this run did not start is never stopped; the TUI still # quits and the summary is still printed. with patch.object(runview.servers, "stop") as mk_stop, \ patch.object(runview.common, "record_post_tui_notice") as mk_notice: view, screen = self.make_view([], server_name=None, stop_and_exit=True) view.started_server = False view._queue.put({"kind": "done", "ok": 1, "total": 1}) result = view.run() self.assertTrue(result) mk_stop.assert_not_called() self.assertIn("No books were converted", mk_notice.call_args[0][0]) def test_stop_and_exit_failure_summary_lists_the_error(self): # A failed ending also auto-exits; the failing book's detail line # lands in the post-TUI summary. with patch.object(runview.servers, "alive", return_value=True), \ patch.object(runview.servers, "stop"), \ patch.object(runview.common, "record_post_tui_notice") as mk_notice: view, screen = self.make_view([], autostart_spec="SPEC", stop_and_exit=True) view.started_server = True view._queue.put({"kind": "book_failed", "name": "bad.txt", "error": "chunk 3 failed", "files": ["bad_x.mp3"]}) view._queue.put({"kind": "done", "ok": 0, "total": 1}) result = view.run() self.assertTrue(result) text = mk_notice.call_args[0][0] self.assertIn("[FAIL] bad.txt: bad_x.mp3", text) self.assertIn("chunk 3 failed", text) self.assertIn("0 of 1 book(s) generated successfully", text) class WorkerTests(_FakeTui, unittest.TestCase): """The worker thread's handoff into audiobook.convert.""" def make_view(self, **cfg): screen = FakeScreen() return runview.RunView(screen, _config(**cfg), clock=lambda: 1000.0) def _drain(self, view): events = [] while True: try: events.append(view._queue.get_nowait()) except Empty: return events def test_book_files_reach_convert_once(self): # Regression: _preflight stashes book_files/planned in the form # kwargs and RunConfig carries them as fields too; passing both to # convert() raised "got multiple values for keyword argument". view = self.make_view( kwargs={"voice": "alloy", "book_files": ["b.txt"], "planned": ["b.txt"]}, book_files=["b.txt"], planned=["b.txt"]) with patch("audiobook.convert") as mk_convert: view._worker_main() _, kw = mk_convert.call_args self.assertEqual(kw["book_files"], ["b.txt"]) self.assertEqual(kw["planned"], ["b.txt"]) self.assertEqual(kw["voice"], "alloy") self.assertNotIn("error", [e["kind"] for e in self._drain(view)]) def test_worker_error_is_appended_to_the_dated_log(self): # A crash before convert() configures logging must still leave its # trace in the file the failure screen points at. with tempfile.TemporaryDirectory() as tmp: log_path = os.path.join(tmp, "audiobook_20260825.log") view = self.make_view(log_path=log_path) with patch("audiobook.convert", side_effect=TypeError("boom")): view._worker_main() with open(log_path, encoding="utf-8") as handle: text = handle.read() self.assertIn("ERROR - boom", text) self.assertIn("error", [e["kind"] for e in self._drain(view)]) def test_restart_first_stops_then_boots_the_new_model(self): # The managed qwen server hosts another model than this run picked: # the worker stops it (releasing the single port) before booting # the spec again — whose argv now names the newly-selected model. from types import SimpleNamespace spec = SimpleNamespace(name="qwen") events = [] def fake_start(spec_, progress=None, cancel=None): events.append(("start", progress)) progress({"kind": "ready", "name": "qwen", "url": "http://127.0.0.1:7860"}) return True with patch("audiobook.convert") as mk_convert, \ patch.object(runview.servers, "stop") as mk_stop, \ patch.object(runview.servers, "start", side_effect=fake_start): view = self.make_view(autostart_spec=spec, restart_first=True, server_name="qwen") view._worker_main() mk_stop.assert_called_once_with("qwen") events = self._drain(view) kinds = [e["kind"] for e in events] self.assertIn("ready", kinds) self.assertNotIn("error", kinds) mk_convert.assert_called_once() def test_no_restart_without_the_flag(self): # A plain autostart never stops a server first. from types import SimpleNamespace spec = SimpleNamespace(name="qwen") with patch("audiobook.convert"), \ patch.object(runview.servers, "stop") as mk_stop, \ patch.object(runview.servers, "start", return_value=True): view = self.make_view(autostart_spec=spec, server_name="qwen") view._worker_main() mk_stop.assert_not_called() if __name__ == "__main__": unittest.main()