Class to capture stderr from C++ shared library.
| 1885 | def testDefaultLogDevicePlacement(self): |
| 1886 | |
| 1887 | class CaptureStderr(str): |
| 1888 | """Class to capture stderr from C++ shared library.""" |
| 1889 | |
| 1890 | def __enter__(self): |
| 1891 | self._esc = compat.as_str('\b') |
| 1892 | self._output = compat.as_str('') |
| 1893 | self._stderr = sys.stderr |
| 1894 | self._fd = self._stderr.fileno() |
| 1895 | self._out_pipe, in_pipe = os.pipe() |
| 1896 | # Save the original io stream. |
| 1897 | self._dup_fd = os.dup(self._fd) |
| 1898 | # Replace the original io stream with in pipe. |
| 1899 | os.dup2(in_pipe, self._fd) |
| 1900 | return self |
| 1901 | |
| 1902 | def __exit__(self, *args): |
| 1903 | self._stderr.write(self._esc) |
| 1904 | self._stderr.flush() |
| 1905 | self.read() |
| 1906 | os.close(self._out_pipe) |
| 1907 | # Restore the original io stream. |
| 1908 | os.dup2(self._dup_fd, self._fd) |
| 1909 | |
| 1910 | def read(self): |
| 1911 | while True: |
| 1912 | data = os.read(self._out_pipe, 1) |
| 1913 | if not data or compat.as_str(data) == self._esc: |
| 1914 | break |
| 1915 | self._output += compat.as_str(data) |
| 1916 | |
| 1917 | def __str__(self): |
| 1918 | return self._output |
| 1919 | |
| 1920 | if context.executing_eagerly(): |
| 1921 | context.set_log_device_placement(True) |
no outgoing calls