diff --git a/AUTHORS b/AUTHORS index e2fad5e8364..fe5bdfa33a2 100644 --- a/AUTHORS +++ b/AUTHORS @@ -134,6 +134,7 @@ Daniil Galiev Dave Hunt David Díaz-Barquero David Mohr +DawnofGenX David Paul Röthlisberger David Peled David Szotten diff --git a/changelog/15064.bugfix.rst b/changelog/15064.bugfix.rst new file mode 100644 index 00000000000..4769671495f --- /dev/null +++ b/changelog/15064.bugfix.rst @@ -0,0 +1,17 @@ +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. + +The handler standing in for a capture handler on those loggers is now also +attached and detached by identity rather than by equality, so a user handler +which compares equal to one of pytest's own is no longer removed in its place, +and two loggers sharing one ``handlers`` list are handled correctly. It no +longer holds its own lock while forwarding, which could deadlock against the +real handler's lock. Filters and formatters installed through such a handler +keep applying while the logger is non-propagating and are removed again when +capture ends, and a failure during capture setup no longer leaves pytest's +handler attached to the root logger. diff --git a/src/_pytest/logging.py b/src/_pytest/logging.py index 51c4954fb38..0c197b1b3f6 100644 --- a/src/_pytest/logging.py +++ b/src/_pytest/logging.py @@ -3,6 +3,7 @@ from __future__ import annotations +from collections.abc import Callable from collections.abc import Generator from collections.abc import Mapping from collections.abc import Set as AbstractSet @@ -22,8 +23,11 @@ from typing import final from typing import Generic from typing import Literal +from typing import NamedTuple +from typing import Protocol from typing import TYPE_CHECKING from typing import TypeVar +import weakref from _pytest import nodes from _pytest._io import TerminalWriter @@ -44,6 +48,15 @@ if TYPE_CHECKING: logging_StreamHandler = logging.StreamHandler[StringIO] + + class _SupportsFilterProtocol(Protocol): + """Structural stand-in for typeshed's private ``_SupportsFilter``.""" + + def filter(self, record: LogRecord) -> bool: ... + + # Same element type ``logging.Filterer.filters`` uses, which also admits a + # plain callable or an object exposing ``.filter()``. + _FilterLike = logging.Filter | Callable[[LogRecord], bool] | _SupportsFilterProtocol else: logging_StreamHandler = logging.StreamHandler @@ -333,17 +346,372 @@ def add_option_ini(option, dest, default=None, type=None, **kwargs): _HandlerType = TypeVar("_HandlerType", bound=logging.Handler) +# One capture-target group: the loggers sharing one ``handlers`` list, plus +# that list. Identity is what groups loggers; hashability is not assumed. +_TargetGroup = tuple[tuple[logging.Logger, ...], list[logging.Handler]] + + +def _remove_handler_by_identity( + logger: logging.Logger, handler: logging.Handler +) -> None: + """Remove ``handler`` from ``logger.handlers`` comparing by identity. + + ``Logger.removeHandler()`` uses ``list.remove()``, i.e. ``__eq__``, so a + user handler which compares equal to one of pytest's own would be removed + in its place. Two loggers may also share a single ``handlers`` list, in + which case the handler must come out of that one list exactly once. + """ + handlers = logger.handlers + for index, existing in enumerate(handlers): + if existing is handler: + del handlers[index] + return + + +def _remove_handler_by_identity_from_list( + handlers: list[logging.Handler], handler: logging.Handler +) -> None: + """Remove ``handler`` from an explicit ``handlers`` list, by identity. + + Two loggers may share one ``handlers`` list. The proxy has to come out of + that one list exactly once, keyed by the list object itself rather than + through a logger, so neither logger is left with a stale entry and a list + replacement on one of them cannot strand the proxy. + """ + for index, existing in enumerate(handlers): + if existing is handler: + del handlers[index] + return + + +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). + + The proxy is a *view* of the real handler: its level, filters and + formatter are the real handler's, so anything installed through + ``logger.handlers`` (e.g. by ``caplog.filtering()``) keeps applying and + nothing has to be undone at teardown. A filter already on the real + handler stays on the real handler even if it is also added here. + """ + + __slots__ = ( + "_closed", + "_detached", + "_group", + "_list_owner", + "_refcount", + "logger", + "real_handler", + ) + + def __init__( + self, + logger: logging.Logger, + real_handler: logging.Handler, + group: tuple[logging.Logger, ...] = (), + ) -> None: + self.logger: logging.Logger | None = logger + # Cleared on final detach so a retained proxy cannot keep the capture + # handler alive; every reader treats ``None`` as "do not forward". + self.real_handler: logging.Handler | None = real_handler + # Every logger sharing this proxy's ``handlers`` list. A shared list + # needs one proxy, and the proxy is asked to forward on behalf of all + # of them, so it has to know the full group to decide correctly. + self._group = group or (logger,) + self._detached = False + # The ``handlers`` list this proxy was appended to. Two loggers may + # share one list, so the proxy has to remember the object it was added + # to: a later list replacement on either logger must not strand the + # proxy or make teardown remove somebody else's handler. + self._list_owner: list[logging.Handler] | None = None + # Number of catching_logs scopes currently sharing this proxy. Overlapping + # scopes (e.g. caplog + report handler nested per test phase, or two + # contexts sharing one handler) must attach a single proxy, or the record + # is forwarded once per scope and captured twice. Starts at 0: the + # owning scope's ``__enter__`` accounts for the first reference. + self._refcount = 0 + # Deliberately skip ``logging.Handler.__init__``'s registration in the + # global ``_handlerList``: a proxy is a short-lived internal object + # which pytest detaches itself, it owns no stream, and it must never be + # closed by ``logging.shutdown()`` (that would close the real handler + # it forwards to). Everything else the base initialiser sets up is + # replicated here. + self._name = None + # ``logging.Handler.__init__`` also sets ``self._closed`` and creates + # the lock; ``level`` and ``filters`` are not replicated -- the + # properties above read and write the real handler directly. + self._closed = False + self.createLock() + + # ------------------------------------------------------------- live view + @property + def level(self) -> int: + # The real handler's level, which may change after attachment (e.g. via + # caplog.set_level()). Read through to it so ``logger.handlers`` shows + # the level that is actually in force. + real = self.real_handler + if real is None: + return logging.NOTSET + return 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]: + # The real handler's filter list, so a filter added through the proxy is + # applied by the real handler and remains visible on the real one. + real = self.real_handler + if real is None: + return [] + return 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 + # (same invariance mismatch as the getter above). + real.filters = value # pyright: ignore[reportAttributeAccessIssue] + + @property + def formatter(self) -> logging.Formatter | None: + real = self.real_handler + if real is None: + return None + return real.formatter + + @formatter.setter + def formatter(self, value: logging.Formatter | None) -> None: + real = getattr(self, "real_handler", None) + if real is not None: + real.formatter = value + + def setLevel(self, level: int | str) -> None: + # ``logger.handlers[...].setLevel()`` must reach the real handler -- + # silently dropping the write made the level a lie. + super().setLevel(level) + 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. ``handle()`` + # forwards to the real handler, which is what applies filters, so the + # filter goes there. Deliberately no bookkeeping: the filter belongs to + # the real handler now and stays until explicitly removed, which is + # what the un-proxied handler would do. + real = self.real_handler + if real is None: + return + if not any(f is filter for f in real.filters): + real.addFilter(filter) + + def removeFilter(self, filter: logging.Filter) -> None: # type: ignore[override] + # Removed from the real handler by identity, so a filter removed + # through the proxy (as ``caplog.filtering()`` does on exit) does not + # stay active for the next test. + 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 without holding the proxy's own lock. + + ``logging.Handler.handle()`` holds ``self.lock`` across ``emit()``. + Calling ``real_handler.handle()`` from there would additionally take the + real handler's lock, establishing a ``proxy -> real`` lock order that + another thread can invert (``real -> proxy``) and deadlock on. Instead, + take the real handler's lock directly, so only one lock is ever held. + + No reentrancy guard is needed: the real handler's lock is an RLock, so + a handler which logs one finite nested record from its own ``emit()`` + is handled the same way it would be without a proxy. An unconditional + log-from-emit handler recurses on the un-proxied handler too. + + ``Logger.callHandlers()`` ignores the return value, so returning a + falsy value here is safe; the stdlib's ``found`` bookkeeping is + unaffected. + """ + if self._is_detached: + return False + real = self.real_handler + if real is None: + return False + # Decide from the logger which actually emitted the record where that + # logger is one of the ones sharing this proxy, because a shared + # ``handlers`` list means the proxy forwards on behalf of all of them. + # A record from a DESCENDANT reached the proxy by walking up the + # hierarchy, so for that case the bound logger's own ``propagate`` + # decides (it is what stopped the walk here). + emitter = logging.getLogger(record.name) + deciding: logging.Logger | None + if any(candidate is emitter for candidate in self._group): + deciding = emitter + else: + deciding = self.logger + if deciding is None: + return False + if deciding.propagate: + # The emitting logger propagates now, so the record continues up the + # hierarchy (to the real handler attached to root) -- do not + # deliver it a second time. + return False + if any(h is real for h in deciding.handlers): + # The real handler is attached to this logger directly (e.g. a + # handler installed through ``logger.addHandler``), so this record + # is already going to be handled by it on this same walk -- forward + # and it would be captured twice. + return False + # Delegate whole-record handling (filters, level, lock, handleError) + # to the real handler. Only the real handler's lock is taken, so no + # lock ordering with the proxy's own lock exists to invert. + return real.handle(record) + + def emit(self, record: logging.LogRecord) -> None: + # Only reached when the handler is driven directly (``emit()`` by user + # code); ``handle()`` forwards before acquiring this handler's lock. + self.handle(record) + + def close(self) -> None: + """Detach, then release the proxy's 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", so a + late call on a detached proxy is a safe no-op rather than an + ``AttributeError``. + """ + self._detached = True + self.logger = None + self.real_handler = None + self._group = () + # Release the remembered list too: a closed proxy is no longer a live + # occupant of any ``handlers`` list, and must not keep a (possibly + # already-replaced) list alive (#15075 review round 3). + self._list_owner = None + # Drop the level too: logging.Handler.close() removes it from the + # module-level level registry. + super().close() + + @property + def _is_detached(self) -> bool: + return getattr(self, "_detached", False) + + +class _TargetCacheEntry(NamedTuple): + """One cached ``catching_logs`` target snapshot, keyed by ``id(handler)``. + + ``handler_ref`` is a weak back-reference used to reject a recycled + ``id()`` and its callback evicts the entry when the handler dies. Each + group is the loggers sharing one ``handlers`` list, held weakly so the + cache cannot keep a removed logger alive, plus the ``id()`` of that list + -- *identity metadata only*, never a strong reference, so the cache can + never retain a replaced list and the unrelated user handlers inside it. + Lookup re-derives the live list from the group's members and validates + that it still has the stored ``id()`` and is still the current + ``handlers`` attribute of every member, and that the list's owner count + still matches the bucket (proxy vs direct) it was classified into; any + mismatch forces a rebuild (#15075 review round 3). + """ + + handler_ref: weakref.ref[logging.Handler] + # Groups classified for bound-proxy attachment: lists with exactly one + # registered owner. Each entry is the group's weak loggers plus the + # ``id()`` of their shared ``handlers`` list. + proxy_groups: tuple[tuple[tuple[weakref.ref[logging.Logger], ...], int], ...] + # Groups classified for direct attachment of the real handler: lists + # with more than one registered owner (shared or aliased to root). + direct_groups: tuple[tuple[tuple[weakref.ref[logging.Logger], ...], int], ...] + registry_size: int + + +def _count_list_owners( + current: Mapping[str, object], root_logger: logging.Logger +) -> dict[int, int]: + """Map ``id(logger.handlers)`` to the number of loggers owning it. + + All registered loggers count, including root and propagating ones: the + proxy-vs-direct classification must see every owner of a list, not just + the ones selected as capture targets, or a later owner's records get + mis-routed (#15075 review). + """ + owners: dict[int, int] = {} + for candidate in current.values(): + if isinstance(candidate, logging.Logger): + list_id = id(candidate.handlers) + owners[list_id] = owners.get(list_id, 0) + 1 + root_id = id(root_logger.handlers) + owners[root_id] = owners.get(root_id, 0) + 1 + return owners + # 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") + # Proxies are reused across the enclosing capture scope, so the set of + # loggers needing one is cached per (manager, handler) and recomputed only + # when a *new* non-propagating logger appears. pytest re-enters + # ``catching_logs`` for every test phase, so this turns a per-entry + # O(n^2) rescan into a single walk. + __slots__ = ( + "attached_claims", + "attached_loggers", + "attached_proxies", + "handler", + "level", + "orig_level", + ) + + # Keyed by ``id(handler)`` rather than by the handler itself: a custom + # capture handler may set ``__hash__ = None``, and a WeakKeyDictionary + # would raise ``TypeError`` when used as a key. The entry holds a weak + # reference back to the handler plus weak references to the loggers, so + # it can neither keep a dead handler alive nor keep a removed logger + # alive. A recycled ``id()`` is rejected via the back-reference, and a + # stale group is rejected on lookup because the stored ``handlers`` list + # must still be the current list of every group member. It is rebuilt + # whenever the logger population, ``propagate`` state, or any cached + # group's list identity changes. + _target_cache: dict[int, _TargetCacheEntry] = {} 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] + ] = [] + # ``handlers`` lists onto which the real handler was appended + # directly, paired with whether this scope did the appending -- the + # conservative path for shared/aliased lists (#15075 review). An + # enclosing scope's or a user's existing copy is claimed without + # appending and must not be removed; only the appending scope removes + # it, from the remembered list object. + self.attached_claims: list[tuple[list[logging.Handler], bool]] = [] def __enter__(self) -> _HandlerType: root_logger = logging.getLogger() @@ -352,17 +720,22 @@ 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 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. + try: + self._attach_proxies(root_logger) + except BaseException: + # Entry must be transactional: if anything above fails (e.g. an + # unhashable Logger), undo the partial setup instead of leaving + # pytest's handler attached to root with no owner to remove it. + self._detach() + raise 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 +743,309 @@ 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 proxies / direct claims for every ``handlers`` list needing one. + + Loggers are grouped by the *identity* of their ``handlers`` list, and + each group is then classified by list ownership: + + * A list owned by exactly one logger gets a bound proxy: the proxy + consults the live ``propagate`` value per record, so flipping it + during a test neither duplicates nor misses records (#15064). + * A list with more than one owner (shared between loggers, or aliased + to ``root.handlers``) cannot host a sound proxy: one proxy cannot + know which owner a shared-list visit is for, and two proxies in one + list duplicate every record. Such lists get the real handler + attached *directly* instead -- the behaviour predating the proxy -- + claimed by identity so nested scopes share one copy and a handler + the user installed on the list is never removed by teardown. This + deliberately retains the pre-proxy limitation for aliased lists: + records of an owner which flips ``propagate`` mid-scope behave + exactly as they did before the proxy existed. + """ + proxy_groups, direct_groups = self._proxy_targets(root_logger, self.handler) + for targets, shared_list in proxy_groups: + # One proxy for this group. Reuse a live one already installed in + # this list so that overlapping catching_logs scopes -- pytest + # nests one per handler and re-enters them for every test phase -- + # share a single proxy. Without reuse the same record is forwarded + # once per scope and captured more than once. + proxy: _BoundProxyHandler | None = None + for existing in shared_list: + if ( + isinstance(existing, _BoundProxyHandler) + and existing.real_handler is self.handler + ): + proxy = existing + break + reused = proxy is not None + if proxy is None: + proxy = _BoundProxyHandler(targets[0], self.handler, targets) + # ``Logger.addHandler`` refuses a handler that compares equal to + # one already attached, so attach by identity and remember exactly + # what we added -- otherwise a user handler which compares equal + # would suppress the proxy *and* be removed in its place on exit. + if not reused and not any(h is proxy for h in shared_list): + shared_list.append(proxy) + proxy._refcount += 1 + proxy._list_owner = shared_list + # Record the claim keyed by the list object, not by logger: a + # shared list is one claim, so nested contexts over the same list + # refcount it once and cannot strand the proxy. + self.attached_proxies.append((shared_list, proxy)) + for _targets, shared_list in direct_groups: + # A scope appends the real handler at most once per list. An + # enclosing scope's (or a user's) copy already in the list is + # found here and left alone, so exit must not remove it either; + # only the scope which actually appended removes it, and it + # removes it from *this* list object even if an owner has since + # replaced its ``handlers`` attribute. + added = not any(h is self.handler for h in shared_list) + if added: + shared_list.append(self.handler) + self.attached_claims.append((shared_list, added)) + + @classmethod + def _proxy_targets( + cls, + root_logger: logging.Logger, + handler: logging.Handler, + ) -> tuple[tuple[_TargetGroup, ...], tuple[_TargetGroup, ...]]: + """Groups of loggers needing capture, grouped by ``handlers`` list. + + Returns ``(proxy_groups, direct_groups)``: each group pairs the + loggers sharing one ``handlers`` list with that list. A group needs a + bound proxy only when its list has exactly one owner; lists with more + than one owner (including lists aliased to ``root.handlers``) go to + the direct-attachment path instead, whose members are only the + loggers that were non-propagating at entry -- a propagating logger + on such a list reaches root already and must not gain a second copy + of the handler through the shared list. + + Cached per handler and revalidated on every entry: the cache is only + reused when the logger population is unchanged, every cached group is + still registered, every member is still non-propagating, and every + cached list is still the *live* list of all its members (validated by + stored ``id()``, never by holding the list itself). A logger which + appears, flips ``propagate``, or replaces its ``handlers`` list + therefore forces a recompute, while the common case -- pytest + re-entering the same scope for every test phase with nothing changed + -- is an O(n) revalidation. + + Identity is used throughout: loggers are not guaranteed to be hashable + (``Logger`` subclasses may set ``__hash__ = None``) and two distinct + loggers can share one ``handlers`` list, so neither a dict keyed by + logger nor a set is safe here. + """ + manager = root_logger.manager + current = manager.loggerDict + cached = cls._target_cache.get(id(handler)) + if ( + cached is not None + and cached.handler_ref() is handler + and cached.registry_size == len(current) + ): + # Revalidate the snapshot: the population is unchanged, so every + # logger it names is usable only while it is still registered + # and non-propagating; the live list re-derived from each group + # must still carry the stored list ``id()`` and be the *current* + # ``handlers`` attribute of all group members; and the list's + # owner count must still match the bucket it was classified + # into. A logger whose ``handlers`` attribute was replaced (or + # which shares a list with a new logger) after the snapshot was + # taken therefore forces a rebuild, so a stale entry never + # attaches to an obsolete list or to a list whose ownership has + # changed (#15075 review round 3). + registered: dict[int, None] = {} + for existing in current.values(): + registered[id(existing)] = None + owners = _count_list_owners(current, root_logger) + derefed: list[tuple[logging.Logger, ...]] = [] + live_lists: list[list[logging.Handler]] = [] + usable = True + for bucket_is_proxy, groups in ( + (True, cached.proxy_groups), + (False, cached.direct_groups), + ): + for group, stored_list_id in groups: + group_loggers: list[logging.Logger] = [] + for ref in group: + logger = ref() + if ( + logger is None + or logger.propagate + or id(logger) not in registered + ): + usable = False + break + group_loggers.append(logger) + if not usable: + break + assert group_loggers + live_list = group_loggers[0].handlers + if id(live_list) != stored_list_id or any( + logger.handlers is not live_list for logger in group_loggers[1:] + ): + usable = False + break + owned_alone = owners.get(id(live_list), 0) == 1 + if owned_alone is not bucket_is_proxy: + usable = False + break + derefed.append(tuple(group_loggers)) + live_lists.append(live_list) + if not usable: + break + if usable: + n_proxy = len(cached.proxy_groups) + return ( + tuple((derefed[i], live_lists[i]) for i in range(n_proxy)), + tuple( + (derefed[i], live_lists[i]) + for i in range(n_proxy, len(derefed)) + ), + ) + + # Collect the non-propagating loggers and their ancestors. + targets: list[logging.Logger] = [] + # Identity set: avoids hashing loggers and gives O(1) membership. + seen: dict[int, logging.Logger] = {} + for candidate in list(current.values()): + if ( + not isinstance(candidate, logging.Logger) + or candidate is root_logger + or candidate.propagate + ): + continue + if id(candidate) not in seen: + seen[id(candidate)] = candidate + targets.append(candidate) + parent = candidate.parent + while parent is not None and parent is not root_logger: + if id(parent) not in seen: + seen[id(parent)] = parent + targets.append(parent) + parent = parent.parent + + # Group by the identity of the handlers list, then classify each + # group by how many *registered* loggers own that list -- including + # root and propagating loggers, which may share a list with a target. + owners = _count_list_owners(current, root_logger) + list_groups: dict[int, tuple[list[logging.Logger], list[logging.Handler]]] = {} + for logger in targets: + key = id(logger.handlers) + entry = list_groups.get(key) + if entry is None: + list_groups[key] = ([logger], logger.handlers) + else: + entry[0].append(logger) + + proxy_groups: list[ + tuple[tuple[logging.Logger, ...], list[logging.Handler]] + ] = [] + direct_groups: list[ + tuple[tuple[logging.Logger, ...], list[logging.Handler]] + ] = [] + for loggers, handlers_list in list_groups.values(): + if owners.get(id(handlers_list), 1) == 1: + proxy_groups.append((tuple(loggers), handlers_list)) + elif any(not logger.propagate for logger in loggers): + # Shared/aliased list: direct attachment for the members which + # need it; a propagating member (an ancestor added by the + # walk) reaches root on its own and gets no copy. + needed = tuple(m for m in loggers if not m.propagate) + direct_groups.append((needed, handlers_list)) + return cls._store_targets( + handler, tuple(proxy_groups), tuple(direct_groups), len(current) + ) + + @classmethod + def _store_targets( + cls, + handler: logging.Handler, + proxy_groups: tuple[_TargetGroup, ...], + direct_groups: tuple[_TargetGroup, ...], + registry_size: int, + ) -> tuple[tuple[_TargetGroup, ...], tuple[_TargetGroup, ...]]: + """Snapshot the groups into the cache and return them. + + The weak back-reference to ``handler`` rejects a recycled ``id()``; + its callback evicts the entry once the handler dies (before the + memory -- and the ``id()`` -- can be reused, so a later entry is + never evicted by the previous handler's death). Only the ``id()`` of + each shared ``handlers`` list is stored, never the list itself: a + strong reference would retain a replaced list and every unrelated + user handler inside it, so staleness is caught on lookup, where the + group's live list must still carry the stored ``id()`` and still be + the current ``handlers`` attribute of every member (#15075 review + round 3). + """ + key = id(handler) + + def _evict( + _ref: weakref.ref[logging.Handler], + key: int = key, + cache: dict[int, _TargetCacheEntry] = cls._target_cache, + ) -> None: + # ``_ref`` is the weakref that fired, i.e. the one stored on the + # entry. Comparing by identity rejects a recycled ``id()`` whose + # entry already belongs to a different handler. + entry = cache.get(key) + if entry is not None and entry.handler_ref is _ref: + del cache[key] + + # The callback must be attached to the weakref that the entry actually + # keeps. A separate throwaway ``weakref.ref(handler, _evict)`` would be + # collected immediately, dropping the callback with it and never + # evicting the entry. + handler_ref = weakref.ref(handler, _evict) + cls._target_cache[key] = _TargetCacheEntry( + handler_ref, + tuple( + (tuple(weakref.ref(logger) for logger in loggers), id(handlers_list)) + for loggers, handlers_list in proxy_groups + ), + tuple( + (tuple(weakref.ref(logger) for logger in loggers), id(handlers_list)) + for loggers, handlers_list in direct_groups + ), + registry_size, + ) + return proxy_groups, direct_groups + + def _detach(self) -> None: + """Release everything this context attached, by identity. + + A proxy shared with an outer (still-active) scope is only released when + the last owner lets go of it, so nested contexts over the same handler + neither detach early nor double-forward. Removal happens from the exact + list the proxy was appended to, so a list shared between two loggers + loses the proxy once and neither logger is left with a stale entry. + """ + for logger in self.attached_loggers: + if any(h is self.handler for h in logger.handlers): + _remove_handler_by_identity(logger, self.handler) + self.attached_loggers.clear() + for shared_list, proxy in self.attached_proxies: + proxy._refcount -= 1 + if proxy._refcount > 0: + # Still owned by an enclosing scope; leave it attached. + continue + if any(h is proxy for h in shared_list): + _remove_handler_by_identity_from_list(shared_list, proxy) + # Stop a retained proxy from forwarding (or keeping the capture + # handler and logger alive) once capture is over. + proxy.close() + self.attached_proxies.clear() + for shared_list, added in self.attached_claims: + # Only a copy this scope appended is removed -- an enclosing + # scope's or a user's copy stays -- and it is removed from the + # remembered list object even if an owner replaced its + # ``handlers`` attribute since entry. + if added and any(h is self.handler for h in shared_list): + _remove_handler_by_identity_from_list(shared_list, self.handler) + self.attached_claims.clear() + def __exit__( self, exc_type: type[BaseException] | None, @@ -379,9 +1055,7 @@ def __exit__( root_logger = logging.getLogger() if self.level is not None: root_logger.setLevel(self.orig_level) - for logger in self.attached_loggers: - logger.removeHandler(self.handler) - self.attached_loggers.clear() + self._detach() 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..bbd2aaf635e --- /dev/null +++ b/testing/logging/test_bound_proxy.py @@ -0,0 +1,1204 @@ +"""Unit-level regression tests for the bound proxy handler (#15064). + +These drive ``catching_logs`` directly, which is where the proxy lifecycle +lives, and assert the exact behaviours the review asked for. 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 gc +import io +import logging +import threading +import weakref + +from _pytest.logging import _BoundProxyHandler +from _pytest.logging import _remove_handler_by_identity +from _pytest.logging import _remove_handler_by_identity_from_list +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(handler_level: int = logging.DEBUG) -> tuple[io.StringIO, logging.Handler]: + stream = io.StringIO() + handler = logging.StreamHandler(stream) + handler.setLevel(handler_level) + return stream, handler + + +def _count(stream: io.StringIO, needle: str) -> int: + return stream.getvalue().count(needle) + + +def _proxy_for(logger: logging.Logger) -> _BoundProxyHandler: + """The bound proxy pytest attached to ``logger`` in the current scope.""" + proxies = [h for h in logger.handlers if isinstance(h, _BoundProxyHandler)] + assert proxies, f"no proxy attached to {logger.name}" + return proxies[0] + + +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 _count(stream, "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 _count(stream, "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 _count(stream, "m") == 1 + assert not [ + h + for h in logging.getLogger("c.plain").handlers + if isinstance(h, _BoundProxyHandler) + ] + + +def test_directly_attached_real_handler_is_not_duplicated() -> None: + """If the real handler is on the logger too, it must not be handled twice.""" + logger = _make_logger("d") + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + logger.addHandler(handler) + logger.warning("m") + assert _count(stream, "m") == 1 + + +def test_shared_handlers_list_is_not_duplicated() -> None: + """Two non-propagating loggers sharing one handlers list.""" + shared: list[logging.Handler] = [] + a = _make_logger("e.a") + b = _make_logger("e.b") + a.handlers = shared + b.handlers = shared + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + a.addHandler(handler) + a.warning("m") + assert _count(stream, "m") == 1 + + +def test_aliased_root_handlers_is_not_duplicated() -> None: + """A logger whose handlers list *is* root.handlers.""" + logger = _make_logger("f") + logger.handlers = logging.getLogger().handlers + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + logger.warning("m") + assert _count(stream, "m") == 1 + + +def test_identity_hostile_user_handler_survives_capture() -> None: + """A user handler that compares equal to everything must not be removed.""" + + class Hostile(logging.Handler): + def __eq__(self, other: object) -> bool: + return True # pragma: no cover - never called; that is the point + + __hash__ = object.__hash__ + + def emit(self, record: logging.LogRecord) -> None: + pass + + logger = _make_logger("g") + hostile = Hostile() + logger.addHandler(hostile) + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + logger.warning("m") + assert _count(stream, "m") == 1 + assert any(h is hostile for h in logger.handlers), "user handler was removed" + assert not any(h is handler for h in logger.handlers) + + +def test_detached_proxy_does_not_forward() -> None: + """A proxy retained after the context must stop forwarding, and must not + keep the capture handler alive.""" + logger = _make_logger("h") + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + proxies = [h for h in logger.handlers if isinstance(h, _BoundProxyHandler)] + assert proxies + proxy = proxies[0] + # Detached on exit: the proxy stops forwarding, and a retained reference + # cannot resurrect capture or keep the capture handler alive. + assert proxy._is_detached + logger.warning("late") + assert _count(stream, "late") == 0 + + +def test_detach_does_not_close_real_handler() -> None: + """The real handler is reused across phases, so it must survive detach.""" + _make_logger("i") + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + pass + assert not getattr(handler, "_closed", False) + handler.handle( + logging.LogRecord("i", logging.WARNING, __file__, 1, "after", (), None) + ) + assert _count(stream, "after") == 1 + + +def test_remove_filter_through_proxy_deactivates_it() -> None: + """Removing a filter through the proxy must stop it applying, both while + capture is live and on the real handler afterwards.""" + logger = _make_logger("rmfilter") + stream, handler = _capture() + + class Reject(logging.Filter): + def __init__(self) -> None: + super().__init__() + self.seen: list[str] = [] + + def filter(self, record: logging.LogRecord) -> bool: + self.seen.append(record.getMessage()) + return False + + reject = Reject() + with catching_logs(handler, level=logging.DEBUG): + proxy = _proxy_for(logger) + proxy.addFilter(reject) + logger.warning("blocked") + assert _count(stream, "blocked") == 0 + assert reject.seen == ["blocked"] + + proxy.removeFilter(reject) + logger.warning("allowed") + assert _count(stream, "allowed") == 1 + + # And it is gone from the real handler too, so a later scope is unaffected. + assert not any(f is reject for f in handler.filters) + handler.handle( + logging.LogRecord("rmfilter", logging.WARNING, __file__, 1, "later", (), None) + ) + assert _count(stream, "later") == 1 + + +def test_set_formatter_through_proxy_applies_to_captured_output() -> None: + """A formatter set through the proxy must shape the captured text.""" + logger = _make_logger("fmt") + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + _proxy_for(logger).setFormatter( + logging.Formatter("FMT:%(levelname)s:%(message)s") + ) + logger.warning("shaped") + assert "FMT:WARNING:shaped" in stream.getvalue() + + +def test_set_formatter_through_proxy_accepts_none() -> None: + """Passing None clears the formatter rather than raising.""" + logger = _make_logger("fmt.none") + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + proxy = _proxy_for(logger) + proxy.setFormatter(logging.Formatter("X:%(message)s")) + proxy.setFormatter(None) + logger.warning("plain") + assert "X:plain" not in stream.getvalue() + assert "plain" in stream.getvalue() + + +def test_emit_directly_on_proxy_forwards_while_non_propagating() -> None: + """Calling ``emit()`` directly (as opposed to ``handle()``) must still + forward for a non-propagating logger and be a no-op once it propagates.""" + logger = _make_logger("direct.emit") + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + proxy = _proxy_for(logger) + proxy.emit( + logging.LogRecord( + "direct.emit", logging.WARNING, __file__, 1, "viaemit", (), None + ) + ) + assert _count(stream, "viaemit") == 1 + + logger.propagate = True + proxy.emit( + logging.LogRecord( + "direct.emit", logging.WARNING, __file__, 1, "skipped", (), None + ) + ) + # A direct emit() does not walk the hierarchy, so the proxy must simply + # not forward: the record was already sent to root's handler by the + # logger's own callHandlers pass, and forwarding again would duplicate + # it. + assert _count(stream, "skipped") == 0 + + +def test_emit_on_detached_proxy_is_a_no_op() -> None: + """A detached proxy must not forward even when emit() is called directly.""" + logger = _make_logger("detached.emit") + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + proxy = _proxy_for(logger) + assert proxy._is_detached + proxy.emit( + logging.LogRecord( + "detached.emit", logging.WARNING, __file__, 1, "nope", (), None + ) + ) + assert _count(stream, "nope") == 0 + + +def test_entry_failure_during_proxy_attachment_rolls_back() -> None: + """A failure *while attaching proxies* must undo the partial setup. + + This is the path the transactional ``__enter__`` exists for: root already + has pytest's handler by the time proxies are attached, so an error there + would otherwise leave it attached with no scope left to remove it. + """ + good = _make_logger("rollback.good") + stream, handler = _capture() + root = logging.getLogger() + + class Exploding(list): # type: ignore[type-arg] + """A handlers list that refuses new entries.""" + + def append(self, item: object) -> None: + raise RuntimeError("boom") + + bad = _make_logger("rollback.bad") + bad.handlers = Exploding() + try: + with pytest.raises(RuntimeError): + with catching_logs(handler, level=logging.DEBUG): + pass # pragma: no cover - __enter__ raises before this line + finally: + # Restore a real list: leaving the poisoned one installed would break + # every later capturing_logs scope, including pytest's own session ones. + bad.handlers = [] + + # Nothing of pytest's is left behind on root or on the loggers. + assert not any(h is handler for h in root.handlers) + assert not [h for h in good.handlers if isinstance(h, _BoundProxyHandler)] + assert _count(stream, "anything") == 0 + + +def test_detached_proxy_ignores_direct_handle_call() -> None: + """Calling ``handle()`` on a retained proxy after detach does nothing. + + The proxy is gone from ``logger.handlers`` once capture ends, so a later + ``logger.warning()`` never reaches it. This covers someone who kept a + reference to the proxy and calls it directly: it must not resurrect + capture, nor raise. + """ + logger = _make_logger("retained") + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + proxy = _proxy_for(logger) + proxy.handle( + logging.LogRecord("retained", logging.WARNING, __file__, 1, "late", (), None) + ) + assert _count(stream, "late") == 0 + + +def test_finite_nested_record_is_captured_not_dropped() -> None: + """A real handler logging one finite nested record must not lose it. + + The proxy forwards into the real handler, which may log one further record + from its own ``emit()``. That nested record is a distinct record and the + un-proxied handler would have captured it, so it must be captured here too: + the real handler's lock is an ``RLock``, so the finite nesting terminates + on its own and needs no reentrancy guard. + + A handler that logs *unconditionally* from its own ``emit()`` still + recurses, but it does so identically without a proxy (the stdlib has the + same reentrant lock), so that is not something the proxy changes. + """ + emitted: list[str] = [] + + class LoggingBack(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + emitted.append(record.getMessage()) + if record.getMessage() == "outer": + # One finite nested record, then stop. + logger.warning("inner") + + logger = _make_logger("recursive") + real = LoggingBack() + + with catching_logs(real, level=logging.DEBUG): + logger.warning("outer") + + assert emitted == ["outer", "inner"] + + +def test_setlevel_on_proxy_reaches_the_real_handler() -> None: + """``setLevel`` on the proxy must reach the real handler. + + The proxy is a view of the real handler, so a level written through it is + the level that is actually in force -- dropping the write would leave + ``logger.handlers`` advertising a level that does not apply. + """ + logger = _make_logger("j") + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + proxy = next(h for h in logger.handlers if isinstance(h, _BoundProxyHandler)) + proxy.setLevel(logging.CRITICAL) + assert proxy.level == logging.CRITICAL + logger.warning("m") + # The record is now below the real handler's level, so it is not handled. + assert _count(stream, "m") == 0 + + +def test_proxy_does_not_deadlock_with_real_handler_across_threads() -> None: + """The proxy must not hold its own lock while invoking the real handler. + + ``logging.Handler.handle`` holds ``self.lock`` across ``emit``. If the proxy + forwards to the real handler from there, it takes the real handler's lock + while already holding its own, giving a ``proxy -> real`` order that another + thread inverts as ``real -> proxy`` -- an ABBA deadlock. + + The check is the lock ORDER (deterministic), not the absence of a hang: a + thread that already holds the real handler's lock must never find the proxy + lock held by a thread waiting for the real one. + """ + logger = _make_logger("k") + handler = logging.StreamHandler(io.StringIO()) + handler.setLevel(logging.DEBUG) + trace: list[str] = [] + trace_lock = threading.Lock() + + def note(message: str) -> None: + with trace_lock: + trace.append(message) + + class TracingRLock: + """An RLock that records acquisition order across threads. + + Implements the context-manager protocol as well, because + ``logging.Handler.handle()`` uses ``with self.lock:`` on newer + Pythons (3.14+). + """ + + def __init__(self, name: str) -> None: + self._rlock = threading.RLock() + self._name = name + + def __enter__(self) -> None: + self.acquire() # pragma: no cover - only taken on Python 3.14+ + + def __exit__(self, *exc: object) -> None: + self.release() # pragma: no cover - only taken on Python 3.14+ + + def acquire(self, *args: object, **kwargs: object) -> bool: + note(f"{threading.current_thread().name}:want:{self._name}") + acquired = self._rlock.acquire() + note(f"{threading.current_thread().name}:got:{self._name}") + return acquired + + def release(self) -> None: + self._rlock.release() + note(f"{threading.current_thread().name}:rel:{self._name}") + + handler.lock = TracingRLock("REAL") # type: ignore[assignment] + proxy = _BoundProxyHandler(logger, handler) + proxy.lock = TracingRLock("PROXY") # type: ignore[assignment] + logger.addHandler(proxy) + + ready = threading.Event() + + def other_thread() -> None: + # Simulate a thread already inside the real handler (it holds its lock). + handler.acquire() + try: + ready.set() + logger.warning("from-B") + finally: + handler.release() + + def first_thread() -> None: + ready.wait(5) + logger.warning("from-A") + + other = threading.Thread(target=other_thread, daemon=True) + first = threading.Thread(target=first_thread, daemon=True) + other.start() + first.start() + first.join(timeout=10) + other.join(timeout=10) + + assert not first.is_alive(), "deadlock: first thread did not finish" + assert not other.is_alive(), "deadlock: other thread did not finish" + + # The inversion: a thread holding PROXY while waiting for REAL. + for index, entry in enumerate(trace[:-1]): + if entry.endswith(":got:PROXY") and trace[index + 1].endswith(":want:REAL"): + pytest.fail( # pragma: no cover - only on a reintroduced deadlock + "proxy held its own lock while invoking the real handler: " + f"{entry} -> {trace[index + 1]}" + ) + + +def test_entry_rolls_back_when_logger_is_unhashable() -> None: + """A custom Logger with ``__hash__ = None`` must not break entry, and a + failure must not leave pytest's handler attached to root. + + The logger's class is swapped in place because ``catching_logs`` walks + ``manager.loggerDict``; the original class is restored in the ``finally`` + so the poisoned logger cannot leak into other tests. + """ + + class UnhashableLogger(logging.Logger): + __hash__ = None # type: ignore[assignment] + + _make_logger("l") + broken = logging.getLogger("l.broken") + broken.handlers.clear() + broken.propagate = False + original_class = broken.__class__ + broken.__class__ = UnhashableLogger + stream, handler = _capture() + try: + with catching_logs(handler, level=logging.DEBUG): + broken.warning("m") + finally: + broken.__class__ = original_class + broken.handlers.clear() + assert _count(stream, "m") == 1 + assert not any(h is handler for h in logging.getLogger().handlers) + + +def test_entry_failure_rolls_back_root_attachment() -> None: + """A failure while attaching proxies must not leave pytest's handler on + root with no scope left to remove it. + + The failure is injected by passing a handler whose ``setLevel`` raises -- + that happens inside ``__enter__`` before any proxy work -- and the logger + population is restored by the autouse ``_clean_logging`` fixture, so no + state leaks into the surrounding session. + """ + + class FailingLevel(logging.Handler): + def setLevel(self, level: int | str) -> None: + raise RuntimeError("boom") + + def emit(self, record: logging.LogRecord) -> None: + pass # pragma: no cover - setLevel raises before any record + + _make_logger("o") + handler = FailingLevel() + root = logging.getLogger() + with pytest.raises(RuntimeError): + with catching_logs(handler, level=logging.DEBUG): + pass # pragma: no cover - __enter__ raises before this line + assert not any(h is handler for h in root.handlers) + + +def test_nested_contexts_with_same_target_capture_once() -> None: + """Overlapping contexts sharing one handler must not duplicate records.""" + logger = _make_logger("m") + stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + with catching_logs(handler, level=logging.DEBUG): + logger.warning("m") + assert _count(stream, "m") == 1 + + +def test_exception_inside_context_still_detaches() -> None: + """__exit__ must clean up when the body raises.""" + logger = _make_logger("n") + _stream, handler = _capture() + with pytest.raises(RuntimeError): + with catching_logs(handler, level=logging.DEBUG): + raise RuntimeError("boom") + assert not any(h is handler for h in logging.getLogger().handlers) + assert not [h for h in logger.handlers if isinstance(h, _BoundProxyHandler)] + + +def test_logger_created_after_first_capture_still_gets_a_proxy() -> None: + """The proxy-target cache must not hide a logger that appears later. + + pytest re-enters ``catching_logs`` for every test phase, so the target set + is cached; a logger created between two phases must still be picked up. + """ + _make_logger("cache.first") + _stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + pass + + late = _make_logger("cache.late") + with catching_logs(handler, level=logging.DEBUG): + proxies = [h for h in late.handlers if isinstance(h, _BoundProxyHandler)] + assert len(proxies) == 1 + late.warning("m") + assert _count(_stream, "m") == 1 + + +def test_removed_logger_is_not_held_by_the_target_cache() -> None: + """A logger deleted from ``loggerDict`` must not stay cached forever.""" + _make_logger("cache.gone") + _stream, handler = _capture() + with catching_logs(handler, level=logging.DEBUG): + pass + + gone = logging.getLogger("cache.gone") + del logging.getLogger().manager.loggerDict["cache.gone"] + gone.handlers.clear() + + with catching_logs(handler, level=logging.DEBUG): + assert not [h for h in gone.handlers if isinstance(h, _BoundProxyHandler)] + + +def test_dead_handler_is_evicted_from_the_target_cache() -> None: + """The entry's weakref callback must evict it once the handler dies. + + The callback has to be attached to the weakref the entry actually keeps -- + a separate throwaway ``weakref.ref(handler, cb)`` is collected immediately + and never fires. Without the eviction the entry lingers until its ``id()`` + is recycled. + """ + from _pytest.logging import catching_logs as _catching_logs + + _make_logger("cache.dead") + _stream, handler = _capture() + + with _catching_logs(handler, level=logging.DEBUG): + pass + + cache = _catching_logs._target_cache + key = id(handler) + assert key in cache, "the scope should have cached the target snapshot" + + del handler + gc.collect() + + assert key not in cache, "a dead handler's cache entry must be evicted" + + +def test_eviction_callback_tolerates_an_already_removed_entry() -> None: + """A dead handler whose cache entry is already gone must not raise. + + The entry can disappear before the handler does (a later scope, an explicit + clear). The callback's ``cache.get`` then returns ``None`` and it must fall + through instead of touching the missing entry. + """ + from _pytest.logging import catching_logs as _catching_logs + + _make_logger("cache.precleared") + _stream, handler = _capture() + + with _catching_logs(handler, level=logging.DEBUG): + pass + + cache = _catching_logs._target_cache + key = id(handler) + assert key in cache + + # Hold the entry's weakref so the callback still has a live owner after the + # entry itself is removed, then let the handler die: the callback fires with + # ``cache.get`` returning ``None`` and must fall through. + entry_ref = cache[key].handler_ref + del cache[key] + del handler + gc.collect() + + assert entry_ref() is None + assert key not in cache + + +# -------------------------------------------------------------------------- +# Regression coverage named in review #2 on #15075. Each of these fails on the +# previous head (or on the merge base) and pins the corrected behaviour. + + +def test_shared_handlers_list_transition_does_not_lose_sibling() -> None: + """Two loggers sharing one ``handlers`` list, one flips ``propagate``. + + A list with more than one owner gets the real handler attached directly + (the conservative fallback -- a single proxy cannot know which owner a + shared-list visit is for), so the sibling's record is captured through + the list itself and is never lost when the other owner starts + propagating. + """ + a = _make_logger("shared.a") + b = _make_logger("shared.b") + shared: list[logging.Handler] = [] + a.handlers = shared + b.handlers = shared + + stream, handler = _capture() + with catching_logs(handler): + # Direct attachment for the shared list, not a proxy: the real + # handler is in the list once, claimed by identity. + assert not [h for h in shared if isinstance(h, _BoundProxyHandler)] + assert sum(h is handler for h in shared) == 1 + a.propagate = True + b.error("from-b") + + assert _count(stream, "from-b") == 1 + + +def test_shared_handlers_list_transition_keeps_merge_base_behaviour() -> None: + """Shared lists deliberately retain the pre-proxy transition limitation. + + With direct attachment on a shared list, a sibling that stays + non-propagating always has its record captured (no lost records), while + a record from the owner that flipped ``propagate`` mid-scope is handled + once via the list and once via root -- exactly what the merge base did. + Proxying shared lists cannot do better without per-owner routing the + stdlib dispatch does not provide, so the conservative behaviour wins. + """ + a = _make_logger("dup.a") + b = _make_logger("dup.b") + shared: list[logging.Handler] = [] + a.handlers = shared + b.handlers = shared + + stream, handler = _capture() + with catching_logs(handler): + a.propagate = True + b.error("only-once") + a.error("via-root") + + assert _count(stream, "only-once") == 1 + # Merge-base behaviour for a mid-scope False->True flip on a shared + # list: the list attachment and the root attachment both see it. + assert _count(stream, "via-root") == 2 + + +def test_finite_nested_record_from_real_handler_emit() -> None: + """A handler logging one distinct record from its own ``emit()`` keeps it. + + The record is finite, so the reentrant lock terminates the nesting on its + own; the proxy must not swallow it. + """ + emitted: list[str] = [] + + class Nested(logging.Handler): + def __init__(self) -> None: + super().__init__() + self.once = False + + def emit(self, record: logging.LogRecord) -> None: + emitted.append(record.getMessage()) + if not self.once and record.getMessage() == "outer": + self.once = True + logger.warning("inner") + + logger = _make_logger("nested.emit") + with catching_logs(Nested()): + logger.error("outer") + + assert emitted == ["outer", "inner"] + + +def test_pre_existing_target_filter_survives_teardown() -> None: + """A filter already on the real handler is not removed at detach. + + Adding it again through the proxy must not make the proxy claim ownership + of it: teardown would then delete a filter the caller installed. + """ + logger = _make_logger("preexisting") + _stream, handler = _capture() + pre = logging.Filter("pre-existing") + handler.addFilter(pre) + + with catching_logs(handler): + proxy = _proxy_for(logger) + proxy.addFilter(pre) # same object, added again via the proxy + assert any(f is pre for f in handler.filters) + + assert any(f is pre for f in handler.filters), ( + "teardown removed a filter the proxy did not add" + ) + + +def test_pre_existing_filter_still_applies_after_teardown() -> None: + """The surviving filter must keep filtering, not merely survive.""" + logger = _make_logger("preexisting2") + stream, handler = _capture() + + def only_warnings(record: logging.LogRecord) -> bool: + return record.levelno >= logging.WARNING + + handler.addFilter(only_warnings) + with catching_logs(handler): + logger.info("dropped") + logger.warning("kept") + assert _count(stream, "dropped") == 0 + assert _count(stream, "kept") == 1 + # Still in force after the context ended. + logger.info("dropped-too") + assert _count(stream, "dropped-too") == 0 + + +def test_proxy_delegates_level_filters_and_formatter() -> None: + """The proxy is a view: level, filters and formatter are the real ones.""" + logger = _make_logger("view") + _stream, handler = _capture() + with catching_logs(handler): + proxy = _proxy_for(logger) + # level is a view in both directions + assert proxy.level == logging.DEBUG + handler.setLevel(logging.ERROR) + assert proxy.level == logging.ERROR + # filters list is the real handler's + f = logging.Filter("f") + proxy.addFilter(f) + assert any(x is f for x in handler.filters) + assert proxy.filters is handler.filters + # formatter is the real handler's + proxy.setFormatter(logging.Formatter("VIEW:%(message)s")) + assert handler.formatter is not None + assert handler.formatter._fmt == "VIEW:%(message)s" + proxy.removeFilter(f) + assert not any(x is f for x in handler.filters) + + +def test_detach_releases_strong_references_and_is_inert() -> None: + """Final detach clears the proxy's strong refs and makes it inert.""" + logger = _make_logger("detach") + proxies: list[_BoundProxyHandler] = [] + with catching_logs(logging.StreamHandler(io.StringIO())): + proxies.append(_proxy_for(logger)) + proxy = proxies[0] + + assert proxy._is_detached + assert proxy.logger is None + assert proxy.real_handler is None + assert proxy._group == () + # A late handle() is a safe no-op, not an AttributeError. + record = logging.LogRecord("detach", logging.ERROR, "p", 1, "late", (), None) + assert proxy.handle(record) is False + + +def test_capture_handler_is_not_held_by_detached_proxy() -> None: + """A detached proxy must not pin the capture handler alive.""" + import gc + import weakref + + logger = _make_logger("gcpin") + handler = logging.StreamHandler(io.StringIO()) + with catching_logs(handler): + # Assert the proxy exists so the handler is exercised; the point of the + # test is what happens to ``handler`` once the proxy is detached. + assert _proxy_for(logger) is not None + logger.error("x") + ref = weakref.ref(handler) + del handler + gc.collect() + assert ref() is None, "detached proxy kept the capture handler alive" + + +def test_unhashable_logger_can_be_captured() -> None: + """A ``Logger`` subclass with ``__hash__ = None`` must not break entry. + + Keying the proxy-target cache by the logger would raise ``TypeError`` here. + """ + root = logging.getLogger() + + class UnhashableLogger(logging.Logger): + __hash__ = None # type: ignore[assignment] + + logger = UnhashableLogger("unhashable") + logger.setLevel(logging.DEBUG) + logger.propagate = False + root.manager.loggerDict["unhashable"] = logger + + stream, handler = _capture() + with catching_logs(handler): + logger.error("from-unhashable") + + assert _count(stream, "from-unhashable") == 1 + + +def test_nested_contexts_share_one_direct_claim_for_shared_list() -> None: + """Overlapping scopes over a shared list claim one direct attachment.""" + a = _make_logger("nested_ctx.a") + b = _make_logger("nested_ctx.b") + shared: list[logging.Handler] = [] + a.handlers = shared + b.handlers = shared + + stream, handler = _capture() + with catching_logs(handler): + with catching_logs(handler): + # Shared list -> direct attachment; the inner scope reuses the + # outer scope's copy instead of adding a second one. + assert not [h for h in shared if isinstance(h, _BoundProxyHandler)] + assert sum(h is handler for h in shared) == 1, ( + "nested scopes must share one claim" + ) + b.error("inner-record") + # Outer scope still active: the claim is still attached. + assert any(h is handler for h in shared) + b.error("outer-record") + + # Distinct needles: "inner-record" is a substring of neither, and the two + # records are counted separately. + assert _count(stream, "inner-record") == 1 + assert _count(stream, "outer-record") == 1 + # Both contexts exited: nothing left behind. + assert not [h for h in shared if h is handler] + + +def test_handlers_list_replaced_between_catches_is_not_stale() -> None: + """Replacing a logger's ``handlers`` list between captures is detected. + + The target cache may not attach to the list that was cached during the + first capture; the second capture must reach the logger's *current* + list. The cache only stores the ``id()`` of each list, so the replaced + list (and any unrelated user handler inside it) is never retained. + """ + logger = _make_logger("cache.replace") + old_list: list[logging.Handler] = logger.handlers + user_handler = logging.NullHandler() + old_list.append(user_handler) + + stream, handler = _capture() + with catching_logs(handler): + logger.error("first") + assert _count(stream, "first") == 1 + + fresh: list[logging.Handler] = [] + logger.handlers = fresh + with catching_logs(handler): + assert any(isinstance(h, _BoundProxyHandler) for h in fresh), ( + "capture must attach to the logger's current list" + ) + logger.error("second") + assert _count(stream, "second") == 1 + # The cache must not retain the replaced list. Plain lists are not + # weak-referenceable, so use the user handler still sitting in it as a + # liveness probe: if the cache held the old list, the handler (and the + # list) could not be collected. + handler_alive = weakref.ref(user_handler) + del old_list, logger, user_handler + gc.collect() + assert handler_alive() is None + + +# -------------------------------------------------------------------------- +# Coverage for the proxy's direct property assignments, defensive guards, +# cache revalidation mismatches, and the identity-based removal helpers. + + +def test_direct_level_assignment_through_proxy_reaches_real_handler() -> None: + """``proxy.level = X`` (not just ``setLevel()``) must write through.""" + logger = _make_logger("prop.level") + stream, handler = _capture() + with catching_logs(handler): + proxy = _proxy_for(logger) + proxy.level = logging.CRITICAL + assert handler.level == logging.CRITICAL + assert proxy.level == logging.CRITICAL + logger.warning("quiet") + assert _count(stream, "quiet") == 0 + + +def test_direct_filters_assignment_through_proxy_replaces_real_list() -> None: + """``proxy.filters = [...]`` must replace the real handler's filter list.""" + logger = _make_logger("prop.filters") + stream, handler = _capture() + keep = logging.Filter() + with catching_logs(handler): + proxy = _proxy_for(logger) + proxy.filters = [keep] + assert handler.filters is proxy.filters + assert any(f is keep for f in handler.filters) + logger.warning("seen") + assert _count(stream, "seen") == 1 + + +def test_direct_formatter_assignment_through_proxy_shapes_output() -> None: + """``proxy.formatter = ...`` must write through to the real handler.""" + logger = _make_logger("prop.formatter") + stream, handler = _capture() + with catching_logs(handler): + proxy = _proxy_for(logger) + proxy.formatter = logging.Formatter("PROP:%(message)s") + assert proxy.formatter is handler.formatter + logger.warning("shaped") + proxy.formatter = None + assert proxy.formatter is None + assert "PROP:shaped" in stream.getvalue() + + +def test_detached_proxy_property_accessors_are_inert() -> None: + """Every proxy property and setter must be a safe no-op once detached.""" + logger = _make_logger("detached.props") + stream, handler = _capture() + with catching_logs(handler): + proxy = _proxy_for(logger) + assert proxy._is_detached + assert proxy.level == logging.NOTSET + proxy.level = logging.ERROR # no target to write to + assert proxy.filters == [] + proxy.filters = [logging.Filter()] # no target to write to + assert proxy.formatter is None + proxy.formatter = logging.Formatter() # no target to write to + proxy.setLevel(logging.ERROR) # must not raise + proxy.addFilter(logging.Filter()) # must not raise + proxy.removeFilter(logging.Filter()) # must not raise + proxy.setFormatter(logging.Formatter()) # must not raise + logger.warning("after") + assert _count(stream, "after") == 0 + + +def test_remove_filter_scans_past_non_matching_filters() -> None: + """``removeFilter`` must scan by identity and tolerate a missing filter.""" + logger = _make_logger("rmfilter.scan") + _stream, handler = _capture() + first = logging.Filter("first") + target = logging.Filter("target") + absent = logging.Filter("absent") + handler.addFilter(first) + with catching_logs(handler): + proxy = _proxy_for(logger) + proxy.addFilter(target) + # 'first' is skipped by identity comparison before 'target' matches. + proxy.removeFilter(target) + assert any(f is first for f in handler.filters) + assert not any(f is target for f in handler.filters) + # No match anywhere: a no-op, not an error. + proxy.removeFilter(absent) + assert any(f is first for f in handler.filters) + + +def test_proxy_handle_guards_missing_references() -> None: + """``handle()`` must return False, not crash, when a reference is gone. + + ``close()`` clears the references together with the detached flag, so a + half-cleared proxy is a defensive state the guards exist for; it is + assembled directly here because no public path produces it. + """ + logger = _make_logger("guard.refs") + real = logging.StreamHandler(io.StringIO()) + proxy = _BoundProxyHandler(logger, real) + record = logging.LogRecord( + "guard.unrelated", logging.ERROR, __file__, 1, "x", (), None + ) + proxy.real_handler = None + assert proxy.handle(record) is False # real handler missing + assert proxy.level == logging.NOTSET + assert proxy.filters == [] + assert proxy.formatter is None + proxy.real_handler = real + proxy.logger = None # emitter is outside the group and the binding is gone + assert proxy.handle(record) is False + + +def test_proxy_forwards_record_from_a_propagating_child() -> None: + """A record from a propagating child reaches the parent's proxy. + + The child is not part of the proxy's group, so the *bound* logger's + ``propagate`` decides whether the proxy forwards the record. + """ + parent = _make_logger("route.par") + kid = _make_logger("route.par.kid", propagate=True) + stream, handler = _capture() + with catching_logs(handler): + assert not [h for h in kid.handlers if isinstance(h, _BoundProxyHandler)] + assert _proxy_for(parent) + kid.error("via-parent") + assert _count(stream, "via-parent") == 1 + + +def test_stale_cache_entry_is_rejected_via_the_back_reference() -> None: + """A cached snapshot whose handler died is never reused. + + The target cache is keyed by ``id(handler)``, which the interpreter can + recycle for a different handler. The stored ``handler_ref`` back- + reference resolves to the *original* handler, so after it dies the guard + at lookup time rejects the stale snapshot and forces a rebuild for the + new handler -- asserted here directly on the cache, since reliably + reproducing a recycled id is not possible. + """ + _make_logger("stale.first") + handler = logging.StreamHandler(io.StringIO()) + with catching_logs(handler): + pass + key = id(handler) + assert catching_logs._target_cache[key].handler_ref() is handler + + # Simulate the handler's death for the cache without freeing the object + # (freeing it could let its id be recycled under an unrelated handler). + stale_ref = catching_logs._target_cache[key].handler_ref + other = logging.StreamHandler(io.StringIO()) + with catching_logs(other): + # ``other`` may or may not be cached under a different key; what must + # hold is that the original snapshot can no longer validate: its + # back-reference would not resolve to whichever handler looks it up. + # The entry for ``handler`` is still present (asserted above and not + # evicted while ``handler`` is alive), so no guard is needed here. + entry = catching_logs._target_cache[key] + assert entry.handler_ref is stale_ref + assert entry.handler_ref() is handler # still alive here + + # Once the handler is really gone, the back-reference resolves to nothing + # and the entry can never pass the ``handler_ref() is handler`` check. + gone = weakref.ref(handler) + del handler + gc.collect() + assert gone() is None + assert stale_ref() is None + + +def test_owner_count_change_invalidates_the_proxy_bucket() -> None: + """A cached single-owner list that gains a second owner forces a rebuild. + + The proxy-vs-direct classification depends on list *ownership*, not just + list identity: when another logger starts sharing the cached list without + the population changing, the cached entry must be rejected and the list + re-classified for direct attachment. + """ + a = _make_logger("owners.a") + b = _make_logger("owners.b", propagate=True) + _stream, handler = _capture() + with catching_logs(handler): + assert any(isinstance(h, _BoundProxyHandler) for h in a.handlers) + b.handlers = a.handlers # a second owner, without a population change + with catching_logs(handler): + assert not [h for h in a.handlers if isinstance(h, _BoundProxyHandler)] + assert sum(h is handler for h in a.handlers) == 1 + a.error("owned") + assert _count(_stream, "owned") == 1 + + +def test_ancestor_already_collected_is_not_added_twice() -> None: + """A non-propagating logger reached as an ancestor first is collected once. + + ``loggerDict`` iteration order decides which visit comes first; re- + inserting a parent behind its child simulates a parent that the ancestor + walk had already picked up, pinning the dedup-by-identity branch. + """ + root = logging.getLogger() + parent = _make_logger("twice") + kid = _make_logger("twice.kid") + registry = root.manager.loggerDict + # Re-insert the parent last so iteration visits the child first: the + # child's ancestor walk collects the parent into ``seen``, and when + # iteration later reaches the parent it is already collected, taking the + # dedup-by-identity branch. + del registry["twice"] + registry["twice"] = parent + + stream, handler = _capture() + with catching_logs(handler): + proxies = [h for h in parent.handlers if isinstance(h, _BoundProxyHandler)] + assert len(proxies) == 1 + parent.error("from-parent") + kid.error("from-kid") + assert _count(stream, "from-parent") == 1 + assert _count(stream, "from-kid") == 1 + + +def test_shared_list_group_with_only_propagating_members_is_skipped() -> None: + """A direct-classified group whose needed members are empty attaches nothing. + + The ancestor walk adds a propagating parent to the targets; if its list + is shared with another logger the group lands in the direct bucket, but + every member propagates and reaches root on its own, so the filtered + ``needed`` set is empty and the group is skipped entirely. + """ + parent = _make_logger("skip.par", propagate=True) + kid = _make_logger("skip.par.kid") # non-propagating + other = _make_logger("skip.other", propagate=True) + other.handlers = parent.handlers # two owners of parent's list + + stream, handler = _capture() + with catching_logs(handler): + assert not [h for h in parent.handlers if isinstance(h, _BoundProxyHandler)] + assert not any(h is handler for h in parent.handlers) + kid.error("kid") + parent.error("from-par") + assert _count(stream, "kid") == 1 + assert _count(stream, "from-par") == 1 + + +def test_parent_and_child_sharing_one_handlers_list() -> None: + """Parent and child sharing one ``handlers`` list get one direct claim.""" + parent = _make_logger("pcs") + child = _make_logger("pcs.kid") + shared: list[logging.Handler] = [] + parent.handlers = shared + child.handlers = shared + stream, handler = _capture() + with catching_logs(handler): + assert sum(h is handler for h in shared) == 1 + child.error("from-child") + parent.error("from-parent") + assert _count(stream, "from-child") == 1 + assert _count(stream, "from-parent") == 1 + + +def test_proxy_removed_externally_is_still_closed_on_exit() -> None: + """A proxy pulled out of the list mid-scope must not break the teardown. + + ``_detach`` removes the proxy by identity; if a user already removed it, + the removal is skipped and the proxy is closed all the same. + """ + logger = _make_logger("strand") + stream, handler = _capture() + with catching_logs(handler): + proxy = _proxy_for(logger) + _remove_handler_by_identity_from_list(logger.handlers, proxy) + assert proxy._is_detached + logger.warning("late") + assert _count(stream, "late") == 0 + + +def test_identity_removers_leave_foreign_lists_alone() -> None: + """Both identity-based removers are no-ops when the handler is absent.""" + logger = _make_logger("removers") + keep = logging.NullHandler() + logger.addHandler(keep) + _remove_handler_by_identity(logger, logging.NullHandler()) + assert len(logger.handlers) == 1 + assert logger.handlers[0] is keep + + listed: list[logging.Handler] = [keep] + _remove_handler_by_identity_from_list(listed, logging.NullHandler()) + assert listed == [keep] diff --git a/testing/logging/test_fixture.py b/testing/logging/test_fixture.py index 95c0f44b7f5..7e1896cdc31 100644 --- a/testing/logging/test_fixture.py +++ b/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( """ diff --git a/testing/logging/test_reporting.py b/testing/logging/test_reporting.py index 11013fdd749..4931c404847 100644 --- a/testing/logging/test_reporting.py +++ b/testing/logging/test_reporting.py @@ -2,8 +2,10 @@ from __future__ import annotations import io +import logging import os import re +import sys from typing import cast from _pytest.capture import CaptureManager @@ -1287,6 +1289,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( @@ -1579,3 +1612,167 @@ def test_foo(): assert "info text going to logger" not in contents assert "warning text going to logger" not in contents assert "error text going to logger" in contents + + +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) + + +@pytest.mark.skipif( + not hasattr(logging, "LogRecord") or sys.version_info < (3, 12), + reason="filter() returning a replacement record is only honoured from 3.12", +) +def test_report_capture_replacement_record_on_non_propagating_logger( + pytester: Pytester, +) -> None: + """A filter returning a replacement record still applies through the proxy + (#15064). + + ``logging.Handler.handle()`` only honours a replacement record (a filter + returning a ``LogRecord`` rather than a bool) from Python 3.12; on 3.10 and + 3.11 the return value is used as a plain truthiness test by the stdlib + itself, so the record is emitted unchanged. Skipped there accordingly. + """ + pytester.makepyfile( + """ + import logging + + logger = logging.getLogger("replace.example") + logger.propagate = False + + def replace(record): + return logging.makeLogRecord({ + "msg": "replaced message", + "levelname": "ERROR", + "levelno": logging.ERROR, + }) + + def test_replacement(caplog): + for handler in logger.handlers: + handler.addFilter(replace) + logger.warning("original message") + assert caplog.messages == ["replaced message"] + """ + ) + + result = pytester.runpytest() + result.assert_outcomes(passed=1)