1 /*
2 tests/test_iostream.cpp -- Usage of scoped_output_redirect
3
4 Copyright (c) 2017 Henry F. Schreiner
5
6 All rights reserved. Use of this source code is governed by a
7 BSD-style license that can be found in the LICENSE file.
8 */
9
10 #if defined(_MSC_VER) && _MSC_VER < 1910 // VS 2015's MSVC
11 # pragma warning(disable: 4702) // unreachable code in system header (xatomic.h(382))
12 #endif
13
14 #include <pybind11/iostream.h>
15 #include "pybind11_tests.h"
16 #include <atomic>
17 #include <iostream>
18 #include <mutex>
19 #include <string>
20 #include <thread>
21
noisy_function(const std::string & msg,bool flush)22 void noisy_function(const std::string &msg, bool flush) {
23
24 std::cout << msg;
25 if (flush)
26 std::cout << std::flush;
27 }
28
noisy_funct_dual(const std::string & msg,const std::string & emsg)29 void noisy_funct_dual(const std::string &msg, const std::string &emsg) {
30 std::cout << msg;
31 std::cerr << emsg;
32 }
33
34 // object to manage C++ thread
35 // simply repeatedly write to std::cerr until stopped
36 // redirect is called at some point to test the safety of scoped_estream_redirect
37 struct TestThread {
TestThreadTestThread38 TestThread() : stop_{false} {
__anon79d877d10102null39 auto thread_f = [this] {
40 static std::mutex cout_mutex;
41 while (!stop_) {
42 {
43 // #HelpAppreciated: Work on iostream.h thread safety.
44 // Without this lock, the clang ThreadSanitizer (tsan) reliably reports a
45 // data race, and this test is predictably flakey on Windows.
46 // For more background see the discussion under
47 // https://github.com/pybind/pybind11/pull/2982 and
48 // https://github.com/pybind/pybind11/pull/2995.
49 const std::lock_guard<std::mutex> lock(cout_mutex);
50 std::cout << "x" << std::flush;
51 }
52 std::this_thread::sleep_for(std::chrono::microseconds(50));
53 } };
54 t_ = new std::thread(std::move(thread_f));
55 }
56
~TestThreadTestThread57 ~TestThread() {
58 delete t_;
59 }
60
stopTestThread61 void stop() { stop_ = true; }
62
joinTestThread63 void join() const {
64 py::gil_scoped_release gil_lock;
65 t_->join();
66 }
67
sleepTestThread68 void sleep() {
69 py::gil_scoped_release gil_lock;
70 std::this_thread::sleep_for(std::chrono::milliseconds(50));
71 }
72
73 std::thread *t_{nullptr};
74 std::atomic<bool> stop_;
75 };
76
77
TEST_SUBMODULE(iostream,m)78 TEST_SUBMODULE(iostream, m) {
79
80 add_ostream_redirect(m);
81
82 // test_evals
83
84 m.def("captured_output_default", [](const std::string &msg) {
85 py::scoped_ostream_redirect redir;
86 std::cout << msg << std::flush;
87 });
88
89 m.def("captured_output", [](const std::string &msg) {
90 py::scoped_ostream_redirect redir(std::cout, py::module_::import("sys").attr("stdout"));
91 std::cout << msg << std::flush;
92 });
93
94 m.def("guard_output", &noisy_function,
95 py::call_guard<py::scoped_ostream_redirect>(),
96 py::arg("msg"), py::arg("flush")=true);
97
98 m.def("captured_err", [](const std::string &msg) {
99 py::scoped_ostream_redirect redir(std::cerr, py::module_::import("sys").attr("stderr"));
100 std::cerr << msg << std::flush;
101 });
102
103 m.def("noisy_function", &noisy_function, py::arg("msg"), py::arg("flush") = true);
104
105 m.def("dual_guard", &noisy_funct_dual,
106 py::call_guard<py::scoped_ostream_redirect, py::scoped_estream_redirect>(),
107 py::arg("msg"), py::arg("emsg"));
108
109 m.def("raw_output", [](const std::string &msg) { std::cout << msg << std::flush; });
110
111 m.def("raw_err", [](const std::string &msg) { std::cerr << msg << std::flush; });
112
113 m.def("captured_dual", [](const std::string &msg, const std::string &emsg) {
114 py::scoped_ostream_redirect redirout(std::cout, py::module_::import("sys").attr("stdout"));
115 py::scoped_ostream_redirect redirerr(std::cerr, py::module_::import("sys").attr("stderr"));
116 std::cout << msg << std::flush;
117 std::cerr << emsg << std::flush;
118 });
119
120 py::class_<TestThread>(m, "TestThread")
121 .def(py::init<>())
122 .def("stop", &TestThread::stop)
123 .def("join", &TestThread::join)
124 .def("sleep", &TestThread::sleep);
125 }
126