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 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_running_event_escapes_starting_without_a_boot(self):
        # Regression for the stuck "Status: starting" run view: when this
        # run did not boot the server (already running locally, or remote)
        # the worker's probe reports "running" — and the book events that
        # follow must still move the panel to "processing".
        view, _ = self.make_view()
        view.handle_event({"kind": "running", "name": "audiocpp",
                           "url": "http://x"})
        self.assertEqual(view.server, "ready")
        self.assertFalse(view.started_server)
        view.handle_event({"kind": "book", "index": 1, "total": 1,
                           "name": "b"})
        self.assertEqual(view.phase, "convert")
        self.assertEqual(view.server, "processing")
        view.handle_event({"kind": "book_done", "name": "b", "ok": True})
        view.handle_event({"kind": "done", "ok": 1, "total": 1})
        self.assertEqual(view.phase, "done")
        self.assertNotEqual(view.server, "starting")

    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_chunk_failure_message_names_the_model_and_reason(self):
        # An "All" run stamps chunk_failed with the generating model and
        # the server's error detail; the message carries both, and the
        # results row keeps the reason (book_done(ok=False) has none).
        view, _ = self.make_view()
        view.handle_event({"kind": "book", "index": 1, "total": 3,
                           "name": "b.txt", "model": "m1"})
        view.handle_event({"kind": "chunk_failed", "chunk": 1, "total": 1,
                           "error": "vocoder backend buffer allocation "
                                    "failed", "model": "m1"})
        self.assertEqual(view.error_message,
                         "m1: chunk 1/1 failed — vocoder backend buffer "
                         "allocation failed")
        view.handle_event({"kind": "book_done", "name": "b.txt",
                           "ok": False, "model": "m1"})
        self.assertEqual(view.book_results,
                         [("b.txt", False, [],
                           "vocoder backend buffer allocation failed",
                           "m1")])

    def test_chunk_failure_without_detail_keeps_the_count(self):
        # No server error detail: the message (and the row reason) still
        # say which chunk of how many failed.
        view, _ = self.make_view()
        view.handle_event({"kind": "book", "index": 1, "total": 1,
                           "name": "b", "model": "m1"})
        view.handle_event({"kind": "chunk_failed", "chunk": 2, "total": 4,
                           "model": "m1"})
        self.assertEqual(view.error_message, "m1: chunk 2/4 failed")
        view.handle_event({"kind": "book_done", "name": "b", "ok": False,
                           "model": "m1"})
        self.assertEqual(view.book_results[0][3], "chunk 2/4 failed")

    def test_new_book_clears_a_stale_failure_message(self):
        # The heart of the "Chunk 1/1 failed persisted for every later
        # model" bug: a failure message from one book of an "All" run must
        # not linger under the next book's progress.
        view, _ = self.make_view()
        view.handle_event({"kind": "book", "index": 1, "total": 2,
                           "name": "b.txt", "model": "m1"})
        view.handle_event({"kind": "chunk_failed", "chunk": 1, "total": 1,
                           "error": "boom", "model": "m1"})
        view.handle_event({"kind": "book_done", "name": "b.txt",
                           "ok": False, "model": "m1"})
        self.assertTrue(view.error_message)
        view.handle_event({"kind": "book", "index": 2, "total": 2,
                           "name": "b.txt", "model": "m2"})
        self.assertEqual(view.error_message, "")
        self.assertEqual(view._book_error, "")

    def test_done_message_lists_the_failed_models(self):
        # The terminal "done" screen names what failed instead of showing
        # the last chunk failure (or a bare count).
        view, _ = self.make_view()
        for index, (model, ok) in enumerate(
                [("m1", False), ("m2", True), ("m3", False)], 1):
            view.handle_event({"kind": "book", "index": index, "total": 3,
                               "name": "b.txt", "model": model})
            view.handle_event({"kind": "book_done", "name": "b.txt",
                               "ok": ok, "model": model})
        view.handle_event({"kind": "done", "ok": 1, "total": 3})
        self.assertEqual(view.phase, "error")
        self.assertEqual(view.error_message,
                         "2 of 3 book(s) failed: m1, m3")

    def test_done_message_caps_the_failed_model_list(self):
        # More failures than fit the two detail lines: the list is capped
        # with the leftover count (the rows carry the full list).
        view, _ = self.make_view()
        for index in range(1, 8):
            view.handle_event({"kind": "book", "index": index, "total": 7,
                               "name": "b.txt", "model": f"m{index}"})
            view.handle_event({"kind": "book_done", "name": "b.txt",
                               "ok": False, "model": f"m{index}"})
        view.handle_event({"kind": "done", "ok": 0, "total": 7})
        self.assertEqual(
            view.error_message,
            "7 of 7 book(s) failed: m1, m2, m3, m4, m5, … +2 more")

    def test_done_all_failed_rows_only_names_models_of_failures(self):
        # A model that cannot even start (audiobook.py's constructor
        # failure path) reports book_failed without a model field: the
        # name stands in for the model in the summary.
        view, _ = self.make_view()
        view.handle_event({"kind": "book_failed", "name": "m1",
                           "error": "voice 'x' is not available",
                           "files": []})
        view.handle_event({"kind": "done", "ok": 0, "total": 1})
        self.assertEqual(view.error_message,
                         "1 of 1 book(s) failed: m1")

    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_exit_keeps_a_known_crash_hint(self):
        view, _ = self.make_view()
        view.handle_event({"kind": "exited", "name": "sglomni",
                           "returncode": 1, "log_tail": ["fp8..."],
                           "hint": "FP8 needs compute capability 8.9+"})
        self.assertEqual(view.phase, "error")
        self.assertEqual(view.boot_hint,
                         "FP8 needs compute capability 8.9+")

    def test_boot_failure_is_recorded_in_the_dated_log(self):
        # A failed boot never reaches the converter, so without this the
        # dated log the failure pointers name would stay blank.
        with tempfile.TemporaryDirectory() as tmp:
            log_path = os.path.join(tmp, "audiobook_test.log")
            view, _ = self.make_view(log_path=log_path)
            view.handle_event({"kind": "starting", "name": "sglomni",
                               "log_path": "/tmp/sglomni-server.log"})
            view.handle_event({"kind": "exited", "name": "sglomni",
                               "returncode": 1, "log_tail": ["boom"],
                               "hint": "FP8 needs compute capability 8.9+"})
            with open(log_path, encoding="utf-8") as logf:
                text = logf.read()
        self.assertIn("ERROR - server exited with code 1", text)
        self.assertIn("WARNING - hint: FP8 needs compute capability 8.9+",
                      text)
        self.assertIn("the server's own output is in /tmp/sglomni-server.log",
                      text)

    def test_boot_timeout_is_recorded_without_optional_detail(self):
        with tempfile.TemporaryDirectory() as tmp:
            log_path = os.path.join(tmp, "audiobook_test.log")
            view, _ = self.make_view(log_path=log_path)
            view.handle_event({"kind": "timeout", "name": "sglomni",
                               "seconds": 1200, "log_tail": []})
            with open(log_path, encoding="utf-8") as logf:
                text = logf.read()
        self.assertIn("ERROR - server did not become ready in time", text)
        self.assertNotIn("hint:", text)
        self.assertNotIn("the server's own output is in", text)

    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_down_during_boot_marks_not_responding(self):
        # The worker's boot-time probe of a server this run did not start:
        # a target that never answers is reported "down" (not the eternal
        # "starting"), and never offered for a stop (not started here).
        view, _ = self.make_view()
        view.handle_event({"kind": "server_down"})
        self.assertEqual(view.server, "down")
        self.assertEqual(view.phase, "boot")
        self.assertFalse(view.started_server)

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

    def test_all_run_book_events_carry_the_model(self):
        # "All (multiple generation)" runs stamp the generating model onto
        # the book and book_done/book_failed events; the view keeps it for
        # the progress line and the summary rows.
        view, _ = self.make_view()
        view.handle_event({"kind": "book", "index": 2, "total": 4,
                           "name": "book.txt", "model": "m1"})
        self.assertEqual(view.book, (2, 4, "book.txt", "m1"))
        view.handle_event({"kind": "book_done", "name": "book.txt",
                           "ok": True, "files": ["book_m1_Vivian.mp3"],
                           "model": "m1"})
        self.assertEqual(view.book_results,
                         [("book.txt", True, ["book_m1_Vivian.mp3"], "",
                           "m1")])
        view.handle_event({"kind": "book", "index": 3, "total": 4,
                           "name": "book.txt", "model": "m2"})
        view.handle_event({"kind": "book_failed", "name": "book.txt",
                           "error": "chunk failed", "model": "m2"})
        self.assertEqual(view.book_results[-1],
                         ("book.txt", False, [], "chunk failed", "m2"))
        # A plain (single-model) run carries no model: the field stays None.
        view.handle_event({"kind": "book", "index": 4, "total": 4,
                           "name": "book.txt"})
        self.assertEqual(view.book, (4, 4, "book.txt", None))


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_carriage_return_progress_splits_into_lines(self):
        # tqdm/git-style \r-only progress: each segment becomes its own
        # log line instead of one ever-growing buffered line.
        path, appender = self._appender()
        appender.write("pct 0\rpct 1\rpct 2\r")
        appender.flush()
        with open(path, encoding="utf-8") as handle:
            lines = handle.read().splitlines()
        self.assertEqual(len(lines), 3)
        self.assertTrue(lines[2].endswith(" - pct 2"))

    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")
        # server_url=None: no boot and no probe, so the run log test sees
        # only the worker's own events.
        view, _ = self.make_view(log_path=log_path, server_url=None)

        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_ready_without_autostart_shows_the_not_started_tag(self):
        # A run that talks to a server it did not start (already running
        # locally, or remote) tags the ready state so the panel does not
        # imply this run booted it.
        view, screen = self.make_view()
        view.handle_event({"kind": "running", "name": "audiocpp",
                           "url": "http://x"})
        view.render()
        self.assertIn("ready (not started by this run)", self._strings(screen))

    def test_not_responding_message_shown_while_down(self):
        view, screen = self.make_view()
        view.handle_event({"kind": "server_down"})
        view.render()
        text = self._strings(screen)
        self.assertIn("not responding", text)
        self.assertIn("the server is not responding", text)

    def test_error_summary_draws_the_boot_hint(self):
        view, screen = self.make_view()
        view.handle_event({"kind": "exited", "name": "sglomni",
                           "returncode": 1, "log_tail": [],
                           "hint": "FP8 needs compute capability 8.9+"})
        view.render()
        self.assertIn("FP8 needs compute capability 8.9+",
                      self._strings(screen))

    def test_error_screen_names_the_server_log(self):
        view, screen = self.make_view()
        view.handle_event({"kind": "starting", "name": "sglomni",
                           "log_path": "/tmp/sglomni-server.log"})
        view.handle_event({"kind": "exited", "name": "sglomni",
                           "returncode": 1, "log_tail": ["boom"]})
        view.render()
        self.assertIn("server log: /tmp/sglomni-server.log",
                      self._strings(screen))

    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_summary_screen_names_the_generating_model(self):
        # An "All" run's summary rows distinguish which model generated
        # each result (the progress line shows the model too).
        view, screen = self.make_view()
        view.handle_event({"kind": "book", "index": 1, "total": 2,
                           "name": "book.txt", "model": "m1"})
        view.handle_event({"kind": "book_done", "name": "book.txt",
                           "ok": True, "files": ["book_m1_Vivian.mp3"],
                           "model": "m1"})
        view.handle_event({"kind": "done", "ok": 1, "total": 2})
        view.render()
        text = self._strings(screen)
        self.assertIn("book.txt — m1", text)

    def test_summary_shows_failed_rows_first_with_their_reason(self):
        # With dozens of "All"-run results, the failures must be visible
        # without scrolling past the successes, and each failed row says
        # why it failed.
        view, screen = self.make_view()
        view.handle_event({"kind": "book", "index": 1, "total": 3,
                           "name": "book.txt", "model": "m1"})
        view.handle_event({"kind": "book_done", "name": "book.txt",
                           "ok": True, "model": "m1"})
        view.handle_event({"kind": "book", "index": 2, "total": 3,
                           "name": "book.txt", "model": "m2"})
        view.handle_event({"kind": "chunk_failed", "chunk": 1, "total": 1,
                           "error": "missing model root: dac",
                           "model": "m2"})
        view.handle_event({"kind": "book_done", "name": "book.txt",
                           "ok": False, "model": "m2"})
        view.handle_event({"kind": "book", "index": 3, "total": 3,
                           "name": "book.txt", "model": "m3"})
        view.handle_event({"kind": "book_done", "name": "book.txt",
                           "ok": True, "model": "m3"})
        view.handle_event({"kind": "done", "ok": 2, "total": 3})
        view.render()
        text = self._strings(screen)
        fail_pos = text.index("[FAIL]")
        ok_pos = text.index("[OK]")
        self.assertLess(fail_pos, ok_pos)
        self.assertIn("book.txt — m2: missing model root: dac", text)

    def test_progress_line_shows_the_model_of_a_chunk_failure(self):
        # While the run is live, the message line names the model that
        # failed (not just the chunk counters).
        view, screen = self.make_view()
        view.handle_event({"kind": "book", "index": 4, "total": 30,
                           "name": "book.txt", "model": "DramaBox-GGUF"})
        view.handle_event({"kind": "chunk_failed", "chunk": 1, "total": 1,
                           "error": "vocoder backend buffer allocation "
                                    "failed", "model": "DramaBox-GGUF"})
        view.render()
        text = self._strings(screen)
        self.assertIn("DramaBox-GGUF: chunk 1/1 failed", 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. A
        # successful run's summary does not point at the log file.
        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,
                                          log_path="/tmp/runs/a.log")
            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)
        self.assertNotIn("Full details in the log file", 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, which ends with the converter's
        # log file path where the full details live.
        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,
                                          log_path="/tmp/runs/a.log")
            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)
        self.assertIn("Full details in the log file: /tmp/runs/a.log", text)

    def test_stop_and_exit_empty_failure_summary_lists_the_log(self):
        # A run that errors before any book result (worker crash) still
        # points the post-TUI summary at the log file.
        with patch.object(runview.servers, "stop"), \
                patch.object(runview.common,
                             "record_post_tui_notice") as mk_notice:
            view, screen = self.make_view([], stop_and_exit=True,
                                          log_path="/tmp/runs/a.log")
            view._queue.put({"kind": "error", "message": "boom"})
            view.run()
        self.assertIn("No books were converted",
                      mk_notice.call_args[0][0])
        self.assertIn("Full details in the log file: /tmp/runs/a.log",
                      mk_notice.call_args[0][0])

    def test_stop_and_exit_boot_failure_names_reason_hint_and_server_log(self):
        # A run that dies in the boot phase must not summarize as a bare
        # "No books were converted": the reason, the known-crash hint,
        # and the server's own log path all land in the summary.
        with patch.object(runview.servers, "stop"), \
                patch.object(runview.common,
                             "record_post_tui_notice") as mk_notice:
            view, screen = self.make_view([], stop_and_exit=True,
                                          log_path="/tmp/runs/a.log")
            view._queue.put({"kind": "starting", "name": "sglomni",
                             "log_path": "/tmp/sglomni-server.log"})
            view._queue.put({"kind": "exited", "name": "sglomni",
                             "returncode": 1, "log_tail": [],
                             "hint": "FP8 needs compute capability 8.9+"})
            view.run()
        text = mk_notice.call_args[0][0]
        self.assertIn("No books were converted", text)
        self.assertIn("Failure: server exited with code 1", text)
        self.assertIn("hint: FP8 needs compute capability 8.9+", text)
        self.assertIn("server log: /tmp/sglomni-server.log", text)
        self.assertIn("Full details in the log file: /tmp/runs/a.log", 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, \
                patch.object(runview.common, "server_running",
                             return_value=True):
            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")), \
                    patch.object(runview.common, "server_running",
                                 return_value=False):
                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()

    def test_worker_probes_the_server_it_does_not_boot(self):
        # No autostart (the server is already running locally, or remote):
        # the worker still reports the target as "running" so the status
        # leaves "starting", then converts against it.
        with patch("audiobook.convert") as mk_convert, \
                patch.object(runview.common, "server_running",
                             return_value=True) as mk_running:
            view = self.make_view()
            view._worker_main()
        mk_running.assert_called_once_with("http://127.0.0.1:8080")
        mk_convert.assert_called_once()
        events = self._drain(view)
        self.assertEqual(events[0],
                         {"kind": "running", "name": "audiocpp",
                          "url": "http://127.0.0.1:8080"})
        self.assertEqual(events[-1]["kind"], "worker_exit")

    def test_worker_reports_down_when_the_probe_fails(self):
        # A target that never answers (dead remote, server stopped between
        # the form and the run) is reported "down"; the conversion still
        # runs and fails through the normal chunk/error path.
        with patch("audiobook.convert") as mk_convert, \
                patch.object(runview.common, "server_running",
                             return_value=False):
            view = self.make_view()
            view._worker_main()
        mk_convert.assert_called_once()
        self.assertEqual(self._drain(view)[0]["kind"], "server_down")

    def test_worker_skips_the_probe_without_a_url(self):
        # No server URL at all (backend vanished between menu and run):
        # there is nothing to probe, so no server events are queued.
        with patch("audiobook.convert"), \
                patch.object(runview.common, "server_running") as mk_running:
            view = self.make_view(server_url=None, server_name=None,
                                  server_identity=None)
            view._worker_main()
        mk_running.assert_not_called()
        self.assertEqual([e["kind"] for e in self._drain(view)],
                         ["worker_exit"])


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