"""Tests for the in-TUI task view (ui/taskview.py). The view is driven the same way as the other TUI widgets: the fake curses module and recording screen from test_tui stand in for a terminal. The step-sequencing logic is exercised through ``run_steps_inline`` (no thread), 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 from ui import taskview def _step(title, rc=0): def work(emit, cancel): return rc return taskview.TaskStep(title, work) class _FakeTui: def setUp(self): self.curses = FakeCurses() patcher = patch.dict(sys.modules, {"curses": self.curses}) patcher.start() self.addCleanup(patcher.stop) taskview.tui._THEME.clear() self.addCleanup(taskview.tui._THEME.clear) def make_view(self, steps=(), width=80, height=24): screen = FakeScreen(width=width, height=height) with patch.object(taskview.TaskView, "_worker_main", lambda self: None): view = taskview.TaskView(screen, "Setup", list(steps), clock=lambda: 1000.0) return view, screen class RunStepsInlineTests(unittest.TestCase): def test_runs_steps_in_order_and_returns_zero(self): order = [] steps = [ taskview.TaskStep("a", lambda emit, cancel: order.append("a") or 0), taskview.TaskStep("b", lambda emit, cancel: order.append("b") or 0), ] self.assertEqual(taskview.run_steps_inline(steps), 0) self.assertEqual(order, ["a", "b"]) def test_returns_first_bad_rc_and_continues(self): order = [] steps = [ taskview.TaskStep("a", lambda emit, cancel: order.append("a") or 1), taskview.TaskStep("b", lambda emit, cancel: order.append("b") or 2), ] self.assertEqual(taskview.run_steps_inline(steps), 1) # The second step still ran (warn-and-continue semantics). self.assertEqual(order, ["a", "b"]) def test_passes_emit_and_cancel_to_each_step(self): seen = [] emit = object() cancel = object() steps = [taskview.TaskStep( "a", lambda e, c: seen.append((e, c)) or 0)] taskview.run_steps_inline(steps, emit=emit, cancel=cancel) self.assertEqual(seen, [(emit, cancel)]) class LineWriterTests(unittest.TestCase): def _split(self, text): lines = [] writer = taskview._LineWriter(lines.append) writer.write(text) writer.flush() return lines def test_splits_on_newline(self): self.assertEqual(self._split("one\ntwo\n"), ["one", "two"]) def test_splits_on_carriage_return(self): # git/tqdm progress updates use \r; each update is its own line. self.assertEqual(self._split("a\rb\rc"), ["a", "b", "c"]) def test_handles_mixed_terminators_and_no_final_newline(self): self.assertEqual(self._split("x\ny\r\nz"), ["x", "y", "z"]) class ProgressParsingTests(_FakeTui, unittest.TestCase): def _ingest(self, line): view, _ = self.make_view(steps=[_step("a")]) view._ingest_line(line) return view def test_bytes_progress_is_hidden_from_the_log(self): view = self._ingest("AUDIOCPP_PROGRESS downloaded=512 total=2048") self.assertEqual(view._progress, (512, 2048)) self.assertEqual(view._progress_kind, "bytes") self.assertEqual(view.log_tail, []) def test_percent_progress_kept_in_log(self): view = self._ingest("[ 45%] Building CXX object foo.o") self.assertEqual(view._progress, (45, 100)) self.assertEqual(view._progress_kind, "percent") self.assertEqual(view.log_tail, ["[ 45%] Building CXX object foo.o"]) def test_count_progress_from_ninja(self): view = self._ingest("[123/456] Compiling bar.cpp") self.assertEqual(view._progress, (123, 456)) self.assertEqual(view._progress_kind, "count") def test_git_percent_progress(self): view = self._ingest("Receiving objects: 33% (99/300), 1.2 MiB") self.assertEqual(view._progress, (33, 100)) def test_percent_above_one_hundred_ignored(self): view = self._ingest("CPU 150% usage") self.assertIsNone(view._progress) def test_plain_line_only_logs(self): view = self._ingest("[INFO] doing work") self.assertIsNone(view._progress) self.assertEqual(view.log_tail, ["[INFO] doing work"]) def test_log_tail_is_capped(self): view, _ = self.make_view(steps=[_step("a")]) for i in range(taskview._LOG_TAIL + 5): view._ingest_line(f"line {i}") self.assertEqual(len(view.log_tail), taskview._LOG_TAIL) self.assertEqual(view.log_tail[-1], f"line {taskview._LOG_TAIL + 4}") class StateTransitionTests(_FakeTui, unittest.TestCase): def _steps(self): return [_step("one"), _step("two")] def test_success_flow_marks_steps_ok(self): view, _ = self.make_view(steps=self._steps()) view.handle_event({"kind": "step_start", "index": 0, "title": "one"}) self.assertEqual(view.current, 0) view.handle_event({"kind": "step_done", "index": 0, "rc": 0}) view.handle_event({"kind": "step_start", "index": 1, "title": "two"}) view.handle_event({"kind": "step_done", "index": 1, "rc": 0}) view.handle_event({"kind": "finish", "phase": "done", "rc": 0}) self.assertEqual(view.phase, "done") self.assertEqual(view._result_rc(), 0) self.assertEqual(view._step_mark(0), ("[OK]", "ok")) self.assertEqual(view._step_mark(1), ("[OK]", "ok")) def test_failure_marks_step_failed_and_returns_bad_rc(self): view, _ = self.make_view(steps=self._steps()) view.handle_event({"kind": "step_start", "index": 0, "title": "one"}) view.handle_event({"kind": "step_done", "index": 0, "rc": 7}) view.handle_event({"kind": "finish", "phase": "error", "rc": 7}) self.assertEqual(view.phase, "error") self.assertEqual(view._result_rc(), 7) self.assertEqual(view._step_mark(0), ("[FAIL]", "err")) self.assertEqual(view._step_mark(1), ("[ ]", "dim")) def test_cancelled_run_returns_nonzero(self): view, _ = self.make_view(steps=self._steps()) view.handle_event({"kind": "step_start", "index": 0, "title": "one"}) view.handle_event({"kind": "step_cancelled", "index": 0}) view.handle_event({"kind": "finish", "phase": "cancelled", "rc": 1}) self.assertEqual(view.phase, "cancelled") self.assertEqual(view._result_rc(), 1) # The step interrupted by cancel is marked cancelled, not failed. self.assertEqual(view._step_mark(0), ("[x]", "warn")) self.assertEqual(view._step_mark(1), ("[ ]", "dim")) def test_running_step_shows_a_spinner_mark(self): view, _ = self.make_view(steps=self._steps()) view.handle_event({"kind": "step_start", "index": 0, "title": "one"}) mark, kind = view._step_mark(0) self.assertEqual(kind, "warn") self.assertIn("[", mark) class RenderTests(_FakeTui, unittest.TestCase): def _strings(self, screen): return " ".join(text for _, _, text, _ in screen.strings) def test_running_screen_lists_steps_and_cancel_footer(self): view, screen = self.make_view(steps=[_step("one"), _step("two")]) view.handle_event({"kind": "step_start", "index": 0, "title": "one"}) view.render() text = self._strings(screen) self.assertIn("one", text) self.assertIn("two", text) self.assertIn("Esc or q: cancel", text) def test_done_screen_shows_the_completion_footer(self): view, screen = self.make_view(steps=[_step("one")]) view.handle_event({"kind": "step_start", "index": 0, "title": "one"}) view.handle_event({"kind": "step_done", "index": 0, "rc": 0}) view.handle_event({"kind": "finish", "phase": "done", "rc": 0}) view.render() text = self._strings(screen) self.assertIn("completed", text) self.assertIn("press any key", text) def test_progress_bar_drawn_when_known(self): view, screen = self.make_view(steps=[_step("one")]) view.handle_event({"kind": "step_start", "index": 0, "title": "one"}) view._ingest_line("AUDIOCPP_PROGRESS downloaded=512 total=2048") view.render() text = self._strings(screen) self.assertIn("Progress", text) class NoWaitTests(_FakeTui, unittest.TestCase): """wait_on_finish=False: the view returns to the caller on finish.""" def make_view(self, steps=(), width=80, height=24, wait_on_finish=False): screen = FakeScreen(width=width, height=height) with patch.object(taskview.TaskView, "_worker_main", lambda self: None): view = taskview.TaskView(screen, "Setup", list(steps), clock=lambda: 1000.0, wait_on_finish=wait_on_finish) return view, screen def _strings(self, screen): return " ".join(text for _, _, text, _ in screen.strings) def _run_view(self, wait_on_finish): """Run() with a synchronous worker that posts a successful finish.""" screen = FakeScreen(width=80, height=24) step = _step("one") def fake_worker_main(self): self._queue.put({"kind": "step_start", "index": 0, "title": "one"}) self._queue.put({"kind": "step_done", "index": 0, "rc": 0}) self._queue.put({"kind": "finish", "phase": "done", "rc": 0}) with patch.object(taskview.TaskView, "_worker_main", fake_worker_main): view = taskview.TaskView(screen, "Setup", [step], clock=lambda: 1000.0, wait_on_finish=wait_on_finish) # Replace the daemon thread with a synchronous runner so the # events are queued before run() enters its loop (deterministic). view._worker = _SyncWorker(view._worker_main) return view.run() def test_returns_immediately_when_not_waiting(self): # No scripted keys: run() must return on finish without asking for one. self.assertEqual(self._run_view(wait_on_finish=False), 0) def test_done_footer_omits_key_hint_when_not_waiting(self): view, screen = self.make_view(steps=[_step("one")]) view.handle_event({"kind": "step_start", "index": 0, "title": "one"}) view.handle_event({"kind": "step_done", "index": 0, "rc": 0}) view.handle_event({"kind": "finish", "phase": "done", "rc": 0}) view.render() text = self._strings(screen) self.assertIn("completed", text) self.assertNotIn("press any key", text) class _SyncWorker: """A stand-in for threading.Thread that runs the target synchronously.""" def __init__(self, fn): self._fn = fn def start(self): self._fn() class LabelTests(unittest.TestCase): def test_fmt_bytes(self): self.assertEqual(taskview._fmt_bytes(512), "512B") self.assertEqual(taskview._fmt_bytes(2048), "2.0KB") self.assertEqual(taskview._fmt_bytes(5 * 1024 * 1024), "5.0MB") def test_progress_label_bytes(self): self.assertEqual(taskview._progress_label((512, 2048), "bytes"), "512B / 2.0KB") def test_progress_label_count(self): self.assertEqual(taskview._progress_label((3, 10), "count"), "3/10") def test_progress_label_percent(self): self.assertEqual(taskview._progress_label((45, 100), "percent"), "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()