"""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) 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 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 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()