diff options
Diffstat (limited to 'app/tests/test_taskview.py')
| -rw-r--r-- | app/tests/test_taskview.py | 210 |
1 files changed, 210 insertions, 0 deletions
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() |