"""
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
import threading
from logging.handlers import BufferingHandler
from nose2.events import Plugin
from nose2.util import ln, parse_log_level
log = logging.getLogger(__name__)
__unittest = True
[docs]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()
[docs] def startTestRun(self, event):
"""Set up logging handler"""
self._setupLoghandler()
[docs] def startTest(self, event):
"""Set up handler for new test"""
self._setupLoghandler()
[docs] def setTestOutcome(self, event):
"""Store captured log messages in ``event.metadata``"""
self._addCapturedLogs(event)
[docs] def stopTest(self, event):
"""Clear captured messages, ready for next test"""
self.handler.truncate()
[docs] 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()