Commit bfc1cdf8 authored by Manuel Günther's avatar Manuel Günther
Browse files

Moved the cxx_disable test case to C++ (did not get it to work with Python3 otherwise)

parent e85397e3
Pipeline #4987 passed with stages
in 11 minutes and 51 seconds
......@@ -206,7 +206,8 @@ static auto _output_disable_doc = bob::extension::FunctionDoc(
"_test_output_disable",
"Writes C++ messages with and without being visible"
)
.add_prototype("")
.add_prototype("", "success")
.add_return("success", "bool", "The success of the test")
;
PyObject* output_disable(PyObject*, PyObject* args, PyObject* kwds) {
BOB_TRY
......@@ -214,27 +215,56 @@ BOB_TRY
if (!PyArg_ParseTupleAndKeywords(args, kwds, "", kwlist)) return 0;
// redirect std::cout and std::cerr, see http://stackoverflow.com/a/5419388/3301902
std::stringstream out, err;
std::streambuf * oldout = std::cout.rdbuf(out.rdbuf());
std::streambuf * olderr = std::cerr.rdbuf(err.rdbuf());
bool success = true;
try {
bob::core::log_level(bob::core::DEBUG);
bob::core::debug << "This is a debug message" << std::endl;
bob::core::info << "This is an info message" << std::endl;
bob::core::warn << "This is a warning message" << std::endl;
bob::core::error << "This is an error message" << std::endl;
success = success && out.str() == "This is a debug message\nThis is an info message\n";
success = success && err.str() == "This is a warning message\nThis is an error message\n";
out.str("");
err.str("");
bob::core::log_level(bob::core::ERROR);
bob::core::debug << "This is a debug message" << std::endl;
bob::core::info << "This is an info message" << std::endl;
bob::core::warn << "This is a warning message" << std::endl;
bob::core::error << "This is an error message" << std::endl;
success = success && out.str() == "";
success = success && err.str() == "This is an error message\n";
out.str("");
err.str("");
bob::core::log_level(bob::core::DISABLE);
bob::core::debug << "This is a debug message" << std::endl;
bob::core::info << "This is an info message" << std::endl;
bob::core::warn << "This is a warning message" << std::endl;
bob::core::error << "This is an error message" << std::endl;
success = success && out.str() == "";
success = success && err.str() == "";
} catch(...){
success = false;
}
// make sure that cout and cerr are redirected to their original streams
std::cout.rdbuf( oldout );
std::cerr.rdbuf( olderr );
bob::core::log_level(bob::core::DEBUG);
Py_RETURN_NONE;
if (success)
Py_RETURN_TRUE;
else
Py_RETURN_FALSE;
BOB_CATCH_FUNCTION("_test_output_disable", 0)
}
......
......@@ -77,102 +77,4 @@ def test_from_cxx_multithreaded():
def test_from_cxx_disable():
from bob.core._test import _test_output_disable
import sys, os
import threading
class OutputGrabber(object):
"""
Class used to grab standard output or another stream.
This is a slight modification of what was proposed here:
http://stackoverflow.com/a/29834357/3301902
"""
escape_char = "\b"
def __init__(self, stream=None, threaded=False):
self.origstream = stream
self.threaded = threaded
if self.origstream is None:
self.origstream = sys.stdout
self.origstreamfd = self.origstream.fileno()
self.capturedtext = ""
# Create a pipe so the stream can be captured:
self.pipe_out, self.pipe_in = os.pipe()
def start(self):
"""
Start capturing the stream data.
"""
self.capturedtext = ""
# Save a copy of the stream:
self.streamfd = os.dup(self.origstreamfd)
# Replace the Original stream with our write pipe
os.dup2(self.pipe_in, self.origstreamfd)
if self.threaded:
# Start thread that will read the stream:
self.workerThread = threading.Thread(target=self.readOutput)
self.workerThread.start()
# Make sure that the thread is running and os.read is executed:
time.sleep(0.01)
def stop(self):
"""
Stop capturing the stream data and save the text in `capturedtext`.
"""
# Flush the stream to make sure all our data goes in before
# the escape character.
self.origstream.flush()
# Print the escape character to make the readOutput method stop:
self.origstream.write(self.escape_char)
self.origstream.flush()
if self.threaded:
# wait until the thread finishes so we are sure that
# we have until the last character:
self.workerThread.join()
else:
self.readOutput()
# Close the pipe:
os.close(self.pipe_out)
# Restore the original stream:
os.dup2(self.streamfd, self.origstreamfd)
def readOutput(self):
"""
Read the stream data (one byte at a time)
and save the text in `capturedtext`.
"""
while True:
data = os.read(self.pipe_out, 1) # Read One Byte Only
if self.escape_char in data:
break
if not data:
break
self.capturedtext += data
try:
# redirect output and error streams
out = OutputGrabber(sys.stdout)
err = OutputGrabber(sys.stderr)
out.start()
err.start()
# run our code
_test_output_disable()
finally:
# Clean up the pipe and restore the original stdout
out.stop()
err.stop()
# output should contain two lines
outs = out.capturedtext.rstrip().split("\n")
assert len(outs) == 2
assert outs[0] == "This is a debug message"
assert outs[1] == "This is an info message"
# error should contain three lines; error message is printed twice
errs = err.capturedtext.rstrip().split("\n")
assert len(errs) == 3
assert errs[0] == "This is a warning message"
assert errs[1] == "This is an error message"
assert errs[2] == "This is an error message"
assert _test_output_disable(), "The C++ test function returned false, indicating an (unknonw) error"
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment