pybind11-2.10.3/tests/test_iostream.py

0001 from contextlib import redirect_stderr, redirect_stdout
0002 from io import StringIO
0003
0004 from pybind11_tests import iostream as m
0005
0006
0007 def test_captured(capsys):
0008     msg = "I've been redirected to Python, I hope!"
0009     m.captured_output(msg)
0010     stdout, stderr = capsys.readouterr()
0011     assert stdout == msg
0012     assert stderr == ""
0013
0014     m.captured_output_default(msg)
0015     stdout, stderr = capsys.readouterr()
0016     assert stdout == msg
0017     assert stderr == ""
0018
0019     m.captured_err(msg)
0020     stdout, stderr = capsys.readouterr()
0021     assert stdout == ""
0022     assert stderr == msg
0023
0024
0025 def test_captured_large_string(capsys):
0026     # Make this bigger than the buffer used on the C++ side: 1024 chars
0027     msg = "I've been redirected to Python, I hope!"
0028     msg = msg * (1024 // len(msg) + 1)
0029
0030     m.captured_output_default(msg)
0031     stdout, stderr = capsys.readouterr()
0032     assert stdout == msg
0033     assert stderr == ""
0034
0035
0036 def test_captured_utf8_2byte_offset0(capsys):
0037     msg = "\u07FF"
0038     msg = "" + msg * (1024 // len(msg) + 1)
0039
0040     m.captured_output_default(msg)
0041     stdout, stderr = capsys.readouterr()
0042     assert stdout == msg
0043     assert stderr == ""
0044
0045
0046 def test_captured_utf8_2byte_offset1(capsys):
0047     msg = "\u07FF"
0048     msg = "1" + msg * (1024 // len(msg) + 1)
0049
0050     m.captured_output_default(msg)
0051     stdout, stderr = capsys.readouterr()
0052     assert stdout == msg
0053     assert stderr == ""
0054
0055
0056 def test_captured_utf8_3byte_offset0(capsys):
0057     msg = "\uFFFF"
0058     msg = "" + msg * (1024 // len(msg) + 1)
0059
0060     m.captured_output_default(msg)
0061     stdout, stderr = capsys.readouterr()
0062     assert stdout == msg
0063     assert stderr == ""
0064
0065
0066 def test_captured_utf8_3byte_offset1(capsys):
0067     msg = "\uFFFF"
0068     msg = "1" + msg * (1024 // len(msg) + 1)
0069
0070     m.captured_output_default(msg)
0071     stdout, stderr = capsys.readouterr()
0072     assert stdout == msg
0073     assert stderr == ""
0074
0075
0076 def test_captured_utf8_3byte_offset2(capsys):
0077     msg = "\uFFFF"
0078     msg = "12" + msg * (1024 // len(msg) + 1)
0079
0080     m.captured_output_default(msg)
0081     stdout, stderr = capsys.readouterr()
0082     assert stdout == msg
0083     assert stderr == ""
0084
0085
0086 def test_captured_utf8_4byte_offset0(capsys):
0087     msg = "\U0010FFFF"
0088     msg = "" + msg * (1024 // len(msg) + 1)
0089
0090     m.captured_output_default(msg)
0091     stdout, stderr = capsys.readouterr()
0092     assert stdout == msg
0093     assert stderr == ""
0094
0095
0096 def test_captured_utf8_4byte_offset1(capsys):
0097     msg = "\U0010FFFF"
0098     msg = "1" + msg * (1024 // len(msg) + 1)
0099
0100     m.captured_output_default(msg)
0101     stdout, stderr = capsys.readouterr()
0102     assert stdout == msg
0103     assert stderr == ""
0104
0105
0106 def test_captured_utf8_4byte_offset2(capsys):
0107     msg = "\U0010FFFF"
0108     msg = "12" + msg * (1024 // len(msg) + 1)
0109
0110     m.captured_output_default(msg)
0111     stdout, stderr = capsys.readouterr()
0112     assert stdout == msg
0113     assert stderr == ""
0114
0115
0116 def test_captured_utf8_4byte_offset3(capsys):
0117     msg = "\U0010FFFF"
0118     msg = "123" + msg * (1024 // len(msg) + 1)
0119
0120     m.captured_output_default(msg)
0121     stdout, stderr = capsys.readouterr()
0122     assert stdout == msg
0123     assert stderr == ""
0124
0125
0126 def test_guard_capture(capsys):
0127     msg = "I've been redirected to Python, I hope!"
0128     m.guard_output(msg)
0129     stdout, stderr = capsys.readouterr()
0130     assert stdout == msg
0131     assert stderr == ""
0132
0133
0134 def test_series_captured(capture):
0135     with capture:
0136         m.captured_output("a")
0137         m.captured_output("b")
0138     assert capture == "ab"
0139
0140
0141 def test_flush(capfd):
0142     msg = "(not flushed)"
0143     msg2 = "(flushed)"
0144
0145     with m.ostream_redirect():
0146         m.noisy_function(msg, flush=False)
0147         stdout, stderr = capfd.readouterr()
0148         assert stdout == ""
0149
0150         m.noisy_function(msg2, flush=True)
0151         stdout, stderr = capfd.readouterr()
0152         assert stdout == msg + msg2
0153
0154         m.noisy_function(msg, flush=False)
0155
0156     stdout, stderr = capfd.readouterr()
0157     assert stdout == msg
0158
0159
0160 def test_not_captured(capfd):
0161     msg = "Something that should not show up in log"
0162     stream = StringIO()
0163     with redirect_stdout(stream):
0164         m.raw_output(msg)
0165     stdout, stderr = capfd.readouterr()
0166     assert stdout == msg
0167     assert stderr == ""
0168     assert stream.getvalue() == ""
0169
0170     stream = StringIO()
0171     with redirect_stdout(stream):
0172         m.captured_output(msg)
0173     stdout, stderr = capfd.readouterr()
0174     assert stdout == ""
0175     assert stderr == ""
0176     assert stream.getvalue() == msg
0177
0178
0179 def test_err(capfd):
0180     msg = "Something that should not show up in log"
0181     stream = StringIO()
0182     with redirect_stderr(stream):
0183         m.raw_err(msg)
0184     stdout, stderr = capfd.readouterr()
0185     assert stdout == ""
0186     assert stderr == msg
0187     assert stream.getvalue() == ""
0188
0189     stream = StringIO()
0190     with redirect_stderr(stream):
0191         m.captured_err(msg)
0192     stdout, stderr = capfd.readouterr()
0193     assert stdout == ""
0194     assert stderr == ""
0195     assert stream.getvalue() == msg
0196
0197
0198 def test_multi_captured(capfd):
0199     stream = StringIO()
0200     with redirect_stdout(stream):
0201         m.captured_output("a")
0202         m.raw_output("b")
0203         m.captured_output("c")
0204         m.raw_output("d")
0205     stdout, stderr = capfd.readouterr()
0206     assert stdout == "bd"
0207     assert stream.getvalue() == "ac"
0208
0209
0210 def test_dual(capsys):
0211     m.captured_dual("a", "b")
0212     stdout, stderr = capsys.readouterr()
0213     assert stdout == "a"
0214     assert stderr == "b"
0215
0216
0217 def test_redirect(capfd):
0218     msg = "Should not be in log!"
0219     stream = StringIO()
0220     with redirect_stdout(stream):
0221         m.raw_output(msg)
0222     stdout, stderr = capfd.readouterr()
0223     assert stdout == msg
0224     assert stream.getvalue() == ""
0225
0226     stream = StringIO()
0227     with redirect_stdout(stream):
0228         with m.ostream_redirect():
0229             m.raw_output(msg)
0230     stdout, stderr = capfd.readouterr()
0231     assert stdout == ""
0232     assert stream.getvalue() == msg
0233
0234     stream = StringIO()
0235     with redirect_stdout(stream):
0236         m.raw_output(msg)
0237     stdout, stderr = capfd.readouterr()
0238     assert stdout == msg
0239     assert stream.getvalue() == ""
0240
0241
0242 def test_redirect_err(capfd):
0243     msg = "StdOut"
0244     msg2 = "StdErr"
0245
0246     stream = StringIO()
0247     with redirect_stderr(stream):
0248         with m.ostream_redirect(stdout=False):
0249             m.raw_output(msg)
0250             m.raw_err(msg2)
0251     stdout, stderr = capfd.readouterr()
0252     assert stdout == msg
0253     assert stderr == ""
0254     assert stream.getvalue() == msg2
0255
0256
0257 def test_redirect_both(capfd):
0258     msg = "StdOut"
0259     msg2 = "StdErr"
0260
0261     stream = StringIO()
0262     stream2 = StringIO()
0263     with redirect_stdout(stream):
0264         with redirect_stderr(stream2):
0265             with m.ostream_redirect():
0266                 m.raw_output(msg)
0267                 m.raw_err(msg2)
0268     stdout, stderr = capfd.readouterr()
0269     assert stdout == ""
0270     assert stderr == ""
0271     assert stream.getvalue() == msg
0272     assert stream2.getvalue() == msg2
0273
0274
0275 def test_threading():
0276     with m.ostream_redirect(stdout=True, stderr=False):
0277         # start some threads
0278         threads = []
0279
0280         # start some threads
0281         for _j in range(20):
0282             threads.append(m.TestThread())
0283
0284         # give the threads some time to fail
0285         threads[0].sleep()
0286
0287         # stop all the threads
0288         for t in threads:
0289             t.stop()
0290
0291         for t in threads:
0292             t.join()
0293
0294         # if a thread segfaults, we don't get here
0295         assert True