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 pytest-dev#3697.

Author: Tobias Deiminger

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,88 @@
 _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:
+    __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):
+        if logger not in self.loggers:
+            logger.addHandler(self.handler)
+            if self.level is not None:
+                logger.setLevel(min(logger.level, self.level))
+                self.loggers[logger] = logger.level  # remember original log level
+            else:
+                self.loggers[logger] = None
+
+    def release_logger(self, logger: logging.Logger):
+        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 +377,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):

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."""