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_guard_capture(capsys):
73    msg = "I've been redirected to Python, I hope!"
74    m.guard_output(msg)
75    stdout, stderr = capsys.readouterr()
76    assert stdout == msg
77    assert stderr == ""
78
79
80def test_series_captured(capture):
81    with capture:
82        m.captured_output("a")
83        m.captured_output("b")
84    assert capture == "ab"
85
86
87def test_flush(capfd):
88    msg = "(not flushed)"
89    msg2 = "(flushed)"
90
91    with m.ostream_redirect():
92        m.noisy_function(msg, flush=False)
93        stdout, stderr = capfd.readouterr()
94        assert stdout == ""
95
96        m.noisy_function(msg2, flush=True)
97        stdout, stderr = capfd.readouterr()
98        assert stdout == msg + msg2
99
100        m.noisy_function(msg, flush=False)
101
102    stdout, stderr = capfd.readouterr()
103    assert stdout == msg
104
105
106def test_not_captured(capfd):
107    msg = "Something that should not show up in log"
108    stream = StringIO()
109    with redirect_stdout(stream):
110        m.raw_output(msg)
111    stdout, stderr = capfd.readouterr()
112    assert stdout == msg
113    assert stderr == ""
114    assert stream.getvalue() == ""
115
116    stream = StringIO()
117    with redirect_stdout(stream):
118        m.captured_output(msg)
119    stdout, stderr = capfd.readouterr()
120    assert stdout == ""
121    assert stderr == ""
122    assert stream.getvalue() == msg
123
124
125def test_err(capfd):
126    msg = "Something that should not show up in log"
127    stream = StringIO()
128    with redirect_stderr(stream):
129        m.raw_err(msg)
130    stdout, stderr = capfd.readouterr()
131    assert stdout == ""
132    assert stderr == msg
133    assert stream.getvalue() == ""
134
135    stream = StringIO()
136    with redirect_stderr(stream):
137        m.captured_err(msg)
138    stdout, stderr = capfd.readouterr()
139    assert stdout == ""
140    assert stderr == ""
141    assert stream.getvalue() == msg
142
143
144def test_multi_captured(capfd):
145    stream = StringIO()
146    with redirect_stdout(stream):
147        m.captured_output("a")
148        m.raw_output("b")
149        m.captured_output("c")
150        m.raw_output("d")
151    stdout, stderr = capfd.readouterr()
152    assert stdout == "bd"
153    assert stream.getvalue() == "ac"
154
155
156def test_dual(capsys):
157    m.captured_dual("a", "b")
158    stdout, stderr = capsys.readouterr()
159    assert stdout == "a"
160    assert stderr == "b"
161
162
163def test_redirect(capfd):
164    msg = "Should not be in log!"
165    stream = StringIO()
166    with redirect_stdout(stream):
167        m.raw_output(msg)
168    stdout, stderr = capfd.readouterr()
169    assert stdout == msg
170    assert stream.getvalue() == ""
171
172    stream = StringIO()
173    with redirect_stdout(stream):
174        with m.ostream_redirect():
175            m.raw_output(msg)
176    stdout, stderr = capfd.readouterr()
177    assert stdout == ""
178    assert stream.getvalue() == msg
179
180    stream = StringIO()
181    with redirect_stdout(stream):
182        m.raw_output(msg)
183    stdout, stderr = capfd.readouterr()
184    assert stdout == msg
185    assert stream.getvalue() == ""
186
187
188def test_redirect_err(capfd):
189    msg = "StdOut"
190    msg2 = "StdErr"
191
192    stream = StringIO()
193    with redirect_stderr(stream):
194        with m.ostream_redirect(stdout=False):
195            m.raw_output(msg)
196            m.raw_err(msg2)
197    stdout, stderr = capfd.readouterr()
198    assert stdout == msg
199    assert stderr == ""
200    assert stream.getvalue() == msg2
201
202
203def test_redirect_both(capfd):
204    msg = "StdOut"
205    msg2 = "StdErr"
206
207    stream = StringIO()
208    stream2 = StringIO()
209    with redirect_stdout(stream):
210        with redirect_stderr(stream2):
211            with m.ostream_redirect():
212                m.raw_output(msg)
213                m.raw_err(msg2)
214    stdout, stderr = capfd.readouterr()
215    assert stdout == ""
216    assert stderr == ""
217    assert stream.getvalue() == msg
218    assert stream2.getvalue() == msg2
219