/usr/share/pyshared/nose2/plugins/buffer.py is in python-nose2 0.4.7-2.
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 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 | """
Buffer stdout and/or stderr during test execution, appending any
output to the error reports of failed tests.
This allows you to use print for debugging in tests without making
your test runs noisy.
This plugin implements :func:`startTest`, :func:`stopTest`,
:func:`setTestOutcome`, :func:`outcomeDetail`, :func:`beforeInteraction`
and :func:`afterInteraction` to manage capturing sys.stdout and/or
sys.stderr into buffers, attaching the buffered output to test error
report detail, and getting out of the way when other plugins want to
talk to the user.
"""
import sys
from six import StringIO
from nose2 import events
from nose2.util import ln
__unittest = True
class _Buffer(object):
def __init__(self, stream):
self._stream = stream
self._buffer = StringIO()
def fileno(self):
return self._stream.fileno()
def __getattr__(self, attr):
# this happens on unpickling
if attr == '_buffer':
raise AttributeError("No _buffer yet")
return getattr(self._buffer, attr)
def __le__(self, obj):
return self._buffer.getvalue() == obj
def __eq__(self, obj):
return self._buffer.getvalue() == obj
def __str__(self):
return self._buffer.getvalue()
def __repr__(self):
return repr(self._buffer.getvalue())
class OutputBufferPlugin(events.Plugin):
"""Buffer output during test execution"""
commandLineSwitch = ('B', 'output-buffer', 'Enable output buffer')
configSection = 'output-buffer'
def __init__(self):
self.captureStdout = self.config.as_bool('stdout', default=True)
self.captureStderr = self.config.as_bool('stderr', default=False)
self.bufStdout = self.bufStderr = None
self.realStdout = sys.__stdout__
self.realStderr = sys.__stderr__
self._disable = False
def registerInSubprocess(self, event):
event.pluginClasses.append(self.__class__)
# turn off in this process: the subproc will run the tests
self._disable = True
def startSubprocess(self, event):
self.realStdout = sys.__stdout__
self.realStderr = sys.__stderr__
def startTest(self, event):
"""Start buffering selected stream(s)"""
self._buffer()
def stopTest(self, event):
"""Stop buffering"""
self._restore()
def setTestOutcome(self, event):
"""Attach buffer(s) to event.metadata"""
if self._disable:
return
if self.captureStdout and 'stdout' not in event.metadata:
event.metadata['stdout'] = self.bufStdout
if self.captureStderr and 'stderr' not in event.metadata:
event.metadata['stderr'] = self.bufStderr
def outcomeDetail(self, event):
"""Add buffered output to event.extraDetail"""
for stream in ('stdout', 'stderr'):
if stream in event.outcomeEvent.metadata:
buf = event.outcomeEvent.metadata[stream].getvalue()
if not buf:
continue
event.extraDetail.append(
ln('>> begin captured %s <<' % stream))
event.extraDetail.append(buf)
event.extraDetail.append(ln('>> end captured %s <<' % stream))
def beforeInteraction(self, event):
"""Stop buffering so users can see stdout"""
self._restore()
def afterInteraction(self, event):
"""Start buffering again (does not clear buffers)"""
self._buffer(fresh=False)
def stopSubprocess(self, event):
self._restore()
def _restore(self):
if self._disable:
return
if self.captureStdout:
sys.stdout = self.realStdout
if self.captureStderr:
sys.stderr = self.realStderr
def _buffer(self, fresh=True):
if self._disable:
return
if self.captureStdout:
if fresh or self.bufStdout is None:
self.bufStdout = _Buffer(sys.stdout)
sys.stdout = self.bufStdout
if self.captureStderr:
if fresh or self.bufStderr is None:
self.bufStderr = _Buffer(sys.stderr)
sys.stderr = self.bufStderr
|