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