/usr/lib/python2.7/dist-packages/nose2/plugins/logcapture.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 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 | """
Capture log messages during test execution, appending them to the
error reports of failed tests.
This plugin implements :func:`startTestRun`, :func:`startTest`,
:func:`stopTest`, :func:`setTestOutcome`, and :func:`outcomeDetail` to
set up a logging configuration that captures log messages during test
execution, and appends them to error reports for tests that fail or
raise exceptions.
"""
import logging
from logging.handlers import BufferingHandler
import threading
from nose2.events import Plugin
from nose2.util import ln, parse_log_level
log = logging.getLogger(__name__)
__unittest = True
class LogCapture(Plugin):
"""Capture log messages during test execution"""
configSection = 'log-capture'
commandLineSwitch = (None, 'log-capture', 'Enable log capture')
logformat = '%(name)s: %(levelname)s: %(message)s'
logdatefmt = None
clear = False
filters = ['-nose']
def __init__(self):
self.logformat = self.config.as_str('format', self.logformat)
self.logdatefmt = self.config.as_str('date-format', self.logdatefmt)
self.filters = self.config.as_list('filter', self.filters)
self.clear = self.config.as_bool('clear-handlers', self.clear)
self.loglevel = parse_log_level(
self.config.as_str('log-level', 'NOTSET'))
self.handler = MyMemoryHandler(1000, self.logformat, self.logdatefmt,
self.filters)
def registerInSubprocess(self, event):
event.pluginClasses.append(self.__class__)
def startSubprocess(self, event):
self._setupLoghandler()
def startTestRun(self, event):
"""Set up logging handler"""
self._setupLoghandler()
def startTest(self, event):
"""Set up handler for new test"""
self._setupLoghandler()
def setTestOutcome(self, event):
"""Store captured log messages in ``event.metadata``"""
self._addCapturedLogs(event)
def stopTest(self, event):
"""Clear captured messages, ready for next test"""
self.handler.truncate()
def outcomeDetail(self, event):
"""Append captured log messages to ``event.extraDetail``"""
logs = event.outcomeEvent.metadata.get('logs', None)
if logs:
event.extraDetail.append(ln('>> begin captured logging <<'))
event.extraDetail.extend(logs)
event.extraDetail.append(ln('>> end captured logging <<'))
def _setupLoghandler(self):
# setup our handler with root logger
root_logger = logging.getLogger()
if self.clear:
if hasattr(root_logger, "handlers"):
for handler in root_logger.handlers:
root_logger.removeHandler(handler)
for logger in logging.Logger.manager.loggerDict.values():
if hasattr(logger, "handlers"):
for handler in logger.handlers:
logger.removeHandler(handler)
# make sure there isn't one already
# you can't simply use "if self.handler not in root_logger.handlers"
# since at least in unit tests this doesn't work --
# LogCapture() is instantiated for each test case while root_logger
# is module global
# so we always add new MyMemoryHandler instance
for handler in root_logger.handlers[:]:
if isinstance(handler, MyMemoryHandler):
root_logger.handlers.remove(handler)
root_logger.addHandler(self.handler)
root_logger.setLevel(self.loglevel)
def _addCapturedLogs(self, event):
format = self.handler.format
records = [format(r) for r in self.handler.buffer]
if 'logs' in event.metadata:
event.metadata['logs'].extend(records)
else:
event.metadata['logs'] = records
class FilterSet(object):
def __init__(self, filter_components):
self.inclusive, self.exclusive = self._partition(filter_components)
@staticmethod
def _partition(components):
inclusive, exclusive = [], []
for component in components:
if component.startswith('-'):
exclusive.append(component[1:])
else:
inclusive.append(component)
return inclusive, exclusive
def allow(self, record):
"""returns whether this record should be printed"""
if not self:
# nothing to filter
return True
return self._allow(record) and not self._deny(record)
@staticmethod
def _any_match(matchers, record):
"""return the bool of whether `record` starts with
any item in `matchers`"""
def record_matches_key(key):
return record == key or record.startswith(key + '.')
return any(map(record_matches_key, matchers))
def _allow(self, record):
if not self.inclusive:
return True
return self._any_match(self.inclusive, record)
def _deny(self, record):
if not self.exclusive:
return False
return self._any_match(self.exclusive, record)
class MyMemoryHandler(BufferingHandler):
def __init__(self, capacity, logformat, logdatefmt, filters):
BufferingHandler.__init__(self, capacity)
fmt = logging.Formatter(logformat, logdatefmt)
self.setFormatter(fmt)
self.filterset = FilterSet(filters)
def flush(self):
pass # do nothing
def truncate(self):
self.buffer = []
def filter(self, record):
return self.filterset.allow(record.name)
def emit(self, record):
# take a snapshot of the potentially mutable arguments
record.msg = record.getMessage()
record.args = {}
BufferingHandler.emit(self, record)
def __getstate__(self):
state = self.__dict__.copy()
del state['lock']
return state
def __setstate__(self, state):
self.__dict__.update(state)
self.lock = threading.RLock()
|