diff options
Diffstat (limited to 'app/tests/test_audiobook_cli.py')
| -rw-r--r-- | app/tests/test_audiobook_cli.py | 135 |
1 files changed, 131 insertions, 4 deletions
diff --git a/app/tests/test_audiobook_cli.py b/app/tests/test_audiobook_cli.py index d947c5d..f13cf9e 100644 --- a/app/tests/test_audiobook_cli.py +++ b/app/tests/test_audiobook_cli.py @@ -1,12 +1,13 @@ -"""Tests for the audiobook.py CLI — single-book flags and arg validation. +"""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 and -the pre-flight overrides are tested against the real functions with -temporary directories. +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 @@ -250,6 +251,23 @@ class MainHappyPathTests(MainTestCase): 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.""" @@ -314,6 +332,115 @@ class ConvertWiringTests(unittest.TestCase): 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.""" |