From 083def4a63cc3817d2a83a32e6862dfc7f86c0fd Mon Sep 17 00:00:00 2001 From: DawnofGenX Date: Tue, 6 Oct 2026 19:44:18 +0000 Subject: [PATCH] logging: capture each record once when propagate changes during a test A logger which is non-propagating when capture starts never reaches the root logger, so the capture handler is attached to that logger too. If the logger then enables propagate during the test, the record is handled both by that handler and by the one on the root logger and is captured twice (#15064). Attach a small proxy instead of the real handler to each logger which cannot reach root (and to its ancestors). The proxy forwards to the real handler only while its logger is still non-propagating, deciding per record from the live propagate value, so a flip in either direction is captured exactly once with no cache or teardown bookkeeping. It is a live view of the real handler, so caplog.set_level(), caplog.filtering() and formatters installed through logger.handlers keep applying. --- AUTHORS | 2 + changelog/15064.bugfix.rst | 8 + src/_pytest/logging.py | 260 ++++++++++++++++++++++++-- testing/logging/test_bound_proxy.py | 279 ++++++++++++++++++++++++++++ testing/logging/test_fixture.py | 94 ++++++++++ testing/logging/test_reporting.py | 154 +++++++++++++++ 6 files changed, 784 insertions(+), 13 deletions(-) create mode 100644 changelog/15064.bugfix.rst create mode 100644 testing/logging/test_bound_proxy.py diff --git a/AUTHORS b/AUTHORS index e2fad5e8364..a70059f01f3 100644 --- a/AUTHORS +++ b/AUTHORS @@ -139,6 +139,7 @@ David Peled David Szotten David Vierra Daw-Ran Liou +DawnofGenX Debi Mishra Denis Cherednichenko Denis Kirisov @@ -402,6 +403,7 @@ Prakhar Gurunani Praneeth Kodumagulla Prashant Anand Prashant Sharma +Priyansh Kansara Pulkit Goyal Punyashloka Biswal Quentin Pradet diff --git a/changelog/15064.bugfix.rst b/changelog/15064.bugfix.rst new file mode 100644 index 00000000000..220136351f5 --- /dev/null +++ b/changelog/15064.bugfix.rst @@ -0,0 +1,8 @@ +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. A logger which becomes non-propagating after capture started, and an +ancestor which becomes the new propagation barrier mid-test, are now handled +too, so their records are neither duplicated nor lost. diff --git a/src/_pytest/logging.py b/src/_pytest/logging.py index 51c4954fb38..ca488c3550a 100644 --- a/src/_pytest/logging.py +++ b/src/_pytest/logging.py @@ -54,6 +54,20 @@ caplog_records_key = StashKey[dict[str, list[logging.LogRecord]]]() +if TYPE_CHECKING: + from collections.abc import Callable + from typing import Protocol + + class _SupportsFilterProtocol(Protocol): + """Structural stand-in for typeshed's private ``_SupportsFilter``.""" + + def filter(self, record: LogRecord) -> bool: ... + + # The element type ``logging.Filterer.filters`` uses, which also admits a + # plain callable or an object exposing ``.filter()``. + _FilterLike = logging.Filter | Callable[[LogRecord], bool] | _SupportsFilterProtocol + + def _remove_ansi_escape_sequences(text: str) -> str: return _ANSI_ESCAPE_SEQ.sub("", text) @@ -334,16 +348,187 @@ def add_option_ini(option, dest, default=None, type=None, **kwargs): _HandlerType = TypeVar("_HandlerType", bound=logging.Handler) +def _remove_handler_by_identity( + logger: logging.Logger, handler: logging.Handler +) -> None: + """Remove ``handler`` from ``logger.handlers`` comparing by identity. + + ``Logger.removeHandler()`` removes by equality, so a user handler which + compares equal to one of pytest's own would be removed in its place. + """ + handlers = logger.handlers + for index, existing in enumerate(handlers): + if existing is handler: + del handlers[index] + return + + +class _BoundProxyHandler(logging.Handler): + """A stand-in for a pytest capture handler on a non-propagating logger. + + A logger which is non-propagating when capture starts never reaches the + root logger, so the capture handler has to be attached to that logger as + well. If the logger then enables ``Logger.propagate`` during the test, the + record is seen by both that handler and the one on the root logger and is + captured twice (#15064). + + This proxy is attached instead of the real handler. It forwards to the + real handler only while its logger is *still* non-propagating, so the + record is captured exactly once whether or not ``propagate`` was flipped. + The decision is per record and reads the bound logger's live ``propagate`` + value, so nothing needs to be cached or recomputed between test phases. + + The proxy is a live *view* of the real handler: its level, filters and + formatter are the real handler's, so ``caplog.set_level()``, + ``caplog.filtering()`` and a formatter installed through ``logger.handlers`` + keep applying, and nothing has to be undone at teardown. + """ + + def __init__(self, logger: logging.Logger, real_handler: logging.Handler) -> None: + # Deliberately skip ``logging.Handler.__init__``: it would set + # ``level``/``filters``/``formatter``, which are live views onto the + # real handler here, clobbering its state, and it registers the proxy + # in the global handler list, so ``logging.shutdown()`` could close the + # real handler. Replicate only the base state the stdlib relies on. + self._name = None + self._closed = False + self.createLock() + self.logger: logging.Logger | None = logger + # Both are cleared on detach so a retained proxy cannot keep the + # capture handler or logger alive; readers treat ``None`` as + # "do not forward". + self.real_handler: logging.Handler | None = real_handler + # Number of catching_logs scopes currently sharing this proxy; the + # last one out removes it. Overlapping scopes (e.g. caplog and the + # report handler, or two contexts sharing one handler) must attach a + # single proxy, or the record is forwarded once per scope. + self._refcount = 0 + + # ------------------------------------------------------------- live view + @property + def level(self) -> int: + real = self.real_handler + return logging.NOTSET if real is None else real.level + + @level.setter + def level(self, value: int) -> None: + real = self.real_handler + if real is not None: + real.level = value + + @property + def filters(self) -> list[_FilterLike]: + real = self.real_handler + return [] if real is None else real.filters + + @filters.setter + def filters(self, value: list[_FilterLike]) -> None: + real = self.real_handler + if real is not None: + # ``Handler.filters`` is typed invariantly as ``list[Filter]``; + # the wider filter kinds are accepted by the stdlib at runtime. + real.filters = value # pyright: ignore[reportAttributeAccessIssue] + + @property + def formatter(self) -> logging.Formatter | None: + real = self.real_handler + return None if real is None else real.formatter + + @formatter.setter + def formatter(self, value: logging.Formatter | None) -> None: + real = self.real_handler + if real is not None: + real.formatter = value + + def setLevel(self, level: int | str) -> None: + real = self.real_handler + if real is not None: + real.setLevel(level) + + def addFilter(self, filter: logging.Filter) -> None: # type: ignore[override] + # Filters installed through ``logger.handlers`` (e.g. by + # ``caplog.filtering()``) must keep affecting capture, and the real + # handler is what applies them, so the filter goes there. + real = self.real_handler + if real is not None and not any(f is filter for f in real.filters): + real.addFilter(filter) + + def removeFilter(self, filter: logging.Filter) -> None: # type: ignore[override] + real = self.real_handler + if real is not None: + for index, existing in enumerate(real.filters): + if existing is filter: + del real.filters[index] + break + + def setFormatter(self, fmt: logging.Formatter | None) -> None: + real = self.real_handler + if real is not None: + real.setFormatter(fmt) + + # ------------------------------------------------------------ forwarding + def handle(self, record: logging.LogRecord) -> bool: + """Forward to the real handler while the bound logger is non-propagating. + + Called directly by ``Logger.callHandlers()``. The proxy's own lock is + deliberately not taken: only the real handler's lock is held while + forwarding, so there is no ``proxy -> real`` lock order for another + thread to invert and deadlock on. ``Logger.callHandlers()`` ignores the + return value. + """ + real = self.real_handler + logger = self.logger + if real is None or logger is None: + return False + if logger.propagate: + # The bound logger propagates now, so the walk continues past it -- + # either to the real handler on root or to a non-propagating + # ancestor which has its own proxy. Forwarding here too would + # deliver a second copy. + return False + if any(h is real for h in logger.handlers): + # The real handler is attached to this logger directly, so this + # same walk will already handle the record with it. + return False + return real.handle(record) + + def emit(self, record: logging.LogRecord) -> None: + # Only reached when the handler is driven directly by user code; + # ``handle()`` forwards before this handler's own lock is taken. + self.handle(record) + + def close(self) -> None: + """Detach, then release strong references. + + Detaching must not close the real handler, which pytest reuses across + phases. The strong ``logger``/``real_handler`` references are dropped + so a retained proxy cannot keep either object alive; ``handle()`` and + ``emit()`` treat a missing reference as "do not forward". + """ + self.logger = None + self.real_handler = None + super().close() + + # 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[list[logging.Handler], _BoundProxyHandler] + ] = [] def __enter__(self) -> _HandlerType: root_logger = logging.getLogger() @@ -352,17 +537,11 @@ 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. - 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) + # Attach a stand-in to all non-propagating loggers (their records + # won't reach root) and to their ancestors, so that a logger which + # flips ``propagate`` during the test -- or an ancestor which becomes + # the new barrier -- is still captured exactly once (#15064). + self._attach_proxies(root_logger) 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. @@ -370,6 +549,47 @@ def __enter__(self) -> _HandlerType: root_logger.setLevel(min(self.orig_level, self.level)) return self.handler + def _attach_proxies(self, root_logger: logging.Logger) -> None: + """Attach a proxy for every logger which cannot reach the root logger. + + A non-propagating logger never reaches root, so the real handler on + root does not see its records; the proxy stands in for it there. Its + ancestors get one too: a record emitted below an ancestor which + becomes non-propagating mid-test stops at that ancestor, so it needs a + stand-in as well. + """ + for logger in root_logger.manager.loggerDict.values(): + if ( + not isinstance(logger, logging.Logger) + or logger is root_logger + or logger.propagate + ): + continue + node: logging.Logger | None = logger + while node is not None and node is not root_logger: + self._attach_proxy(node) + node = node.parent + + def _attach_proxy(self, logger: logging.Logger) -> None: + # Reuse a proxy an enclosing scope already attached for this handler, + # so overlapping scopes forward a record once rather than once each. + for existing in logger.handlers: + if ( + isinstance(existing, _BoundProxyHandler) + and existing.real_handler is self.handler + ): + existing._refcount += 1 + self.attached_proxies.append((logger.handlers, existing)) + return + proxy = _BoundProxyHandler(logger, self.handler) + # ``Logger.addHandler`` refuses a handler comparing equal to one + # already attached, so append by identity and remember the exact list + # so teardown removes it from there even if the attribute is replaced. + handlers = logger.handlers + handlers.append(proxy) + proxy._refcount = 1 + self.attached_proxies.append((handlers, proxy)) + def __exit__( self, exc_type: type[BaseException] | None, @@ -380,8 +600,22 @@ def __exit__( if self.level is not None: root_logger.setLevel(self.orig_level) for logger in self.attached_loggers: - logger.removeHandler(self.handler) + if any(h is self.handler for h in logger.handlers): + _remove_handler_by_identity(logger, self.handler) self.attached_loggers.clear() + for handlers, proxy in self.attached_proxies: + proxy._refcount -= 1 + if proxy._refcount > 0: + # Still owned by an enclosing scope; leave it attached. + continue + # Remove by identity from the exact list the proxy was appended + # to; a no-op if it was already removed externally. + for index, existing in enumerate(handlers): + if existing is proxy: + del handlers[index] + break + proxy.close() + self.attached_proxies.clear() class LogCaptureHandler(logging_StreamHandler): diff --git a/testing/logging/test_bound_proxy.py b/testing/logging/test_bound_proxy.py new file mode 100644 index 00000000000..8cb9c32c892 --- /dev/null +++ b/testing/logging/test_bound_proxy.py @@ -0,0 +1,279 @@ +"""Unit tests for the bound proxy handler used by ``catching_logs`` (#15064). + +These drive ``catching_logs`` directly, which is where the proxy lifecycle +lives. End-to-end behaviour (caplog, reports, live logs, ``--log-file``) is +covered in ``test_fixture.py`` / ``test_reporting.py``. +""" + +from __future__ import annotations + +from collections.abc import Iterator +import io +import logging + +from _pytest.logging import _BoundProxyHandler +from _pytest.logging import catching_logs +import pytest + + +@pytest.fixture(autouse=True) +def _clean_logging() -> Iterator[None]: + """Isolate each test from the ambient logging configuration.""" + root = logging.getLogger() + saved_root_handlers = list(root.handlers) + saved_level = root.level + saved_dict = dict(root.manager.loggerDict) + for name in list(root.manager.loggerDict): + del root.manager.loggerDict[name] + root.handlers.clear() + root.setLevel(logging.WARNING) + try: + yield + finally: + for name in list(root.manager.loggerDict): + del root.manager.loggerDict[name] + root.manager.loggerDict.update(saved_dict) + root.handlers.clear() + root.handlers.extend(saved_root_handlers) + root.setLevel(saved_level) + + +def _make_logger(name: str, *, propagate: bool = False) -> logging.Logger: + logger = logging.getLogger(name) + logger.handlers.clear() + logger.setLevel(logging.DEBUG) + logger.propagate = propagate + return logger + + +def _capture() -> tuple[io.StringIO, logging.Handler]: + stream = io.StringIO() + handler = logging.StreamHandler(stream) + handler.setLevel(logging.DEBUG) + return stream, handler + + +def test_proxy_captures_once_when_propagation_enabled() -> None: + """The original bug: a non-propagating logger which starts propagating.""" + logger = _make_logger("a") + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + logger.propagate = True + logger.warning("m") + assert stream.getvalue().count("m") == 1 + + +def test_proxy_captures_while_non_propagating() -> None: + """A logger which stays non-propagating is still captured (#3697).""" + logger = _make_logger("b") + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + logger.warning("m") + assert stream.getvalue().count("m") == 1 + + +def test_no_proxy_for_fully_propagating_logger() -> None: + """Loggers that propagate throughout must not get a proxy at all.""" + _make_logger("c.plain", propagate=True) + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + logging.getLogger("c.plain").warning("m") + assert stream.getvalue().count("m") == 1 + assert not [ + h + for h in logging.getLogger("c.plain").handlers + if isinstance(h, _BoundProxyHandler) + ] + + +def test_ancestor_of_non_propagating_logger_gets_a_proxy() -> None: + """An ancestor which becomes the new barrier mid-test still captures once.""" + parent = _make_logger("p", propagate=True) + child = _make_logger("p.child") + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + child.propagate = True + parent.propagate = False + child.warning("m") + assert stream.getvalue().count("m") == 1 + + +def test_logger_becoming_non_propagating_between_scopes_is_captured() -> None: + """Selection is not cached, so a logger made non-propagating after a first + scope is still captured in the next one (#15064).""" + logger = _make_logger("late", propagate=True) + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + logger.warning("first") + logger.propagate = False + with catching_logs(handler, level=logging.DEBUG): + logger.warning("second") + assert stream.getvalue().count("first") == 1 + assert stream.getvalue().count("second") == 1 + + +def test_proxy_forwards_level_filters_and_formatter() -> None: + """The proxy is a live view of the real handler, not a copy.""" + logger = _make_logger("view") + _, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + proxy = next(h for h in logger.handlers if isinstance(h, _BoundProxyHandler)) + proxy.setLevel(logging.ERROR) + assert handler.level == logging.ERROR + assert proxy.level == logging.ERROR + proxy.addFilter(logging.Filter()) + assert proxy.filters == handler.filters + fmt = logging.Formatter("%(message)s!") + proxy.setFormatter(fmt) + assert handler.formatter is fmt + + +def test_proxy_property_setters_reach_the_real_handler() -> None: + """Direct attribute assignment through the proxy updates the real handler.""" + logger = _make_logger("setters") + _, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + proxy = next(h for h in logger.handlers if isinstance(h, _BoundProxyHandler)) + proxy.level = logging.ERROR + assert handler.level == logging.ERROR + filters = [logging.Filter()] + proxy.filters = filters # type: ignore[assignment] + assert handler.filters == filters + proxy.formatter = logging.Formatter("%(message)s!") + assert handler.formatter is not None + + +def test_proxy_forwards_when_emitted_directly() -> None: + """A record driven straight through ``emit()`` is forwarded to the real + handler while the bound logger is non-propagating.""" + logger = _make_logger("emit") + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + proxy = next(h for h in logger.handlers if isinstance(h, _BoundProxyHandler)) + proxy.emit(logging.LogRecord("emit", logging.WARNING, "", 0, "m", (), None)) + assert stream.getvalue().count("m") == 1 + + +def test_proxy_skips_forwarding_when_real_handler_attached_directly() -> None: + """The real handler on the logger itself already handles the record, so the + proxy must not forward a second copy.""" + logger = _make_logger("direct") + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + logger.addHandler(handler) + logger.warning("m") + assert stream.getvalue().count("m") == 1 + + +def test_detached_proxy_is_inert() -> None: + """A proxy used after its scope ended does not forward or raise.""" + logger = _make_logger("inert") + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + proxy = next(h for h in logger.handlers if isinstance(h, _BoundProxyHandler)) + record = logging.LogRecord("inert", logging.WARNING, "", 0, "m", (), None) + assert proxy.handle(record) is False + proxy.emit(record) + assert stream.getvalue() == "" + # Property accessors on a detached proxy are also safe no-ops. + assert proxy.level == logging.NOTSET + assert proxy.filters == [] + assert proxy.formatter is None + proxy.level = logging.ERROR + proxy.filters = [] + proxy.formatter = None + # ...as are the handler methods, which forward to the (now absent) real + # handler. + proxy.setLevel(logging.ERROR) + proxy.addFilter(logging.Filter()) + proxy.removeFilter(logging.Filter()) + proxy.setFormatter(logging.Formatter("%(message)s")) + + +def test_proxy_add_filter_is_idempotent() -> None: + """Adding the same filter object twice only installs it once.""" + logger = _make_logger("addfilter") + _, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + proxy = next(h for h in logger.handlers if isinstance(h, _BoundProxyHandler)) + filt = logging.Filter() + proxy.addFilter(filt) + proxy.addFilter(filt) + assert handler.filters.count(filt) == 1 + + +def test_proxy_remove_filter_matches_and_ignores_absent() -> None: + """``removeFilter`` removes a present filter by identity and is a no-op for + one that was never added.""" + logger = _make_logger("removefilter") + _, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + proxy = next(h for h in logger.handlers if isinstance(h, _BoundProxyHandler)) + present = logging.Filter() + proxy.addFilter(present) + proxy.removeFilter(logging.Filter()) # not present: no-op + assert present in handler.filters + proxy.removeFilter(present) + assert present not in handler.filters + + +def test_remove_handler_by_identity_ignores_absent_handler() -> None: + """The identity remover is a no-op when the handler is not on the logger.""" + from _pytest.logging import _remove_handler_by_identity + + logger = _make_logger("remover") + other = logging.StreamHandler() + existing = logging.StreamHandler() + logger.addHandler(existing) + _remove_handler_by_identity(logger, other) # absent: nothing removed + assert existing in logger.handlers + + +def test_proxy_removed_externally_is_still_closed_on_exit() -> None: + """A proxy a user stripped from the logger mid-scope is still closed at + teardown, and removal from the remembered list is a no-op.""" + logger = _make_logger("external") + _, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + proxy = next(h for h in logger.handlers if isinstance(h, _BoundProxyHandler)) + logger.removeHandler(proxy) + assert proxy.real_handler is None + + +def test_proxy_detached_on_exit_does_not_forward() -> None: + """A proxy kept past the context is inert and releases the real handler.""" + logger = _make_logger("gone") + _, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + proxy = next(h for h in logger.handlers if isinstance(h, _BoundProxyHandler)) + assert not any(h is proxy for h in logger.handlers) + assert proxy.real_handler is None + assert ( + proxy.handle(logging.LogRecord("gone", logging.WARNING, "", 0, "m", (), None)) + is False + ) + + +def test_detach_does_not_close_real_handler() -> None: + _make_logger("keep") + _, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + pass + # The real handler is reused across phases; detaching the proxy must not + # close it. + assert not getattr(handler, "_closed") + + +def test_nested_scopes_share_one_proxy() -> None: + """Overlapping scopes over the same handler must not double-forward.""" + logger = _make_logger("nested") + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + with catching_logs(handler, level=logging.DEBUG): + logger.warning("m") + # Still one proxy while the outer scope is active. + assert ( + len([h for h in logger.handlers if isinstance(h, _BoundProxyHandler)]) == 1 + ) + assert stream.getvalue().count("m") == 1 + assert not [h for h in logger.handlers if isinstance(h, _BoundProxyHandler)] diff --git a/testing/logging/test_fixture.py b/testing/logging/test_fixture.py index 95c0f44b7f5..8388a102271 100644 --- a/testing/logging/test_fixture.py +++ b/testing/logging/test_fixture.py @@ -472,6 +472,100 @@ 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_capture_once_when_logger_becomes_non_propagating_between_tests( + pytester: Pytester, +) -> None: + """A logger made non-propagating by a session-scoped fixture (i.e. between + capture scopes) is still captured in every test (#15064). + + The selection of loggers needing a stand-in must not be cached across + capture scopes: a logger which becomes non-propagating after the first + scope started would otherwise never be captured again. + """ + pytester.makepyfile( + """ + import logging + + import pytest + + app = logging.getLogger("myapp") + + @pytest.fixture(scope="session", autouse=True) + def configure_logging(): + app.propagate = False + yield + app.propagate = True + + def test_one(caplog): + app.warning("one") + assert caplog.messages == ["one"] + + def test_two(caplog): + app.warning("two") + assert caplog.messages == ["two"] + """ + ) + + result = pytester.runpytest() + result.assert_outcomes(passed=2) + + def test_captures_despite_exception(pytester: Pytester) -> None: pytester.makepyfile( """ diff --git a/testing/logging/test_reporting.py b/testing/logging/test_reporting.py index 11013fdd749..36fd7297594 100644 --- a/testing/logging/test_reporting.py +++ b/testing/logging/test_reporting.py @@ -1287,6 +1287,160 @@ 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_log_cli_output_capture_once_when_propagation_enabled_during_test( + pytester: Pytester, +) -> None: + """Live ``--log-cli-level`` output shows a record once, not twice, when a + logger enables propagation mid-test (#15064).""" + pytester.makepyfile( + """ + import logging + + logger = logging.getLogger("cli.example") + logger.propagate = False + + def test_log_once(): + logger.propagate = True + logger.warning("live only once") + """ + ) + + result = pytester.runpytest("--log-cli-level=WARNING") + result.assert_outcomes(passed=1) + result.stdout.no_fnmatch_line("*live only once*live only once*") + assert result.stdout.str().count("live only once") == 1 + + +def test_log_file_output_capture_once_when_propagation_enabled_during_test( + pytester: Pytester, +) -> None: + """``--log-file`` output records the message once, not twice (#15064).""" + pytester.makepyfile( + """ + import logging + + logger = logging.getLogger("file.example") + logger.propagate = False + + def test_log_once(): + logger.propagate = True + logger.warning("file only once") + """ + ) + + log_file = str(pytester.path.joinpath("pytest.log")) + result = pytester.runpytest(f"--log-file={log_file}", "--log-file-level=WARNING") + result.assert_outcomes(passed=1) + + with open(log_file, encoding="utf-8") as rfh: + contents = rfh.read() + assert contents.count("file only once") == 1 + + +def test_report_capture_with_level_filter_on_non_propagating_logger( + pytester: Pytester, +) -> None: + """A level set through the proxy on a non-propagating logger still applies + to records forwarded to the real capture handler (#15064).""" + pytester.makepyfile( + """ + import logging + + logger = logging.getLogger("level.example") + logger.propagate = False + + def test_level_filter(caplog): + with caplog.at_level(logging.WARNING, logger="level.example"): + logger.info("suppressed info") + logger.warning("kept warning") + assert caplog.messages == ["kept warning"] + """ + ) + + result = pytester.runpytest() + result.assert_outcomes(passed=1) + + +def test_report_capture_with_handler_filter_on_non_propagating_logger( + pytester: Pytester, +) -> None: + """A filter installed through the proxy on a non-propagating logger still + affects capture (#15064). + + The filter is the caller's own: the proxy is a view of the real handler, so + the filter stays on the real handler until it is explicitly removed, which + is what the un-proxied handler would do. Teardown deliberately does NOT + remove it -- a proxy must not claim ownership of state it did not create. + """ + pytester.makepyfile( + """ + import logging + + logger = logging.getLogger("filter.example") + logger.propagate = False + + def only_warnings(record): + return record.levelno >= logging.WARNING + + def test_filter_applies(caplog): + for handler in logger.handlers: + handler.addFilter(only_warnings) + logger.info("dropped") + logger.warning("kept") + assert "dropped" not in caplog.text + assert "kept" in caplog.text + + def test_filter_still_applies_until_removed(caplog): + # The filter was not removed at the end of the previous test: it + # belongs to the caller, so it persists exactly as it would on the + # real handler. It is removed here explicitly, which is the only + # thing that should take it out of the capture path. + logger.info("still dropped") + assert "still dropped" not in caplog.text + for handler in logger.handlers: + if hasattr(handler, "removeFilter"): + handler.removeFilter(only_warnings) + caplog.set_level(logging.INFO) + logger.info("not filtered now") + assert "not filtered now" in caplog.text + """ + ) + + result = pytester.runpytest() + result.assert_outcomes(passed=2) + + def test_colored_ansi_esc_caplogtext(pytester: Pytester) -> None: """Make sure that caplog.text does not contain ANSI escape sequences.""" pytester.makepyfile(