"""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.clients import ( BACKEND_AUDIOCPP, BACKEND_FASTER, BACKEND_QWEN, VOICE_MODE_CLONE, VOICE_MODE_CUSTOM, ) 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(BACKEND_FASTER), VOICE_MODE_CLONE) def test_audiocpp_voice_clones(self): self.assertEqual(voice_mode_for(BACKEND_AUDIOCPP, voice="narrator"), VOICE_MODE_CLONE) def test_audiocpp_no_voice_is_custom(self): self.assertEqual(voice_mode_for(BACKEND_AUDIOCPP), VOICE_MODE_CUSTOM) def test_qwen_clone_wav_clones(self): self.assertEqual(voice_mode_for(BACKEND_QWEN, clone="x.wav"), VOICE_MODE_CLONE) def test_qwen_no_clone_is_custom(self): self.assertEqual(voice_mode_for(BACKEND_QWEN), 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=VOICE_MODE_CUSTOM, backend=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) def test_book_done_reports_output_files(self): # book_done carries the output file names the run view's summary # lists after the TUI closes. events = [] converter = self.fixture.build(progress=events.append) converter.run() done = next(e for e in events if e["kind"] == "book_done") self.assertEqual(done["files"], ["book_Vivian.mp3"]) self.assertEqual(converter.current_outputs, ["book_Vivian.mp3"]) def test_multi_chapter_book_lists_every_chapter_file(self): # A multi-section book (no --single-file) produces one file per # chapter, all reported on the event. sections = [MagicMock(text=f"chapter {n} text.", title=t) for n, t in enumerate(("One", "Two"), 1)] book = MagicMock(title="Book", author="Author", sections=sections) events = [] with patch.object(converter_mod.extractors, "extract_book", return_value=book): converter = self.fixture.build(progress=events.append) converter.single_file = False converter.run() done = next(e for e in events if e["kind"] == "book_done") self.assertTrue(done["ok"]) self.assertEqual(done["files"], ["book_Vivian_01_One.mp3", "book_Vivian_02_Two.mp3"]) 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()