Utility for redirecting C++ streams to Python (#1009)
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 1e68a74..0d6ebfc 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -40,6 +40,7 @@
test_eval.cpp
test_exceptions.cpp
test_factory_constructors.cpp
+ test_iostream.cpp
test_kwargs_and_defaults.cpp
test_local_bindings.cpp
test_methods_and_attributes.cpp
diff --git a/tests/test_iostream.cpp b/tests/test_iostream.cpp
new file mode 100644
index 0000000..e67f88a
--- /dev/null
+++ b/tests/test_iostream.cpp
@@ -0,0 +1,73 @@
+/*
+ tests/test_iostream.cpp -- Usage of scoped_output_redirect
+
+ Copyright (c) 2017 Henry F. Schreiner
+
+ All rights reserved. Use of this source code is governed by a
+ BSD-style license that can be found in the LICENSE file.
+*/
+
+
+#include <pybind11/iostream.h>
+#include "pybind11_tests.h"
+#include <iostream>
+
+
+void noisy_function(std::string msg, bool flush) {
+
+ std::cout << msg;
+ if (flush)
+ std::cout << std::flush;
+}
+
+void noisy_funct_dual(std::string msg, std::string emsg) {
+ std::cout << msg;
+ std::cerr << emsg;
+}
+
+TEST_SUBMODULE(iostream, m) {
+
+ add_ostream_redirect(m);
+
+ // test_evals
+
+ m.def("captured_output_default", [](std::string msg) {
+ py::scoped_ostream_redirect redir;
+ std::cout << msg << std::flush;
+ });
+
+ m.def("captured_output", [](std::string msg) {
+ py::scoped_ostream_redirect redir(std::cout, py::module::import("sys").attr("stdout"));
+ std::cout << msg << std::flush;
+ });
+
+ m.def("guard_output", &noisy_function,
+ py::call_guard<py::scoped_ostream_redirect>(),
+ py::arg("msg"), py::arg("flush")=true);
+
+ m.def("captured_err", [](std::string msg) {
+ py::scoped_ostream_redirect redir(std::cerr, py::module::import("sys").attr("stderr"));
+ std::cerr << msg << std::flush;
+ });
+
+ m.def("noisy_function", &noisy_function, py::arg("msg"), py::arg("flush") = true);
+
+ m.def("dual_guard", &noisy_funct_dual,
+ py::call_guard<py::scoped_ostream_redirect, py::scoped_estream_redirect>(),
+ py::arg("msg"), py::arg("emsg"));
+
+ m.def("raw_output", [](std::string msg) {
+ std::cout << msg << std::flush;
+ });
+
+ m.def("raw_err", [](std::string msg) {
+ std::cerr << msg << std::flush;
+ });
+
+ m.def("captured_dual", [](std::string msg, std::string emsg) {
+ py::scoped_ostream_redirect redirout(std::cout, py::module::import("sys").attr("stdout"));
+ py::scoped_ostream_redirect redirerr(std::cerr, py::module::import("sys").attr("stderr"));
+ std::cout << msg << std::flush;
+ std::cerr << emsg << std::flush;
+ });
+}
diff --git a/tests/test_iostream.py b/tests/test_iostream.py
new file mode 100644
index 0000000..3364849
--- /dev/null
+++ b/tests/test_iostream.py
@@ -0,0 +1,203 @@
+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):
+ msg = "I've been redirected to Python, I hope!"
+ m.captured_output(msg)
+ stdout, stderr = capsys.readouterr()
+ assert stdout == msg
+ assert stderr == ''
+
+ m.captured_output_default(msg)
+ stdout, stderr = capsys.readouterr()
+ assert stdout == msg
+ assert stderr == ''
+
+ m.captured_err(msg)
+ stdout, stderr = capsys.readouterr()
+ assert stdout == ''
+ assert stderr == msg
+
+
+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 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 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 stderr == ''
+ assert stream.getvalue() == ''
+
+ stream = StringIO()
+ with redirect_stdout(stream):
+ m.captured_output(msg)
+ stdout, stderr = capfd.readouterr()
+ assert stdout == ''
+ assert 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 stdout == ''
+ assert stderr == msg
+ assert stream.getvalue() == ''
+
+ stream = StringIO()
+ with redirect_stderr(stream):
+ m.captured_err(msg)
+ stdout, stderr = capfd.readouterr()
+ assert stdout == ''
+ assert 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 stream.getvalue() == ''
+
+ stream = StringIO()
+ with redirect_stdout(stream):
+ with m.ostream_redirect():
+ m.raw_output(msg)
+ stdout, stderr = capfd.readouterr()
+ assert stdout == ''
+ assert stream.getvalue() == msg
+
+ stream = StringIO()
+ with redirect_stdout(stream):
+ m.raw_output(msg)
+ stdout, stderr = capfd.readouterr()
+ assert stdout == msg
+ assert stream.getvalue() == ''
+
+
+def test_redirect_err(capfd):
+ msg = "StdOut"
+ msg2 = "StdErr"
+
+ stream = StringIO()
+ with redirect_stderr(stream):
+ with m.ostream_redirect(stdout=False):
+ m.raw_output(msg)
+ m.raw_err(msg2)
+ stdout, stderr = capfd.readouterr()
+ assert stdout == msg
+ assert stderr == ''
+ assert stream.getvalue() == msg2
+
+
+def test_redirect_both(capfd):
+ msg = "StdOut"
+ msg2 = "StdErr"
+
+ stream = StringIO()
+ stream2 = StringIO()
+ with redirect_stdout(stream):
+ with redirect_stderr(stream2):
+ with m.ostream_redirect():
+ m.raw_output(msg)
+ m.raw_err(msg2)
+ stdout, stderr = capfd.readouterr()
+ assert stdout == ''
+ assert stderr == ''
+ assert stream.getvalue() == msg
+ assert stream2.getvalue() == msg2