feat: tui backend server progress and generate script progress

1 files changed, 183 insertions, 0 deletions

diff --git a/app/tests/test_converter_progress.py b/app/tests/test_converter_progress.py
new file mode 100644
index 0000000..1041173
--- /dev/null
+++ b/app/tests/test_converter_progress.py
@@ -0,0 +1,183 @@
+"""Tests for the converter's progress-event and cancellation plumbing.
+
+These exercise the wiring the TUI run view relies on: a ``progress``
+callback receiving book/chunk/done events, a ``cancel`` (threading.Event)
+aborting the run between chunks (raising ConversionCancelled), and the
+injectable ``confirm`` hook on the overwrite prompt.
+"""
+
+import io
+import tempfile
+import threading
+import unittest
+from contextlib import redirect_stdout
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from converter import config, tts
+from converter import converter as converter_mod
+from converter.converter import (
+    AudiobookConverter,
+    ConversionCancelled,
+    prompt_overwrite,
+    voice_mode_for,
+)
+
+
+class VoiceModeForTests(unittest.TestCase):
+    def test_faster_always_clones(self):
+        self.assertEqual(voice_mode_for(tts.BACKEND_FASTER),
+                         tts.VOICE_MODE_CLONE)
+
+    def test_audiocpp_voice_clones(self):
+        self.assertEqual(voice_mode_for(tts.BACKEND_AUDIOCPP, voice="narrator"),
+                         tts.VOICE_MODE_CLONE)
+
+    def test_audiocpp_no_voice_is_custom(self):
+        self.assertEqual(voice_mode_for(tts.BACKEND_AUDIOCPP),
+                         tts.VOICE_MODE_CUSTOM)
+
+    def test_qwen_clone_wav_clones(self):
+        self.assertEqual(voice_mode_for(tts.BACKEND_QWEN, clone="x.wav"),
+                         tts.VOICE_MODE_CLONE)
+
+    def test_qwen_no_clone_is_custom(self):
+        self.assertEqual(voice_mode_for(tts.BACKEND_QWEN),
+                         tts.VOICE_MODE_CUSTOM)
+
+
+class PromptOverwriteConfirmTests(unittest.TestCase):
+    def test_confirm_callback_receives_message_and_default(self):
+        calls = []
+        result = prompt_overwrite([Path("out.mp3")], "out",
+                                  confirm=lambda m, d: calls.append((m, d)) or False)
+        self.assertFalse(result)
+        self.assertEqual(len(calls), 1)
+        self.assertTrue(calls[0][1])  # default yes
+        self.assertIn("out.mp3", calls[0][0])
+
+
+class _ConvertFixture:
+    """A real AudiobookConverter whose TTS client is stubbed."""
+
+    def __init__(self, test_case):
+        self.test = test_case
+        self._books_tmp = tempfile.TemporaryDirectory()
+        self._output_tmp = tempfile.TemporaryDirectory()
+        self._orig = (converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER)
+        converter_mod.BOOKS_FOLDER = Path(self._books_tmp.name)
+        converter_mod.AUDIOBOOKS_FOLDER = Path(self._output_tmp.name)
+        (converter_mod.BOOKS_FOLDER / "book.txt").write_text(
+            "one two three four five", encoding="utf-8")
+        # The stub returns a path that does not exist on disk, so the
+        # final assembly (and cover art) is patched out of the run() path.
+        self._patchers = [
+            patch.object(converter_mod.audio, "combine_chunks",
+                         return_value=True),
+            patch.object(converter_mod.audio, "combine_chapters_to_m4b",
+                         return_value=True),
+            patch.object(converter_mod.cover, "generate_cover",
+                         return_value=None),
+        ]
+        for patcher in self._patchers:
+            patcher.start()
+        test_case.addCleanup(self.cleanup)
+
+    def cleanup(self):
+        converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = self._orig
+        for patcher in self._patchers:
+            patcher.stop()
+        self._books_tmp.cleanup()
+        self._output_tmp.cleanup()
+
+    def build(self, progress=None, cancel=None):
+        # Patch the TTS client construction so the real constructor runs
+        # (exercising the progress/cancel wiring) without dialing a server.
+        with patch.object(converter_mod, "QwenTTSClient",
+                          return_value=MagicMock()):
+            converter = AudiobookConverter(
+                voice_mode=tts.VOICE_MODE_CUSTOM, backend=tts.BACKEND_QWEN,
+                output_format="mp3", language="English",
+                progress=progress, cancel=cancel)
+        converter.tts.process_chunk_with_retry.return_value = "chunk_0001.wav"
+        converter._book_files = [converter_mod.BOOKS_FOLDER / "book.txt"]
+        converter._planned = [(converter_mod.BOOKS_FOLDER / "book.txt",
+                               "book_Vivian")]
+        return converter
+
+
+class ProgressEventTests(unittest.TestCase):
+    def setUp(self):
+        self.fixture = _ConvertFixture(self)
+
+    def test_run_emits_book_chunks_done(self):
+        events = []
+        converter = self.fixture.build(progress=events.append)
+        ok = converter.run()
+        self.assertTrue(ok)
+        kinds = [event["kind"] for event in events]
+        self.assertEqual(kinds, ["book", "chunks", "chunk_done",
+                                 "book_done", "done"])
+        self.assertEqual(events[0]["name"], "book.txt")
+        self.assertEqual(events[-1]["ok"], 1)
+
+    def test_run_suppresses_console_prints_when_progress_set(self):
+        buf = io.StringIO()
+        converter = self.fixture.build(progress=lambda e: None)
+        with redirect_stdout(buf):
+            converter.run()
+        # The banner/summary/chunk prints are replaced by events.
+        out = buf.getvalue()
+        self.assertNotIn("CONVERSION SUMMARY", out)
+        self.assertNotIn("PROCESSING", out)
+        self.assertNotIn("completed", out)
+
+    def test_chunk_failed_sets_error_state(self):
+        events = []
+        converter = self.fixture.build(progress=events.append)
+        converter.tts.process_chunk_with_retry.return_value = None
+        converter.run()
+        self.assertIn("chunk_failed",
+                      [event["kind"] for event in events])
+        self.assertEqual(events[-1]["kind"], "done")
+        self.assertEqual(events[-1]["ok"], 0)
+
+
+class CancelTests(unittest.TestCase):
+    def setUp(self):
+        self.fixture = _ConvertFixture(self)
+
+    def test_cancel_between_chunks_aborts_and_emits_cancelled(self):
+        events = []
+        cancel = threading.Event()
+        converter = self.fixture.build(progress=events.append, cancel=cancel)
+        # Cancel as the first chunk completes; the next chunk's pre-check
+        # must raise ConversionCancelled before requesting it.
+        def generate(chunk_num, text):
+            cancel.set()
+            return "chunk_0001.wav"
+
+        converter.tts.process_chunk_with_retry.side_effect = generate
+        with patch.object(converter_mod, "chunking") as mk_chunking:
+            mk_chunking.split_into_chunks.return_value = [
+                "one two", "three four", "five"]
+            converter.run()
+        kinds = [event["kind"] for event in events]
+        self.assertIn("cancelled", kinds)
+        self.assertEqual(events[-1]["kind"], "done")
+        self.assertTrue(events[-1]["cancelled"])
+
+    def test_check_cancelled_raises_when_event_set(self):
+        cancel = threading.Event()
+        cancel.set()
+        converter = self.fixture.build(cancel=cancel)
+        with self.assertRaises(ConversionCancelled):
+            converter._check_cancelled()
+
+    def test_check_cancelled_silent_when_not_set(self):
+        converter = self.fixture.build(cancel=threading.Event())
+        converter._check_cancelled()  # no raise
+
+
+if __name__ == "__main__":
+    unittest.main()