"""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 sys import unittest 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_wrap_wraps_on_word_boundaries(self): self.assertEqual(runview._wrap("aaaa bbbb cccc dddd", 12), ["aaaa bbbb", "cccc dddd"]) 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") 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_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_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() if __name__ == "__main__": unittest.main()