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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
|
import locale
import os
import sys
import unittest
from io import StringIO
from unittest import skipUnless, skipIf, expectedFailure
import blessed
import pyte
from pyte import control as ctrl, Stream, Screen
from curtsies.window import BaseWindow, FullscreenWindow, CursorAwareWindow
class FakeStdin(StringIO):
encoding = "ascii"
# thanks superbobry for this code: https://github.com/selectel/pyte/issues/13
class ReportingStream(Stream):
report_escape = {"6": "report_cursor_position"}
def _arguments(self, char):
if char == "n":
# DSR command looks like 'CSI<N>n'. So all we need to do
# is wait for the 'n' argument.
return self.dispatch(self.report_escape[self.current])
else:
return super()._arguments(char)
class ReportingScreen(Screen):
def __init__(self, *args, **kwargs):
self._report_file = FakeStdin()
super().__init__(*args, **kwargs)
def report_cursor_position(self):
# cursor position is 1-indexed in the ANSI escape sequence API
s = ctrl.CSI + "%d;%sR" % (self.cursor.y + 1, self.cursor.x + 1)
self._report_file.seek(0)
self._report_file.write(s)
self._report_file.seek(0)
class ReportingScreenWithExtra(ReportingScreen):
def report_cursor_position(self):
# cursor position is 1-indexed in the ANSI escape sequence API
extra = "qwerty\nasdf\nzxcv"
s = ctrl.CSI + "%d;%sR" % (self.cursor.y + 1, self.cursor.x + 1)
self._report_file.seek(0)
self._report_file.write(extra + s)
self._report_file.seek(0)
class Bugger:
__before__ = __after__ = lambda *args: None
def __getattr__(self, event):
to = sys.stdout
def inner(*args, **flags):
to.write(event.upper() + " ")
to.write("; ".join(map(repr, args)))
to.write(" ")
to.write(", ".join(f"{name}: {repr(arg)}" for name, arg in flags.items()))
to.write(os.linesep)
return inner
class ScreenStdout:
def __init__(self, stream):
self.stream = stream
def write(self, s):
if sys.version_info[0] == 3:
self.stream.feed(s)
else:
self.stream.feed(s.decode(locale.getpreferredencoding()))
def flush(self):
pass
@skipUnless(sys.stdin.isatty(), "blessed Terminal needs streams open")
class TestFullscreenWindow(unittest.TestCase):
def setUp(self):
self.screen = pyte.Screen(10, 3)
self.stream = pyte.Stream()
self.stream.attach(self.screen)
stdout = ScreenStdout(self.stream)
self.window = FullscreenWindow(stdout)
def test_render(self):
with self.window:
self.window.render_to_terminal(["hi", "there"])
self.assertEqual(
self.screen.display, ["hi ", "there ", " "]
)
def test_scroll(self):
with self.window:
self.window.render_to_terminal(["hi", "there"])
self.window.scroll_down()
self.assertEqual(
self.screen.display, ["there ", " ", " "]
)
class NopContext:
def __enter__(*args):
pass
def __exit__(*args):
pass
@skipUnless(sys.stdin.isatty(), "blessed Terminal needs streams open")
class TestCursorAwareWindow(unittest.TestCase):
def setUp(self):
self.screen = ReportingScreen(6, 3)
self.stream = ReportingStream()
self.stream.attach(self.screen)
self.stream.attach(Bugger())
stdout = ScreenStdout(self.stream)
self.window = CursorAwareWindow(
out_stream=stdout, in_stream=self.screen._report_file
)
self.window.cbreak = NopContext()
blessed.Terminal.height = 3
blessed.Terminal.width = 6
# This isn't passing locally for me anymore :/
@expectedFailure
def test_render(self):
with self.window:
self.assertEqual(self.window.top_usable_row, 0)
self.window.render_to_terminal(["hi", "there"])
self.assertEqual(self.screen.display, ["hi ", "there ", " "])
# This isn't passing locally for me anymore :/
@expectedFailure
def test_cursor_position(self):
with self.window:
self.window.render_to_terminal(["hi", "there"], cursor_pos=(2, 4))
self.assertEqual(self.window.get_cursor_position(), (2, 4))
# This isn't passing locally for me anymore :/
@expectedFailure
def test_inital_cursor_position(self):
self.screen.cursor.y += 1
with self.window:
self.assertEqual(self.window.top_usable_row, 1)
self.window.render_to_terminal(["hi", "there"])
self.assertEqual(self.screen.display, [" ", "hi ", "there "])
@skipUnless(sys.stdin.isatty(), "blessed Terminal needs streams open")
class TestCursorAwareWindowWithExtraInput(unittest.TestCase):
def setUp(self):
self.screen = ReportingScreenWithExtra(6, 3)
self.stream = ReportingStream()
self.stream.attach(self.screen)
self.stream.attach(Bugger())
stdout = ScreenStdout(self.stream)
self.extra_bytes = []
self.window = CursorAwareWindow(
out_stream=stdout,
in_stream=self.screen._report_file,
extra_bytes_callback=self.extra_bytes_callback,
)
self.window.cbreak = NopContext()
blessed.Terminal.height = 3
blessed.Terminal.width = 6
def extra_bytes_callback(self, bytes):
self.extra_bytes.append(bytes)
# This isn't passing locally for me anymore :/
@expectedFailure
def test_report_extra_bytes(self):
with self.window:
pass # should have triggered getting initial cursor position
self.assertEqual(b"".join(self.extra_bytes), b"qwerty\nasdf\nzxcv")
|
