test_ok1/_pytest/logging.py

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.')
    parser.addini(
        'log_cli', default=False, type='bool',
        help='enable log display during test run (also known as "live logging").')
    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=None):
    """Context manager that prepares the whole logging machinery properly."""
    root_logger = logging.getLogger()

    if formatter is not None:
        handler.setFormatter(formatter)
    if level is not None:
        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)
    if level is not None:
        orig_level = root_logger.level
        root_logger.setLevel(level)
    try:
        yield handler
    finally:
        if level is not None:
            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
        self._initial_log_levels = {}  # type: Dict[str, int] # dict of log name -> log level

    def _finalize(self):
        """Finalizes the fixture.

        This restores the log levels changed by :meth:`set_level`.
        """
        # restore log levels
        for logger_name, level in self._initial_log_levels.items():
            logger = logging.getLogger(logger_name)
            logger.setLevel(level)

    @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. The level will be restored to its previous value at the end of
        the test.

        :param int level: the logger to level.
        :param str logger: the logger to update the level. If not given, the root logger level is updated.

        .. versionchanged:: 3.4
            The levels of the loggers changed by this function will be restored to their initial values at the
            end of the test.
        """
        logger_name = logger
        logger = logging.getLogger(logger_name)
        self._initial_log_levels.setdefault(logger_name, logger.level)
        logger.setLevel(level)

    @contextmanager
    def at_level(self, level, logger=None):
        """Context manager that sets the level for capturing of logs. After the end of the 'with' statement the
        level is restored to its original value.

        :param int level: the logger to level.
        :param str logger: the logger to update the level. If not given, the root logger level is updated.
        """
        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
    """
    result = LogCaptureFixture(request.node)
    yield result
    result._finalize()


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')


@contextmanager
def _dummy_context_manager():
    yield


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.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'))
        self.log_level = get_actual_log_level(config, 'log_level')

        if config.getini('log_cli'):
            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)
            log_cli_level = get_actual_log_level(config, 'log_cli_level', 'log_level')
            self.log_cli_handler = log_cli_handler  # needed for a single unittest
            self.live_logs_context = catching_logs(log_cli_handler,
                                                   formatter=log_cli_formatter,
                                                   level=log_cli_level)
        else:
            self.log_cli_handler = None
            self.live_logs_context = _dummy_context_manager()

        log_file = get_option_ini(config, 'log_file')
        if log_file:
            self.log_file_level = get_actual_log_level(config, 'log_file_level')

            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, level=self.log_level) 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_context:
            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