1from pybind11_tests import iostream as m
2import sys
3
4from contextlib import contextmanager
5
6try:
7    # Python 3
8    from io import StringIO
9except ImportError:
10    # Python 2
11    try:
12        from cStringIO import StringIO
13    except ImportError:
14        from StringIO import StringIO
15
16try:
17    # Python 3.4
18    from contextlib import redirect_stdout
19except ImportError:
20    @contextmanager
21    def redirect_stdout(target):
22        original = sys.stdout
23        sys.stdout = target
24        yield
25        sys.stdout = original
26
27try:
28    # Python 3.5
29    from contextlib import redirect_stderr
30except ImportError:
31    @contextmanager
32    def redirect_stderr(target):
33        original = sys.stderr
34        sys.stderr = target
35        yield
36        sys.stderr = original
37
38
39def test_captured(capsys):
40    msg = "I've been redirected to Python, I hope!"
41    m.captured_output(msg)
42    stdout, stderr = capsys.readouterr()
43    assert stdout == msg
44    assert stderr == ''
45
46    m.captured_output_default(msg)
47    stdout, stderr = capsys.readouterr()
48    assert stdout == msg
49    assert stderr == ''
50
51    m.captured_err(msg)
52    stdout, stderr = capsys.readouterr()
53    assert stdout == ''
54    assert stderr == msg
55
56
57def test_captured_large_string(capsys):
58    # Make this bigger than the buffer used on the C++ side: 1024 chars
59    msg = "I've been redirected to Python, I hope!"
60    msg = msg * (1024 // len(msg) + 1)
61
62    m.captured_output_default(msg)
63    stdout, stderr = capsys.readouterr()
64    assert stdout == msg
65    assert stderr == ''
66
67
68def test_guard_capture(capsys):
69    msg = "I've been redirected to Python, I hope!"
70    m.guard_output(msg)
71    stdout, stderr = capsys.readouterr()
72    assert stdout == msg
73    assert stderr == ''
74
75
76def test_series_captured(capture):
77    with capture:
78        m.captured_output("a")
79        m.captured_output("b")
80    assert capture == "ab"
81
82
83def test_flush(capfd):
84    msg = "(not flushed)"
85    msg2 = "(flushed)"
86
87    with m.ostream_redirect():
88        m.noisy_function(msg, flush=False)
89        stdout, stderr = capfd.readouterr()
90        assert stdout == ''
91
92        m.noisy_function(msg2, flush=True)
93        stdout, stderr = capfd.readouterr()
94        assert stdout == msg + msg2
95
96        m.noisy_function(msg, flush=False)
97
98    stdout, stderr = capfd.readouterr()
99    assert stdout == msg
100
101
102def test_not_captured(capfd):
103    msg = "Something that should not show up in log"
104    stream = StringIO()
105    with redirect_stdout(stream):
106        m.raw_output(msg)
107    stdout, stderr = capfd.readouterr()
108    assert stdout == msg
109    assert stderr == ''
110    assert stream.getvalue() == ''
111
112    stream = StringIO()
113    with redirect_stdout(stream):
114        m.captured_output(msg)
115    stdout, stderr = capfd.readouterr()
116    assert stdout == ''
117    assert stderr == ''
118    assert stream.getvalue() == msg
119
120
121def test_err(capfd):
122    msg = "Something that should not show up in log"
123    stream = StringIO()
124    with redirect_stderr(stream):
125        m.raw_err(msg)
126    stdout, stderr = capfd.readouterr()
127    assert stdout == ''
128    assert stderr == msg
129    assert stream.getvalue() == ''
130
131    stream = StringIO()
132    with redirect_stderr(stream):
133        m.captured_err(msg)
134    stdout, stderr = capfd.readouterr()
135    assert stdout == ''
136    assert stderr == ''
137    assert stream.getvalue() == msg
138
139
140def test_multi_captured(capfd):
141    stream = StringIO()
142    with redirect_stdout(stream):
143        m.captured_output("a")
144        m.raw_output("b")
145        m.captured_output("c")
146        m.raw_output("d")
147    stdout, stderr = capfd.readouterr()
148    assert stdout == 'bd'
149    assert stream.getvalue() == 'ac'
150
151
152def test_dual(capsys):
153    m.captured_dual("a", "b")
154    stdout, stderr = capsys.readouterr()
155    assert stdout == "a"
156    assert stderr == "b"
157
158
159def test_redirect(capfd):
160    msg = "Should not be in log!"
161    stream = StringIO()
162    with redirect_stdout(stream):
163        m.raw_output(msg)
164    stdout, stderr = capfd.readouterr()
165    assert stdout == msg
166    assert stream.getvalue() == ''
167
168    stream = StringIO()
169    with redirect_stdout(stream):
170        with m.ostream_redirect():
171            m.raw_output(msg)
172    stdout, stderr = capfd.readouterr()
173    assert stdout == ''
174    assert stream.getvalue() == msg
175
176    stream = StringIO()
177    with redirect_stdout(stream):
178        m.raw_output(msg)
179    stdout, stderr = capfd.readouterr()
180    assert stdout == msg
181    assert stream.getvalue() == ''
182
183
184def test_redirect_err(capfd):
185    msg = "StdOut"
186    msg2 = "StdErr"
187
188    stream = StringIO()
189    with redirect_stderr(stream):
190        with m.ostream_redirect(stdout=False):
191            m.raw_output(msg)
192            m.raw_err(msg2)
193    stdout, stderr = capfd.readouterr()
194    assert stdout == msg
195    assert stderr == ''
196    assert stream.getvalue() == msg2
197
198
199def test_redirect_both(capfd):
200    msg = "StdOut"
201    msg2 = "StdErr"
202
203    stream = StringIO()
204    stream2 = StringIO()
205    with redirect_stdout(stream):
206        with redirect_stderr(stream2):
207            with m.ostream_redirect():
208                m.raw_output(msg)
209                m.raw_err(msg2)
210    stdout, stderr = capfd.readouterr()
211    assert stdout == ''
212    assert stderr == ''
213    assert stream.getvalue() == msg
214    assert stream2.getvalue() == msg2
215