pre-commit-ci[bot]pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>Henry Schreiner* chore(deps): update pre-commit hooks updates: - [github.com/pre-commit/mirrors-clang-format: v22.1.5 → v22.1.8](https://github.com/pre-commit/mirrors-clang-format/compare/v22.1.5...v22.1.8) - [github.com/astral-sh/ruff-pre-commit: v0.15.20 → v0.16.1](https://github.com/astral-sh/ruff-pre-commit/compare/v0.15.20...v0.16.1) - [github.com/pre-commit/mirrors-mypy: v2.1.0 → v2.3.0](https://github.com/pre-commit/mirrors-mypy/compare/v2.1.0...v2.3.0) - [github.com/codespell-project/codespell: v2.4.2 → v2.4.3](https://github.com/codespell-project/codespell/compare/v2.4.2...v2.4.3) * style: pre-commit fixes * fix: adapt to the ruff 0.16 default rule set The hook update pulled in ruff 0.16, which expanded its default rules and autofixed `self: S -> S` to `-> Self`. That fix added a runtime `typing_extensions` import to setup_helpers.py, which must stay dependency-free; it broke every test job whose environment does not supply that package. The import is now guarded by TYPE_CHECKING. Also replace the `exec` in docs/conf.py with an importlib module load, parenthesize implicit string concatenations in list literals, mark unused unpacked variables with a leading underscore, and make noxfile.py executable to match its shebang. The remaining new rules are silenced per-file, with reasons: the chrono tests check naive local time on purpose, and the other rules conflict with how the tests are written. Assisted-by: ClaudeCode:claude-opus-5 --------- Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: Henry Schreiner <henryfs@princeton.edu>
330 lines
8.0 KiB
Python
330 lines
8.0 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
from contextlib import redirect_stderr, redirect_stdout
|
|
from io import StringIO
|
|
|
|
import pytest
|
|
|
|
import env
|
|
from pybind11_tests import iostream as m
|
|
|
|
if env.WIN:
|
|
wv_build = sys.getwindowsversion().build
|
|
skip_if_ge = 26100
|
|
if wv_build >= skip_if_ge:
|
|
pytest.skip(
|
|
f"Windows build {wv_build} >= {skip_if_ge}:"
|
|
" Skipping iostream capture (redirection regression needs investigation)",
|
|
allow_module_level=True,
|
|
)
|
|
|
|
|
|
def test_captured(capsys):
|
|
msg = "I've been redirected to Python, I hope!"
|
|
m.captured_output(msg)
|
|
stdout, stderr = capsys.readouterr()
|
|
assert stdout == msg
|
|
assert not stderr
|
|
|
|
m.captured_output_default(msg)
|
|
stdout, stderr = capsys.readouterr()
|
|
assert stdout == msg
|
|
assert not stderr
|
|
|
|
m.captured_err(msg)
|
|
stdout, stderr = capsys.readouterr()
|
|
assert not stdout
|
|
assert stderr == msg
|
|
|
|
|
|
def test_captured_large_string(capsys):
|
|
# Make this bigger than the buffer used on the C++ side: 1024 chars
|
|
msg = "I've been redirected to Python, I hope!"
|
|
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_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):
|
|
msg = "I've been redirected to Python, I hope!"
|
|
m.guard_output(msg)
|
|
stdout, stderr = capsys.readouterr()
|
|
assert stdout == msg
|
|
assert not stderr
|
|
|
|
|
|
def test_series_captured(capture):
|
|
with capture:
|
|
m.captured_output("a")
|
|
m.captured_output("b")
|
|
assert capture == "ab"
|
|
|
|
|
|
def test_flush(capfd):
|
|
msg = "(not flushed)"
|
|
msg2 = "(flushed)"
|
|
|
|
with m.ostream_redirect():
|
|
m.noisy_function(msg, flush=False)
|
|
stdout, _stderr = capfd.readouterr()
|
|
assert not stdout
|
|
|
|
m.noisy_function(msg2, flush=True)
|
|
stdout, _stderr = capfd.readouterr()
|
|
assert stdout == msg + msg2
|
|
|
|
m.noisy_function(msg, flush=False)
|
|
|
|
stdout, _stderr = capfd.readouterr()
|
|
assert stdout == msg
|
|
|
|
|
|
def test_not_captured(capfd):
|
|
msg = "Something that should not show up in log"
|
|
stream = StringIO()
|
|
with redirect_stdout(stream):
|
|
m.raw_output(msg)
|
|
stdout, stderr = capfd.readouterr()
|
|
assert stdout == msg
|
|
assert not stderr
|
|
assert not stream.getvalue()
|
|
|
|
stream = StringIO()
|
|
with redirect_stdout(stream):
|
|
m.captured_output(msg)
|
|
stdout, stderr = capfd.readouterr()
|
|
assert not stdout
|
|
assert not stderr
|
|
assert stream.getvalue() == msg
|
|
|
|
|
|
def test_err(capfd):
|
|
msg = "Something that should not show up in log"
|
|
stream = StringIO()
|
|
with redirect_stderr(stream):
|
|
m.raw_err(msg)
|
|
stdout, stderr = capfd.readouterr()
|
|
assert not stdout
|
|
assert stderr == msg
|
|
assert not stream.getvalue()
|
|
|
|
stream = StringIO()
|
|
with redirect_stderr(stream):
|
|
m.captured_err(msg)
|
|
stdout, stderr = capfd.readouterr()
|
|
assert not stdout
|
|
assert not stderr
|
|
assert stream.getvalue() == msg
|
|
|
|
|
|
def test_multi_captured(capfd):
|
|
stream = StringIO()
|
|
with redirect_stdout(stream):
|
|
m.captured_output("a")
|
|
m.raw_output("b")
|
|
m.captured_output("c")
|
|
m.raw_output("d")
|
|
stdout, _stderr = capfd.readouterr()
|
|
assert stdout == "bd"
|
|
assert stream.getvalue() == "ac"
|
|
|
|
|
|
def test_dual(capsys):
|
|
m.captured_dual("a", "b")
|
|
stdout, stderr = capsys.readouterr()
|
|
assert stdout == "a"
|
|
assert stderr == "b"
|
|
|
|
|
|
def test_redirect(capfd):
|
|
msg = "Should not be in log!"
|
|
stream = StringIO()
|
|
with redirect_stdout(stream):
|
|
m.raw_output(msg)
|
|
stdout, _stderr = capfd.readouterr()
|
|
assert stdout == msg
|
|
assert not stream.getvalue()
|
|
|
|
stream = StringIO()
|
|
with redirect_stdout(stream), m.ostream_redirect():
|
|
m.raw_output(msg)
|
|
stdout, _stderr = capfd.readouterr()
|
|
assert not stdout
|
|
assert stream.getvalue() == msg
|
|
|
|
stream = StringIO()
|
|
with redirect_stdout(stream):
|
|
m.raw_output(msg)
|
|
stdout, _stderr = capfd.readouterr()
|
|
assert stdout == msg
|
|
assert not stream.getvalue()
|
|
|
|
|
|
def test_redirect_err(capfd):
|
|
msg = "StdOut"
|
|
msg2 = "StdErr"
|
|
|
|
stream = StringIO()
|
|
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 not stderr
|
|
assert stream.getvalue() == msg2
|
|
|
|
|
|
def test_redirect_both(capfd):
|
|
msg = "StdOut"
|
|
msg2 = "StdErr"
|
|
|
|
stream = StringIO()
|
|
stream2 = StringIO()
|
|
with redirect_stdout(stream), redirect_stderr(stream2), m.ostream_redirect():
|
|
m.raw_output(msg)
|
|
m.raw_err(msg2)
|
|
stdout, stderr = capfd.readouterr()
|
|
assert not stdout
|
|
assert not stderr
|
|
assert stream.getvalue() == msg
|
|
assert stream2.getvalue() == msg2
|
|
|
|
|
|
def test_move_redirect(capsys):
|
|
m.move_redirect_output("before_move", "after_move")
|
|
stdout, stderr = capsys.readouterr()
|
|
assert stdout == "before_moveafter_move"
|
|
assert not stderr
|
|
|
|
|
|
def test_move_redirect_unflushed(capsys):
|
|
m.move_redirect_output_unflushed("before_move", "after_move")
|
|
stdout, stderr = capsys.readouterr()
|
|
assert stdout == "before_moveafter_move"
|
|
assert not stderr
|
|
|
|
|
|
def test_move_redirect_null_rdbuf(capsys):
|
|
m.move_redirect_null_rdbuf("hello")
|
|
stdout, stderr = capsys.readouterr()
|
|
assert stdout == "hellohello"
|
|
assert not stderr
|
|
|
|
|
|
def test_null_rdbuf_restored():
|
|
assert m.get_null_rdbuf_restored("test")
|
|
|
|
|
|
@pytest.mark.skipif(sys.platform.startswith("emscripten"), reason="Requires threads")
|
|
def test_threading():
|
|
with m.ostream_redirect(stdout=True, stderr=False):
|
|
# start some threads
|
|
threads = [m.TestThread() for _j in range(20)]
|
|
|
|
# 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
|