1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
|
"""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
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()
|
