1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
"""Tests for the shared full-screen view plumbing (ui/viewkit.py).
ScreenView is the base behind TaskView and RunView; those views' suites
cover it end to end. These tests pin the template-loop contract itself:
worker start-up, event draining, terminal/key handling, and the getch
mode switching.
"""
import sys
import threading
import unittest
from unittest.mock import patch
from tests.test_tui import FakeCurses, FakeScreen
from ui import viewkit
from ui.viewkit import ScreenView
class _SyncThread:
"""Thread stand-in that runs its target synchronously on start()."""
def __init__(self, target, daemon=None):
self._target = target
self._alive = False
def start(self):
self._alive = True
self._target()
self._alive = False
def is_alive(self):
return self._alive
def join(self, timeout=None):
pass
class EchoView(ScreenView):
"""Minimal concrete view: records keys, ends after N events."""
def __init__(self, scr, *, terminal_after=None, clock=lambda: 1000.0):
super().__init__(scr, clock=clock)
self.keys = []
self.terminal_after = terminal_after
def work():
for index in range(self.terminal_after or 0):
self._queue.put({"kind": "tick", "index": index})
if self.terminal_after is not None:
self._finish("done")
# Created here; the base template starts it.
self._worker = _SyncThread(work)
def handle_event(self, event):
pass
def render(self):
pass
def _terminal_result(self):
return "terminal"
class TemplateLoopTests(unittest.TestCase):
def setUp(self):
self.curses = FakeCurses()
patcher = patch.dict(sys.modules, {"curses": self.curses})
patcher.start()
self.addCleanup(patcher.stop)
viewkit.tui._THEME.clear()
self.addCleanup(viewkit.tui._THEME.clear)
def _view(self, keys, **kwargs):
screen = FakeScreen(keys=keys)
return EchoView(screen, **kwargs), screen
def test_worker_runs_and_terminal_key_ends_the_view(self):
view, screen = self._view(keys=[ord("x")], terminal_after=2)
self.assertEqual(view.run(), "terminal")
# Both queued events were drained before the key was read.
self.assertEqual(view.phase, "done")
def test_timeout_getch_keeps_the_loop_going(self):
# -1 (redraw timeout) events never end the loop.
view, screen = self._view(keys=[-1, ord("x")], terminal_after=1)
self.assertEqual(view.run(), "terminal")
def test_esc_confirms_cancel_then_ends_via_after_cancel(self):
view, screen = self._view(keys=[27])
with patch.object(view, "_prompt_cancel", return_value=True) as mk, \
patch.object(view, "_after_cancel",
return_value="cancelled") as mk_after:
self.assertEqual(view.run(), "cancelled")
mk.assert_called_once()
mk_after.assert_called_once()
def test_esc_declined_keeps_the_view_running(self):
view, screen = self._view(keys=[27, ord("x")], terminal_after=1)
with patch.object(view, "_prompt_cancel", return_value=False):
self.assertEqual(view.run(), "terminal")
def test_early_exit_wins_before_render(self):
view, screen = self._view(keys=[], terminal_after=1)
with patch.object(view, "_early_exit", return_value="early"), \
patch.object(view, "render") as mk_render:
self.assertEqual(view.run(), "early")
mk_render.assert_not_called()
def test_stop_hook_restores_blocking_getch(self):
view, screen = self._view(keys=[ord("x")], terminal_after=1)
view.run()
self.assertEqual(screen.timeouts[-1], -1)
class DefaultPromptCancelTests(unittest.TestCase):
def setUp(self):
self.curses = FakeCurses()
patcher = patch.dict(sys.modules, {"curses": self.curses})
patcher.start()
self.addCleanup(patcher.stop)
viewkit.tui._THEME.clear()
self.addCleanup(viewkit.tui._THEME.clear)
def test_confirm_sets_cancel_and_joins_the_worker(self):
screen = FakeScreen(keys=[])
view = EchoView(screen)
joined = threading.Event()
class Worker:
def is_alive(self):
return True
def join(self, timeout=None):
joined.set()
view._worker = Worker()
with patch.object(viewkit.tui, "confirm", return_value=True):
self.assertTrue(view._prompt_cancel())
self.assertTrue(joined.is_set())
self.assertTrue(view.cancelling)
self.assertTrue(view._cancel.is_set())
def test_declined_cancel_leaves_the_run_alone(self):
screen = FakeScreen(keys=[])
view = EchoView(screen)
with patch.object(viewkit.tui, "confirm", return_value=False):
self.assertFalse(view._prompt_cancel())
self.assertFalse(view.cancelling)
self.assertFalse(view._cancel.is_set())
if __name__ == "__main__":
unittest.main()
|
