1# -*- coding: utf-8 -*- 2import sys 3from contextlib import contextmanager 4 5from pybind11_tests import iostream as m 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