path: root/tests/test_converter.py

"""Tests for the audiobook converter orchestration helpers."""

import tempfile
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch

from converter import config
from converter.converter import AudiobookConverter, find_existing_outputs, prompt_overwrite


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 = config.AUDIOBOOKS_FOLDER
        config.AUDIOBOOKS_FOLDER = self.folder

    def tearDown(self):
        config.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(config.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(config.VOICE_MODE_CUSTOM)._narrator_tag(),
                             "Uncle_Fu")

    def test_clone_uses_reference_audio_stem(self):
        self.assertEqual(self._converter(config.VOICE_MODE_CLONE, "/x/ref.wav")._narrator_tag(),
                         "ref")

    def test_clone_stem_spaces_become_underscores(self):
        self.assertEqual(self._converter(config.VOICE_MODE_CLONE, "/x/my voice.wav")._narrator_tag(),
                         "my_voice")

    def test_invalid_characters_sanitized(self):
        self.assertEqual(self._converter(config.VOICE_MODE_CLONE, "/x/bad:name?.wav")._narrator_tag(),
                         "bad_name")

    def test_empty_after_sanitize_falls_back(self):
        self.assertEqual(self._converter(config.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(config, "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 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 = (config.BOOKS_FOLDER, config.AUDIOBOOKS_FOLDER)
        config.BOOKS_FOLDER = Path(self._books_tmp.name)
        config.AUDIOBOOKS_FOLDER = Path(self._output_tmp.name)
        (config.BOOKS_FOLDER / "book.txt").write_text("hello world", encoding="utf-8")
        self.converter = AudiobookConverter.__new__(AudiobookConverter)
        self.converter.voice_mode = config.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):
        config.BOOKS_FOLDER, config.AUDIOBOOKS_FOLDER = self._original_folders
        self._books_tmp.cleanup()
        self._output_tmp.cleanup()

    def test_declined_book_is_skipped(self):
        (config.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((config.AUDIOBOOKS_FOLDER / "book_Vivian.mp3").exists())

    def test_accepted_book_is_converted(self):
        (config.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()