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(