From bfc1cdf8f8541d8ef8c4ff88382d3d28eae815a7 Mon Sep 17 00:00:00 2001
From: Manuel Gunther <siebenkopf@googlemail.com>
Date: Wed, 19 Oct 2016 18:42:54 -0600
Subject: [PATCH] Moved the cxx_disable test case to C++ (did not get it to
work with Python3 otherwise)
---
bob/core/test.cpp | 68 ++++++++++++++++++--------
bob/core/test_logging.py | 100 +--------------------------------------
2 files changed, 50 insertions(+), 118 deletions(-)
diff --git a/bob/core/test.cpp b/bob/core/test.cpp
index 6554760..ed6cd4d 100644
--- a/bob/core/test.cpp
+++ b/bob/core/test.cpp
@@ -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;
- 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;
-
- 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;
-
- 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;
+ // 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)
}
diff --git a/bob/core/test_logging.py b/bob/core/test_logging.py
index 2f27288..71d35f5 100644
--- a/bob/core/test_logging.py
+++ b/bob/core/test_logging.py
@@ -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"
--
GitLab