path: root/app/tests/test_audiobook_cli.py

"""Tests for the audiobook.py CLI — single-book flags, arg validation, and
the managed-server wiring.

audiobook.py lives at the repo root (one level above app/), so the tests
bootstrap the root onto sys.path to import it. main() runs with the
managed-environment bootstrap stubbed (it would otherwise re-exec the
process into envs/tts) and convert() mocked, asserting only argparse
behavior and what reaches convert(); convert()'s single-book wiring, the
pre-flight overrides, and the manage_server lifecycle are tested against
the real functions with temporary directories.
"""

import contextlib
import io
import logging
import shutil
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch

# audiobook.py sits at the repo root, two levels above this test module.
REPO_ROOT = Path(__file__).resolve().parents[2]
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))

import audiobook  # noqa: E402
from converter import config  # noqa: E402
from converter import converter as converter_mod  # noqa: E402
from converter.converter import AudiobookConverter  # noqa: E402


def _make_book(tmp: Path, name: str = "dune.txt") -> Path:
    book = tmp / name
    book.write_text("A beginning is a very delicate time.", encoding="utf-8")
    return book


class MainTestCase(unittest.TestCase):
    """Base: run audiobook.main() with argv, bootstrap stubbed, convert mocked."""

    def setUp(self):
        self.tmp = Path(tempfile.mkdtemp(prefix="audiobook_cli_"))
        self.addCleanup(shutil.rmtree, self.tmp, True)

    def run_main(self, argv, backend="audiocpp"):
        """Run main() with the given argv; returns (code, stderr, convert mock).

        The envs bootstrap (which re-execs into the managed venv via
        os.execv when active) and convert() are stubbed, so no TTS work
        happens and the process survives. BACKEND (a --backend value, or
        None to omit the required flag) is prepended unless the argv
        already carries --backend.
        """
        if backend is not None and "--backend" not in argv:
            argv = ["--backend", backend, *argv]
        err = io.StringIO()
        convert = MagicMock(return_value=0)
        with patch.object(sys, "argv", ["audiobook.py", *argv]), \
                contextlib.redirect_stderr(err), \
                patch.object(audiobook._envs, "bootstrap"), \
                patch.object(audiobook, "convert", convert):
            try:
                audiobook.main()
                code = None
            except SystemExit as exc:
                code = exc.code
        return code, err.getvalue(), convert


class MainBackendTests(MainTestCase):
    """The backend/model/voice are per-run choices with no config defaults."""

    def test_backend_is_required(self):
        code, err, convert = self.run_main(["--debug"], backend=None)
        self.assertEqual(code, 2)
        self.assertIn("--backend", err)
        self.assertIn("required", err)
        convert.assert_not_called()

    def test_backend_reaches_convert(self):
        code, _, convert = self.run_main([])
        self.assertEqual(code, 0)
        self.assertEqual(convert.call_args.kwargs["backend"], "audiocpp")

    def test_qwen_accepts_a_builtin_speaker_voice(self):
        code, _, convert = self.run_main(
            ["--backend", "qwen", "--voice", "Vivian"], backend=None)
        self.assertEqual(code, 0)
        self.assertEqual(convert.call_args.kwargs["backend"], "qwen")
        self.assertEqual(convert.call_args.kwargs["voice"], "Vivian")

    def test_qwen_rejects_a_non_speaker_voice(self):
        code, err, convert = self.run_main(
            ["--backend", "qwen", "--voice", "narrator"], backend=None)
        self.assertEqual(code, 2)
        self.assertIn("not a built-in speaker", err)
        convert.assert_not_called()

    def test_qwen_requires_a_voice_without_clone_or_instructions(self):
        code, err, convert = self.run_main(["--backend", "qwen"], backend=None)
        self.assertEqual(code, 2)
        self.assertIn("--backend qwen needs a voice", err)
        convert.assert_not_called()

    def test_qwen_clone_run_needs_no_voice(self):
        code, _, convert = self.run_main(
            ["--backend", "qwen", "--clone", "ref.wav"], backend=None)
        self.assertEqual(code, 0)
        convert.assert_called_once()

    def test_faster_requires_a_voice(self):
        code, err, convert = self.run_main(["--backend", "faster"],
                                           backend=None)
        self.assertEqual(code, 2)
        self.assertIn("--backend faster requires --voice", err)
        convert.assert_not_called()

    def test_noninteractive_no_args_stops_with_guidance(self):
        # No args in a non-interactive session cannot guess a backend:
        # point the user at --backend / the TUI instead of converting.
        out, err = io.StringIO(), io.StringIO()
        with patch.object(sys, "argv", ["audiobook.py"]), \
                patch.object(sys, "stdin", io.StringIO()), \
                patch.object(sys, "stdout", io.StringIO()), \
                contextlib.redirect_stdout(out), \
                contextlib.redirect_stderr(err), \
                patch.object(audiobook._envs, "bootstrap"):
            with self.assertRaises(SystemExit) as ctx:
                audiobook.main()
        self.assertEqual(ctx.exception.code, 2)
        self.assertIn("No --backend given", out.getvalue())
        self.assertIn("No --backend given", err.getvalue() + out.getvalue())


class MainFlagConflictTests(MainTestCase):
    """Mixing the directory and single-book flag pairs stops with an error."""

    def test_input_and_input_file_conflict(self):
        code, err, convert = self.run_main(
            ["--input", str(self.tmp), "--input-file", str(self.tmp / "dune.txt")])
        self.assertEqual(code, 2)
        self.assertIn("--input", err)
        self.assertIn("--input-file", err)
        self.assertIn("cannot be used together", err)
        convert.assert_not_called()

    def test_output_and_output_file_conflict(self):
        code, err, convert = self.run_main(
            ["--output", str(self.tmp / "out"),
             "--output-file", str(self.tmp / "out" / "dune.mp3")])
        self.assertEqual(code, 2)
        self.assertIn("--output", err)
        self.assertIn("--output-file", err)
        self.assertIn("cannot be used together", err)
        convert.assert_not_called()

    def test_output_file_requires_input_file(self):
        code, err, convert = self.run_main(
            ["--output-file", str(self.tmp / "dune.mp3")])
        self.assertEqual(code, 2)
        self.assertIn("--output-file", err)
        self.assertIn("--input-file", err)
        convert.assert_not_called()

    def test_conflict_wins_over_bad_directory(self):
        # The flag explanation fires even when --input is also invalid.
        code, err, convert = self.run_main(
            ["--input", str(self.tmp / "nope"),
             "--input-file", str(self.tmp / "dune.txt")])
        self.assertEqual(code, 2)
        self.assertIn("cannot be used together", err)
        convert.assert_not_called()


class MainPathValidationTests(MainTestCase):
    """--input-file/--output-file values are validated before converting."""

    def test_missing_input_file(self):
        code, err, convert = self.run_main(
            ["--input-file", str(self.tmp / "nope.txt")])
        self.assertEqual(code, 2)
        self.assertIn("no such book file", err)
        convert.assert_not_called()

    def test_unsupported_input_file_format(self):
        book = _make_book(self.tmp, "dune.docx")
        code, err, convert = self.run_main(["--input-file", str(book)])
        self.assertEqual(code, 2)
        self.assertIn("unsupported book format", err)
        self.assertIn(".docx", err)
        convert.assert_not_called()

    def test_output_file_extension_mismatch_stops_the_run(self):
        book = _make_book(self.tmp)
        code, err, convert = self.run_main(
            ["--input-file", str(book),
             "--output-file", str(self.tmp / "dune.mp3"),
             "--format", "m4b"])
        self.assertEqual(code, 2)
        self.assertIn("does not match the output format", err)
        self.assertIn("--format mp3", err)
        convert.assert_not_called()

    def test_output_file_unsupported_extension(self):
        book = _make_book(self.tmp)
        code, err, convert = self.run_main(
            ["--input-file", str(book),
             "--output-file", str(self.tmp / "dune.xyz")])
        self.assertEqual(code, 2)
        self.assertIn("unsupported extension", err)
        convert.assert_not_called()


class MainHappyPathTests(MainTestCase):
    """Valid single-book flags reach convert() resolved and typed."""

    def test_input_and_output_file_forwarded(self):
        book = _make_book(self.tmp)
        target = self.tmp / "out" / "dune.mp3"
        code, _, convert = self.run_main(
            ["--input-file", str(book), "--output-file", str(target),
             "--format", "mp3"])
        self.assertEqual(code, 0)
        convert.assert_called_once()
        kwargs = convert.call_args.kwargs
        self.assertEqual(kwargs["input_file"], book)
        self.assertEqual(kwargs["output_file"], target)
        self.assertEqual(kwargs["output_format"], "mp3")
        self.assertIsNone(kwargs["input_dir"])
        self.assertIsNone(kwargs["output_dir"])

    def test_input_file_alone_keeps_output_defaults(self):
        book = _make_book(self.tmp)
        code, _, convert = self.run_main(["--input-file", str(book)])
        self.assertEqual(code, 0)
        kwargs = convert.call_args.kwargs
        self.assertEqual(kwargs["input_file"], book)
        self.assertIsNone(kwargs["output_file"])
        self.assertEqual(kwargs["output_format"], config.AUDIO_FORMAT)

    def test_directory_flags_still_forwarded(self):
        out = self.tmp / "out"
        code, _, convert = self.run_main(
            ["--input", str(self.tmp), "--output", str(out)])
        self.assertEqual(code, 0)
        kwargs = convert.call_args.kwargs
        self.assertEqual(kwargs["input_dir"], self.tmp)
        self.assertEqual(kwargs["output_dir"], out)
        self.assertIsNone(kwargs["input_file"])
        self.assertIsNone(kwargs["output_file"])

    def test_manage_server_requested_without_api_url(self):
        # Without --api-url the CLI asks convert() to manage the server
        # lifecycle (convert() performs the actual boot/stop).
        code, _, convert = self.run_main([])
        self.assertEqual(code, 0)
        self.assertIs(convert.call_args.kwargs["manage_server"], True)

    def test_api_url_run_still_carries_the_manage_flag(self):
        # The flag travels too; convert() itself skips management when an
        # explicit api_url targets an external server.
        code, _, convert = self.run_main(
            ["--api-url", "10.20.30.40:8080"])
        self.assertEqual(code, 0)
        self.assertEqual(convert.call_args.kwargs["api_url"],
                         "http://10.20.30.40:8080")
        self.assertIs(convert.call_args.kwargs["manage_server"], True)


class ConvertWiringTests(unittest.TestCase):
    """convert() turns the single-book flags into the pre-flight overrides."""

    def setUp(self):
        self.tmp = Path(tempfile.mkdtemp(prefix="audiobook_wiring_"))
        self.addCleanup(shutil.rmtree, self.tmp, True)
        self.book = _make_book(self.tmp)
        # convert() repoints the converter module's folder globals; restore
        # them so other tests keep seeing the configured folders.
        self._old_folders = (converter_mod.BOOKS_FOLDER,
                             converter_mod.AUDIOBOOKS_FOLDER)
        self.addCleanup(self._restore_folders)

    def _restore_folders(self):
        converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = \
            self._old_folders

    def _convert(self, **kwargs):
        kwargs.setdefault("backend", "audiocpp")
        preflight = MagicMock(
            return_value=([self.book], [(self.book, "dune")]))
        fake_instance = MagicMock()
        fake_instance.run.return_value = True
        fake_class = MagicMock(return_value=fake_instance)
        fake_class.preflight_overwrites = preflight
        with patch.object(audiobook, "setup_logging"), \
                patch.object(audiobook, "setup_directories"), \
                patch.object(audiobook, "AudiobookConverter", fake_class):
            code = audiobook.convert(**kwargs)
        return code, preflight, fake_class, fake_instance

    def test_output_file_redirects_folder_and_names_the_book(self):
        out = self.tmp / "out"
        code, preflight, _, fake_instance = self._convert(
            input_file=self.book, output_file=out / "dune.mp3",
            output_format="mp3")
        self.assertEqual(code, 0)
        self.assertEqual(converter_mod.AUDIOBOOKS_FOLDER, out)
        self.assertEqual(preflight.call_args.kwargs["book_files"], [self.book])
        self.assertEqual(preflight.call_args.kwargs["output_name"], "dune")
        self.assertEqual(fake_instance._book_files, [self.book])
        self.assertEqual(fake_instance._planned, [(self.book, "dune")])

    def test_output_file_without_extension_uses_stem(self):
        out = self.tmp / "out"
        _, preflight, _, _ = self._convert(
            input_file=self.book, output_file=out / "dune")
        self.assertEqual(preflight.call_args.kwargs["output_name"], "dune")

    def test_input_file_alone_keeps_output_folder_and_tagged_name(self):
        _, preflight, _, _ = self._convert(input_file=self.book)
        self.assertEqual(preflight.call_args.kwargs["book_files"],
                         [self.book])
        self.assertIsNone(preflight.call_args.kwargs["output_name"])
        self.assertEqual(
            converter_mod.AUDIOBOOKS_FOLDER,
            converter_mod.resolve_dir(config.OUTPUT_DIR, "output"))

    def test_output_file_requires_input_file(self):
        with self.assertRaises(ValueError):
            self._convert(output_file=self.tmp / "dune.mp3")


class ManagedServerWiringTests(unittest.TestCase):
    """convert(manage_server=True) boots and stops the server around the run.

    The lifecycle decisions live in backends.managed (tested there); these
    pin convert()'s wiring: when the boot happens relative to pre-flight
    and the conversion, that shutdown runs even on failure or Ctrl-C, and
    that an external api_url or the hub's progress-callback path never
    touch the server.
    """

    def setUp(self):
        self.tmp = Path(tempfile.mkdtemp(prefix="audiobook_managed_"))
        self.addCleanup(shutil.rmtree, self.tmp, True)
        self.book = _make_book(self.tmp)
        self._old_folders = (converter_mod.BOOKS_FOLDER,
                             converter_mod.AUDIOBOOKS_FOLDER)
        self.addCleanup(self._restore_folders)

    def _restore_folders(self):
        converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = \
            self._old_folders

    def _convert(self, *, server_ok=True, run_result=True, run_raises=None,
                 preflight_result=None, **kwargs):
        """Run convert() with the managed-server and converter mocked.

        Returns (code, ensure mock, server mock, events, preflight mock);
        EVENTS records the order of ensure_running / run / shutdown.
        """
        kwargs.setdefault("backend", "audiocpp")
        kwargs.setdefault("manage_server", True)
        if preflight_result is None:
            preflight_result = ([self.book], [(self.book, "dune")])
        preflight = MagicMock(return_value=preflight_result)
        fake_instance = MagicMock()
        events = []

        def _run():
            events.append("run")
            if run_raises is not None:
                raise run_raises
            return run_result
        fake_instance.run.side_effect = _run
        fake_class = MagicMock(return_value=fake_instance)
        fake_class.preflight_overwrites = preflight
        server = MagicMock()
        server.ok = server_ok
        server.shutdown.side_effect = lambda: events.append("shutdown")
        ensure = MagicMock(return_value=server)

        def _ensure(backend, voice_mode):
            events.append(("ensure", backend, voice_mode))
            return server
        ensure.side_effect = _ensure
        with patch.object(audiobook, "setup_logging"), \
                patch.object(audiobook, "setup_directories"), \
                patch.object(audiobook, "AudiobookConverter", fake_class), \
                patch("backends.managed.ensure_running", ensure):
            code = audiobook.convert(**kwargs)
        return code, ensure, server, events, preflight

    def test_server_boots_before_the_run_and_stops_after(self):
        code, ensure, _, events, _ = self._convert()
        self.assertEqual(code, 0)
        self.assertEqual(events, [("ensure", "audiocpp", "custom_voice"),
                                  "run", "shutdown"])

    def test_qwen_run_needs_its_voice_mode_model(self):
        _, ensure, _, events, _ = self._convert(
            backend="qwen", clone="ref.wav")
        self.assertEqual(events[0], ("ensure", "qwen", "voice_clone"))

    def test_not_ok_boot_stops_before_converting(self):
        code, ensure, server, events, _ = self._convert(server_ok=False)
        self.assertEqual(code, 1)
        self.assertEqual(events, [("ensure", "audiocpp", "custom_voice"),
                                  "shutdown"])

    def test_shutdown_runs_when_the_conversion_fails(self):
        code, _, _, events, _ = self._convert(
            run_raises=RuntimeError("server unreachable"))
        self.assertEqual(code, 1)
        self.assertEqual(events, [("ensure", "audiocpp", "custom_voice"),
                                  "run", "shutdown"])

    def test_shutdown_runs_on_ctrl_c(self):
        code, _, _, events, _ = self._convert(run_raises=KeyboardInterrupt)
        self.assertEqual(code, 130)
        self.assertEqual(events, [("ensure", "audiocpp", "custom_voice"),
                                  "run", "shutdown"])

    def test_no_management_for_an_explicit_api_url(self):
        code, ensure, _, _, _ = self._convert(
            api_url="http://10.20.30.40:8080")
        self.assertEqual(code, 0)
        ensure.assert_not_called()

    def test_no_management_for_the_hub_path(self):
        # The run view boots/stops the server itself: manage_server False.
        _, ensure, _, _, _ = self._convert(manage_server=False)
        ensure.assert_not_called()

    def test_no_server_boot_when_nothing_to_convert(self):
        code, ensure, _, _, preflight = self._convert(
            preflight_result=([], []))
        self.assertEqual(code, 0)
        ensure.assert_not_called()


class PreflightOverrideTests(unittest.TestCase):
    """preflight_overwrites honors the explicit book list and output name."""

    def setUp(self):
        self.tmp = Path(tempfile.mkdtemp(prefix="audiobook_preflight_"))
        self.addCleanup(shutil.rmtree, self.tmp, True)
        self.book = _make_book(self.tmp)
        # The overwrite check globs AUDIOBOOKS_FOLDER; point it at the
        # temporary folder so the repo's real output dir stays untouched.
        patcher = patch.object(converter_mod, "AUDIOBOOKS_FOLDER", self.tmp)
        patcher.start()
        self.addCleanup(patcher.stop)

    def _preflight(self, **kwargs):
        options = dict(backend="audiocpp", voice="Vivian",
                       voice_mode="custom", voice_clone_ref_audio=None,
                       output_format="mp3")
        options.update(kwargs)
        return AudiobookConverter.preflight_overwrites(**options)

    def test_explicit_book_and_output_name_used_verbatim(self):
        book_files, planned = self._preflight(book_files=[self.book],
                                              output_name="dune")
        self.assertEqual(book_files, [self.book])
        self.assertEqual(planned, [(self.book, "dune")])

    def test_explicit_output_name_skips_narrator_tag(self):
        _, planned = self._preflight(book_files=[self.book],
                                     output_name="dune")
        # A directory scan would append the narrator tag (dune_Vivian).
        self.assertNotIn("Vivian", planned[0][1])

    def test_unsupported_books_filtered_from_explicit_list(self):
        stray = self.tmp / "notes.docx"
        stray.write_text("nope", encoding="utf-8")
        book_files, planned = self._preflight(
            book_files=[self.book, stray], output_name="dune")
        self.assertEqual(book_files, [self.book])
        self.assertEqual(planned, [(self.book, "dune")])

    def test_declined_overwrite_yields_empty_planned(self):
        existing = self.tmp / "dune.mp3"
        existing.write_bytes(b"prior audio")
        with patch.object(converter_mod, "prompt_overwrite",
                          return_value=False):
            book_files, planned = self._preflight(book_files=[self.book],
                                                  output_name="dune")
        self.assertEqual(book_files, [self.book])
        self.assertEqual(planned, [])

    def test_accepted_overwrite_plans_the_book(self):
        existing = self.tmp / "dune.mp3"
        existing.write_bytes(b"prior audio")
        with patch.object(converter_mod, "prompt_overwrite",
                          return_value=True):
            book_files, planned = self._preflight(book_files=[self.book],
                                                  output_name="dune")
        self.assertEqual(planned, [(self.book, "dune")])

    def test_empty_explicit_list_nothing_to_convert(self):
        book_files, planned = self._preflight(book_files=[],
                                              output_name="dune")
        self.assertEqual((book_files, planned), ([], []))


class FatalErrorReportingTests(unittest.TestCase):
    """convert() reports failures once; tracebacks stay in the log file.

    A RuntimeError/ValueError is an expected, user-facing failure (the
    clients raise them with actionable hints): the console gets a single
    [FATAL] line and the full traceback only lands in the dated log file.
    Any other exception is an actual crash, so its traceback is shown too.
    """

    def setUp(self):
        self.tmp = Path(tempfile.mkdtemp(prefix="audiobook_fatal_"))
        self.addCleanup(shutil.rmtree, self.tmp, True)
        self.book = _make_book(self.tmp)
        self.log_dir = self.tmp / "logs"
        # setup_logging opens the dated log file from this global.
        patcher = patch.object(converter_mod, "LOGS_FOLDER", self.log_dir)
        patcher.start()
        self.addCleanup(patcher.stop)
        self.addCleanup(self._reset_logging)
        # convert() repoints the converter module's folder globals; restore
        # them so other tests keep seeing the configured folders.
        self._old_folders = (converter_mod.BOOKS_FOLDER,
                             converter_mod.AUDIOBOOKS_FOLDER)
        self.addCleanup(self._restore_folders)

    def _reset_logging(self):
        root = logging.getLogger()
        for handler in list(root.handlers):
            root.removeHandler(handler)
            handler.close()
        logging.getLogger("converter").setLevel(logging.INFO)

    def _restore_folders(self):
        converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = \
            self._old_folders

    def _convert(self, exc, progress=None):
        """Run convert() with the converter class raising EXC in __init__.

        Returns (exit code, stdout, stderr); setup_logging runs for real
        with LOGS_FOLDER pointed at the temporary directory, so the dated
        log file's contents can be asserted on.
        """
        preflight = MagicMock(
            return_value=([self.book], [(self.book, "dune")]))
        fake_class = MagicMock(side_effect=exc)
        fake_class.preflight_overwrites = preflight
        out, err = io.StringIO(), io.StringIO()
        with patch.object(audiobook, "setup_directories"), \
                contextlib.redirect_stdout(out), \
                contextlib.redirect_stderr(err), \
                patch.object(audiobook, "AudiobookConverter", fake_class):
            code = audiobook.convert(backend="audiocpp", progress=progress)
        return code, out.getvalue(), err.getvalue()

    def _log_text(self) -> str:
        (log_path,) = self.log_dir.glob("audiobook_*.log")
        return log_path.read_text(encoding="utf-8")

    def test_expected_failure_shows_one_friendly_line(self):
        message = ("The audio.cpp model 'Qwen3-TTS-12Hz-1.7B-Base-GGUF' "
                   "(family 'qwen3_tts') has no built-in speakers "
                   "(see README).")
        code, out, err = self._convert(RuntimeError(message))
        self.assertEqual(code, 1)
        self.assertEqual(out.count("[FATAL]"), 1)
        self.assertIn(f"[FATAL] Fatal error: {message}", out)
        self.assertNotIn("Traceback (most recent call last)", out)
        self.assertNotIn("Traceback (most recent call last)", err)
        log_text = self._log_text()
        self.assertIn("Traceback (most recent call last):", log_text)
        self.assertIn(f"RuntimeError: {message}", log_text)

    def test_expected_failure_reports_error_event(self):
        events = []
        code, _, _ = self._convert(ValueError("bad input"),
                                   progress=events.append)
        self.assertEqual(code, 1)
        self.assertEqual(events,
                         [{"kind": "error", "message": "bad input"}])

    def test_unexpected_crash_also_shows_traceback(self):
        code, out, err = self._convert(TypeError("boom"))
        self.assertEqual(code, 1)
        self.assertEqual(out.count("[FATAL]"), 1)
        self.assertIn("Traceback (most recent call last)", err)
        self.assertIn("TypeError: boom", err)
        self.assertIn("Traceback (most recent call last)", self._log_text())


if __name__ == "__main__":
    unittest.main()