"""Tests for the audiobook converter orchestration helpers.""" import logging import tempfile import unittest 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_language_defaults_to_config(self): with patch("converter.converter.QwenTTSClient") as mock_tts: AudiobookConverter() self.assertEqual(mock_tts.call_args.kwargs["language"], "English") def test_output_format_defaults_to_config(self): with patch("converter.converter.QwenTTSClient"): converter = AudiobookConverter() self.assertEqual(converter.output_format, config.AUDIO_FORMAT) self.assertEqual(config.AUDIO_FORMAT, "m4b") def test_language_normalized_before_tts_client(self): with patch("converter.converter.QwenTTSClient") as mock_tts: converter = AudiobookConverter(language="ja") 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): converter = AudiobookConverter.__new__(AudiobookConverter) converter.voice_mode = voice_mode converter.voice_clone_ref_audio = ref_audio converter.faster = False converter.faster_voice = None 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, "CUSTOM_VOICE_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") 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.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_debug_flag_wiring(self): with patch("converter.converter.QwenTTSClient"): self.assertFalse(AudiobookConverter().debug) self.assertTrue(AudiobookConverter(debug=True).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.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 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, 3) 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 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.faster = False self.converter.faster_voice = 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()