diff options
Diffstat (limited to 'app/tests')
| -rw-r--r-- | app/tests/test_backends_audiocpp.py | 130 | ||||
| -rw-r--r-- | app/tests/test_taskview.py | 210 |
2 files changed, 330 insertions, 10 deletions
diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py index 8f79085..2f41131 100644 --- a/app/tests/test_backends_audiocpp.py +++ b/app/tests/test_backends_audiocpp.py @@ -649,6 +649,11 @@ class InstallModelsTests(unittest.TestCase): self.assertTrue(make_server._decide_download(self.checkout, confirm)) confirm.assert_called_once() + def test_decide_download_defaults_to_yes(self): + confirm = MagicMock(return_value=True) + make_server._decide_download(self.checkout, confirm) + self.assertIs(confirm.call_args[0][1], True) + class TranscribeWavDirTests(unittest.TestCase): def setUp(self): @@ -1793,27 +1798,132 @@ if __name__ == "__main__": class SetupScreenTests(unittest.TestCase): """setup_screen: the wizard run on the hub's screen, setup tail via the - in-TUI task view.""" + in-TUI task view (two parallel lanes on a fresh install).""" def test_abort_returns_one_without_executing(self): with patch.object(make_server, "_wizard", return_value=None) as mk_wizard, \ - patch.object(make_server, "_execute_steps") as mk_steps: + patch.object(make_server, "_execute_lanes") as mk_lanes: rc = make_server.setup_screen(None) self.assertEqual(rc, 1) mk_wizard.assert_called_once() - mk_steps.assert_not_called() + mk_lanes.assert_not_called() def test_success_runs_the_tail_in_the_task_view(self): settings = {"audiocpp_dir": Path("/x")} - steps = [make_server.taskview.TaskStep("t", lambda emit, cancel: 0)] + lanes = [make_server.taskview.TaskLane( + "Build", [make_server.taskview.TaskStep("t", lambda emit, cancel: 0)])] with patch.object(make_server, "_wizard", return_value=settings), \ - patch.object(make_server, "_execute_steps", - return_value=steps) as mk_steps, \ - patch.object(make_server.taskview, "run_steps", + patch.object(make_server, "_execute_lanes", + return_value=lanes) as mk_lanes, \ + patch.object(make_server.taskview, "run_lanes", return_value=0) as mk_run: rc = make_server.setup_screen(None) self.assertEqual(rc, 0) - mk_steps.assert_called_once() - self.assertIs(mk_steps.call_args[0][0], settings) + mk_lanes.assert_called_once() + self.assertIs(mk_lanes.call_args[0][0], settings) + self.assertTrue(mk_lanes.call_args[1]["parallel"]) mk_run.assert_called_once() - self.assertEqual(mk_run.call_args[0][2], steps) + self.assertEqual(mk_run.call_args[0][2], lanes) + + +class ExecuteLanesTests(unittest.TestCase): + """_execute_lanes: two lanes (build + configure/download) and the + flattened console order.""" + + def _settings(self, **overrides): + settings = { + "audiocpp_dir": Path("/x"), + "backend": "cuda", + "build": True, + "download": True, + "include_clone": False, + "wav_dir": None, + "plan": None, + "sync_port": None, + "sync_model_ids": None, + "delete_unused": False, + "unused_entries": [], + "model_entries": [], + "entry_ids": [], + "install_guidance": [], + "output_path": Path("/x/server.json"), + "host": "127.0.0.1", + "port": 8080, + "lazy_load": True, + } + settings.update(overrides) + return settings + + def test_two_lanes_when_building(self): + args = make_server.build_parser().parse_args([]) + lanes = make_server._execute_lanes(self._settings(), args) + self.assertEqual([lane.title for lane in lanes], + ["Build", "Configure & download"]) + self.assertEqual([s.title for s in lanes[0].steps], + ["Build audiocpp_server (cuda)"]) + self.assertEqual([s.title for s in lanes[1].steps], + ["Transcribe reference voices", + "Write server.json & sync config", + "Download models"]) + + def test_single_lane_when_not_building(self): + args = make_server.build_parser().parse_args([]) + lanes = make_server._execute_lanes( + self._settings(build=False), args) + self.assertEqual([lane.title for lane in lanes], + ["Configure & download"]) + + def test_flattened_console_steps_keep_the_build_first(self): + args = make_server.build_parser().parse_args([]) + steps = make_server._execute_steps(self._settings(), args) + self.assertEqual([s.title for s in steps], + ["Build audiocpp_server (cuda)", + "Transcribe reference voices", + "Write server.json & sync config", + "Download models"]) + + def test_download_step_prints_the_parallel_launch_hint(self): + args = make_server.build_parser().parse_args([]) + lanes = make_server._execute_lanes(self._settings(), args, + parallel=True) + install_step = lanes[1].steps[2] + with patch.object(make_server, "_install_models") as mk_install, \ + patch.object(make_server, "_print_launch_hint") as mk_hint: + install_step.work(lambda line: None, threading.Event()) + mk_hint.assert_called_once() + self.assertTrue(mk_hint.call_args[1]["pending_build"]) + + def test_download_step_console_hint_is_not_pending(self): + args = make_server.build_parser().parse_args([]) + lanes = make_server._execute_lanes(self._settings(), args) + install_step = lanes[1].steps[2] + with patch.object(make_server, "_install_models") as mk_install, \ + patch.object(make_server, "_print_launch_hint") as mk_hint: + install_step.work(lambda line: None, threading.Event()) + mk_hint.assert_called_once() + self.assertFalse(mk_hint.call_args[1]["pending_build"]) + + +class LaunchHintTests(unittest.TestCase): + """_print_launch_hint: exact command vs. the pending-build message.""" + + def _capture(self, audiocpp_dir, output_path, pending_build=False): + buf = io.StringIO() + with redirect_stdout(buf), \ + patch.object(make_server, "find_audiocpp_server_bin", + return_value=None): + make_server._print_launch_hint(audiocpp_dir, output_path, + pending_build=pending_build) + return buf.getvalue() + + def test_pending_build_names_the_post_build_command(self): + out = self._capture(Path("/tmp/acpp"), Path("/tmp/acpp/server.json"), + pending_build=True) + self.assertIn("still building", out) + self.assertNotIn("Build it first", out) + self.assertIn("audiocpp_server --config /tmp/acpp/server.json", out) + + def test_missing_binary_gives_build_remediation(self): + out = self._capture(Path("/tmp/acpp"), Path("/tmp/acpp/server.json")) + self.assertIn("Build it first", out) + self.assertNotIn("still building", out) diff --git a/app/tests/test_taskview.py b/app/tests/test_taskview.py index 15fc501..d1bc658 100644 --- a/app/tests/test_taskview.py +++ b/app/tests/test_taskview.py @@ -7,8 +7,11 @@ the progress-line parsing through ``TaskView._ingest_line``, and state transitions through ``handle_event`` + ``_step_mark`` + ``_result_rc``. """ +import io import sys +import threading import unittest +from queue import Empty from unittest.mock import patch from tests.test_tui import FakeCurses, FakeScreen @@ -227,5 +230,212 @@ class LabelTests(unittest.TestCase): "45%") +class RunLanesTests(_FakeTui, unittest.TestCase): + """run_lanes: one lane falls back to run_steps, two use the split view.""" + + def test_single_lane_delegates_to_run_steps(self): + lane = taskview.TaskLane("Build", [_step("one")]) + with patch.object(taskview, "run_steps", return_value=0) as mk_run: + rc = taskview.run_lanes(None, "Setup", [lane]) + self.assertEqual(rc, 0) + mk_run.assert_called_once() + self.assertEqual(mk_run.call_args[0][2], lane.steps) + + def test_empty_lanes_are_dropped(self): + lane = taskview.TaskLane("Build", [_step("one")]) + with patch.object(taskview, "run_steps", return_value=0) as mk_run: + rc = taskview.run_lanes( + None, "Setup", + [taskview.TaskLane("Empty", []), lane]) + self.assertEqual(rc, 0) + self.assertEqual(mk_run.call_args[0][2], lane.steps) + + def test_no_lanes_returns_zero_without_running(self): + with patch.object(taskview, "run_steps") as mk_run, \ + patch.object(taskview, "LanesView") as mk_view: + rc = taskview.run_lanes(None, "Setup", []) + self.assertEqual(rc, 0) + mk_run.assert_not_called() + mk_view.assert_not_called() + + def test_two_lanes_uses_the_split_view(self): + lanes = [taskview.TaskLane("Build", [_step("one")]), + taskview.TaskLane("Download", [_step("two")])] + with patch.object(taskview, "LanesView") as mk_view: + mk_view.return_value.run.return_value = 0 + rc = taskview.run_lanes(None, "Setup", lanes) + self.assertEqual(rc, 0) + mk_view.assert_called_once_with(None, "Setup", lanes) + mk_view.return_value.run.assert_called_once_with() + + +class ThreadRouterTests(unittest.TestCase): + def test_routes_to_registered_thread_and_falls_back(self): + fallback = io.StringIO() + router = taskview._ThreadRouter(fallback) + captured = [] + writer = taskview._LineWriter(captured.append) + with router.for_thread(writer): + router.write("hello\n") + self.assertEqual(captured, ["hello"]) + # Unregistered thread falls through to the original stream. + router.write("fallback\n") + self.assertEqual(fallback.getvalue(), "fallback\n") + + def test_concurrent_prints_land_in_their_own_writer(self): + registry = {} + router = taskview._ThreadRouter(io.StringIO(), registry) + lines_a, lines_b = [], [] + wa = taskview._LineWriter(lines_a.append) + wb = taskview._LineWriter(lines_b.append) + ready_a, ready_b = threading.Event(), threading.Event() + go = threading.Event() + + def body(writer, tag, ready): + with router.for_thread(writer): + ready.set() + go.wait() + for i in range(50): + print(f"{tag}-{i}") + + with patch.object(sys, "stdout", router): + threads = [threading.Thread(target=body, + args=(wa, "A", ready_a)), + threading.Thread(target=body, + args=(wb, "B", ready_b))] + for t in threads: + t.start() + ready_a.wait() + ready_b.wait() + go.set() + for t in threads: + t.join() + self.assertTrue(lines_a) + self.assertTrue(lines_b) + self.assertTrue(all(line.startswith("A-") for line in lines_a)) + self.assertTrue(all(line.startswith("B-") for line in lines_b)) + + +class LanesViewTests(_FakeTui, unittest.TestCase): + """LanesView: parallel lanes, isolated logs/progress, split rendering.""" + + def make_view(self, lanes, width=80, height=24): + screen = FakeScreen(width=width, height=height) + view = taskview.LanesView(screen, "Setup", lanes, + clock=lambda: 1000.0) + return view, screen + + def _two_lanes(self): + return [taskview.TaskLane("Build", [_step("one"), _step("two")]), + taskview.TaskLane("Download models", [_step("dl")])] + + def test_two_lane_workers_run_both_steps(self): + ran = [] + + def work(name): + def _w(emit, cancel): + ran.append(name) + return 0 + return _w + + lanes = [taskview.TaskLane("A", [taskview.TaskStep("a", work("A"))]), + taskview.TaskLane("B", [taskview.TaskStep("b", work("B"))])] + view, _ = self.make_view(lanes) + registry = {} + router = taskview._ThreadRouter(sys.stdout, registry) + threads = [threading.Thread(target=view._lane_worker, + args=(lane_state, router, view._cancel)) + for lane_state in view._lanes] + for t in threads: + t.start() + for t in threads: + t.join() + view._drain() + self.assertCountEqual(ran, ["A", "B"]) + self.assertTrue(all(lane.finished for lane in view._lanes)) + + def test_lane_worker_routes_prints_to_its_lane(self): + def work(emit, cancel): + print("lane-A-log") + return 0 + + lanes = [taskview.TaskLane("A", [taskview.TaskStep("a", work)])] + view, _ = self.make_view(lanes) + registry = {} + router = taskview._ThreadRouter(io.StringIO(), registry) + lane_state = view._lanes[0] + with patch.object(sys, "stdout", router): + view._lane_worker(lane_state, router, view._cancel) + events = [] + while True: + try: + events.append(lane_state.queue.get_nowait()) + except Empty: + break + lines = [e["text"] for e in events if e.get("kind") == "line"] + self.assertIn("lane-A-log", lines) + self.assertTrue(any(e.get("kind") == "lane_finish" for e in events)) + + def test_progress_is_isolated_per_lane(self): + view, _ = self.make_view(self._two_lanes()) + view._ingest_lane_line(view._lanes[0], + "AUDIOCPP_PROGRESS downloaded=512 total=2048") + self.assertEqual(view._lanes[0].progress, (512, 2048)) + self.assertIsNone(view._lanes[1].progress) + view._ingest_lane_line(view._lanes[1], "[ 45%] building") + self.assertEqual(view._lanes[1].progress, (45, 100)) + self.assertEqual(view._lanes[0].progress_kind, "bytes") + + def test_result_rc_returns_first_nonzero_across_lanes(self): + view, _ = self.make_view(self._two_lanes()) + # Lane 0 step 0 succeeds, lane 0 step 1 fails, lane 1 succeeds. + view._lanes[0].results = [0, 7] + view._lanes[1].results = [0] + self.assertEqual(view._result_rc(), 7) + + def test_all_lanes_finished_flips_phase_to_done(self): + view, _ = self.make_view(self._two_lanes()) + for lane in view._lanes: + lane.finished = True + lane.rc = 0 + view._drain() + self.assertEqual(view.phase, "done") + self.assertEqual(view._result_rc(), 0) + + def test_error_phase_when_a_lane_reports_failure(self): + view, _ = self.make_view(self._two_lanes()) + view._lanes[0].finished = True + view._lanes[0].rc = 3 + view._lanes[1].finished = True + view._lanes[1].rc = 0 + view._drain() + self.assertEqual(view.phase, "error") + + def test_cancelled_run_returns_nonzero(self): + view, _ = self.make_view(self._two_lanes()) + view._cancel.set() + for lane in view._lanes: + lane.finished = True + view._drain() + self.assertEqual(view.phase, "cancelled") + self.assertTrue(view.cancelled) + self.assertEqual(view._result_rc(), 1) + + def test_split_render_draws_both_lane_titles(self): + view, screen = self.make_view(self._two_lanes()) + view.render() + text = " ".join(t for _, _, t, _ in screen.strings) + self.assertIn("Build", text) + self.assertIn("Download models", text) + self.assertIn("Esc or q: cancel", text) + + def test_split_render_stacks_on_narrow_terminal(self): + view, screen = self.make_view(self._two_lanes(), width=60) + view.render() + text = " ".join(t for _, _, t, _ in screen.strings) + self.assertIn("Build", text) + self.assertIn("Download models", text) + + if __name__ == "__main__": unittest.main() |