μHAL (v2.8.17)
Part of the IPbus software repository
All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Modules Pages
test_iostream.py
Go to the documentation of this file.
1from contextlib import redirect_stderr, redirect_stdout
2from io import StringIO
3
4from pybind11_tests import iostream as m
5
6
7def test_captured(capsys):
8 msg = "I've been redirected to Python, I hope!"
9 m.captured_output(msg)
10 stdout, stderr = capsys.readouterr()
11 assert stdout == msg
12 assert stderr == ""
13
14 m.captured_output_default(msg)
15 stdout, stderr = capsys.readouterr()
16 assert stdout == msg
17 assert stderr == ""
18
19 m.captured_err(msg)
20 stdout, stderr = capsys.readouterr()
21 assert stdout == ""
22 assert stderr == msg
23
24
25def test_captured_large_string(capsys):
26 # Make this bigger than the buffer used on the C++ side: 1024 chars
27 msg = "I've been redirected to Python, I hope!"
28 msg = msg * (1024 // len(msg) + 1)
29
30 m.captured_output_default(msg)
31 stdout, stderr = capsys.readouterr()
32 assert stdout == msg
33 assert stderr == ""
34
35
36def test_captured_utf8_2byte_offset0(capsys):
37 msg = "\u07FF"
38 msg = "" + msg * (1024 // len(msg) + 1)
39
40 m.captured_output_default(msg)
41 stdout, stderr = capsys.readouterr()
42 assert stdout == msg
43 assert stderr == ""
44
45
46def test_captured_utf8_2byte_offset1(capsys):
47 msg = "\u07FF"
48 msg = "1" + msg * (1024 // len(msg) + 1)
49
50 m.captured_output_default(msg)
51 stdout, stderr = capsys.readouterr()
52 assert stdout == msg
53 assert stderr == ""
54
55
56def test_captured_utf8_3byte_offset0(capsys):
57 msg = "\uFFFF"
58 msg = "" + msg * (1024 // len(msg) + 1)
59
60 m.captured_output_default(msg)
61 stdout, stderr = capsys.readouterr()
62 assert stdout == msg
63 assert stderr == ""
64
65
66def test_captured_utf8_3byte_offset1(capsys):
67 msg = "\uFFFF"
68 msg = "1" + msg * (1024 // len(msg) + 1)
69
70 m.captured_output_default(msg)
71 stdout, stderr = capsys.readouterr()
72 assert stdout == msg
73 assert stderr == ""
74
75
76def test_captured_utf8_3byte_offset2(capsys):
77 msg = "\uFFFF"
78 msg = "12" + msg * (1024 // len(msg) + 1)
79
80 m.captured_output_default(msg)
81 stdout, stderr = capsys.readouterr()
82 assert stdout == msg
83 assert stderr == ""
84
85
86def test_captured_utf8_4byte_offset0(capsys):
87 msg = "\U0010FFFF"
88 msg = "" + msg * (1024 // len(msg) + 1)
89
90 m.captured_output_default(msg)
91 stdout, stderr = capsys.readouterr()
92 assert stdout == msg
93 assert stderr == ""
94
95
96def test_captured_utf8_4byte_offset1(capsys):
97 msg = "\U0010FFFF"
98 msg = "1" + msg * (1024 // len(msg) + 1)
99
100 m.captured_output_default(msg)
101 stdout, stderr = capsys.readouterr()
102 assert stdout == msg
103 assert stderr == ""
104
105
106def test_captured_utf8_4byte_offset2(capsys):
107 msg = "\U0010FFFF"
108 msg = "12" + msg * (1024 // len(msg) + 1)
109
110 m.captured_output_default(msg)
111 stdout, stderr = capsys.readouterr()
112 assert stdout == msg
113 assert stderr == ""
114
115
116def test_captured_utf8_4byte_offset3(capsys):
117 msg = "\U0010FFFF"
118 msg = "123" + msg * (1024 // len(msg) + 1)
119
120 m.captured_output_default(msg)
121 stdout, stderr = capsys.readouterr()
122 assert stdout == msg
123 assert stderr == ""
124
125
126def test_guard_capture(capsys):
127 msg = "I've been redirected to Python, I hope!"
128 m.guard_output(msg)
129 stdout, stderr = capsys.readouterr()
130 assert stdout == msg
131 assert stderr == ""
132
133
134def test_series_captured(capture):
135 with capture:
136 m.captured_output("a")
137 m.captured_output("b")
138 assert capture == "ab"
139
140
141def test_flush(capfd):
142 msg = "(not flushed)"
143 msg2 = "(flushed)"
144
145 with m.ostream_redirect():
146 m.noisy_function(msg, flush=False)
147 stdout, stderr = capfd.readouterr()
148 assert stdout == ""
149
150 m.noisy_function(msg2, flush=True)
151 stdout, stderr = capfd.readouterr()
152 assert stdout == msg + msg2
153
154 m.noisy_function(msg, flush=False)
155
156 stdout, stderr = capfd.readouterr()
157 assert stdout == msg
158
159
160def test_not_captured(capfd):
161 msg = "Something that should not show up in log"
162 stream = StringIO()
163 with redirect_stdout(stream):
164 m.raw_output(msg)
165 stdout, stderr = capfd.readouterr()
166 assert stdout == msg
167 assert stderr == ""
168 assert stream.getvalue() == ""
169
170 stream = StringIO()
171 with redirect_stdout(stream):
172 m.captured_output(msg)
173 stdout, stderr = capfd.readouterr()
174 assert stdout == ""
175 assert stderr == ""
176 assert stream.getvalue() == msg
177
178
179def test_err(capfd):
180 msg = "Something that should not show up in log"
181 stream = StringIO()
182 with redirect_stderr(stream):
183 m.raw_err(msg)
184 stdout, stderr = capfd.readouterr()
185 assert stdout == ""
186 assert stderr == msg
187 assert stream.getvalue() == ""
188
189 stream = StringIO()
190 with redirect_stderr(stream):
191 m.captured_err(msg)
192 stdout, stderr = capfd.readouterr()
193 assert stdout == ""
194 assert stderr == ""
195 assert stream.getvalue() == msg
196
197
198def test_multi_captured(capfd):
199 stream = StringIO()
200 with redirect_stdout(stream):
201 m.captured_output("a")
202 m.raw_output("b")
203 m.captured_output("c")
204 m.raw_output("d")
205 stdout, stderr = capfd.readouterr()
206 assert stdout == "bd"
207 assert stream.getvalue() == "ac"
208
209
210def test_dual(capsys):
211 m.captured_dual("a", "b")
212 stdout, stderr = capsys.readouterr()
213 assert stdout == "a"
214 assert stderr == "b"
215
216
217def test_redirect(capfd):
218 msg = "Should not be in log!"
219 stream = StringIO()
220 with redirect_stdout(stream):
221 m.raw_output(msg)
222 stdout, stderr = capfd.readouterr()
223 assert stdout == msg
224 assert stream.getvalue() == ""
225
226 stream = StringIO()
227 with redirect_stdout(stream):
228 with m.ostream_redirect():
229 m.raw_output(msg)
230 stdout, stderr = capfd.readouterr()
231 assert stdout == ""
232 assert stream.getvalue() == msg
233
234 stream = StringIO()
235 with redirect_stdout(stream):
236 m.raw_output(msg)
237 stdout, stderr = capfd.readouterr()
238 assert stdout == msg
239 assert stream.getvalue() == ""
240
241
242def test_redirect_err(capfd):
243 msg = "StdOut"
244 msg2 = "StdErr"
245
246 stream = StringIO()
247 with redirect_stderr(stream):
248 with m.ostream_redirect(stdout=False):
249 m.raw_output(msg)
250 m.raw_err(msg2)
251 stdout, stderr = capfd.readouterr()
252 assert stdout == msg
253 assert stderr == ""
254 assert stream.getvalue() == msg2
255
256
257def test_redirect_both(capfd):
258 msg = "StdOut"
259 msg2 = "StdErr"
260
261 stream = StringIO()
262 stream2 = StringIO()
263 with redirect_stdout(stream):
264 with redirect_stderr(stream2):
265 with m.ostream_redirect():
266 m.raw_output(msg)
267 m.raw_err(msg2)
268 stdout, stderr = capfd.readouterr()
269 assert stdout == ""
270 assert stderr == ""
271 assert stream.getvalue() == msg
272 assert stream2.getvalue() == msg2
273
274
275def test_threading():
276 with m.ostream_redirect(stdout=True, stderr=False):
277 # start some threads
278 threads = []
279
280 # start some threads
281 for _j in range(20):
282 threads.append(m.TestThread())
283
284 # give the threads some time to fail
285 threads[0].sleep()
286
287 # stop all the threads
288 for t in threads:
289 t.stop()
290
291 for t in threads:
292 t.join()
293
294 # if a thread segfaults, we don't get here
295 assert True
size_t len(handle h)
Get the length of a Python object.
Definition: pytypes.h:2002