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 threading
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
import logging_kit  # 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 AllModelsConvertTests(unittest.TestCase):
    """convert(model_ids=...): the "All (multiple generation)" loop.

    One AudiobookConverter per model, model-major, each forced to unload
    previously-loaded server models (clean VRAM between models); a failed
    book or a model that cannot start does not sink the remaining models;
    the progress events are renumbered into one global book sequence
    stamped with the generating model and one merged "done" is emitted.
    """

    def setUp(self):
        self.tmp = Path(tempfile.mkdtemp(prefix="audiobook_all_"))
        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, *, run_results=None, make_run=None, progress=None,
                 cancel=None, planned_by_model="default", instances=None,
                 **kwargs):
        """Run convert() with AudiobookConverter mocked per model.

        RUN_RESULTS gives each model's converter.run() return value in
        construction order; MAKE_RUN, when given, builds each instance's
        run() from the constructor kwargs (for event-emitting fakes).
        INSTANCES, when given, is a list the per-model instances are
        appended to. Returns (code, class mock, per-model ctor kwargs).
        """
        kwargs.setdefault("backend", "audiocpp")
        kwargs.setdefault("model_ids", ["m1", "m2"])
        kwargs.setdefault("model_voices", {"m1": "narrator", "m2": None})
        if planned_by_model == "default":
            kwargs.setdefault("planned_by_model", {
                "m1": [(self.book, "book_m1_narrator")],
                "m2": [(self.book, "book_m2_none")]})
        else:
            kwargs["planned_by_model"] = planned_by_model
        kwargs.setdefault("book_files", [self.book])
        if progress is not None:
            kwargs["progress"] = progress
        if cancel is not None:
            kwargs["cancel"] = cancel
        states = list(run_results or [])
        ctor_kwargs = []
        made = instances if instances is not None else []

        def make_instance(*args, **ckwargs):
            ckwargs = dict(ckwargs)
            ctor_kwargs.append(ckwargs)
            inst = MagicMock()
            made.append(inst)
            result = True
            if make_run is not None:
                inst.run.side_effect = make_run(ckwargs)
            else:
                result = states[len(made) - 1] \
                    if len(made) <= len(states) else True
                inst.run.return_value = result
            # Per-book outcomes the All-run loop counts on the CLI path
            # (see audiobook._convert_each_model).
            inst.results = {"book.txt": bool(result)}
            return inst

        fake_class = MagicMock(side_effect=make_instance)
        with patch.object(audiobook, "setup_logging"), \
                patch.object(audiobook, "setup_directories"), \
                patch.object(audiobook, "AudiobookConverter", fake_class):
            code = audiobook.convert(**kwargs)
        return code, fake_class, ctor_kwargs

    def test_one_converter_per_model_with_its_own_voice(self):
        # Model-major: every model runs its planned books before the next
        # model starts, each with the voice the form adapted for it and a
        # forced pre-run model unload (clean VRAM between models).
        code, fake_class, ctors = self._convert()
        self.assertEqual(code, 0)
        self.assertEqual(fake_class.call_count, 2)
        self.assertEqual(ctors[0]["model_id"], "m1")
        self.assertEqual(ctors[0]["voice"], "narrator")
        self.assertTrue(ctors[0]["unload_models"])
        self.assertEqual(ctors[1]["model_id"], "m2")
        self.assertIsNone(ctors[1]["voice"])
        self.assertTrue(ctors[1]["unload_models"])

    def test_planned_entries_reach_each_converter(self):
        instances = []
        self._convert(instances=instances)
        self.assertEqual(instances[0]._planned,
                         [(self.book, "book_m1_narrator")])
        self.assertEqual(instances[1]._planned,
                         [(self.book, "book_m2_none")])

    def test_failed_model_does_not_sink_the_next(self):
        # m1's conversion fails: the loop still constructs and runs m2,
        # and the overall run reports failure (not every book succeeded).
        code, fake_class, _ = self._convert(run_results=[False, True])
        self.assertEqual(code, 1)
        self.assertEqual(fake_class.call_count, 2)

    def test_model_that_cannot_start_is_skipped(self):
        # A connect-time failure (constructor raise) is reported and the
        # remaining models still run.
        started = []

        def make_instance(*args, **ckwargs):
            started.append(ckwargs["model_id"])
            if ckwargs["model_id"] == "m1":
                raise RuntimeError("voice 'x' is not available")
            inst = MagicMock()
            inst.run.return_value = True
            return inst

        fake_class = MagicMock(side_effect=make_instance)
        with patch.object(audiobook, "setup_logging"), \
                patch.object(audiobook, "setup_directories"), \
                patch.object(audiobook, "AudiobookConverter", fake_class), \
                patch.object(logging_kit, "log_traceback"):
            code = audiobook.convert(
                backend="audiocpp", model_ids=["m1", "m2"],
                model_voices={"m1": "narrator", "m2": None},
                planned_by_model={"m1": [(self.book, "book_m1_narrator")],
                                  "m2": [(self.book, "book_m2_none")]},
                book_files=[self.book])
        self.assertEqual(started, ["m1", "m2"])
        self.assertEqual(code, 1)

    def test_events_are_renumbered_and_stamped_with_the_model(self):
        # Book events carry one global index across all models; done/cancel
        # events from the per-model converters are swallowed and one merged
        # "done" is emitted at the end.
        events = []

        def make_run(ckwargs):
            emit = ckwargs["progress"]

            def run():
                emit({"kind": "book", "index": 1, "total": 1,
                      "name": "book.txt"})
                emit({"kind": "chunks", "total": 3})
                emit({"kind": "chunk_done", "chunk": 1, "total": 3})
                emit({"kind": "book_done", "name": "book.txt", "ok": True,
                      "files": [f"book_{ckwargs['model_id']}.mp3"]})
                emit({"kind": "done", "ok": 1, "total": 1})
                return True
            return run

        code, _, _ = self._convert(progress=events.append,
                                   make_run=make_run)
        self.assertEqual(code, 0)
        kinds = [e["kind"] for e in events]
        self.assertEqual(kinds, ["book", "chunks", "chunk_done", "book_done",
                                 "book", "chunks", "chunk_done", "book_done",
                                 "done"])
        self.assertEqual(events[0],
                         {"kind": "book", "index": 1, "total": 2,
                          "name": "book.txt", "model": "m1"})
        self.assertEqual(events[4],
                         {"kind": "book", "index": 2, "total": 2,
                          "name": "book.txt", "model": "m2"})
        # Passthrough events are stamped with the model too, so the run
        # view can attribute a chunk failure to its model.
        self.assertEqual(events[1],
                         {"kind": "chunks", "total": 3, "model": "m1"})
        self.assertEqual(events[2],
                         {"kind": "chunk_done", "chunk": 1, "total": 3,
                          "model": "m1"})
        self.assertEqual(events[3]["model"], "m1")
        self.assertEqual(events[5]["model"], "m2")
        self.assertEqual(events[7]["model"], "m2")
        self.assertEqual(events[8],
                         {"kind": "done", "ok": 2, "total": 2,
                          "cancelled": False})

    def test_cancellation_stops_the_remaining_models(self):
        # m1's run sets the cancel event: the loop stops before m2 and the
        # merged done reports the cancellation.
        cancel = threading.Event()
        events = []

        def make_run(ckwargs):
            emit = ckwargs["progress"]

            def run():
                emit({"kind": "book", "index": 1, "total": 1,
                      "name": "book.txt"})
                emit({"kind": "book_done", "name": "book.txt", "ok": True,
                      "files": ["book_m1_narrator.mp3"]})
                cancel.set()
                return False
            return run

        code, fake_class, _ = self._convert(progress=events.append,
                                            make_run=make_run, cancel=cancel)
        self.assertEqual(fake_class.call_count, 1)
        self.assertEqual(code, 1)
        self.assertEqual(events[-1],
                         {"kind": "done", "ok": 1, "total": 2,
                          "cancelled": True})

    def test_plans_are_computed_when_not_provided(self):
        # Without planned_by_model (a scripted call) each model plans its
        # own model-tagged outputs, with its own voice for the narrator
        # tag the overwrite questions are asked about.
        preflight = MagicMock(return_value=([self.book],
                                            [(self.book, "dune")]))
        fake_class = MagicMock()
        fake_class.preflight_overwrites = preflight
        # The converter class is mocked wholesale; the name-tag helper is
        # a pure static method, so stand in the real behavior.
        fake_class.compute_model_tag = staticmethod(
            AudiobookConverter.compute_model_tag)

        def make_instance(*args, **ckwargs):
            inst = MagicMock()
            inst.run.return_value = True
            inst.results = {"book.txt": True}
            return inst
        fake_class.side_effect = make_instance
        with patch.object(audiobook, "setup_logging"), \
                patch.object(audiobook, "setup_directories"), \
                patch.object(audiobook, "AudiobookConverter", fake_class):
            code = audiobook.convert(
                backend="audiocpp", model_ids=["m1", "m2"],
                model_voices={"m1": "narrator", "m2": None},
                book_files=None)
        self.assertEqual(code, 0)
        self.assertEqual(preflight.call_count, 2)
        first, second = preflight.call_args_list
        self.assertEqual(first.kwargs["voice"], "narrator")
        self.assertEqual(first.kwargs["name_tag"], "m1")
        self.assertEqual(second.kwargs["voice"], None)
        self.assertEqual(second.kwargs["name_tag"], "m2")

    def test_nothing_planned_is_a_clean_noop(self):
        code, fake_class, _ = self._convert(
            planned_by_model={"m1": [], "m2": []})
        self.assertEqual(code, 0)
        fake_class.assert_not_called()


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 _log_path(self) -> str:
        """The dated log file's full path, as the console reports it."""
        return str(logging_kit.stream_path("audiobook", self.log_dir))

    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.assertIn(f"[INFO] Full details in the log file: "
                      f"{self._log_path()}", 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, out, _ = self._convert(ValueError("bad input"),
                                     progress=events.append)
        self.assertEqual(code, 1)
        self.assertEqual(events,
                         [{"kind": "error", "message": "bad input"}])
        # The TUI run view owns the console and points failures at the log
        # itself; no console pointer on this path.
        self.assertNotIn("Full details in the log file", out)

    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())
        self.assertIn(f"[INFO] Full details in the log file: "
                      f"{self._log_path()}", out)

    def test_failed_run_prints_the_log_path(self):
        # A run that fails without raising (a book aborted the rest) ends
        # with the log file path too.
        preflight = MagicMock(
            return_value=([self.book], [(self.book, "dune")]))
        fake_class = MagicMock()
        fake_class.preflight_overwrites = preflight
        fake_class.return_value.run.return_value = False
        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")
        self.assertEqual(code, 1)
        self.assertIn(f"[INFO] Full details in the log file: "
                      f"{self._log_path()}", out.getvalue())


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