Used OutputGrabber to try to capture std::cout outputs of DPCTLSyclInterface

The stream `std::cout` that the library writes to in a function like
`DPCTLDeviceMgr_PrintDeviceInfo` is not the same as Python's sys.stdout

It makes it difficult to verify that printed output is what should be
expected. The changes in this commit allow the test to pass, while ensuring
that the captured output is non-empty, but pytest must be envoked with
--capture=no option

Author: oleksandr-pavlyk

dpctl/tests/_redirector.py

@@ -0,0 +1,89 @@
+#                      Data Parallel Control (dpctl)
+#
+# Copyright 2020-2021 Intel Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+""" Defines class to capture std::cout/std::cerr outputs of loaded
+library (DPCTLSyclInterface) from Python.
+"""
+
+import os
+import sys
+
+# Source: https://stackoverflow.com/a/29834357/594376
+# License: https://creativecommons.org/licenses/by-sa/4.0/
+
+
+class OutputGrabber(object):
+    """
+    Class used to grab standard output.
+    """
+
+    escape_char = "\b"
+
+    def __init__(self, stream=None):
+        self.origstream = stream
+        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 __enter__(self):
+        self.start()
+        return self
+
+    def __exit__(self, type, value, traceback):
+        self.stop()
+
+    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)
+
+    def stop(self):
+        """
+        Stop capturing the stream data and save the text in `capturedtext`.
+        """
+        # Print the escape character to make the readOutput method stop:
+        self.origstream.write(self.escape_char)
+        # Flush the stream to make sure all our data goes in before
+        # the escape character:
+        self.origstream.flush()
+        self.readOutput()
+        # Close the pipe:
+        os.close(self.pipe_in)
+        os.close(self.pipe_out)
+        # Restore the original stream:
+        os.dup2(self.streamfd, self.origstreamfd)
+        # Close the duplicate stream:
+        os.close(self.streamfd)
+
+    def readOutput(self):
+        """
+        Read the stream data (one byte at a time)
+        and save the text in `capturedtext`.
+        """
+        while True:
+            char = os.read(self.pipe_out, 1).decode(self.origstream.encoding)
+            if not char or self.escape_char in char:
+                print(char)
+                break
+            self.capturedtext += char

dpctl/tests/test_sycl_queue.py

@@ -17,11 +17,15 @@
 """ Defines unit test cases for the SyclQueue class.
 """
 
+import ctypes
+import ctypes.util
+
 import pytest
 
 import dpctl
 
 from ._helper import create_invalid_capsule
+from ._redirector import OutputGrabber
 
 list_of_standard_selectors = [
     dpctl.select_accelerator_device,
@@ -401,16 +405,24 @@ def test_channeling_device_properties():
         dev = q.sycl_device
     except dpctl.SyclQueueCreationError:
         pytest.fail("Failed to create device from default selector")
-    import io
-    from contextlib import redirect_stdout
 
-    f1 = io.StringIO()
-    with redirect_stdout(f1):
+    libc = ctypes.cdll.LoadLibrary(ctypes.util.find_library("c"))
+
+    libc.puts(b"fff")
+    out = OutputGrabber()
+    with out:
+        libc.puts(b"fff")
+
+    q_fh = OutputGrabber()
+    with q_fh:
         q.print_device_info()  # should execute without raising
-    f2 = io.StringIO()
-    with redirect_stdout(f2):
+    d_fh = OutputGrabber()
+    with d_fh:
         dev.print_device_info()
-    assert f1.getvalue() == f2.getvalue(), "Mismatch in print_device_info"
+    q_output = q_fh.capturedtext
+    d_output = d_fh.capturedtext
+    assert q_output, "No output captured"
+    assert q_output == d_output, "Mismatch in print_device_info"
     for pr in ["backend", "name", "driver_version"]:
         assert getattr(q, pr) == getattr(
             dev, pr