logging: capture each record once when propagate changes during a test (#15064)

changelog/15064.bugfix.rst

@@ -0,0 +1,7 @@
+Fixed duplicate log records when a logger which was non-propagating at capture
+start enables ``logging.Logger.propagate`` during the test: the record was
+handled both by pytest's handler attached directly to that logger and by the
+one attached to the root logger. Capture (``caplog``, failure-report sections,
+``--log-cli-level`` and ``--log-file`` output) now sees each record exactly
+once, and the previously-missed ``True -> False`` transition on the affected
+loggers and their ancestors is now handled as well.

src/_pytest/logging.py

@@ -334,16 +334,58 @@ def add_option_ini(option, dest, default=None, type=None, **kwargs):
 _HandlerType = TypeVar("_HandlerType", bound=logging.Handler)
 
 
+class _BoundProxyHandler(logging.Handler):
+    """A proxy for a pytest capture handler, bound to one logger.
+
+    The proxy forwards records to the real handler only while its logger
+    currently does not propagate. This is attached (instead of the real
+    handler) to loggers which were non-propagating when capture started, and
+    to their ancestors, so that flipping ``Logger.propagate`` during a test
+    neither duplicates the record (direct handler plus root handler) nor
+    misses it (#15064, #3697).
+    """
+
+    __slots__ = ("logger", "real_handler")
+
+    def __init__(self, logger: logging.Logger, real_handler: logging.Handler) -> None:
+        self.logger = logger
+        self.real_handler = real_handler
+        super().__init__()
+
+    @property
+    def level(self) -> int:
+        # Always defer to the real handler, whose level may change after
+        # attachment (e.g. via caplog.set_level()).
+        return self.real_handler.level
+
+    @level.setter
+    def level(self, value: int) -> None:
+        # Only assigned by logging.Handler.__init__(); the level is tracked
+        # through the real handler instead.
+        pass
+
+    def emit(self, record: logging.LogRecord) -> None:
+        if not self.logger.propagate:
+            self.real_handler.handle(record)
+
+
 # Not using @contextmanager for performance reasons.
 class catching_logs(Generic[_HandlerType]):
     """Context manager that prepares the whole logging machinery properly."""
 
-    __slots__ = ("attached_loggers", "handler", "level", "orig_level")
+    __slots__ = (
+        "attached_loggers",
+        "attached_proxies",
+        "handler",
+        "level",
+        "orig_level",
+    )
 
     def __init__(self, handler: _HandlerType, level: int | None = None) -> None:
         self.handler = handler
         self.level = level
         self.attached_loggers: list[logging.Logger] = []
+        self.attached_proxies: list[tuple[logging.Logger, _BoundProxyHandler]] = []
 
     def __enter__(self) -> _HandlerType:
         root_logger = logging.getLogger()
@@ -352,17 +394,30 @@ def __enter__(self) -> _HandlerType:
         # Attach to root logger.
         root_logger.addHandler(self.handler)
         self.attached_loggers.append(root_logger)
-        # Attach to all non-propagating loggers (won't reach root).
-        # Note that will miss loggers that *become* non-propagating
-        # after the `__enter__`. Not worth the trouble for now.
+        # Attach bound proxy handlers to all non-propagating loggers
+        # (their records won't reach root) and to their ancestors, so that
+        # records which *do* reach root after a `propagate` change are only
+        # handled once. The proxies consult the live `propagate` value per
+        # record (#15064).
+        # Note that this still misses loggers (outside those ancestor
+        # chains) which *become* non-propagating after the `__enter__`.
+        # Not worth the trouble for now.
+        proxy_targets: dict[logging.Logger, None] = {}
         for logger in root_logger.manager.loggerDict.values():
             if (
                 isinstance(logger, logging.Logger)
                 and not logger.propagate
                 and logger is not root_logger
             ):
-                logger.addHandler(self.handler)
-                self.attached_loggers.append(logger)
+                proxy_targets[logger] = None
+                parent = logger.parent
+                while parent is not None and parent is not root_logger:
+                    proxy_targets.setdefault(parent)
+                    parent = parent.parent
+        for logger in proxy_targets:
+            proxy = _BoundProxyHandler(logger, self.handler)
+            logger.addHandler(proxy)
+            self.attached_proxies.append((logger, proxy))
         if self.level is not None:
             # Non-propagating loggers still inherit the level (unless a logger
             # explicitly set level), so only do this on the root logger.
@@ -382,6 +437,9 @@ def __exit__(
         for logger in self.attached_loggers:
             logger.removeHandler(self.handler)
         self.attached_loggers.clear()
+        for logger, proxy in self.attached_proxies:
+            logger.removeHandler(proxy)
+        self.attached_proxies.clear()
 
 
 class LogCaptureHandler(logging_StreamHandler):

testing/logging/test_fixture.py

@@ -472,6 +472,62 @@ def test_non_propagating_logger(caplog):
     result.assert_outcomes(passed=1)
 
 
+def test_capture_once_when_propagation_enabled_during_test(
+    pytester: Pytester,
+) -> None:
+    """A logger which is non-propagating at capture start but enables
+    propagation during the test must not have its records captured twice
+    (#15064)."""
+    pytester.makepyfile(
+        """
+        import logging
+
+        logger = logging.getLogger("example")
+        logger.propagate = False
+        child_logger = logging.getLogger("example.child")
+
+        def test_log_is_captured_once(caplog):
+            logger.propagate = True
+
+            logger.warning("only once")
+            child_logger.warning("child only once")
+
+            assert caplog.messages == ["only once", "child only once"]
+        """
+    )
+
+    result = pytester.runpytest()
+    result.assert_outcomes(passed=1)
+
+
+def test_capture_once_when_propagation_barrier_moves_to_ancestor(
+    pytester: Pytester,
+) -> None:
+    """A child which was non-propagating at capture start and propagates to an
+    ancestor which becomes the new barrier mid-test is captured exactly once
+    (#15064)."""
+    pytester.makepyfile(
+        """
+        import logging
+
+        parent = logging.getLogger("mixed.parent")
+        child = logging.getLogger("mixed.parent.child")
+        child.propagate = False
+
+        def test_barrier_moves(caplog):
+            child.propagate = True
+            parent.propagate = False
+
+            child.warning("once at new barrier")
+
+            assert caplog.messages == ["once at new barrier"]
+        """
+    )
+
+    result = pytester.runpytest()
+    result.assert_outcomes(passed=1)
+
+
 def test_captures_despite_exception(pytester: Pytester) -> None:
     pytester.makepyfile(
         """

testing/logging/test_reporting.py

@@ -1287,6 +1287,37 @@ def test_log_file():
     assert not list(report.get_sections("Captured stderr call"))
 
 
+def test_log_propagation_enabled_during_test_captured_once(
+    pytester: Pytester,
+) -> None:
+    """Records from a logger which enables propagation during the test appear
+    exactly once in the report's captured-log sections (#15064)."""
+    pytester.makepyfile(
+        """
+        import logging
+
+        logging.getLogger('foo').propagate = False
+
+        def test_log_once():
+            logging.getLogger('foo').warning("before enabling propagation")
+            logging.getLogger('foo').propagate = True
+            logging.getLogger('foo').warning("after enabling propagation")
+            assert False, "intentionally fail to trigger report logging output"
+    """
+    )
+
+    reprec = pytester.inline_run()
+    reports = reprec.getfailures()
+    assert len(reports) == 1
+    report = reports[0]
+    sections = list(report.get_sections("Captured log call"))
+    assert len(sections) == 1
+    log_text = sections[0][1]
+    assert log_text.count("before enabling propagation") == 1
+    assert log_text.count("after enabling propagation") == 1
+    assert log_text.count("WARNING") == 2
+
+
 def test_colored_ansi_esc_caplogtext(pytester: Pytester) -> None:
     """Make sure that caplog.text does not contain ANSI escape sequences."""
     pytester.makepyfile(