path: root/app/tests/test_runview.py

"""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 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_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")

    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 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)])


if __name__ == "__main__":
    unittest.main()