diff options
Diffstat (limited to 'app/tests/test_converter.py')
| -rw-r--r-- | app/tests/test_converter.py | 619 |
1 files changed, 619 insertions, 0 deletions
diff --git a/app/tests/test_converter.py b/app/tests/test_converter.py new file mode 100644 index 0000000..2fe0f5d --- /dev/null +++ b/app/tests/test_converter.py @@ -0,0 +1,619 @@ +"""Tests for the audiobook converter orchestration helpers.""" + +import io +import logging +import tempfile +import time +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, + find_existing_outputs, + prompt_overwrite, + setup_logging, +) + + +class SanitizeFilenameTests(unittest.TestCase): + def test_removes_invalid_characters(self): + self.assertEqual(AudiobookConverter._sanitize_filename('A "bad" name: here'), + "A bad name here") + + def test_collapses_whitespace(self): + self.assertEqual(AudiobookConverter._sanitize_filename(" spaced\tout "), "spaced out") + + def test_empty_falls_back(self): + self.assertEqual(AudiobookConverter._sanitize_filename("///"), "chapter") + + +class ConfigurationValidationTests(unittest.TestCase): + def test_invalid_voice_mode_rejected(self): + with self.assertRaises(ValueError): + AudiobookConverter(voice_mode="custon_voice") + + def test_nonpositive_speed_rejected(self): + with self.assertRaises(ValueError): + AudiobookConverter(speed=0) + + def test_unknown_format_rejected(self): + with self.assertRaises(ValueError): + AudiobookConverter(output_format="wma") + + def test_unknown_language_rejected(self): + with self.assertRaises(ValueError): + AudiobookConverter(language="klingon") + + def test_unknown_backend_rejected(self): + with self.assertRaises(ValueError) as ctx: + AudiobookConverter(backend="piper") + self.assertIn("piper", str(ctx.exception)) + self.assertIn("audiocpp", str(ctx.exception)) + + def test_language_defaults_to_config(self): + with patch("converter.converter.QwenTTSClient") as mock_tts: + AudiobookConverter(backend=tts.BACKEND_QWEN) + self.assertEqual(mock_tts.call_args.kwargs["language"], config.LANGUAGE) + + def test_output_format_defaults_to_config(self): + with patch("converter.converter.QwenTTSClient"): + converter = AudiobookConverter(backend=tts.BACKEND_QWEN) + self.assertEqual(converter.output_format, config.AUDIO_FORMAT) + + def test_language_normalized_before_tts_client(self): + with patch("converter.converter.QwenTTSClient") as mock_tts: + converter = AudiobookConverter(language="ja", backend=tts.BACKEND_QWEN) + self.assertEqual(converter.language, "Japanese") + self.assertEqual(mock_tts.call_args.kwargs["language"], "Japanese") + + +class FindExistingOutputsTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + self._original = converter_mod.AUDIOBOOKS_FOLDER + converter_mod.AUDIOBOOKS_FOLDER = self.folder + + def tearDown(self): + converter_mod.AUDIOBOOKS_FOLDER = self._original + self._tmp.cleanup() + + def _touch(self, name): + path = self.folder / name + path.write_bytes(b"x") + return path + + def test_no_existing_output(self): + self.assertEqual(find_existing_outputs("dune", "mp3"), []) + + def test_primary_output_detected(self): + self._touch("dune.mp3") + self.assertEqual([p.name for p in find_existing_outputs("dune", "mp3")], + ["dune.mp3"]) + + def test_chapter_and_speed_copies_detected(self): + for name in ("dune_01_Dune.mp3", "dune_02_Barony.mp3", "dune_1.5x.mp3"): + self._touch(name) + self._touch("dune2_01.mp3") # different book stem; must not match + found = [p.name for p in find_existing_outputs("dune", "mp3")] + self.assertEqual(len(found), 3) + + def test_other_extensions_ignored(self): + self._touch("dune.mp3") + self.assertEqual(find_existing_outputs("dune", "m4b"), []) + + def test_glob_metacharacters_in_stem(self): + self._touch("book [1].mp3") + self._touch("book [1]_1.5x.mp3") + found = [p.name for p in find_existing_outputs("book [1]", "mp3")] + self.assertEqual(sorted(found), ["book [1].mp3", "book [1]_1.5x.mp3"]) + + def test_narrator_named_outputs_detected(self): + for name in ("dune_Vivian.mp3", "dune_Vivian_1.5.mp3", "dune_Vivian_01_Dune.mp3"): + self._touch(name) + found = [p.name for p in find_existing_outputs("dune_Vivian", "mp3")] + self.assertEqual(len(found), 3) + + def test_legacy_outputs_without_narrator_ignored(self): + self._touch("dune.mp3") + self._touch("dune_1.5.mp3") + self.assertEqual(find_existing_outputs("dune_Vivian", "mp3"), []) + + +class NarratorTagTests(unittest.TestCase): + def _converter(self, voice_mode, ref_audio=None, instructions=None): + converter = AudiobookConverter.__new__(AudiobookConverter) + converter.voice_mode = voice_mode + converter.voice_clone_ref_audio = ref_audio + converter.backend = tts.BACKEND_QWEN + converter.voice = None + converter.instructions = instructions + return converter + + def test_custom_voice_uses_speaker_display_name(self): + self.assertEqual(self._converter(tts.VOICE_MODE_CUSTOM)._narrator_tag(), + "Vivian") + + def test_multi_word_display_name_gets_underscores(self): + with patch.object(config, "SPEAKER", "uncle_fu"): + self.assertEqual(self._converter(tts.VOICE_MODE_CUSTOM)._narrator_tag(), + "Uncle_Fu") + + def test_clone_uses_reference_audio_stem(self): + self.assertEqual(self._converter(tts.VOICE_MODE_CLONE, "/x/ref.wav")._narrator_tag(), + "ref") + + def test_clone_stem_spaces_become_underscores(self): + self.assertEqual(self._converter(tts.VOICE_MODE_CLONE, "/x/my voice.wav")._narrator_tag(), + "my_voice") + + def test_invalid_characters_sanitized(self): + self.assertEqual(self._converter(tts.VOICE_MODE_CLONE, "/x/bad:name?.wav")._narrator_tag(), + "bad_name") + + def test_empty_after_sanitize_falls_back(self): + self.assertEqual(self._converter(tts.VOICE_MODE_CLONE, "/x/???.wav")._narrator_tag(), + "narrator") + + def _audiocpp_converter(self, voice=None, instructions=None): + converter = self._converter(tts.VOICE_MODE_CUSTOM, + instructions=instructions) + converter.backend = tts.BACKEND_AUDIOCPP + converter.voice = voice + return converter + + def test_audiocpp_design_run_uses_designed_tag(self): + # An instruction without a voice (voice design, or instruction- + # defined voices) must not be named after the built-in speaker. + converter = self._audiocpp_converter(instructions="A warm narrator") + self.assertEqual(converter._narrator_tag(), "designed") + + def test_audiocpp_instruction_with_voice_keeps_voice_tag(self): + converter = self._audiocpp_converter( + voice="narrator", instructions="Calm delivery") + self.assertEqual(converter._narrator_tag(), "narrator") + + def test_audiocpp_speaker_mode_keeps_speaker_tag(self): + converter = self._audiocpp_converter() + self.assertEqual(converter._narrator_tag(), "Vivian") + + def test_preflight_design_run_uses_designed_tag(self): + with tempfile.TemporaryDirectory() as books_tmp, \ + tempfile.TemporaryDirectory() as output_tmp: + original = (converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER) + converter_mod.BOOKS_FOLDER = Path(books_tmp) + converter_mod.AUDIOBOOKS_FOLDER = Path(output_tmp) + try: + (converter_mod.BOOKS_FOLDER / "book.txt").write_text( + "hello world", encoding="utf-8") + with patch("builtins.input", + side_effect=AssertionError("should not prompt")): + _, planned = AudiobookConverter.preflight_overwrites( + tts.BACKEND_AUDIOCPP, None, tts.VOICE_MODE_CUSTOM, + None, "mp3", instructions="A warm narrator") + self.assertEqual(planned, [(converter_mod.BOOKS_FOLDER / "book.txt", + "book_designed")]) + finally: + converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = original + + +class ChapterDebugDirTests(unittest.TestCase): + """Per-chapter debug subfolder naming (chunk numbering restarts per chapter).""" + + def test_none_when_not_debugging(self): + self.assertIsNone(AudiobookConverter._chapter_debug_dir(None, 3, "The Trial")) + + def test_chapter_subfolder_named_by_index_and_title(self): + book_dir = Path("debug") / "dune_Vivian" + chapter_dir = AudiobookConverter._chapter_debug_dir(book_dir, 3, "The Trial") + self.assertEqual(chapter_dir, book_dir / "03_The Trial") + + def test_untitled_chapter_uses_fallback(self): + chapter_dir = AudiobookConverter._chapter_debug_dir(Path("d"), 1, "") + self.assertEqual(chapter_dir, Path("d") / "01_chapter") + + +class DebugDumpTests(unittest.TestCase): + """--debug: per-chunk text/audio dumps and request/response logging.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self._debug_folder = patch.object(converter_mod, "DEBUG_FOLDER", Path(self._tmp.name)) + self._debug_folder.start() + self.debug_root = Path(self._tmp.name) + self.converter = AudiobookConverter.__new__(AudiobookConverter) + self.converter.client_chunks = True + self.converter.tts = MagicMock() + + def tearDown(self): + self._debug_folder.stop() + self._tmp.cleanup() + + def _chunk_source(self, name, body=b"audio"): + path = self.debug_root / "sources" / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(body) + return path + + def test_successful_chunk_dumps_text_and_audio(self): + audio = self._chunk_source("chunk_0001.wav") + self.converter.tts.process_chunk_with_retry.return_value = audio + results = self.converter._synthesize_chunks(["Hello world."], + debug_dir=self.debug_root / "book") + self.assertEqual(results, {1: audio}) + debug_dir = self.debug_root / "book" + self.assertEqual((debug_dir / "chunk_0001.txt").read_text(encoding="utf-8"), + "Hello world.") + self.assertEqual((debug_dir / "chunk_0001.wav").read_bytes(), b"audio") + + def test_failed_chunk_dumps_text_but_no_audio(self): + self.converter.tts.process_chunk_with_retry.return_value = None + results = self.converter._synthesize_chunks(["Hello again."], + debug_dir=self.debug_root / "book") + self.assertEqual(results, {1: None}) + debug_dir = self.debug_root / "book" + self.assertEqual([path.name for path in sorted(debug_dir.iterdir())], + ["chunk_0001.txt"]) + + def test_text_dumped_even_when_request_raises(self): + self.converter.tts.process_chunk_with_retry.side_effect = RuntimeError("boom") + results = self.converter._synthesize_chunks(["Crash text."], + debug_dir=self.debug_root / "book") + self.assertEqual(results, {1: None}) + self.assertEqual((self.debug_root / "book" / "chunk_0001.txt").read_text( + encoding="utf-8"), "Crash text.") + + def test_audio_suffix_preserved_and_nested_dirs_created(self): + audio = self._chunk_source("generated.mp3") + self.converter.tts.process_chunk_with_retry.return_value = audio + self.converter._synthesize_chunks(["Hello."], + debug_dir=self.debug_root / "nested" / "book") + self.assertTrue((self.debug_root / "nested" / "book" / "chunk_0001.mp3").exists()) + + def test_no_debug_dir_writes_nothing(self): + audio = self._chunk_source("chunk_0001.wav") + self.converter.tts.process_chunk_with_retry.return_value = audio + results = self.converter._synthesize_chunks(["Hello world."]) + self.assertEqual(results, {1: audio}) + self.assertEqual([path.name for path in self.debug_root.iterdir()], ["sources"]) + + def test_request_and_response_are_logged(self): + audio = self._chunk_source("chunk_0001.wav") + self.converter.tts.process_chunk_with_retry.return_value = audio + with self.assertLogs("converter.converter", level="DEBUG") as logs: + self.converter._synthesize_chunks(["Hello world."], + debug_dir=self.debug_root / "book") + joined = "\n".join(logs.output) + self.assertIn("Chunk 1/1 request text: Hello world.", joined) + self.assertIn("Chunk 1/1 response in", joined) + self.assertIn("chunk_0001.wav", joined) + + def test_debug_write_failure_does_not_abort_conversion(self): + blocker = self.debug_root / "blocker" + blocker.write_bytes(b"") + audio = self._chunk_source("chunk_0001.wav") + self.converter.tts.process_chunk_with_retry.return_value = audio + results = self.converter._synthesize_chunks(["Hello."], debug_dir=blocker / "book") + self.assertEqual(results, {1: audio}) + + def test_failed_chunk_stops_remaining_chunks(self): + audio = self._chunk_source("chunk_0001.wav") + self.converter.tts.process_chunk_with_retry.side_effect = [audio, None, audio] + results = self.converter._synthesize_chunks(["One.", "Two.", "Three."]) + self.assertEqual(results, {1: audio, 2: None}) + self.assertEqual(self.converter.tts.process_chunk_with_retry.call_count, 2) + + def test_raising_chunk_stops_remaining_chunks(self): + audio = self._chunk_source("chunk_0001.wav") + self.converter.tts.process_chunk_with_retry.side_effect = [audio, RuntimeError("boom")] + results = self.converter._synthesize_chunks(["One.", "Two.", "Three."]) + self.assertEqual(results, {1: audio, 2: None}) + self.assertEqual(self.converter.tts.process_chunk_with_retry.call_count, 2) + + def test_debug_flag_wiring(self): + with patch("converter.converter.QwenTTSClient"): + self.assertFalse(AudiobookConverter(backend=tts.BACKEND_QWEN).debug) + self.assertTrue(AudiobookConverter(debug=True, backend=tts.BACKEND_QWEN).debug) + + +class SetupLoggingTests(unittest.TestCase): + """Console handler stays quiet; the log file keeps the full record.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self._logs_folder = patch.object(converter_mod, "LOGS_FOLDER", Path(self._tmp.name)) + self._logs_folder.start() + self._root = logging.getLogger() + self._saved_handlers = self._root.handlers[:] + self._saved_level = self._root.level + self._saved_converter_level = logging.getLogger("converter").level + self._root.handlers.clear() + + def tearDown(self): + for handler in self._root.handlers: + if handler not in self._saved_handlers: + handler.close() + self._root.handlers[:] = self._saved_handlers + self._root.setLevel(self._saved_level) + logging.getLogger("converter").setLevel(self._saved_converter_level) + self._logs_folder.stop() + self._tmp.cleanup() + + def _console_handler(self): + matches = [h for h in logging.getLogger().handlers + if isinstance(h, logging.StreamHandler) + and not isinstance(h, logging.FileHandler)] + self.assertEqual(len(matches), 1) + return matches[0] + + def _file_handler(self): + matches = [h for h in logging.getLogger().handlers + if isinstance(h, logging.FileHandler)] + self.assertEqual(len(matches), 1) + return matches[0] + + def test_console_quiet_and_file_verbose_by_default(self): + setup_logging() + self.assertEqual(self._console_handler().level, logging.WARNING) + self.assertEqual(self._file_handler().level, logging.INFO) + + def test_debug_flag_lowers_both_handlers(self): + setup_logging(debug=True) + self.assertEqual(self._console_handler().level, logging.DEBUG) + self.assertEqual(self._file_handler().level, logging.DEBUG) + + def test_http_logs_filtered_from_console_only(self): + setup_logging(debug=True) + console = self._console_handler() + http_record = logging.LogRecord("httpx", logging.INFO, "httpx", 1, + "HTTP Request: GET ...", None, None) + self.assertFalse(console.filter(http_record)) + chunk_record = logging.LogRecord("converter.converter", logging.DEBUG, + "converter", 1, + "Chunk 1/1 request text", None, None) + self.assertTrue(console.filter(chunk_record)) + + +class SynthesizeChunkLoggingTests(unittest.TestCase): + """Chunk failures surface as a single ERROR record (no print echo).""" + + def setUp(self): + self.converter = AudiobookConverter.__new__(AudiobookConverter) + self.converter.client_chunks = True + self.converter.tts = MagicMock() + + def test_failed_chunk_logs_single_error(self): + self.converter.tts.process_chunk_with_retry.return_value = None + with self.assertLogs("converter.converter", level="ERROR") as logs: + results = self.converter._synthesize_chunks(["Hello."]) + self.assertEqual(results, {1: None}) + self.assertEqual(len(logs.output), 1) + self.assertIn("Chunk 1/1 failed", logs.output[0]) + + def test_raising_chunk_logs_single_error(self): + self.converter.tts.process_chunk_with_retry.side_effect = RuntimeError("boom") + with self.assertLogs("converter.converter", level="ERROR") as logs: + results = self.converter._synthesize_chunks(["Hello."]) + self.assertEqual(results, {1: None}) + self.assertEqual(len(logs.output), 1) + self.assertIn("Chunk 1/1 error: boom", logs.output[0]) + + +class ServerSideChunkingOutputTests(unittest.TestCase): + """With client-side chunking off (audiocpp default), the console skips + the chunk vocabulary because the whole request is one server call.""" + + def _converter(self, client_chunks: bool): + converter = AudiobookConverter.__new__(AudiobookConverter) + converter.client_chunks = client_chunks + converter.backend = tts.BACKEND_AUDIOCPP + converter.speed = 1.0 + converter.output_format = "mp3" + converter.tts = MagicMock() + converter.tts.process_chunk_with_retry.return_value = "chunk.wav" + return converter + + def test_client_chunking_prints_chunk_progress(self): + buf = io.StringIO() + with redirect_stdout(buf): + self._converter(client_chunks=True)._synthesize_chunks(["Hello."]) + out = buf.getvalue() + self.assertIn("PROCESSING 1 CHUNKS", out) + self.assertIn("Chunk 1/1 completed", out) + self.assertIn("Successful: 1/1", out) + + def test_server_side_chunking_suppresses_chunk_output(self): + buf = io.StringIO() + with redirect_stdout(buf): + self._converter(client_chunks=False)._synthesize_chunks(["Hello."]) + self.assertEqual(buf.getvalue(), "") + + def test_server_side_chunking_suppresses_chapter_chunk_suffix(self): + buf = io.StringIO() + with patch.object(converter_mod.audio, "combine_chunks", return_value=True), \ + redirect_stdout(buf): + ok = self._converter(client_chunks=False)._convert_text( + "Hello world.", Path("out.mp3"), time.time(), chapter=(2, 5)) + self.assertTrue(ok) + out = buf.getvalue() + self.assertIn("Chapter 2/5 converted", out) + self.assertNotIn("chunk", out.lower()) + + def test_single_request_run_notes_long_wait(self): + buf = io.StringIO() + with patch.object(converter_mod.audio, "combine_chunks", return_value=True), \ + redirect_stdout(buf): + ok = self._converter(client_chunks=False)._convert_text( + "Hello world.", Path("out.mp3"), time.time(), chapter=(2, 5)) + self.assertTrue(ok) + out = buf.getvalue() + self.assertIn("Sending the chapter 2/5 to the audio.cpp server as a " + "single request", out) + self.assertIn("expected for this to take a very long time", out) + + def test_client_chunking_run_keeps_chunk_phrasing(self): + buf = io.StringIO() + with patch.object(converter_mod.audio, "combine_chunks", return_value=True), \ + redirect_stdout(buf): + ok = self._converter(client_chunks=True)._convert_text( + "Hello world.", Path("out.mp3"), time.time(), chapter=(2, 5)) + self.assertTrue(ok) + out = buf.getvalue() + self.assertIn("Processing 1 chunks via audio.cpp server", out) + self.assertNotIn("single request", out) + self.assertIn("Chapter 2/5 converted (1/1 chunks)", out) + + def test_partial_chunks_abort_without_assembling(self): + converter = self._converter(client_chunks=True) + converter.tts.process_chunk_with_retry.side_effect = ["chunk_0001.wav", None] + text = " ".join(f"word{i}" for i in range(8)) + with patch.object(config, "CHUNK_SIZE", 5), \ + patch.object(converter_mod.audio, "combine_chunks") as mock_combine: + ok = converter._convert_text(text, Path("out.mp3"), time.time()) + self.assertFalse(ok) + mock_combine.assert_not_called() + + +class PromptOverwriteTests(unittest.TestCase): + def test_single_file_yes(self): + with patch("builtins.input", return_value="y"): + self.assertTrue(prompt_overwrite([Path("dune.mp3")], "dune")) + + def test_single_file_no(self): + with patch("builtins.input", return_value="n"): + self.assertFalse(prompt_overwrite([Path("dune.mp3")], "dune")) + + def test_accepts_full_words(self): + with patch("builtins.input", return_value="yes"): + self.assertTrue(prompt_overwrite([Path("dune.mp3")], "dune")) + with patch("builtins.input", return_value="No"): + self.assertFalse(prompt_overwrite([Path("dune.mp3")], "dune")) + + def test_invalid_answer_reasked(self): + with patch("builtins.input", side_effect=["maybe", "n"]) as mock_input: + self.assertFalse(prompt_overwrite([Path("dune.mp3")], "dune")) + self.assertEqual(mock_input.call_count, 2) + + def test_empty_answer_defaults_yes(self): + # Pressing Enter (empty input) accepts the default of yes, matching + # the make_audiocpp_server_json tool's ask_bool(default=True) prompt. + with patch("builtins.input", return_value=""): + self.assertTrue(prompt_overwrite([Path("dune.mp3")], "dune")) + + def test_eof_keeps_existing_output(self): + with patch("builtins.input", side_effect=EOFError): + self.assertFalse(prompt_overwrite([Path("dune.mp3")], "dune")) + + def test_multiple_files_prompt_names_them(self): + files = [Path("dune_01_Dune.mp3"), Path("dune_02_Barony.mp3")] + with patch("builtins.input", return_value="y") as mock_input: + self.assertTrue(prompt_overwrite(files, "dune")) + prompt_text = mock_input.call_args[0][0] + self.assertIn("2 output files for 'dune'", prompt_text) + self.assertIn("dune_01_Dune.mp3", prompt_text) + self.assertIn("overwrite them", prompt_text) + + +class PreflightOverwritesTests(unittest.TestCase): + """The pre-flight overwrite check runs without a TTS server connection.""" + + def setUp(self): + self._books_tmp = tempfile.TemporaryDirectory() + self._output_tmp = tempfile.TemporaryDirectory() + self._original_folders = (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("hello world", encoding="utf-8") + + def tearDown(self): + converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = self._original_folders + self._books_tmp.cleanup() + self._output_tmp.cleanup() + + def test_no_books_returns_empty(self): + (converter_mod.BOOKS_FOLDER / "book.txt").unlink() + with patch("builtins.input", side_effect=AssertionError("should not prompt")): + book_files, planned = AudiobookConverter.preflight_overwrites( + tts.BACKEND_QWEN, None, tts.VOICE_MODE_CUSTOM, None, "mp3") + self.assertEqual(book_files, []) + self.assertEqual(planned, []) + + def test_new_book_planned_without_prompt(self): + with patch("builtins.input", side_effect=AssertionError("should not prompt")): + book_files, planned = AudiobookConverter.preflight_overwrites( + tts.BACKEND_QWEN, None, tts.VOICE_MODE_CUSTOM, None, "mp3") + self.assertEqual(len(book_files), 1) + self.assertEqual(planned, [(book_files[0], "book_Vivian")]) + + def test_existing_output_enter_defaults_yes(self): + (converter_mod.AUDIOBOOKS_FOLDER / "book_Vivian.mp3").write_bytes(b"existing") + with patch("builtins.input", return_value=""): + book_files, planned = AudiobookConverter.preflight_overwrites( + tts.BACKEND_QWEN, None, tts.VOICE_MODE_CUSTOM, None, "mp3") + self.assertEqual(planned, [(book_files[0], "book_Vivian")]) + + def test_existing_output_declined_is_skipped(self): + (converter_mod.AUDIOBOOKS_FOLDER / "book_Vivian.mp3").write_bytes(b"existing") + with patch("builtins.input", return_value="n"): + book_files, planned = AudiobookConverter.preflight_overwrites( + tts.BACKEND_QWEN, None, tts.VOICE_MODE_CUSTOM, None, "mp3") + self.assertEqual(len(book_files), 1) + self.assertEqual(planned, []) + + +class RunOverwritePromptTests(unittest.TestCase): + """The full run() flow: prompts collected before any conversion starts.""" + + def setUp(self): + self._books_tmp = tempfile.TemporaryDirectory() + self._output_tmp = tempfile.TemporaryDirectory() + self._original_folders = (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("hello world", encoding="utf-8") + self.converter = AudiobookConverter.__new__(AudiobookConverter) + self.converter.voice_mode = tts.VOICE_MODE_CUSTOM + self.converter.voice_clone_ref_audio = None + self.converter.backend = tts.BACKEND_QWEN + self.converter.voice = None + self.converter.instructions = None + self.converter.speed = 1.0 + self.converter.single_file = False + self.converter.output_format = "mp3" + self.converter.language = "English" + self.converter.debug = False + self.converted = [] + self.converter.convert_book = ( + lambda file_path, output_name=None: + not self.converted.append((file_path.name, output_name)) or True) + + def tearDown(self): + converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = self._original_folders + self._books_tmp.cleanup() + self._output_tmp.cleanup() + + def test_declined_book_is_skipped(self): + (converter_mod.AUDIOBOOKS_FOLDER / "book_Vivian.mp3").write_bytes(b"existing") + with patch("builtins.input", return_value="n"): + self.assertTrue(self.converter.run()) + self.assertEqual(self.converted, []) + self.assertTrue((converter_mod.AUDIOBOOKS_FOLDER / "book_Vivian.mp3").exists()) + + def test_accepted_book_is_converted(self): + (converter_mod.AUDIOBOOKS_FOLDER / "book_Vivian.mp3").write_bytes(b"existing") + with patch("builtins.input", return_value="y"): + self.assertTrue(self.converter.run()) + self.assertEqual(self.converted, [("book.txt", "book_Vivian")]) + + def test_new_book_converted_without_prompt(self): + with patch("builtins.input", side_effect=AssertionError("should not prompt")): + self.assertTrue(self.converter.run()) + self.assertEqual(self.converted, [("book.txt", "book_Vivian")]) + + +if __name__ == "__main__": + unittest.main() |