This file is indexed.

/usr/lib/python2.7/dist-packages/nose2/plugins/buffer.py is in python-nose2 0.5.0-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
 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