2
This plugin captures logging statements issued during test execution. When an
3
error or failure occurs, the captured log messages are attached to the running
4
test in the test.capturedLogging attribute, and displayed with the error failure
5
output. It is enabled by default but can be turned off with the option
8
You can filter captured logging statements with the ``--logging-filter`` option.
9
If set, it specifies which logger(s) will be captured; loggers that do not match
10
will be passed. Example: specifying ``--logging-filter=sqlalchemy,myapp``
11
will ensure that only statements logged via sqlalchemy.engine, myapp
12
or myapp.foo.bar logger will be logged.
14
You can remove other installed logging handlers with the
15
``--logging-clear-handlers`` option.
19
from logging.handlers import BufferingHandler
22
from nose.plugins.base import Plugin
23
from nose.util import anyp, ln, safe_str
26
from cStringIO import StringIO
28
from StringIO import StringIO
30
log = logging.getLogger(__name__)
32
class FilterSet(object):
33
def __init__(self, filter_components):
34
self.inclusive, self.exclusive = self._partition(filter_components)
37
def _partition(components):
38
inclusive, exclusive = [], []
39
for component in components:
40
if component.startswith('-'):
41
exclusive.append(component[1:])
43
inclusive.append(component)
44
return inclusive, exclusive
45
_partition = staticmethod(_partition)
47
def allow(self, record):
48
"""returns whether this record should be printed"""
52
return self._allow(record) and not self._deny(record)
55
def _any_match(matchers, record):
56
"""return the bool of whether `record` starts with
57
any item in `matchers`"""
58
def record_matches_key(key):
59
return record == key or record.startswith(key + '.')
60
return anyp(bool, map(record_matches_key, matchers))
61
_any_match = staticmethod(_any_match)
63
def _allow(self, record):
64
if not self.inclusive:
66
return self._any_match(self.inclusive, record)
68
def _deny(self, record):
69
if not self.exclusive:
71
return self._any_match(self.exclusive, record)
74
class MyMemoryHandler(BufferingHandler):
75
def __init__(self, capacity, logformat, logdatefmt, filters):
76
BufferingHandler.__init__(self, capacity)
77
fmt = logging.Formatter(logformat, logdatefmt)
78
self.setFormatter(fmt)
79
self.filterset = FilterSet(filters)
84
def filter(self, record):
85
return self.filterset.allow(record.name)
86
def __getstate__(self):
87
state = self.__dict__.copy()
90
def __setstate__(self, state):
91
self.__dict__.update(state)
92
self.lock = threading.RLock()
95
class LogCapture(Plugin):
97
Log capture plugin. Enabled by default. Disable with --nologcapture.
98
This plugin captures logging statements issued during test execution,
99
appending any output captured to the error or failure output,
100
should the test fail or raise an error.
103
env_opt = 'NOSE_NOLOGCAPTURE'
106
logformat = '%(name)s: %(levelname)s: %(message)s'
111
def options(self, parser, env):
112
"""Register commandline options.
115
"--nologcapture", action="store_false",
116
default=not env.get(self.env_opt), dest="logcapture",
117
help="Disable logging capture plugin. "
118
"Logging configurtion will be left intact."
119
" [NOSE_NOLOGCAPTURE]")
121
"--logging-format", action="store", dest="logcapture_format",
122
default=env.get('NOSE_LOGFORMAT') or self.logformat,
124
help="Specify custom format to print statements. "
125
"Uses the same format as used by standard logging handlers."
128
"--logging-datefmt", action="store", dest="logcapture_datefmt",
129
default=env.get('NOSE_LOGDATEFMT') or self.logdatefmt,
131
help="Specify custom date/time format to print statements. "
132
"Uses the same format as used by standard logging handlers."
133
" [NOSE_LOGDATEFMT]")
135
"--logging-filter", action="store", dest="logcapture_filters",
136
default=env.get('NOSE_LOGFILTER'),
138
help="Specify which statements to filter in/out. "
139
"By default, everything is captured. If the output is too"
140
" verbose,\nuse this option to filter out needless output.\n"
141
"Example: filter=foo will capture statements issued ONLY to\n"
142
" foo or foo.what.ever.sub but not foobar or other logger.\n"
143
"Specify multiple loggers with comma: filter=foo,bar,baz.\n"
144
"If any logger name is prefixed with a minus, eg filter=-foo,\n"
145
"it will be excluded rather than included. Default: "
146
"exclude logging messages from nose itself (-nose)."
147
" [NOSE_LOGFILTER]\n")
149
"--logging-clear-handlers", action="store_true",
150
default=False, dest="logcapture_clear",
151
help="Clear all other logging handlers")
153
def configure(self, options, conf):
157
# Disable if explicitly disabled, or if logging is
158
# configured via logging config file
159
if not options.logcapture or conf.loggingConfig:
161
self.logformat = options.logcapture_format
162
self.logdatefmt = options.logcapture_datefmt
163
self.clear = options.logcapture_clear
164
if options.logcapture_filters:
165
self.filters = options.logcapture_filters.split(',')
167
def setupLoghandler(self):
168
# setup our handler with root logger
169
root_logger = logging.getLogger()
171
if hasattr(root_logger, "handlers"):
172
for handler in root_logger.handlers:
173
root_logger.removeHandler(handler)
174
for logger in logging.Logger.manager.loggerDict.values():
175
if hasattr(logger, "handlers"):
176
for handler in logger.handlers:
177
logger.removeHandler(handler)
178
# make sure there isn't one already
179
# you can't simply use "if self.handler not in root_logger.handlers"
180
# since at least in unit tests this doesn't work --
181
# LogCapture() is instantiated for each test case while root_logger
183
# so we always add new MyMemoryHandler instance
184
for handler in root_logger.handlers[:]:
185
if isinstance(handler, MyMemoryHandler):
186
root_logger.handlers.remove(handler)
187
root_logger.addHandler(self.handler)
188
# to make sure everything gets captured
189
root_logger.setLevel(logging.NOTSET)
192
"""Set up logging handler before test run begins.
197
self.handler = MyMemoryHandler(1000, self.logformat, self.logdatefmt,
199
self.setupLoghandler()
204
def beforeTest(self, test):
205
"""Clear buffers and handlers before test.
207
self.setupLoghandler()
209
def afterTest(self, test):
210
"""Clear buffers after test.
212
self.handler.truncate()
214
def formatFailure(self, test, err):
215
"""Add captured log messages to failure output.
217
return self.formatError(test, err)
219
def formatError(self, test, err):
220
"""Add captured log messages to error output.
222
# logic flow copied from Capture.formatError
223
test.capturedLogging = records = self.formatLogRecords()
227
return (ec, self.addCaptureToErr(ev, records), tb)
229
def formatLogRecords(self):
230
format = self.handler.format
231
return [safe_str(format(r)) for r in self.handler.buffer]
233
def addCaptureToErr(self, ev, records):
234
return '\n'.join([safe_str(ev), ln('>> begin captured logging <<')] + \
236
[ln('>> end captured logging <<')])