"""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 sys import unittest 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 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%") if __name__ == "__main__": unittest.main()