From 13a1f37e8d8383b43a5e5f561838559cb89870d2 Mon Sep 17 00:00:00 2001 From: Tobias Deiminger Date: Fri, 16 Sep 2022 17:41:09 +0200 Subject: [PATCH] Automatically attach to non-propagating loggers Currently, pytest attaches its log handlers only to the root logger, and relies on log propagation to catch logs from loggers deeper in the hierarchy. In consequence we miss logs if one of those child loggers has logger.propagate set to False by user code. This affects caplog, report logging and live logging. This commit showcases a possible fix, where we attach not only to the root logger, but also to non-propagating loggers as they occur. To do so we patch the standard libraries logging module with our own descriptor that intercepts assignments to logger.propagate. This enables us to inject and remove our pytest handlers right on time, even during executing test calls, with neglible performance impact. Attempt to fix #3697. --- src/_pytest/logging.py | 115 ++++++++++++++++++++++++++---- testing/logging/test_reporting.py | 29 ++++++++ 2 files changed, 129 insertions(+), 15 deletions(-) diff --git a/src/_pytest/logging.py b/src/_pytest/logging.py index f9091399f2c..a5dafa92c57 100644 --- a/src/_pytest/logging.py +++ b/src/_pytest/logging.py @@ -8,11 +8,13 @@ from io import StringIO from pathlib import Path from typing import AbstractSet +from typing import cast from typing import Dict from typing import Generator from typing import List from typing import Mapping from typing import Optional +from typing import Set from typing import Tuple from typing import TYPE_CHECKING from typing import TypeVar @@ -47,12 +49,99 @@ _ANSI_ESCAPE_SEQ = re.compile(r"\x1b\[[\d;]+m") caplog_handler_key = StashKey["LogCaptureHandler"]() caplog_records_key = StashKey[Dict[str, List[logging.LogRecord]]]() +_HandlerType = TypeVar("_HandlerType", bound=logging.Handler) def _remove_ansi_escape_sequences(text: str) -> str: return _ANSI_ESCAPE_SEQ.sub("", text) +class CatchedLoggers: + """Add pytest handler to loggers on demand. + + Usually it would be enough to simply add pytest handlers to the root logger at fixed stages and rely + on log propagation to capture logs from child loggers. However, if a user turns off propagation + for a child logger, we would miss their logs. The chin-ups here and in LoggerPropagateInterceptor + are mainly to capture child loggers with propagate = False, by instantly adding pytest handlers to + them. + """ + + __slots__ = ("handler", "level", "loggers") + + def __init__( + self, + initial_set: Set[logging.Logger], + handler: _HandlerType, + level: Optional[int] = None, + ): + self.handler = handler + self.level = level + self.loggers: Dict[logging.Logger, Optional[int]] = {} + if self.level is not None: + self.handler.setLevel(self.level) + for logger in initial_set: + self.catch_logger(logger) + + def catch_logger(self, logger: logging.Logger): + """Add pytest handler and adjust level of logger. Remember state to restore.""" + if logger not in self.loggers: + logger.addHandler(self.handler) + if self.level is not None: + self.loggers[logger] = logger.level + logger.setLevel(min(logger.level, self.level)) + else: + self.loggers[logger] = None + + def release_logger(self, logger: logging.Logger): + """Restore log level and remove pytest handler.""" + if logger in self.loggers: + logger.removeHandler(self.handler) + orig_level = self.loggers.get(logger) + if orig_level is not None: + logger.setLevel(orig_level) + del self.loggers[logger] + + def release_all_loggers(self): + for logger in list(self.loggers.keys()): + self.release_logger(logger) + + +class LoggerPropagateInterceptor: + """Descriptor to be patched into standard libs logging module. + + It intercepts propagate = False assignments from user code. + + While catching_logs is inactive, it tracks non-propagating loggers without side effects. + While catching_logs is active, it also adds and removes pytest handlers instantly. + """ + + non_propagating_loggers: Set[logging.Logger] = set() + catchers: Set[CatchedLoggers] = set() + + def __init__(self): + for item in logging.getLogger().manager.loggerDict: + if isinstance(item, logging.Logger) and not item.propagate: + self.non_propagating_loggers.add(item) + + def __set__(self, logger: logging.Logger, propagate: bool): + if propagate is False: + if self.catchers: + for catcher in self.catchers: + catcher.catch_logger(logger) + self.non_propagating_loggers.add(logger) + elif propagate is True: + if self.catchers: + for catcher in self.catchers: + catcher.release_logger(logger) + if logger in self.non_propagating_loggers: + self.non_propagating_loggers.remove(logger) + logger.__dict__["propagate"] = propagate + + +# From now, we intercept each assignment to logger.propagate +logging.Logger.propagate = cast(bool, LoggerPropagateInterceptor()) + + class ColoredLevelFormatter(logging.Formatter): """A logging formatter which colorizes the %(levelname)..s part of the log format passed to __init__.""" @@ -299,34 +388,30 @@ def add_option_ini(option, dest, default=None, type=None, **kwargs): ) -_HandlerType = TypeVar("_HandlerType", bound=logging.Handler) - - # Not using @contextmanager for performance reasons. class catching_logs: """Context manager that prepares the whole logging machinery properly.""" - __slots__ = ("handler", "level", "orig_level") + __slots__ = ("handler", "level", "catcher") def __init__(self, handler: _HandlerType, level: Optional[int] = None) -> None: self.handler = handler self.level = level + self.catcher: Optional[CatchedLoggers] = None def __enter__(self): - root_logger = logging.getLogger() - if self.level is not None: - self.handler.setLevel(self.level) - root_logger.addHandler(self.handler) - if self.level is not None: - self.orig_level = root_logger.level - root_logger.setLevel(min(self.orig_level, self.level)) + self.catcher = CatchedLoggers( + LoggerPropagateInterceptor.non_propagating_loggers | {logging.getLogger()}, + self.handler, + self.level, + ) + LoggerPropagateInterceptor.catchers.add(self.catcher) return self.handler def __exit__(self, type, value, traceback): - root_logger = logging.getLogger() - if self.level is not None: - root_logger.setLevel(self.orig_level) - root_logger.removeHandler(self.handler) + assert self.catcher + LoggerPropagateInterceptor.catchers.remove(self.catcher) + self.catcher.release_all_loggers() class LogCaptureHandler(logging_StreamHandler): diff --git a/testing/logging/test_reporting.py b/testing/logging/test_reporting.py index 323ff7b2446..6f5d35b634a 100644 --- a/testing/logging/test_reporting.py +++ b/testing/logging/test_reporting.py @@ -1074,6 +1074,35 @@ def test_second(): assert "message from test 2" in content +def test_log_propagation_false(pytester: Pytester) -> None: + pytester.makepyfile( + """ + import pytest + import logging + + logging.getLogger('foo').propagate = False + + def test_log_file(request): + plugin = request.config.pluginmanager.getplugin('logging-plugin') + logging.getLogger().warning("log goes to root logger") + logging.getLogger('foo').warning("log goes to initially non-propagating logger") + logging.getLogger('foo.bar').propagate = False + logging.getLogger('foo.bar').warning("log goes to propagation-disabled-in-test logger") + assert False, "intentionally fail to trigger report logging output" + """ + ) + + result = pytester.runpytest("-s") + result.stdout.re_match_lines( + [ + "-+ Captured log call -+", + r"WARNING\s+root:test_log_propagation_false.py:8\s+log goes to root logger", + r"WARNING\s+foo:test_log_propagation_false.py:9\s+log goes to initially non-propagating logger", + r"WARNING\s+foo.bar:test_log_propagation_false.py:11\s+log goes to propagation-disabled-in-test logger", + ] + ) + + def test_colored_captured_log(pytester: Pytester) -> None: """Test that the level names of captured log messages of a failing test are colored."""