mirror of
https://github.com/thelsing/knx.git
synced 2026-02-23 13:50:35 +01:00
update pybind11
This commit is contained in:
@@ -1,43 +1,9 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from __future__ import annotations
|
||||
|
||||
from contextlib import redirect_stderr, redirect_stdout
|
||||
from io import StringIO
|
||||
|
||||
from pybind11_tests import iostream as m
|
||||
import sys
|
||||
|
||||
from contextlib import contextmanager
|
||||
|
||||
try:
|
||||
# Python 3
|
||||
from io import StringIO
|
||||
except ImportError:
|
||||
# Python 2
|
||||
try:
|
||||
from cStringIO import StringIO
|
||||
except ImportError:
|
||||
from StringIO import StringIO
|
||||
|
||||
try:
|
||||
# Python 3.4
|
||||
from contextlib import redirect_stdout
|
||||
except ImportError:
|
||||
|
||||
@contextmanager
|
||||
def redirect_stdout(target):
|
||||
original = sys.stdout
|
||||
sys.stdout = target
|
||||
yield
|
||||
sys.stdout = original
|
||||
|
||||
|
||||
try:
|
||||
# Python 3.5
|
||||
from contextlib import redirect_stderr
|
||||
except ImportError:
|
||||
|
||||
@contextmanager
|
||||
def redirect_stderr(target):
|
||||
original = sys.stderr
|
||||
sys.stderr = target
|
||||
yield
|
||||
sys.stderr = original
|
||||
|
||||
|
||||
def test_captured(capsys):
|
||||
@@ -45,16 +11,16 @@ def test_captured(capsys):
|
||||
m.captured_output(msg)
|
||||
stdout, stderr = capsys.readouterr()
|
||||
assert stdout == msg
|
||||
assert stderr == ""
|
||||
assert not stderr
|
||||
|
||||
m.captured_output_default(msg)
|
||||
stdout, stderr = capsys.readouterr()
|
||||
assert stdout == msg
|
||||
assert stderr == ""
|
||||
assert not stderr
|
||||
|
||||
m.captured_err(msg)
|
||||
stdout, stderr = capsys.readouterr()
|
||||
assert stdout == ""
|
||||
assert not stdout
|
||||
assert stderr == msg
|
||||
|
||||
|
||||
@@ -66,7 +32,97 @@ def test_captured_large_string(capsys):
|
||||
m.captured_output_default(msg)
|
||||
stdout, stderr = capsys.readouterr()
|
||||
assert stdout == msg
|
||||
assert stderr == ""
|
||||
assert not stderr
|
||||
|
||||
|
||||
def test_captured_utf8_2byte_offset0(capsys):
|
||||
msg = "\u07ff"
|
||||
msg = "" + msg * (1024 // len(msg) + 1)
|
||||
|
||||
m.captured_output_default(msg)
|
||||
stdout, stderr = capsys.readouterr()
|
||||
assert stdout == msg
|
||||
assert not stderr
|
||||
|
||||
|
||||
def test_captured_utf8_2byte_offset1(capsys):
|
||||
msg = "\u07ff"
|
||||
msg = "1" + msg * (1024 // len(msg) + 1)
|
||||
|
||||
m.captured_output_default(msg)
|
||||
stdout, stderr = capsys.readouterr()
|
||||
assert stdout == msg
|
||||
assert not stderr
|
||||
|
||||
|
||||
def test_captured_utf8_3byte_offset0(capsys):
|
||||
msg = "\uffff"
|
||||
msg = "" + msg * (1024 // len(msg) + 1)
|
||||
|
||||
m.captured_output_default(msg)
|
||||
stdout, stderr = capsys.readouterr()
|
||||
assert stdout == msg
|
||||
assert not stderr
|
||||
|
||||
|
||||
def test_captured_utf8_3byte_offset1(capsys):
|
||||
msg = "\uffff"
|
||||
msg = "1" + msg * (1024 // len(msg) + 1)
|
||||
|
||||
m.captured_output_default(msg)
|
||||
stdout, stderr = capsys.readouterr()
|
||||
assert stdout == msg
|
||||
assert not stderr
|
||||
|
||||
|
||||
def test_captured_utf8_3byte_offset2(capsys):
|
||||
msg = "\uffff"
|
||||
msg = "12" + msg * (1024 // len(msg) + 1)
|
||||
|
||||
m.captured_output_default(msg)
|
||||
stdout, stderr = capsys.readouterr()
|
||||
assert stdout == msg
|
||||
assert not stderr
|
||||
|
||||
|
||||
def test_captured_utf8_4byte_offset0(capsys):
|
||||
msg = "\U0010ffff"
|
||||
msg = "" + msg * (1024 // len(msg) + 1)
|
||||
|
||||
m.captured_output_default(msg)
|
||||
stdout, stderr = capsys.readouterr()
|
||||
assert stdout == msg
|
||||
assert not stderr
|
||||
|
||||
|
||||
def test_captured_utf8_4byte_offset1(capsys):
|
||||
msg = "\U0010ffff"
|
||||
msg = "1" + msg * (1024 // len(msg) + 1)
|
||||
|
||||
m.captured_output_default(msg)
|
||||
stdout, stderr = capsys.readouterr()
|
||||
assert stdout == msg
|
||||
assert not stderr
|
||||
|
||||
|
||||
def test_captured_utf8_4byte_offset2(capsys):
|
||||
msg = "\U0010ffff"
|
||||
msg = "12" + msg * (1024 // len(msg) + 1)
|
||||
|
||||
m.captured_output_default(msg)
|
||||
stdout, stderr = capsys.readouterr()
|
||||
assert stdout == msg
|
||||
assert not stderr
|
||||
|
||||
|
||||
def test_captured_utf8_4byte_offset3(capsys):
|
||||
msg = "\U0010ffff"
|
||||
msg = "123" + msg * (1024 // len(msg) + 1)
|
||||
|
||||
m.captured_output_default(msg)
|
||||
stdout, stderr = capsys.readouterr()
|
||||
assert stdout == msg
|
||||
assert not stderr
|
||||
|
||||
|
||||
def test_guard_capture(capsys):
|
||||
@@ -74,7 +130,7 @@ def test_guard_capture(capsys):
|
||||
m.guard_output(msg)
|
||||
stdout, stderr = capsys.readouterr()
|
||||
assert stdout == msg
|
||||
assert stderr == ""
|
||||
assert not stderr
|
||||
|
||||
|
||||
def test_series_captured(capture):
|
||||
@@ -91,7 +147,7 @@ def test_flush(capfd):
|
||||
with m.ostream_redirect():
|
||||
m.noisy_function(msg, flush=False)
|
||||
stdout, stderr = capfd.readouterr()
|
||||
assert stdout == ""
|
||||
assert not stdout
|
||||
|
||||
m.noisy_function(msg2, flush=True)
|
||||
stdout, stderr = capfd.readouterr()
|
||||
@@ -110,15 +166,15 @@ def test_not_captured(capfd):
|
||||
m.raw_output(msg)
|
||||
stdout, stderr = capfd.readouterr()
|
||||
assert stdout == msg
|
||||
assert stderr == ""
|
||||
assert stream.getvalue() == ""
|
||||
assert not stderr
|
||||
assert not stream.getvalue()
|
||||
|
||||
stream = StringIO()
|
||||
with redirect_stdout(stream):
|
||||
m.captured_output(msg)
|
||||
stdout, stderr = capfd.readouterr()
|
||||
assert stdout == ""
|
||||
assert stderr == ""
|
||||
assert not stdout
|
||||
assert not stderr
|
||||
assert stream.getvalue() == msg
|
||||
|
||||
|
||||
@@ -128,16 +184,16 @@ def test_err(capfd):
|
||||
with redirect_stderr(stream):
|
||||
m.raw_err(msg)
|
||||
stdout, stderr = capfd.readouterr()
|
||||
assert stdout == ""
|
||||
assert not stdout
|
||||
assert stderr == msg
|
||||
assert stream.getvalue() == ""
|
||||
assert not stream.getvalue()
|
||||
|
||||
stream = StringIO()
|
||||
with redirect_stderr(stream):
|
||||
m.captured_err(msg)
|
||||
stdout, stderr = capfd.readouterr()
|
||||
assert stdout == ""
|
||||
assert stderr == ""
|
||||
assert not stdout
|
||||
assert not stderr
|
||||
assert stream.getvalue() == msg
|
||||
|
||||
|
||||
@@ -167,14 +223,13 @@ def test_redirect(capfd):
|
||||
m.raw_output(msg)
|
||||
stdout, stderr = capfd.readouterr()
|
||||
assert stdout == msg
|
||||
assert stream.getvalue() == ""
|
||||
assert not stream.getvalue()
|
||||
|
||||
stream = StringIO()
|
||||
with redirect_stdout(stream):
|
||||
with m.ostream_redirect():
|
||||
m.raw_output(msg)
|
||||
with redirect_stdout(stream), m.ostream_redirect():
|
||||
m.raw_output(msg)
|
||||
stdout, stderr = capfd.readouterr()
|
||||
assert stdout == ""
|
||||
assert not stdout
|
||||
assert stream.getvalue() == msg
|
||||
|
||||
stream = StringIO()
|
||||
@@ -182,7 +237,7 @@ def test_redirect(capfd):
|
||||
m.raw_output(msg)
|
||||
stdout, stderr = capfd.readouterr()
|
||||
assert stdout == msg
|
||||
assert stream.getvalue() == ""
|
||||
assert not stream.getvalue()
|
||||
|
||||
|
||||
def test_redirect_err(capfd):
|
||||
@@ -190,13 +245,12 @@ def test_redirect_err(capfd):
|
||||
msg2 = "StdErr"
|
||||
|
||||
stream = StringIO()
|
||||
with redirect_stderr(stream):
|
||||
with m.ostream_redirect(stdout=False):
|
||||
m.raw_output(msg)
|
||||
m.raw_err(msg2)
|
||||
with redirect_stderr(stream), m.ostream_redirect(stdout=False):
|
||||
m.raw_output(msg)
|
||||
m.raw_err(msg2)
|
||||
stdout, stderr = capfd.readouterr()
|
||||
assert stdout == msg
|
||||
assert stderr == ""
|
||||
assert not stderr
|
||||
assert stream.getvalue() == msg2
|
||||
|
||||
|
||||
@@ -206,13 +260,34 @@ def test_redirect_both(capfd):
|
||||
|
||||
stream = StringIO()
|
||||
stream2 = StringIO()
|
||||
with redirect_stdout(stream):
|
||||
with redirect_stderr(stream2):
|
||||
with m.ostream_redirect():
|
||||
m.raw_output(msg)
|
||||
m.raw_err(msg2)
|
||||
with redirect_stdout(stream), redirect_stderr(stream2), m.ostream_redirect():
|
||||
m.raw_output(msg)
|
||||
m.raw_err(msg2)
|
||||
stdout, stderr = capfd.readouterr()
|
||||
assert stdout == ""
|
||||
assert stderr == ""
|
||||
assert not stdout
|
||||
assert not stderr
|
||||
assert stream.getvalue() == msg
|
||||
assert stream2.getvalue() == msg2
|
||||
|
||||
|
||||
def test_threading():
|
||||
with m.ostream_redirect(stdout=True, stderr=False):
|
||||
# start some threads
|
||||
threads = []
|
||||
|
||||
# start some threads
|
||||
for _j in range(20):
|
||||
threads.append(m.TestThread())
|
||||
|
||||
# give the threads some time to fail
|
||||
threads[0].sleep()
|
||||
|
||||
# stop all the threads
|
||||
for t in threads:
|
||||
t.stop()
|
||||
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
# if a thread segfaults, we don't get here
|
||||
assert True
|
||||
|
||||
Reference in New Issue
Block a user