339 lines
11 KiB
Python
339 lines
11 KiB
Python
from __future__ import absolute_import, division, print_function
|
|
|
|
import logging
|
|
from contextlib import closing, contextmanager
|
|
import sys
|
|
import six
|
|
|
|
import pytest
|
|
import py
|
|
|
|
|
|
DEFAULT_LOG_FORMAT = '%(filename)-25s %(lineno)4d %(levelname)-8s %(message)s'
|
|
DEFAULT_LOG_DATE_FORMAT = '%H:%M:%S'
|
|
|
|
|
|
def get_option_ini(config, *names):
|
|
for name in names:
|
|
ret = config.getoption(name) # 'default' arg won't work as expected
|
|
if ret is None:
|
|
ret = config.getini(name)
|
|
if ret:
|
|
return ret
|
|
|
|
|
|
def pytest_addoption(parser):
|
|
"""Add options to control log capturing."""
|
|
group = parser.getgroup('logging')
|
|
|
|
def add_option_ini(option, dest, default=None, type=None, **kwargs):
|
|
parser.addini(dest, default=default, type=type,
|
|
help='default value for ' + option)
|
|
group.addoption(option, dest=dest, **kwargs)
|
|
|
|
add_option_ini(
|
|
'--no-print-logs',
|
|
dest='log_print', action='store_const', const=False, default=True,
|
|
type='bool',
|
|
help='disable printing caught logs on failed tests.')
|
|
add_option_ini(
|
|
'--log-level',
|
|
dest='log_level', default=None,
|
|
help='logging level used by the logging module')
|
|
add_option_ini(
|
|
'--log-format',
|
|
dest='log_format', default=DEFAULT_LOG_FORMAT,
|
|
help='log format as used by the logging module.')
|
|
add_option_ini(
|
|
'--log-date-format',
|
|
dest='log_date_format', default=DEFAULT_LOG_DATE_FORMAT,
|
|
help='log date format as used by the logging module.')
|
|
add_option_ini(
|
|
'--log-cli-level',
|
|
dest='log_cli_level', default=None,
|
|
help='cli logging level.')
|
|
add_option_ini(
|
|
'--log-cli-format',
|
|
dest='log_cli_format', default=None,
|
|
help='log format as used by the logging module.')
|
|
add_option_ini(
|
|
'--log-cli-date-format',
|
|
dest='log_cli_date_format', default=None,
|
|
help='log date format as used by the logging module.')
|
|
add_option_ini(
|
|
'--log-file',
|
|
dest='log_file', default=None,
|
|
help='path to a file when logging will be written to.')
|
|
add_option_ini(
|
|
'--log-file-level',
|
|
dest='log_file_level', default=None,
|
|
help='log file logging level.')
|
|
add_option_ini(
|
|
'--log-file-format',
|
|
dest='log_file_format', default=DEFAULT_LOG_FORMAT,
|
|
help='log format as used by the logging module.')
|
|
add_option_ini(
|
|
'--log-file-date-format',
|
|
dest='log_file_date_format', default=DEFAULT_LOG_DATE_FORMAT,
|
|
help='log date format as used by the logging module.')
|
|
|
|
|
|
@contextmanager
|
|
def catching_logs(handler, formatter=None, level=logging.NOTSET):
|
|
"""Context manager that prepares the whole logging machinery properly."""
|
|
root_logger = logging.getLogger()
|
|
|
|
if formatter is not None:
|
|
handler.setFormatter(formatter)
|
|
handler.setLevel(level)
|
|
|
|
# Adding the same handler twice would confuse logging system.
|
|
# Just don't do that.
|
|
add_new_handler = handler not in root_logger.handlers
|
|
|
|
if add_new_handler:
|
|
root_logger.addHandler(handler)
|
|
orig_level = root_logger.level
|
|
root_logger.setLevel(min(orig_level, level))
|
|
try:
|
|
yield handler
|
|
finally:
|
|
root_logger.setLevel(orig_level)
|
|
if add_new_handler:
|
|
root_logger.removeHandler(handler)
|
|
|
|
|
|
class LogCaptureHandler(logging.StreamHandler):
|
|
"""A logging handler that stores log records and the log text."""
|
|
|
|
def __init__(self):
|
|
"""Creates a new log handler."""
|
|
logging.StreamHandler.__init__(self, py.io.TextIO())
|
|
self.records = []
|
|
|
|
def emit(self, record):
|
|
"""Keep the log records in a list in addition to the log text."""
|
|
self.records.append(record)
|
|
logging.StreamHandler.emit(self, record)
|
|
|
|
|
|
class LogCaptureFixture(object):
|
|
"""Provides access and control of log capturing."""
|
|
|
|
def __init__(self, item):
|
|
"""Creates a new funcarg."""
|
|
self._item = item
|
|
|
|
@property
|
|
def handler(self):
|
|
return self._item.catch_log_handler
|
|
|
|
def get_handler(self, when):
|
|
"""
|
|
Get the handler for a specified state of the tests.
|
|
Valid values for the when parameter are: 'setup', 'call' and 'teardown'.
|
|
"""
|
|
return self._item.catch_log_handlers.get(when)
|
|
|
|
@property
|
|
def text(self):
|
|
"""Returns the log text."""
|
|
return self.handler.stream.getvalue()
|
|
|
|
@property
|
|
def records(self):
|
|
"""Returns the list of log records."""
|
|
return self.handler.records
|
|
|
|
@property
|
|
def record_tuples(self):
|
|
"""Returns a list of a striped down version of log records intended
|
|
for use in assertion comparison.
|
|
|
|
The format of the tuple is:
|
|
|
|
(logger_name, log_level, message)
|
|
"""
|
|
return [(r.name, r.levelno, r.getMessage()) for r in self.records]
|
|
|
|
def clear(self):
|
|
"""Reset the list of log records."""
|
|
self.handler.records = []
|
|
|
|
def set_level(self, level, logger=None):
|
|
"""Sets the level for capturing of logs.
|
|
|
|
By default, the level is set on the handler used to capture
|
|
logs. Specify a logger name to instead set the level of any
|
|
logger.
|
|
"""
|
|
if logger is None:
|
|
logger = self.handler
|
|
else:
|
|
logger = logging.getLogger(logger)
|
|
logger.setLevel(level)
|
|
|
|
@contextmanager
|
|
def at_level(self, level, logger=None):
|
|
"""Context manager that sets the level for capturing of logs.
|
|
|
|
By default, the level is set on the handler used to capture
|
|
logs. Specify a logger name to instead set the level of any
|
|
logger.
|
|
"""
|
|
if logger is None:
|
|
logger = self.handler
|
|
else:
|
|
logger = logging.getLogger(logger)
|
|
|
|
orig_level = logger.level
|
|
logger.setLevel(level)
|
|
try:
|
|
yield
|
|
finally:
|
|
logger.setLevel(orig_level)
|
|
|
|
|
|
@pytest.fixture
|
|
def caplog(request):
|
|
"""Access and control log capturing.
|
|
|
|
Captured logs are available through the following methods::
|
|
|
|
* caplog.text() -> string containing formatted log output
|
|
* caplog.records() -> list of logging.LogRecord instances
|
|
* caplog.record_tuples() -> list of (logger_name, level, message) tuples
|
|
"""
|
|
return LogCaptureFixture(request.node)
|
|
|
|
|
|
def get_actual_log_level(config, *setting_names):
|
|
"""Return the actual logging level."""
|
|
|
|
for setting_name in setting_names:
|
|
log_level = config.getoption(setting_name)
|
|
if log_level is None:
|
|
log_level = config.getini(setting_name)
|
|
if log_level:
|
|
break
|
|
else:
|
|
return
|
|
|
|
if isinstance(log_level, six.string_types):
|
|
log_level = log_level.upper()
|
|
try:
|
|
return int(getattr(logging, log_level, log_level))
|
|
except ValueError:
|
|
# Python logging does not recognise this as a logging level
|
|
raise pytest.UsageError(
|
|
"'{0}' is not recognized as a logging level name for "
|
|
"'{1}'. Please consider passing the "
|
|
"logging level num instead.".format(
|
|
log_level,
|
|
setting_name))
|
|
|
|
|
|
def pytest_configure(config):
|
|
config.pluginmanager.register(LoggingPlugin(config),
|
|
'logging-plugin')
|
|
|
|
|
|
class LoggingPlugin(object):
|
|
"""Attaches to the logging module and captures log messages for each test.
|
|
"""
|
|
|
|
def __init__(self, config):
|
|
"""Creates a new plugin to capture log messages.
|
|
|
|
The formatter can be safely shared across all handlers so
|
|
create a single one for the entire test session here.
|
|
"""
|
|
self.log_cli_level = get_actual_log_level(
|
|
config, 'log_cli_level', 'log_level') or logging.WARNING
|
|
|
|
self.print_logs = get_option_ini(config, 'log_print')
|
|
self.formatter = logging.Formatter(
|
|
get_option_ini(config, 'log_format'),
|
|
get_option_ini(config, 'log_date_format'))
|
|
|
|
log_cli_handler = logging.StreamHandler(sys.stderr)
|
|
log_cli_format = get_option_ini(
|
|
config, 'log_cli_format', 'log_format')
|
|
log_cli_date_format = get_option_ini(
|
|
config, 'log_cli_date_format', 'log_date_format')
|
|
log_cli_formatter = logging.Formatter(
|
|
log_cli_format,
|
|
datefmt=log_cli_date_format)
|
|
self.log_cli_handler = log_cli_handler # needed for a single unittest
|
|
self.live_logs = catching_logs(log_cli_handler,
|
|
formatter=log_cli_formatter,
|
|
level=self.log_cli_level)
|
|
|
|
log_file = get_option_ini(config, 'log_file')
|
|
if log_file:
|
|
self.log_file_level = get_actual_log_level(
|
|
config, 'log_file_level') or logging.WARNING
|
|
|
|
log_file_format = get_option_ini(
|
|
config, 'log_file_format', 'log_format')
|
|
log_file_date_format = get_option_ini(
|
|
config, 'log_file_date_format', 'log_date_format')
|
|
self.log_file_handler = logging.FileHandler(
|
|
log_file,
|
|
# Each pytest runtests session will write to a clean logfile
|
|
mode='w')
|
|
log_file_formatter = logging.Formatter(
|
|
log_file_format,
|
|
datefmt=log_file_date_format)
|
|
self.log_file_handler.setFormatter(log_file_formatter)
|
|
else:
|
|
self.log_file_handler = None
|
|
|
|
@contextmanager
|
|
def _runtest_for(self, item, when):
|
|
"""Implements the internals of pytest_runtest_xxx() hook."""
|
|
with catching_logs(LogCaptureHandler(),
|
|
formatter=self.formatter) as log_handler:
|
|
if not hasattr(item, 'catch_log_handlers'):
|
|
item.catch_log_handlers = {}
|
|
item.catch_log_handlers[when] = log_handler
|
|
item.catch_log_handler = log_handler
|
|
try:
|
|
yield # run test
|
|
finally:
|
|
del item.catch_log_handler
|
|
if when == 'teardown':
|
|
del item.catch_log_handlers
|
|
|
|
if self.print_logs:
|
|
# Add a captured log section to the report.
|
|
log = log_handler.stream.getvalue().strip()
|
|
item.add_report_section(when, 'log', log)
|
|
|
|
@pytest.hookimpl(hookwrapper=True)
|
|
def pytest_runtest_setup(self, item):
|
|
with self._runtest_for(item, 'setup'):
|
|
yield
|
|
|
|
@pytest.hookimpl(hookwrapper=True)
|
|
def pytest_runtest_call(self, item):
|
|
with self._runtest_for(item, 'call'):
|
|
yield
|
|
|
|
@pytest.hookimpl(hookwrapper=True)
|
|
def pytest_runtest_teardown(self, item):
|
|
with self._runtest_for(item, 'teardown'):
|
|
yield
|
|
|
|
@pytest.hookimpl(hookwrapper=True)
|
|
def pytest_runtestloop(self, session):
|
|
"""Runs all collected test items."""
|
|
with self.live_logs:
|
|
if self.log_file_handler is not None:
|
|
with closing(self.log_file_handler):
|
|
with catching_logs(self.log_file_handler,
|
|
level=self.log_file_level):
|
|
yield # run all the tests
|
|
else:
|
|
yield # run all the tests
|