/usr/lib/python3/dist-packages/pyutilib/misc/tee_io.py is in python3-pyutilib 5.3.5-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | # _________________________________________________________________________
#
# PyUtilib: A Python utility library.
# Copyright (c) 2008 Sandia Corporation.
# This software is distributed under the BSD License.
# Under the terms of Contract DE-AC04-94AL85000 with Sandia Corporation,
# the U.S. Government retains certain rights in this software.
# _________________________________________________________________________
import sys
try:
from io import StringIO
except:
from StringIO import StringIO
class TeeStream(object):
"""This class implements a simple 'Tee' of the specified python
stream. Since this presents a full file interface, TeeStream
objects may be arbitrarily nested."""
def __init__(self, stream):
self.stream = stream
self.buffer = StringIO()
def write(self, data):
self.buffer.write(data)
self.stream.write(data)
def writelines(self, sequence):
for x in sequence:
self.write(x)
def reset(self):
self.buffer = StringIO()
class ConsoleBuffer(object):
"""This class implements a simple 'Tee' of the python stdout and
stderr so the output can be captured and reported programmatically.
We need a specialized class here because other applications /
methods explicitly write to sys.stderr and sys.stdout, so it is
insufficient to 'wrap' the streams like the TeeStream class;
instead, we must replace the standard stdout and stderr objects with
our own duplicator."""
class _Duplicate(object):
def __init__(self, a, b):
self.a = a
self.b = b
def write(self, data):
self.a.write(data)
self.b.write(data)
def __init__(self):
self._dup_out = self._dup_err = None
self._raw_out = sys.stdout
self._raw_err = sys.stderr
self.reset()
def __del__(self):
if self._dup_out is not None and self._dup_out is not sys.stdout:
raise RuntimeError("ConsoleBuffer: Nesting violation " \
"(attempting to delete the buffer while stdout is " \
"redirected away from this buffer).")
if self._dup_err is not None and self._dup_err is not sys.stderr:
raise RuntimeError("ConsoleBuffer: Nesting violation " \
"(attempting to delete the buffer while stderr is " \
"redirected away from this buffer).")
sys.stdout = self._raw_out
sys.stderr = self._raw_err
def reset(self):
if self._dup_out is not None and self._dup_out is not sys.stdout:
raise RuntimeError("ConsoleBuffer: Nesting violation " \
"(attempting to reset() when stdout has been redirected " \
"away from this buffer).")
if self._dup_err is not None and self._dup_err is not sys.stderr:
raise RuntimeError("ConsoleBuffer: Nesting violation " \
"(attempting to reset() when stderr has been redirected " \
"away from this buffer).")
self.out = StringIO()
self.err = StringIO()
self._dup_out = sys.stdout = \
ConsoleBuffer._Duplicate(self.out, self._raw_out)
self._dup_err = sys.stderr = \
ConsoleBuffer._Duplicate(self.err, self._raw_err)
|