diff options
Diffstat (limited to 'app/tests')
| -rw-r--r-- | app/tests/test_backends_servers.py | 60 | ||||
| -rw-r--r-- | app/tests/test_converter_progress.py | 27 | ||||
| -rw-r--r-- | app/tests/test_hub.py | 92 | ||||
| -rw-r--r-- | app/tests/test_runview.py | 81 |
4 files changed, 250 insertions, 10 deletions
diff --git a/app/tests/test_backends_servers.py b/app/tests/test_backends_servers.py index 201d5b6..8eaf53f 100644 --- a/app/tests/test_backends_servers.py +++ b/app/tests/test_backends_servers.py @@ -1,5 +1,7 @@ """Tests for the server lifecycle module (backends/servers.py).""" +import os +import signal import tempfile import unittest from pathlib import Path @@ -225,6 +227,64 @@ class StopTests(unittest.TestCase): self.assertFalse((self.dir / "test-server.pid").exists()) +class ReapTests(unittest.TestCase): + """_reap_exited: an exited child must stop counting as alive.""" + + def test_true_when_child_exited(self): + with patch("os.waitpid", return_value=(4242, 0)) as mk: + self.assertTrue(servers._reap_exited(4242)) + mk.assert_called_once_with(4242, os.WNOHANG) + + def test_false_while_still_running(self): + # (0, 0) is WNOHANG's "still running" answer. + with patch("os.waitpid", return_value=(0, 0)): + self.assertFalse(servers._reap_exited(4242)) + + def test_false_when_not_our_child(self): + with patch("os.waitpid", side_effect=ChildProcessError): + self.assertFalse(servers._reap_exited(4242)) + + +class KillPidTests(unittest.TestCase): + """_kill_pid: the reap check ends the grace wait before SIGKILL.""" + + def test_reaped_child_ends_wait_without_sigkill(self): + with patch("os.getpgid", return_value=4242), \ + patch("os.killpg") as mk_killpg, \ + patch("os.waitpid", return_value=(4242, 0)) as mk_waitpid, \ + patch("time.sleep") as mk_sleep: + ok = servers._kill_pid(4242) + self.assertTrue(ok) + mk_killpg.assert_called_once_with(4242, signal.SIGTERM) + mk_waitpid.assert_called_once_with(4242, os.WNOHANG) + mk_sleep.assert_not_called() + + def test_escalates_to_sigkill_when_child_stays_alive(self): + with patch("os.getpgid", return_value=4242), \ + patch("os.killpg") as mk_killpg, \ + patch("os.waitpid", return_value=(0, 0)), \ + patch("time.sleep"): + ok = servers._kill_pid(4242) + self.assertTrue(ok) + calls = mk_killpg.call_args_list + self.assertEqual(calls[0].args, (4242, signal.SIGTERM)) + self.assertEqual(calls[-1].args, (4242, signal.SIGKILL)) + + def test_foreign_child_falls_back_to_group_probe(self): + # ChildProcessError from waitpid (not our child / already reaped): + # the killpg(0) probe decides; a vanished group ends the wait. + with patch("os.getpgid", return_value=4242), \ + patch("os.killpg", + side_effect=[None, ProcessLookupError]) as mk_killpg, \ + patch("os.waitpid", side_effect=ChildProcessError), \ + patch("time.sleep"): + ok = servers._kill_pid(4242) + self.assertTrue(ok) + calls = mk_killpg.call_args_list + self.assertEqual(calls[0].args, (4242, signal.SIGTERM)) + self.assertEqual(calls[-1].args, (4242, 0)) + + class ManagesTests(unittest.TestCase): """manages(): a live recorded pid marks a server as ours.""" diff --git a/app/tests/test_converter_progress.py b/app/tests/test_converter_progress.py index 1041173..0f708a1 100644 --- a/app/tests/test_converter_progress.py +++ b/app/tests/test_converter_progress.py @@ -142,6 +142,33 @@ class ProgressEventTests(unittest.TestCase): self.assertEqual(events[-1]["kind"], "done") self.assertEqual(events[-1]["ok"], 0) + def test_book_done_reports_output_files(self): + # book_done carries the output file names the run view's summary + # lists after the TUI closes. + events = [] + converter = self.fixture.build(progress=events.append) + converter.run() + done = next(e for e in events if e["kind"] == "book_done") + self.assertEqual(done["files"], ["book_Vivian.mp3"]) + self.assertEqual(converter.current_outputs, ["book_Vivian.mp3"]) + + def test_multi_chapter_book_lists_every_chapter_file(self): + # A multi-section book (no --single-file) produces one file per + # chapter, all reported on the event. + sections = [MagicMock(text=f"chapter {n} text.", title=t) + for n, t in enumerate(("One", "Two"), 1)] + book = MagicMock(title="Book", author="Author", sections=sections) + events = [] + with patch.object(converter_mod.extractors, "extract_book", + return_value=book): + converter = self.fixture.build(progress=events.append) + converter.single_file = False + converter.run() + done = next(e for e in events if e["kind"] == "book_done") + self.assertTrue(done["ok"]) + self.assertEqual(done["files"], + ["book_Vivian_01_One.mp3", "book_Vivian_02_Two.mp3"]) + class CancelTests(unittest.TestCase): def setUp(self): diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py index 3547ba8..ebae82b 100644 --- a/app/tests/test_hub.py +++ b/app/tests/test_hub.py @@ -651,7 +651,8 @@ class ConvertFlowTests(unittest.TestCase): def _form_values(self, **overrides): """A fully-populated form result, with sensible defaults.""" values = {"output_format": "m4b", "speed": "1.0", - "single_file": False, "debug": False} + "single_file": False, "debug": False, + "stop_and_exit": True} values.update(overrides) return values @@ -666,23 +667,26 @@ class ConvertFlowTests(unittest.TestCase): """A backend status that is ready to convert with.""" return BackendStatus(key, label, installed=True, configured=True) - def _convert(self, stdscr, statuses): + def _convert(self, stdscr, statuses, run_result=None): """Run the convert flow with STATUSES, returning the command tuple. ``_run_conversion`` is stubbed so the accepted command is captured - instead of launching the run view; None is returned when the flow - aborts before reaching a conversion (nothing ready, a flash). + instead of launching the run view (and reporting RUN_RESULT — True + when the user answered "stop the server and exit"); None is returned + when the flow aborts before reaching a conversion (nothing ready, a + flash). The screen's navigation result lands on ``self.nav``. """ captured = {} def fake_run_conversion(self_, backend, kwargs): captured["backend"] = backend captured["kwargs"] = kwargs + return run_result with patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub._Hub, "_run_conversion", fake_run_conversion): - hub._Hub(None).screen_convert() + self.nav = hub._Hub(None).screen_convert() if "backend" not in captured: return None return ("convert", captured["backend"], captured["kwargs"]) @@ -742,9 +746,14 @@ class ConvertFlowTests(unittest.TestCase): self.assertEqual([f["key"] for f in fields], ["backend", "model_id", "audiocpp_voice", "instructions", "output_format", "speed", - "single_file", "debug"]) + "single_file", "debug", "stop_and_exit"]) self.assertEqual(form_kwargs["buttons"], ("Generate!", "Cancel")) self.assertTrue(form_kwargs["start_on_buttons"]) + # The stop-and-exit toggle ships on by default. + stop_field = self._field("stop_and_exit") + self.assertEqual(stop_field["kind"], "bool") + self.assertTrue(stop_field["value"]) + self.assertTrue(cmd[2]["stop_and_exit"]) # The backend field offers the remote entry under a [remote] label. self.assertEqual(fields[0]["choices"], [("audio.cpp [remote]", "audiocpp-remote")]) @@ -1011,7 +1020,8 @@ class ConvertFlowTests(unittest.TestCase): fields = self.tui.forms_seen[0][1] self.assertEqual([f["key"] for f in fields], ["backend", "mode", "speaker", "clone", - "output_format", "speed", "single_file", "debug"]) + "output_format", "speed", "single_file", "debug", + "stop_and_exit"]) mode_field = self._field("mode") self.assertEqual(mode_field["choices"], [("Built-in speaker", "custom"), @@ -1080,6 +1090,24 @@ class ConvertFlowTests(unittest.TestCase): self.assertEqual(cmd[2]["api_url"], "http://10.0.0.5:8000") self.assertEqual(self._field("faster_voice")["kind"], "text") + def test_stop_and_exit_answer_quits_the_tui(self): + # Yes on the run view's stop-and-exit prompt ends the whole TUI: + # screen_convert returns None instead of Wizard.BACK. + st = self._remote("faster", "faster-qwen3-tts", + url="http://10.0.0.5:8000") + self._answer_form(backend="faster-remote", faster_voice="obama") + cmd = self._convert(None, [st], run_result=True) + self.assertIsNotNone(cmd) + self.assertIsNone(self.nav) + + def test_normal_finish_returns_to_the_menu(self): + st = self._remote("faster", "faster-qwen3-tts", + url="http://10.0.0.5:8000") + self._answer_form(backend="faster-remote", faster_voice="obama") + cmd = self._convert(None, [st]) + self.assertIsNotNone(cmd) + self.assertIs(self.nav, hub.tui.Wizard.BACK) + def test_qwen_remote_limited_modes_and_api_url(self): # A remote qwen with only the Base (clone) demo answering: the form # offers only clone mode and targets the clone remote URL. @@ -1130,7 +1158,7 @@ class ConvertFlowTests(unittest.TestCase): [f["key"] for f in fields], ["backend", "model_id", "audiocpp_voice", "instructions", "mode", "speaker", "clone", "output_format", "speed", - "single_file", "debug"]) + "single_file", "debug", "stop_and_exit"]) # The form opens on the configured default (audio.cpp): its fields # show, the other backend's hide. (Instructions is hidden too: the # default higgs entry is clone-only, which ignores instructions.) @@ -1216,6 +1244,24 @@ class PrepareRunConfigTests(unittest.TestCase): self.assertEqual(cfg.server_name, "qwen-custom") self.assertNotIn("autostart", kwargs) + def test_stop_and_exit_travels_on_the_config_not_the_kwargs(self): + # The run-view toggle is not a converter kwarg: it moves onto the + # config (and defaults to off when the form did not send it). + with tempfile.TemporaryDirectory() as tmp: + with patch.object(hub, "LOGS_FOLDER", Path(tmp)), \ + patch.object(hub, "detect_all", return_value=[]): + cfg = hub._prepare_run_config( + "audiocpp", {"stop_and_exit": True}) + self.assertTrue(cfg.stop_and_exit) + cfg = hub._prepare_run_config("audiocpp", {}) + self.assertFalse(cfg.stop_and_exit) + + def test_remote_config_carries_stop_and_exit(self): + cfg = hub._prepare_run_config( + "audiocpp", {"api_url": "http://10.0.0.5:8080", + "stop_and_exit": True}) + self.assertTrue(cfg.stop_and_exit) + def test_autostart_with_missing_spec_continues_with_notice(self): kwargs = {"autostart": "gone"} with patch.object(hub, "detect_all", return_value=[]), \ @@ -1336,11 +1382,14 @@ class DispatchConversionTests(unittest.TestCase): timeouts.append(ms) class FakeView: + run_result = None + def __init__(self, scr, config): self.config = config self.scr = scr + def run(self): - pass + return type(self).run_result screen = Screen() with patch.object(hub, "_prepare_run_config", @@ -1348,11 +1397,34 @@ class DispatchConversionTests(unittest.TestCase): backend="qwen", backend_label="qwen-tts", kwargs={}, book_files=[], planned=[])) as mk_cfg, \ patch.object(hub.runview, "RunView", FakeView): - hub._Hub(screen)._run_conversion("qwen", {}) + result = hub._Hub(screen)._run_conversion("qwen", {}) mk_cfg.assert_called_once() # The run view leaves a timed getch behind; it is reset so the hub # menus block for keys again. self.assertEqual(timeouts, [-1]) + # A normal finish lands back on the menu (no quit signal). + self.assertFalse(result) + + def test_stop_and_exit_answer_propagates_from_the_view(self): + class Screen: + def timeout(self, ms): + pass + + class FakeView: + def __init__(self, scr, config): + self.config = config + self.scr = scr + + def run(self): + return True + + with patch.object(hub, "_prepare_run_config", + return_value=hub.runview.RunConfig( + backend="qwen", backend_label="qwen-tts", + kwargs={}, book_files=[], planned=[])), \ + patch.object(hub.runview, "RunView", FakeView): + result = hub._Hub(Screen())._run_conversion("qwen", {}) + self.assertTrue(result) def test_no_run_config_skips_the_view(self): with patch.object(hub, "_prepare_run_config", return_value=None) as mk_cfg, \ diff --git a/app/tests/test_runview.py b/app/tests/test_runview.py index 556cb52..31a8ebe 100644 --- a/app/tests/test_runview.py +++ b/app/tests/test_runview.py @@ -171,6 +171,10 @@ class RenderTests(_FakeTui, unittest.TestCase): text = self._strings(screen) self.assertIn("stopping", text) self.assertIn("(1s)", text) + # The stop wait ignores keys, so the footer must not promise that + # pressing one returns to the menu. + self.assertNotIn("press any key", text) + self.assertIn("stopping the audiocpp server...", text) def test_error_screen_shows_detail_and_log(self): view, screen = self.make_view(log_path="/tmp/audiobook.log") @@ -236,6 +240,83 @@ class RunLoopTests(_FakeTui, unittest.TestCase): view.run() mk_stop.assert_not_called() + def test_finished_run_toggle_off_waits_for_key_then_menu(self): + # Toggle off: a key on the summary screen returns to the menu; no + # prompt is asked and the server keeps running. + with patch.object(runview.tui, "confirm") as mk_confirm, \ + patch.object(runview.servers, "alive", return_value=True), \ + patch.object(runview.servers, "stop") as mk_stop: + view, screen = self.make_view(keys=[ord("x")], + autostart_spec="SPEC") + view.started_server = True + view._queue.put({"kind": "done", "ok": 1, "total": 1}) + result = view.run() + self.assertFalse(result) + mk_confirm.assert_not_called() + mk_stop.assert_not_called() + + def test_stop_and_exit_stops_server_and_quits_without_keys(self): + # Toggle on: the moment the run ends the server is stopped and the + # view reports quit — no key press and no prompt anywhere. + with patch.object(runview.servers, "alive", return_value=True), \ + patch.object(runview.servers, "stop") as mk_stop, \ + patch.object(runview.common, + "record_post_tui_notice") as mk_notice: + view, screen = self.make_view([], autostart_spec="SPEC", + stop_and_exit=True) + view.started_server = True + view._queue.put({"kind": "book", "index": 1, "total": 1, + "name": "book.txt"}) + view._queue.put({"kind": "book_done", "name": "book.txt", + "ok": True, + "files": ["book_test_michael.mp3"]}) + view._queue.put({"kind": "done", "ok": 1, "total": 1}) + result = view.run() + self.assertTrue(result) + mk_stop.assert_called_once_with("audiocpp") + text = mk_notice.call_args[0][0] + self.assertIn("Output directory:", text) + self.assertIn("[OK] book.txt: book_test_michael.mp3", text) + self.assertIn("1 of 1 book(s) generated successfully", text) + self.assertIn("Elapsed time:", text) + + def test_stop_and_exit_leaves_external_servers_alone(self): + # A server this run did not start is never stopped; the TUI still + # quits and the summary is still printed. + with patch.object(runview.servers, "stop") as mk_stop, \ + patch.object(runview.common, + "record_post_tui_notice") as mk_notice: + view, screen = self.make_view([], server_name=None, + stop_and_exit=True) + view.started_server = False + view._queue.put({"kind": "done", "ok": 1, "total": 1}) + result = view.run() + self.assertTrue(result) + mk_stop.assert_not_called() + self.assertIn("No books were converted", + mk_notice.call_args[0][0]) + + def test_stop_and_exit_failure_summary_lists_the_error(self): + # A failed ending also auto-exits; the failing book's detail line + # lands in the post-TUI summary. + with patch.object(runview.servers, "alive", return_value=True), \ + patch.object(runview.servers, "stop"), \ + patch.object(runview.common, + "record_post_tui_notice") as mk_notice: + view, screen = self.make_view([], autostart_spec="SPEC", + stop_and_exit=True) + view.started_server = True + view._queue.put({"kind": "book_failed", "name": "bad.txt", + "error": "chunk 3 failed", + "files": ["bad_x.mp3"]}) + view._queue.put({"kind": "done", "ok": 0, "total": 1}) + result = view.run() + self.assertTrue(result) + text = mk_notice.call_args[0][0] + self.assertIn("[FAIL] bad.txt: bad_x.mp3", text) + self.assertIn("chunk 3 failed", text) + self.assertIn("0 of 1 book(s) generated successfully", text) + class WorkerTests(_FakeTui, unittest.TestCase): """The worker thread's handoff into audiobook.convert.""" |