Source code for pytest_logger.plugin

import os
import sys
import re
import py
import pytest
import logging
import time
import datetime
import argparse
from builtins import object, int
from past.builtins import basestring


def pytest_addhooks(pluginmanager):
    pluginmanager.add_hookspecs(LoggerHookspec)


def _late_addoptions(parser, logcfg):
    """Add options to control logger"""
    parser.addini(
        name='logger_logsdir',
        help='base directory with log files for file loggers [basetemp]',
        default=None,
    )
    group = parser.getgroup('logger')
    group.addoption('--logger-logsdir',
                    help='pick you own logs directory instead of default '
                         'directory under session tmpdir')

    if logcfg._enabled:
        group.addoption('--log',
                        default=logcfg._log_option_default,
                        type=_log_option_parser(logcfg._loggers),
                        metavar='LOGGER,LOGGER.LEVEL,...',
                        help='comma delimited list of loggers optionally suffixed with level '
                             'preceded by a dot. Levels can be lower or uppercase, or numeric. '
                             'For example: "logger1,logger2.info,logger3.FATAL,logger4.25"')


@pytest.hookimpl(trylast=True)
def pytest_load_initial_conftests(early_config, parser, args):
    logcfg = LoggerConfig()
    early_config.hook.pytest_logger_config(logger_config=logcfg)
    early_config.pluginmanager.register(EarlyLoggerPlugin(logcfg), '_early_logger')
    _late_addoptions(parser, logcfg)


def pytest_configure(config):
    early_logger = config.pluginmanager.getplugin('_early_logger')
    config.pluginmanager.register(LoggerPlugin(config, early_logger._logcfg), '_logger')


class EarlyLoggerPlugin(object):
    def __init__(self, logcfg):
        self._logcfg = logcfg


class LoggerPlugin(object):
    def __init__(self, config, logcfg):
        self._config = config
        self._logdirlinks = config.hook.pytest_logger_logdirlink(config=config)
        self._loggers = _loggers_from_logcfg(logcfg, config.getoption('log')) if logcfg._enabled else None
        self._logsdir = None

    def logsdir(self):
        ldir = self._logsdir
        if ldir:
            return ldir
        logger_logsdir = self._config.getoption('logger_logsdir')
        if not logger_logsdir:
            logger_logsdir = self._config.getini('logger_logsdir')
        if logger_logsdir:
            ldir = _make_logsdir_dir(logger_logsdir)
        else:
            ldir = _make_logsdir_tmpdir(self._config._tmpdirhandler)

        self._logsdir = ldir

        for link in self._logdirlinks:
            _refresh_link(str(ldir), link)

        return ldir

    def pytest_runtest_setup(self, item):
        loggers = _choose_loggers(self._loggers, _loggers_from_hooks(item))
        item._logger = state = LoggerState(item=item,
                                           stdoutloggers=loggers.stdout,
                                           fileloggers=loggers.file)
        state.on_setup()

    def pytest_runtest_teardown(self, item, nextitem):
        logger = getattr(item, '_logger', None)
        if logger:
            logger.on_teardown()

    def pytest_runtest_makereport(self, item, call):
        logger = getattr(item, '_logger', None)
        if logger:
            if call.when == 'teardown':
                logger.on_makereport()


class LoggerState(object):
    def __init__(self, item, stdoutloggers, fileloggers):
        self._put_newlines = bool(item.config.option.capture == 'no' and stdoutloggers)
        self.handlers = _make_handlers(stdoutloggers, fileloggers, item)
        self.root_enabler = RootEnabler(bool(stdoutloggers and fileloggers))

    def put_newline(self):
        if self._put_newlines:
            sys.stdout.write('\n')

    def on_setup(self):
        self.put_newline()
        _enable(self.handlers)
        self.root_enabler.enable()

    def on_teardown(self):
        self.root_enabler.disable()
        self.put_newline()

    def on_makereport(self):
        _disable(self.handlers)


class RootEnabler(object):
    def __init__(self, enabled):
        self._enabled = enabled
        self._root_level = logging.root.level

    def enable(self):
        if self._enabled:
            self._root_level = logging.root.level
            logging.root.setLevel(logging.NOTSET)  # stops root logger from blocking logs

    def disable(self):
        if self._enabled:
            logging.root.setLevel(self._root_level)


class Loggers(object):
    def __init__(self, stdout, file_):
        self.stdout = stdout
        self.file = file_

    def __bool__(self):
        return bool(self.stdout) or bool(self.file)


[docs]class LoggerConfig(object): """Configuration of logging to stdout and filesystem.""" def __init__(self): self._enabled = False self._loggers = [] self._log_option_default = ''
[docs] def add_loggers(self, loggers, stdout_level=logging.NOTSET, file_level=logging.NOTSET): """Adds loggers for stdout/filesystem handling. Stdout: loggers will log to stdout only when mentioned in `log` option. If they're mentioned without explicit level, `stdout_level` will be used. Filesystem: loggers will log to files at `file_level`. :arg loggers: List of logger names. :arg stdout_level: Default level at which stdout handlers will pass logs. By default: `logging.NOTSET`, which means: pass everything. :arg file_level: Level at which filesystem handlers will pass logs. By default: `logging.NOTSET`, which means: pass everything. """ self._enabled = True self._loggers.append((loggers, _sanitize_level(stdout_level), _sanitize_level(file_level)))
[docs] def set_log_option_default(self, value): """ Sets default value of `log` option.""" self._log_option_default = value
[docs]class LoggerHookspec(object):
[docs] def pytest_logger_config(self, logger_config): """ called before cmdline options parsing. Accepts terse configuration of both stdout and file logging, adds cmdline options to manipulate stdout logging. Cannot be used together with \*loggers hooks. :arg logger_config: instance of :py:class:`LoggerConfig`, allows setting loggers for stdout and file handling and their levels. """
[docs] def pytest_logger_stdoutloggers(self, item): """ called before testcase setup. If implemented, given loggers will emit their output to terminal output. Cannot be used together with logger_config hook. :arg item: test item for which handlers are to be setup. :return list: List should contain logger name strings or tuples with logger name string and logging level. """
[docs] def pytest_logger_fileloggers(self, item): """ called before testcase setup. If implemented, given loggers will emit their output to files within logs temporary directory. Cannot be used together with logger_config hook. :arg item: test item for which handlers are to be setup. :return list: List should contain logger name strings or tuples with logger name string and logging level. """
class Formatter(logging.Formatter): short_level_names = { logging.FATAL: 'ftl', logging.ERROR: 'err', logging.WARN: 'wrn', logging.INFO: 'inf', logging.DEBUG: 'dbg', } def __init__(self, fmt): logging.Formatter.__init__(self, fmt) self._start = time.time() def formatTime(self, record, datefmt=None): ct = record.created - self._start dt = datetime.datetime.utcfromtimestamp(ct) return dt.strftime("%M:%S.%f")[:-3] # omit useconds, leave mseconds def format(self, record): record.levelshortname = Formatter.short_level_names.get(record.levelno, 'l%s' % record.levelno) return logging.Formatter.format(self, record) @pytest.fixture def logdir(request): return _make_logdir(request._pyfuncitem) def _sanitize_nodeid(filename): filename = filename.replace('::()::', '.') filename = filename.replace('::', '/') filename = re.sub(r'\[(.+)\]', r'-\1', filename) return filename def _sanitize_level(level, raises=True): if isinstance(level, basestring): try: return int(level) except ValueError: int_level = getattr(logging, level.upper(), None) if int_level is not None: return int_level elif isinstance(level, int): return level if raises: raise TypeError('bad logging level, expected int or string, got "%s"' % level) def _refresh_link(source, link_name): try: os.unlink(link_name) except OSError: pass try: os.symlink(source, link_name) except (OSError, AttributeError, NotImplementedError): pass def _make_logsdir_tmpdir(tmpdirhandler): logsdir = tmpdirhandler.getbasetemp() if logsdir.basename.startswith('popen-gw'): logsdir = logsdir.join('..') logsdir = logsdir.join('logs').ensure(dir=1) return logsdir def _make_logsdir_dir(dstname, cleandir=True): logsdir = py.path.local(dstname) if cleandir: if logsdir.check(): logsdir.remove() logsdir.mkdir() return logsdir def _make_logdir(item): plugin = item.config.pluginmanager.getplugin('_logger') return plugin.logsdir().join(_sanitize_nodeid(item.nodeid)).ensure(dir=1) def _enable(handlers): for hdlr in handlers: hdlr.logger.addHandler(hdlr) def _disable(handlers): for hdlr in handlers: hdlr.logger.removeHandler(hdlr) hdlr.close() def _log_option_parser(loggers): def parser(arg): def to_out(elem): def find_row(name): return next((row for row in loggers if name in row[0]), None) def bad_logger(name): names = [x for row in loggers for x in row[0]] pretty_names = '(' + ', '.join(names) + ')' raise argparse.ArgumentTypeError( 'wrong logger, expected %s, got "%s"' % (pretty_names, name)) def bad_level(level): raise argparse.ArgumentTypeError( 'wrong level, expected (INFO, warn, 15, ...), got "%s"' % level) row = find_row(elem) if row: return elem, row[1] if '.' in elem: elem_name, elem_level = elem.rsplit('.', 1) row = find_row(elem_name) level = _sanitize_level(elem_level, raises=False) if row and level is not None: return elem_name, level if row: bad_level(elem_level) if level is not None: bad_logger(elem_name) bad_logger(elem) return [to_out(x) for x in arg.split(',') if x] return parser def _loggers_from_logcfg(logcfg, logopt): def to_stdout(loggers, opt): def one(loggers, one): if isinstance(one, basestring): return one, next(row for row in loggers if one in row[0])[1] else: return one return [one(loggers, x) for x in opt] def to_file(loggers): return [(name, row[2]) for row in loggers for name in row[0]] if isinstance(logopt, basestring): return Loggers([], []) return Loggers( stdout=to_stdout(logcfg._loggers, logopt), file_=to_file(logcfg._loggers) ) def _loggers_from_hooks(item): def to_loggers(configs_lists): def to_logger_and_level(cfg): if isinstance(cfg, basestring): name, level = cfg, logging.NOTSET else: name, level = cfg return name, level return [to_logger_and_level(cfg) for configs in configs_lists for cfg in configs] return Loggers( stdout=to_loggers(item.config.hook.pytest_logger_stdoutloggers(item=item)), file_=to_loggers(item.config.hook.pytest_logger_fileloggers(item=item)) ) def _choose_loggers(config_loggers, hook_loggers): assert (not config_loggers) or (not hook_loggers),\ 'pytest_logger_config and pytest_logger_*loggers hooks used at the same time' return config_loggers or hook_loggers def _make_handlers(stdoutloggers, fileloggers, item): FORMAT = '%(asctime)s %(levelshortname)s %(name)s: %(message)s' fmt = Formatter(fmt=FORMAT) handlers = [] if stdoutloggers: handlers += _make_stdout_handlers(stdoutloggers, fmt) if fileloggers: logdir = _make_logdir(item) handlers += _make_file_handlers(fileloggers, fmt, logdir) return handlers def _make_stdout_handlers(loggers, fmt): def make_handler(logger_and_level, fmt): name, level = logger_and_level logger = logging.getLogger(name) handler = logging.StreamHandler(sys.stdout) handler.setFormatter(fmt) handler.setLevel(level) handler.logger = logger return handler return [make_handler(lgr, fmt) for lgr in loggers] def _make_file_handlers(loggers, fmt, logdir): def make_handler(logdir, logger_and_level, fmt): name, level = logger_and_level logger = logging.getLogger(name) name = name or 'logs' logfile = str(logdir.join(name)) handler = logging.FileHandler(filename=logfile, mode='w', delay=True) handler.setFormatter(fmt) handler.setLevel(level) handler.logger = logger return handler return [make_handler(logdir, lgr, fmt) for lgr in loggers]