logging: capture each record once when propagate changes during a test

Records from a logger which was non-propagating at capture start were
handled twice once the test enables Logger.propagate: pytest's capture
handler ran directly on that logger and again via the root logger.

Instead of attaching the real capture handler to initially
non-propagating loggers, attach a lightweight proxy bound to each such
logger and its ancestors which forwards to the real handler only while
that logger currently has propagate disabled. The proxy reads the live
propagate value per record, so a False -> True transition stops the
direct handling as soon as the record can reach root, and a True ->
False transition on those loggers is now captured as well.

Fixes #15064

Author: DawnofGenX

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(