diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 6bc98e1..70d7101 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -38,6 +38,8 @@ jobs: python: - "3.11" - "3.12" + - "3.13" + - "3.14" nox-session: # To speed things up a bit we use the special ci_checks_max session # that uses the same venv to run multiple linting sessions @@ -115,6 +117,8 @@ jobs: python: - "3.11" - "3.12" + - "3.13" + - "3.14" runs-on: ${{ matrix.platform }} steps: diff --git a/README.md b/README.md index 249b9f9..10679d1 100644 --- a/README.md +++ b/README.md @@ -8,8 +8,8 @@ Core utilities to complement Python's standard library. This library provides essential building blocks for Python applications, including mathematical -utilities, datetime constants, typing helpers, strongly-typed identifiers, and -module introspection tools. +utilities, datetime constants, typing helpers, strongly-typed identifiers, +module introspection tools, and warning and deprecation helpers. The `frequenz-core` library is designed to be lightweight, type-safe, and follow modern Python best practices. It fills common gaps in the standard @@ -19,8 +19,8 @@ library with utilities that are frequently needed across different projects. The following platforms are officially supported (tested): -- **Python:** 3.11 -- **Operating System:** Ubuntu Linux 20.04 +- **Python:** 3.11, 3.12, 3.13, 3.14 +- **Operating System:** Ubuntu Linux 24.04 - **Architectures:** amd64, arm64 ## Installation @@ -92,7 +92,7 @@ positive = Interval(0, None) # [0, ∞] assert 1000 in positive # True ``` -### `Enum` with deprecated members +### `Enum` Utilities Define enums with deprecated members that raise deprecation warnings when accessed: @@ -158,6 +158,77 @@ assert describe(1.5) == "number 1.5" assert describe(None) == "nothing" ``` +### Warnings/Deprecations Utilities + +Silence a warning around a call without the side effect +[`warnings.catch_warnings`](https://docs.python.org/3/library/warnings.html#warnings.catch_warnings) +has, which is to reset the deduplication history of the whole program, so +every warning already shown is shown again +([python/cpython#73858](https://github.com/python/cpython/issues/73858)): + +```python +import warnings + +from frequenz.core.warnings import ignoring_deprecations + +def legacy_parse(raw: str) -> int: + warnings.warn("legacy_parse() is deprecated", DeprecationWarning, stacklevel=2) + return int(raw) + +def parse(raw: str) -> int: + # Only around the call that reaches the deprecated symbol. + with ignoring_deprecations(): + return legacy_parse(raw) + +with warnings.catch_warnings(record=True, action="default") as shown: + for _ in range(10): + warnings.warn("said once", UserWarning) + parse("1") + +assert len(shown) == 1 # ✅ `catch_warnings()` in `parse()` would show it 10 times +``` + +Keep the old import path of a symbol that moved working, serving the very same +object so `isinstance` keeps working through both paths: + +```python +from typing import TYPE_CHECKING, TypeAlias + +from frequenz.core.warnings import deprecated_aliases + +if TYPE_CHECKING: + # Type checkers can't see the runtime `__getattr__` in the `else` branch. + from decimal import Decimal as _Decimal + from fractions import Fraction as _Fraction + + Decimal: TypeAlias = _Decimal + Rational: TypeAlias = _Fraction +else: + # In the `else`, never at module level: a `__getattr__` mypy can see makes + # every name it doesn't know in this module `Any`, typos in downstream + # imports included. + __getattr__ = deprecated_aliases( + __name__, + { + "Decimal": "decimal", # decimal.Decimal + "Rational": "fractions:Fraction", # Renamed on the way out + }, + ) +``` + +And check in a test that a piece of code doesn't warn, without the `"error"` +filter that would make `warnings.warn()` raise inside the code under test: + +```python +from frequenz.core.warnings import asserting_no_deprecations + +def parse(raw: str) -> int: + return int(raw) + +with asserting_no_deprecations(): + assert parse("1") == 1 +``` + ### Strongly-Typed IDs Create type-safe identifiers for different entities: diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 67f33e6..23775bd 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -1,17 +1,11 @@ # Frequenz Core Library Release Notes -## Summary - - - -## Upgrading - - - ## New Features - +A new `frequenz.core.warnings` module with: + +- `ignoring_warnings()` and `ignoring_deprecations()` to silence warnings/deprecations around a piece of code. Unlike `warnings.catch_warnings()`, entering and leaving it doesn't reset the warnings deduplication history of the program, so warnings already shown are not shown again, working around [python/cpython#73858](https://github.com/python/cpython/issues/73858). It costs about the same as the standard library block, so it is also usable on a hot path, where repairing the damage afterwards would not be. -## Bug Fixes +- `asserting_no_warnings()`, and its `asserting_no_deprecations()` shortcut, fail when the code in the block raises a matching warning/deprecation, listing each one with the place it came from. They are meant for tests, and are a better tool than an `"error"` filter, which makes `warnings.warn()` raise inside the code under test and so changes the very behaviour the test is checking. - +- `deprecated_aliases()` builds a module `__getattr__` that warns when a symbol that moved to another module is reached through its old import path, serving the very same object so `isinstance` keeps working through both paths. diff --git a/pyproject.toml b/pyproject.toml index 2ea7884..ec13c51 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -161,7 +161,11 @@ disable = [ ] [tool.pytest.ini_options] -addopts = "-W=all -Werror -Wdefault::DeprecationWarning -Wdefault::PendingDeprecationWarning -vv" +filterwarnings = [ + "error", + "once::DeprecationWarning", + "once::PendingDeprecationWarning", +] testpaths = ["tests", "src"] asyncio_mode = "auto" asyncio_default_fixture_loop_scope = "function" diff --git a/src/frequenz/core/warnings/__init__.py b/src/frequenz/core/warnings/__init__.py new file mode 100644 index 0000000..6ad19a1 --- /dev/null +++ b/src/frequenz/core/warnings/__init__.py @@ -0,0 +1,68 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Warnings utilities. + +This module provides [`ignoring_warnings`][.ignoring_warnings], a context manager to +silence warnings around a piece of code, without the side effect that +[`warnings.catch_warnings`][] has, and +[`ignoring_deprecations`][.ignoring_deprecations] for the common case of a library +having to touch a symbol it deprecated itself. + +It also provides [`deprecated_aliases`][.deprecated_aliases], to keep the old import +path of a symbol that moved to another module working, warning whoever uses it, and +[`asserting_no_warnings`][.asserting_no_warnings] with its +[`asserting_no_deprecations`][.asserting_no_deprecations] shortcut, to check in a test +that a piece of code doesn't warn. + +The documented way of silencing a warning locally is a +[`warnings.catch_warnings`][] block, but merely entering and leaving one invalidates +the warnings deduplication history of the whole program, so every warning that was +already shown is shown again, and again on every call. This is +[python/cpython#73858](https://github.com/python/cpython/issues/73858), open since +2017. + +[`ignoring_warnings`][.ignoring_warnings] adds the filter to the list the interpreter +consults and takes it out again on exit, which leaves the deduplication history +untouched, for about half a microsecond more per block than the standard library one +costs: + +```python +import warnings + +from frequenz.core.warnings import ignoring_warnings + + +def legacy_convert(value: str) -> int: + warnings.warn("legacy_convert() is deprecated", DeprecationWarning, stacklevel=2) + return int(value) + + +def convert(value: str) -> int: + with ignoring_warnings(category=DeprecationWarning): + return legacy_convert(value) + + +with warnings.catch_warnings(record=True, action="default") as caught: + for _ in range(10): + warnings.warn("shown only once", UserWarning) + convert("1") + +assert len(caught) == 1 +``` + +With [`warnings.catch_warnings`][] in `convert()` the same loop shows the +`UserWarning` ten times. +""" + +from ._asserting import asserting_no_deprecations, asserting_no_warnings +from ._deprecated_aliases import deprecated_aliases +from ._ignoring import ignoring_deprecations, ignoring_warnings + +__all__ = [ + "asserting_no_deprecations", + "asserting_no_warnings", + "deprecated_aliases", + "ignoring_deprecations", + "ignoring_warnings", +] diff --git a/src/frequenz/core/warnings/_asserting.py b/src/frequenz/core/warnings/_asserting.py new file mode 100644 index 0000000..1bd93f3 --- /dev/null +++ b/src/frequenz/core/warnings/_asserting.py @@ -0,0 +1,309 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Asserting in a test that a piece of code doesn't warn. + +See the package documentation for the background. +""" + +import contextlib +import re +import warnings +from types import TracebackType + + +def _replay(warning: warnings.WarningMessage) -> None: + """Show a recorded warning, as it would have been shown without the recording. + + The filters are not consulted again: this warning already passed them, and + `record=True` is the only reason it wasn't shown. + + Args: + warning: The recorded warning. + """ + # `_showwarnmsg()` is what the warnings machinery itself calls to show a warning, + # with the whole message: it hands it to `warnings.showwarning` if the application + # replaced that (`logging.captureWarnings()` does), and otherwise to the default + # implementation, which is also what `catch_warnings(record=True)` replaces to + # record instead of showing. Calling the public `showwarning()` here would be + # wrong both ways around: it has no `source` argument, so a `ResourceWarning` + # would lose the object it points at, and calling the default implementation + # directly would skip whatever the application installed. + show_message = getattr(warnings, "_showwarnmsg", None) + if show_message is not None: + show_message(warning) + return + warnings.showwarning( + warning.message, + warning.category, + warning.filename, + warning.lineno, + warning.file, + warning.line, + ) + + +# A lowercase name, like `warnings.catch_warnings`, because this is read as the +# statement it is used in and never as a type worth naming. +# pylint: disable-next=invalid-name +class asserting_no_warnings(contextlib.AbstractContextManager[None]): + """A context manager that fails if a matching warning is raised inside its block. + + Use this to test that a call doesn't warn. The usual alternative, an + `"error"` filter, makes [`warnings.warn`][] raise an exception **inside** the + code under test. A broad `except` in that code can then swallow it or turn + it into an unrelated failure, and the code no longer behaves as it would in + production. This block records warnings instead and leaves the code alone. + When the block ends, it raises [`AssertionError`][] listing every + unexpected warning and where it came from. + + Warnings that don't cause a failure are still shown as usual when the block + ends. The unexpected ones are shown too if the block is left because of an + exception. In that case the exception is raised instead of the + [`AssertionError`][], since it's the bigger problem, but the warnings aren't + lost. + + Warning: Resets the warnings deduplication history + Recording needs [`warnings.catch_warnings`][], and using it resets that + history for the whole program. There's no way around it here: a warning + that was already shown is skipped (by the module's + `__warningregistry__`) before the filters are even checked, so the only + way to catch it again is to reset the history. + + In tests this is usually fine, since they are isolated. Anywhere else, + every warning shown before the block will be shown again after it. So + use this in a short test that checks just this, not in a hot path, a + long-running process, or around a whole test suite. + + Warning: Concurrency + [`warnings.catch_warnings`][] affects the whole process unless + `sys.flags.context_aware_warnings` is on. That flag was added in Python + 3.14 and is off by default, except in free-threaded builds. Without it, + a matching warning from another thread, or from another asyncio task + while the block is waiting on an `await`, is recorded here and fails the + block, even if the code under test never warned. Use it only around + synchronous code, and not where background threads can raise matching + warnings. + + With the flag on, the problem is smaller but not gone. New threads start + with an empty context, so their warnings aren't recorded. Tasks created + inside the block, however, inherit its recording and can still fail it. + + Warning: An `ignoring_warnings` block inside wins + [`ignoring_warnings`][..ignoring_warnings] puts its filter in front of + the one this block adds. So if the code under test ignores a warning + itself, that warning is never recorded and can't fail the block. This is + intended, since the code explicitly asked to ignore it. It does mean + that an [`asserting_no_deprecations`][..asserting_no_deprecations] block + around code using [`ignoring_deprecations`][..ignoring_deprecations] + internally only checks the deprecations the code didn't silence, not all + of them. + + Warning: Handlers installed inside the block + While the block is active, it controls how warnings are shown. Handlers + installed *before* the block, like the one [`logging.captureWarnings`][] + sets up, keep working: once the block ends, they receive the warnings + that didn't fail it. But if the code **under test** replaces + [`warnings.showwarning`][] itself, any warning raised after that goes + straight to its own handler, and this block can't see it or fail on it. + A plain `catch_warnings(record=True)` has the same limitation. + + Warning: Not reentrant + Entering an instance that is already inside its block raises + [`RuntimeError`][]. + + Example: + ```python + import warnings + + from frequenz.core.warnings import asserting_no_warnings + + + def convert(value: str) -> int: + return int(value) + + + with asserting_no_warnings(category=DeprecationWarning): + assert convert("1") == 1 + ``` + """ + + def __init__( # noqa: DOC503 + self, *, category: type[Warning] = Warning, message: str = "" + ) -> None: + """Initialize this instance. + + Args: + category: The category of warnings to reject, including its subclasses. + message: A regular expression the start of the warning message must + match, case insensitively. The default matches every message. + + Raises: + TypeError: If any argument has the wrong type. + re.error: If `message` is not a valid regular expression. + """ + if not (isinstance(category, type) and issubclass(category, Warning)): + raise TypeError(f"category must be a Warning subclass, got {category!r}") + if not isinstance(message, str): + raise TypeError(f"message must be a str, got {message!r}") + self._category = category + self._message = message + self._pattern = re.compile(message, re.IGNORECASE) if message else None + self._entered = False + self._stack: contextlib.ExitStack | None = None + self._caught: list[warnings.WarningMessage] = [] + + def _matches(self, warning: warnings.WarningMessage) -> bool: + """Tell whether a recorded warning is one this block rejects. + + Args: + warning: The recorded warning. + + Returns: + Whether it matches. + """ + # There is no `module` argument next to `category` and `message`, even + # though `ignoring_warnings` has one and the "always" filter added in + # `__enter__()` would take it, and it can't be added here as it stands. + # A recorded `WarningMessage` carries `filename` and not the module + # name the filter matched, and those are different strings (`mymod` + # against `/tmp/mymod.py`), so this function can't reproduce the + # decision. Putting it only on the filter doesn't work either: a + # warning from a module the argument was meant to exempt still reaches + # this function through an ambient "default" or "always" filter, and + # would be reported. Giving it filename semantics instead would mean + # one name for two meanings across this module. + if not issubclass(warning.category, self._category): + return False + return self._pattern is None or bool(self._pattern.match(str(warning.message))) + + def __enter__(self) -> None: + """Enter the block, recording the warnings raised inside it. + + Raises: + RuntimeError: If this instance is already inside its block. + """ + if self._entered: + raise RuntimeError(f"Cannot enter {type(self).__name__}() twice") + self._entered = True + + self._stack = contextlib.ExitStack() + self._caught = self._stack.enter_context(warnings.catch_warnings(record=True)) + + # Recording happens in the default warnings output, so it only sees a + # warning while that output is the one installed. `catch_warnings` resets + # `showwarning` itself to make sure of that, but only when the filters are + # process-wide: with context-aware warnings (Python 3.14 and later) the + # recording lives in the context while `showwarning` stays a module global, + # so a handler the application installed, as `logging.captureWarnings()` + # does, keeps taking every warning, nothing is recorded, and every assertion + # here passes. That is python/cpython#151149, so the reset is done here too, + # in every mode. It is undone before the recording block closes, so the + # replay on the way out still reaches the application's handler. + default_show = getattr(warnings, "_showwarning_orig", None) + if default_show is not None: + self._stack.callback(setattr, warnings, "showwarning", warnings.showwarning) + warnings.showwarning = default_show + + # "always" so a warning already shown elsewhere is still recorded here, with + # the message too, so a warning this block doesn't reject keeps whatever the + # surrounding filters say about it. + warnings.filterwarnings( + "always", message=self._message, category=self._category + ) + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + """Exit the block, replaying what it didn't reject and failing on the rest. + + Args: + exc_type: The type of the exception leaving the block, if any. + exc_value: The exception leaving the block, if any. + traceback: The traceback of that exception, if any. + + Raises: + AssertionError: If a matching warning was raised inside the block and no + exception is on its way out, which is the more important news. + """ + self._entered = False + stack, self._stack = self._stack, None + caught, self._caught = self._caught, [] + if stack is not None: + stack.close() + + # An exception is the more important news, so the assertion stands down and + # the warnings it would have reported are shown instead of being reported. + # Dropping them here would lose them entirely, which is the one thing the + # replay below exists to prevent. + failing = exc_type is None + + # One pass, rather than a list of offenders and a membership test per + # warning: the filter added inside is an "always" one, so a regression in a + # loop can easily leave tens of thousands of warnings here, and the failure + # path is the last place to spend quadratic time. + unexpected: list[warnings.WarningMessage] = [] + for warning in caught: + if failing and self._matches(warning): + unexpected.append(warning) + else: + _replay(warning) + + if unexpected: + listing = "\n".join( + f" {warning.category.__name__}: {warning.message} " + f"({warning.filename}:{warning.lineno})" + for warning in unexpected + ) + raise AssertionError(f"Unexpected warnings raised:\n{listing}") + + +def asserting_no_deprecations( # noqa: DOC502 + *, message: str = "" +) -> asserting_no_warnings: + """Fail if a deprecation warning is raised inside the block. + + This is [`asserting_no_warnings`][..asserting_no_warnings] for the usual case, + checking that the replacement for a deprecated symbol doesn't itself go through + the deprecated one. + + Warning: All limitations of `asserting_no_warnings()` apply + Read the documentation of [`asserting_no_warnings`][..asserting_no_warnings], + many limitations apply here too, as this is only a thin wrapper around it. + + Pass `message` when the code under test legitimately deprecates something else, + or reaches a third-party deprecation there is nothing to be done about: the block + wraps the code being tested, so unlike an + [`ignoring_deprecations`][..ignoring_deprecations] block it can't be narrowed by + making it shorter. + + Example: + ```python + import warnings + + from frequenz.core.warnings import asserting_no_deprecations + + + def convert(value: str) -> int: + return int(value) + + + with asserting_no_deprecations(message="Wrapper is deprecated"): + assert convert("1") == 1 + ``` + + Args: + message: A regular expression the start of the warning message must match, + case insensitively. The default rejects every deprecation. + + Returns: + A context manager that fails if a deprecation warning is raised in it. + + Raises: + TypeError: If `message` has the wrong type. + re.error: If `message` is not a valid regular expression. + """ + return asserting_no_warnings(category=DeprecationWarning, message=message) diff --git a/src/frequenz/core/warnings/_deprecated_aliases.py b/src/frequenz/core/warnings/_deprecated_aliases.py new file mode 100644 index 0000000..8881822 --- /dev/null +++ b/src/frequenz/core/warnings/_deprecated_aliases.py @@ -0,0 +1,250 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Keeping the old import path of a symbol that moved to another module. + +See the package documentation for the background. +""" + +import importlib +import warnings +from collections.abc import Callable, Mapping +from types import ModuleType +from typing import Any + + +def _checked_aliases( + module: str, aliases: Mapping[str, str] +) -> dict[str, tuple[str, str]]: + """Check a set of deprecated aliases and resolve them. + + Everything here is checked when the aliases are declared rather than when one + of them is reached, which can be a release later and in somebody else's code, + where the failure no longer looks like a typo in an alias table. + + Args: + module: The fully qualified name of the module defining the aliases. + aliases: The mapping to check. + + Returns: + Each deprecated name mapped to the module its target lives in and the name + it has there, so a later change to `aliases` can't get past these + checks and the string doesn't have to be taken apart on every lookup. + + Raises: + TypeError: If `module` is not a string, or a name or target is not one. + ValueError: If a target is empty, carries more than one `:`, or has + nothing after it. + """ + if not isinstance(module, str): + raise TypeError(f"module must be a str, got {module!r}") + checked: dict[str, tuple[str, str]] = {} + for name, target in dict(aliases).items(): + if not isinstance(name, str): + raise TypeError(f"alias names must be str, got {name!r}") + if not isinstance(target, str): + raise TypeError(f"the target of {name!r} must be a str, got {target!r}") + target_module, renamed, target_name = target.partition(":") + # `partition()` stops at the first colon, so a second one would silently + # end up inside the name, and the warning would point at `a.b:c`. + if ":" in target_name: + raise ValueError( + f"the target of {name!r} has more than one ':': {target!r}" + ) + if not target_module: + raise ValueError(f"the target of {name!r} names no module: {target!r}") + if renamed and not target_name: + raise ValueError( + f"the target of {name!r} has nothing after the ':': {target!r}" + ) + checked[name] = (target_module, target_name or name) + return checked + + +def deprecated_aliases( # noqa: DOC502 + module: str, + aliases: Mapping[str, str], + *, + message: str = "{old} is deprecated. Use {new} instead.", + category: type[Warning] = DeprecationWarning, + stacklevel: int = 2, +) -> Callable[[str], Any]: + """Build a module `__getattr__` that warns about deprecated aliases. + + Use this for symbols that moved to another module but should keep working from + their old import path, when a [`typing_extensions.deprecated`][] decorator is not + an option because the object is not yours to mark: decorating it would deprecate + it for everybody, including the users of its new home. The alias also stays the + very same object, so [`isinstance`][] keeps working through both paths. + + Danger: + Follow the usage example structure strictly. In particular never drop + the `else:`, otherwise a type checker will see the `__getattr__` + assignment and treat every symbol in the module as [`Any`][typing.Any], + effectively disabling type checking for the entire module. + + Example: Usage + This is how a module that used to define `Decimal` itself, and now gets it + from [`decimal`][], keeps the old import path working: + + ```python + from typing import TYPE_CHECKING, TypeAlias + + from frequenz.core.warnings import deprecated_aliases + + if TYPE_CHECKING: + # Private import, only for type checkers + from decimal import Decimal as _Decimal + + Decimal: TypeAlias = _Decimal + else: + __getattr__ = deprecated_aliases(__name__, {"Decimal": "decimal"}) + ``` + + Reaching `Decimal` through this module now emits a `DeprecationWarning` + saying to use `decimal.Decimal` instead. + + Example: Renaming + When the symbol was also renamed on the way out, the new name goes after a + colon, as in an entry point: + + ```python + from typing import TYPE_CHECKING, TypeAlias + + from frequenz.core.warnings import deprecated_aliases + + if TYPE_CHECKING: + from fractions import Fraction as _Fraction + + Rational: TypeAlias = _Fraction + else: + __getattr__ = deprecated_aliases( + __name__, + { + "Rational": "fractions:Fraction", + }, + ) + ``` + + Example: Custom deprecation message + The message is a template that gets the old and new fully qualified names, + so it can carry a version, a link, or anything else the default doesn't + say: + + ```python + from typing import TYPE_CHECKING, TypeAlias + + from frequenz.core.warnings import deprecated_aliases + + if TYPE_CHECKING: + from decimal import Decimal as _Decimal + + Decimal: TypeAlias = _Decimal + else: + __getattr__ = deprecated_aliases( + __name__, + { + "Decimal": "decimal", + }, + message="{old} is deprecated since v2, use {new} instead.", + ) + ``` + + Tip: Deprecating whole modules + This function can't be used to deprecate a whole module. If you need to + do that, you can keep a real `__init__.py` at the old path, with a + `__getattr__ = deprecated_aliases(...)` that includes all the symbols + that used to be reachable there. + + Warning: Security Warning + This function imports the target module, so the mapping is as trusted + as an `import` statement in this module. Write it out as a literal; + don't build it from anything that comes from outside the program. + + Args: + module: The fully qualified name of the module defining the aliases. + aliases: A mapping of each deprecated symbol to where it lives now, as the + fully qualified name of the module that owns it, optionally followed by + `:` and the name it has there, when it is not the deprecated one. It is + copied, so changing it afterwards has no effect. + message: The template for the warning message, formatted with `{old}` and + `{new}`, the fully qualified names of the alias and of the symbol it + resolves to. + category: The category of the warning to emit. + stacklevel: How far up the stack the warning is reported, counting from the + `__getattr__` itself. The default of 2 points at the code reaching for + the alias, and only needs raising if something wraps the returned + function. + + Returns: + A function suitable for use as the module's `__getattr__`. + + Raises: + TypeError: If an argument has the wrong type, including a name or target + that is not a string. Also later, when an alias is reached, if it + resolves to a module rather than to a symbol in one. + ValueError: If a target is empty, carries more than one `:`, or has + nothing after it, if `message` is not a template taking `{old}` and + `{new}`, or if `stacklevel` is smaller than 1. + """ + if not isinstance(message, str): + raise TypeError(f"message must be a str, got {message!r}") + if not (isinstance(category, type) and issubclass(category, Warning)): + raise TypeError(f"category must be a Warning subclass, got {category!r}") + if not isinstance(stacklevel, int): + raise TypeError(f"stacklevel must be an int, got {stacklevel!r}") + if stacklevel < 1: + raise ValueError(f"stacklevel must be 1 or more, got {stacklevel!r}") + try: + # Formatting it once here is the only way to find a stray brace, which + # would otherwise raise from inside `warnings.warn()` at lookup time. + message.format(old="", new="") + except (IndexError, KeyError, ValueError) as error: + raise ValueError( + f"message must be a template taking {{old}} and {{new}}, " + f"got {message!r}" + ) from error + targets = _checked_aliases(module, aliases) + + def module_getattr(name: str) -> Any: + """Return a deprecated alias, warning about its new location. + + Args: + name: The name being looked up in the module. + + Returns: + The aliased object. + + Raises: + AttributeError: If the name is not one of the deprecated aliases. + TypeError: If the alias resolves to a module, which this can't keep + reachable from its old path. + """ + target = targets.get(name) + if target is None: + raise AttributeError(f"module {module!r} has no attribute {name!r}") + target_module, target_name = target + # Resolved before the warning is raised, so an alias that can't work says + # so instead of warning about a move that didn't happen. It also keeps the + # `TypeError` below from being masked by an "error" filter turning that + # warning into an exception first. + value = getattr(importlib.import_module(target_module), target_name) + if isinstance(value, ModuleType): + raise TypeError( + f"the target of {module}.{name} is the module " + f"{target_module}.{target_name}, and this aliases names, not " + "modules: the import system never consults a package's " + f"__getattr__, so `import {module}.{name}` would fail anyway. " + f"Keep a real __init__.py at {module}.{name}, with the aliases " + "for the names it used to hold in it." + ) + warnings.warn( + message.format( + old=f"{module}.{name}", new=f"{target_module}.{target_name}" + ), + category, + stacklevel=stacklevel, + ) + return value + + return module_getattr diff --git a/src/frequenz/core/warnings/_ignoring.py b/src/frequenz/core/warnings/_ignoring.py new file mode 100644 index 0000000..e0c3b1b --- /dev/null +++ b/src/frequenz/core/warnings/_ignoring.py @@ -0,0 +1,366 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Ignoring warnings without resetting the deduplication history. + +See the package documentation for the background. +""" + +import contextlib +import re +import warnings +from types import TracebackType +from typing import Any, TypeAlias + +_Filter: TypeAlias = tuple[ + str, "re.Pattern[str] | None", type[Warning], "re.Pattern[str] | None", int +] +"""A warning filter, in the shape CPython keeps in its filters list.""" + + +def _live_filters() -> list[Any] | None: + """Return the list of warning filters in effect, the live object. + + Returns: + The list CPython consults on each warning, honouring context-aware warnings + when available, or `None` if it can't be found. + """ + # On 3.14+ the filters in effect may live in a context rather than in + # `warnings.filters`, and `_get_filters()` is what the warnings machinery itself + # calls to find them. + get_filters = getattr(warnings, "_get_filters", None) + filters: Any = get_filters() if get_filters is not None else warnings.filters + return filters if isinstance(filters, list) else None + + +def _build_filter( # noqa: DOC503 + category: type[Warning], message: str, module: str +) -> _Filter: + """Build an `"ignore"` filter entry, validating the arguments. + + [`warnings.filterwarnings`][] checks its arguments before building the same + entry, with `assert`s up to Python 3.12 and with real exceptions from 3.13 on. + Building the entry here bypasses both, so the equivalent checks are done here, + always as real exceptions. + + Args: + category: The category of warnings to ignore. + message: A regular expression the start of the warning message must match, + or an empty string to match any message. + module: A regular expression the start of the module name must match, or an + empty string to match any module. + + Returns: + The filter entry. + + Raises: + TypeError: If any argument has the wrong type. + re.error: If `message` or `module` is not a valid regular expression. + """ + if not (isinstance(category, type) and issubclass(category, Warning)): + raise TypeError(f"category must be a Warning subclass, got {category!r}") + if not isinstance(message, str): + raise TypeError(f"message must be a str, got {message!r}") + if not isinstance(module, str): + raise TypeError(f"module must be a str, got {module!r}") + return ( + "ignore", + re.compile(message, re.IGNORECASE) if message else None, + category, + re.compile(module) if module else None, + 0, + ) + + +def _remove_filter(filters: list[Any], item: _Filter) -> None: + """Take a filter entry out of a filters list again. + + Args: + filters: The list the entry was inserted into. + item: The entry to remove. + """ + for index, existing in enumerate(filters): + # By identity: somebody may have inserted an equal filter meanwhile, and that + # one is theirs to keep. + if existing is item: + del filters[index] + return + + +# A lowercase name, like `warnings.catch_warnings`, because this is read as the +# statement it is used in and never as a type worth naming. +# pylint: disable-next=invalid-name +class ignoring_warnings(contextlib.AbstractContextManager[None]): + """A context manager that ignores the matching warnings raised inside its block. + + The constructor arguments select which warnings are ignored, and mean the same as + the arguments of [`warnings.filterwarnings`][] with `action="ignore"`. + + Unlike [`warnings.catch_warnings`][], this doesn't reset the warnings + deduplication history of the program, so warnings shown before the block are not + shown again after it. See the module documentation for the background. + + Example: + ```python + import warnings + + from frequenz.core.warnings import ignoring_warnings + + + # Some third-party function warns about something out of our control. + def third_party() -> int: + warnings.warn("this is fine, actually", RuntimeWarning, stacklevel=2) + return 42 + + + def use_third_party() -> int: + with ignoring_warnings(RuntimeWarning, message="this is fine"): + return third_party() + ``` + + Warning: This is not a `catch_warnings` replacement + This only adds a filter for the duration of the block, it doesn't save and + restore the filters around it. Any change made to the filters inside the + block, by this thread or another one, is still in effect afterwards, exactly + as if the block wasn't there. Use [`warnings.catch_warnings`][] when the + point is to undo filter changes. + + Warning: Scope + The filter is added to the filters the interpreter consults, which are + normally shared by every thread and every asyncio task, so a matching + warning raised by unrelated code running concurrently is ignored too, for as + long as the block lasts. Keep the block around the call that needs it, and + prefer not to hold it across an `await`. + + Wrapping this in a [`warnings.catch_warnings`][] block only narrows that on + Python 3.14 and newer, with `sys.flags.context_aware_warnings` on, which is + off by default except in free-threaded builds. The outer block then gives + the current context a filters list of its own, kept in a + [`contextvars.ContextVar`][], so other threads, and tasks created before + it, are no longer affected. Tasks created inside it still inherit that same + list, so they are silenced too. Without the flag, + [`warnings.catch_warnings`][] reaches for the same shared state and + isolates nothing. Either way it resets the deduplication history, which is + what this class exists to avoid. + + Warning: Not reentrant + Entering an instance that is already inside its block raises + [`RuntimeError`][]. A `with ignoring_warnings(...)` statement builds a new + instance each time, so this only comes up when one instance is kept in a + variable and entered again from inside itself, by a recursive function or a + helper called in the block. Nesting is rejected rather than supported + because the inner block would only ignore what the outer one already + ignores, so there is nothing to gain from making it work; an instance can + be entered again once it has been left. + + Warning: Not thread-safe + Taking the filter out again is a lookup followed by a deletion, with no + locking in between, so another thread removing a filter from the same list + at the same time can make this delete the wrong one, dropping a filter that + was not this block's and leaving this block's behind. Nothing guards against + that, because a lock would cost every block to protect against something + only a program that changes warning filters from several threads at once can + do. If yours does, don't use this. + """ + + def __init__( # noqa: DOC502 + self, + category: type[Warning] = Warning, + *, + message: str = "", + module: str = "", + ) -> None: + """Initialize this instance. + + Args: + category: The category of warnings to ignore, including its subclasses. + message: A regular expression the start of the warning message must + match, case insensitively. The default matches every message. + module: A regular expression the start of the module name must match. The + default matches every module. + + Raises: + TypeError: If any argument has the wrong type. + re.error: If `message` or `module` is not a valid regular expression. + """ + self._arguments = (category, message, module) + self._item = _build_filter(category, message, module) + self._entered = False + self._filters: list[Any] | None = None + self._fallback: contextlib.ExitStack | None = None + + def __enter__(self) -> None: + """Enter the block, adding the filter. + + Raises: + RuntimeError: If this instance is already inside its block. + """ + if self._entered: + raise RuntimeError(f"Cannot enter {type(self).__name__}() twice") + self._entered = True + + filters = _live_filters() + + if filters is None: + # The interpreter doesn't keep its filters where we can reach them, so + # there is nothing to be clever about: do what the standard library + # does, which is correct, only more expensive and with the history + # damage. + category, message, module = self._arguments + self._fallback = contextlib.ExitStack() + self._fallback.enter_context(warnings.catch_warnings()) + warnings.filterwarnings( + "ignore", message=message, category=category, module=module + ) + return + + # The whole point: inserting into the live list directly means the internal + # filters version doesn't move, and it is that version moving which makes + # CPython treat every module's `__warningregistry__` as stale + # (cpython#73858). Adding an "ignore" filter without bumping it is sound, and + # only for "ignore": the registry is consulted *before* the filters, and an + # ignore can only take warnings away, never add them, so a warning recorded + # as already shown outside the block is still correctly recorded as shown + # inside it, and the other way around. `simplefilter()` and + # `filterwarnings()` can't do this, they bump the version on purpose because + # they can add warnings back. + # + # The list object is remembered, rather than only looked up again on exit, + # because code inside the block can replace `warnings.filters` with another + # list and the entry has to come out of both (see `__exit__()`). Only our own + # entry is removed, and nothing else is restored: an entry/exit snapshot + # would make two blocks overlapping in different threads (A enters, B enters, + # A exits, B exits) put each other's filters back, leaving one installed + # forever. + self._filters = filters + filters.insert(0, self._item) + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + """Exit the block, taking the filter out again. + + Args: + exc_type: The type of the exception leaving the block, if any. + exc_value: The exception leaving the block, if any. + traceback: The traceback of that exception, if any. + """ + self._entered = False + fallback, self._fallback = self._fallback, None + if fallback is not None: + fallback.close() + return + filters, self._filters = self._filters, None + if filters is None: + return + + _remove_filter(filters, self._item) + # Code inside the block may have replaced the filters in effect with a copy of + # the list, which carries our entry along with the rest, and it is that copy + # the interpreter consults now. Both lists are cleaned then: the live one + # because leaving an ignore filter in it silences warnings after the block, + # and the one entered with because it can be put back later, by the + # `catch_warnings` block that made the copy. A list that is neither, left + # behind by a second replacement inside the block, keeps our entry, and there + # is no way to reach it to take it out. + live = _live_filters() + if live is not None and live is not filters: + _remove_filter(live, self._item) + + +def ignoring_deprecations( # noqa: DOC502 + *, message: str = "", module: str = "" +) -> ignoring_warnings: + """Ignore the deprecation warnings raised inside the block. + + This is [`ignoring_warnings`][..ignoring_warnings] for the most common case, + library code that has to touch a symbol it deprecated itself. The user was + already warned by the deprecated symbol they used; warning them again from its + internals, about something they can do nothing about, is noise. + + Warning: All limitations of `ignoring_warnings()` apply + Read the documentation of [`ignoring_warnings`][..ignoring_warnings], + many limitations apply here too, as this is only a thin wrapper around it. + + Tip: + Wrap the call that reaches the deprecated symbol, not everything around + it. A block that covers more than that also silences deprecations that have + nothing to do with the one it was added for. + + Example: + ```python + import warnings + + from frequenz.core.warnings import ignoring_deprecations + + + # A type that used to be public and is now deprecated. + class Wrapper: + def __init__(self, raw: str) -> None: + warnings.warn("Wrapper is deprecated", DeprecationWarning, stacklevel=2) + self.raw = raw + + + def from_wire(raw: str) -> Wrapper: + # Whatever else this function does, the deprecations raised out there + # are the user's business, so they stay outside the block. + with ignoring_deprecations(): + return Wrapper(raw) + ``` + + Example: Inside a function that is itself deprecated + A deprecated function has already warned its caller about this code path, so + the deprecations it reaches on the way are noise too. Even there, keep the + block around the calls that raise them rather than putting it around the + whole body, which this module can't do for you: it is a context manager and + refuses to be used as a decorator. + + ```python + import warnings + + from typing_extensions import deprecated + + from frequenz.core.warnings import ignoring_deprecations + + + # A type that used to be public and is now deprecated. + class Wrapper: + def __init__(self, raw: str) -> None: + warnings.warn("Wrapper is deprecated", DeprecationWarning, stacklevel=2) + self.raw = raw + + + async def save(wrapper: Wrapper) -> None: + print(f"saving {wrapper.raw}") + + + @deprecated("Use from_wire() instead") + async def parse(raw: str) -> Wrapper: + with ignoring_deprecations(): + wrapper = Wrapper(raw) + await save(wrapper) # Outside: the scope note applies across an await. + return wrapper + ``` + + Args: + message: A regular expression the start of the warning message must match, + case insensitively. The default matches every deprecation, which is + usually right: the way to be precise here is a short block, not a narrow + filter. Use it when the call being wrapped can also raise a deprecation + that should be heard. + module: A regular expression the start of the module name must match. The + default matches every module. + + Returns: + A context manager that ignores deprecation warnings. + + Raises: + TypeError: If any argument has the wrong type. + re.error: If `message` or `module` is not a valid regular expression. + """ + return ignoring_warnings( + category=DeprecationWarning, message=message, module=module + ) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..b15f429 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1,8 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Tests for the frequenz-core library. + +This is a package so the helper packages the tests import, like +`tests.warnings.documented_aliases`, have the same name for pytest and for mypy. +""" diff --git a/tests/warnings/__init__.py b/tests/warnings/__init__.py new file mode 100644 index 0000000..4071283 --- /dev/null +++ b/tests/warnings/__init__.py @@ -0,0 +1,4 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Tests for the warnings package.""" diff --git a/tests/warnings/conftest.py b/tests/warnings/conftest.py new file mode 100644 index 0000000..057978f --- /dev/null +++ b/tests/warnings/conftest.py @@ -0,0 +1,59 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Fixtures shared by the warnings tests.""" + +import sys +from collections.abc import Callable, Iterator +from types import ModuleType + +import pytest + +from frequenz.core.warnings import _ignoring + +_MODULE_SOURCE = """ +import warnings + +# Messages include the module name so the "once" action, which deduplicates by +# message and category only, doesn't see repeated messages across tests. +DEFAULT_MESSAGE = f"warned in {__name__}" + +def warn(message=DEFAULT_MESSAGE, category=UserWarning): + warnings.warn(message, category, stacklevel=1) + +def warn_then_block(block, message=DEFAULT_MESSAGE, category=UserWarning): + warnings.warn(message, category, stacklevel=1) + with block(): + pass +""" + + +@pytest.fixture(name="make_module") +def make_module_fixture() -> Iterator[Callable[[str], ModuleType]]: + """Create fresh modules, each with its own (initially missing) registry.""" + created: list[str] = [] + + def make(name: str) -> ModuleType: + module = ModuleType(name) + module.__file__ = f"<{name}>" + # pylint: disable-next=exec-used + exec(compile(_MODULE_SOURCE, f"<{name}>", "exec"), module.__dict__) + sys.modules[name] = module + created.append(name) + return module + + yield make + for name in created: + sys.modules.pop(name, None) + + +@pytest.fixture(name="restore_filters", autouse=True) +def restore_filters_fixture() -> Iterator[None]: + """Leave the process filters as they were, whatever a test does to them.""" + live = _ignoring._live_filters() # pylint: disable=protected-access + saved = None if live is None else live[:] + yield + if saved is not None: + live = _ignoring._live_filters() # pylint: disable=protected-access + assert live is not None + live[:] = saved diff --git a/tests/warnings/documented_aliases/__init__.py b/tests/warnings/documented_aliases/__init__.py new file mode 100644 index 0000000..4b751f0 --- /dev/null +++ b/tests/warnings/documented_aliases/__init__.py @@ -0,0 +1,48 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""A package written the way `deprecated_aliases()` documents, for the tests. + +The other tests build a `ModuleType` and assign the `__getattr__` onto it, which +exercises the function but not the shape the docstring tells people to write: the +aliases declared under `if TYPE_CHECKING:`, the assignment in the `else:` branch, +and a real module the import system loads. This package is that shape, so the +tests importing from here fail if the documented pattern stops working. +""" + +from typing import TYPE_CHECKING, TypeAlias + +from frequenz.core.warnings import deprecated_aliases + +__all__ = ["Decimal", "Rational", "kept"] + +if TYPE_CHECKING: + # Only for type checkers, which can't see the runtime `__getattr__` in the + # `else` branch. + from decimal import Decimal as _Decimal + from fractions import Fraction as _Fraction + + Decimal: TypeAlias = _Decimal + """A decimal number. + + Deprecated: + `tests.warnings.documented_aliases.Decimal` is deprecated. Use + [decimal.Decimal][] instead. + """ + + Rational: TypeAlias = _Fraction + """A rational number. + + Deprecated: + `tests.warnings.documented_aliases.Rational` is deprecated. Use + [fractions.Fraction][] instead. + """ +else: + __getattr__ = deprecated_aliases( + __name__, {"Decimal": "decimal", "Rational": "fractions:Fraction"} + ) + + +def kept() -> str: + """Return the name of a symbol this package still defines itself.""" + return "kept" diff --git a/tests/warnings/test_asserting.py b/tests/warnings/test_asserting.py new file mode 100644 index 0000000..f6a508c --- /dev/null +++ b/tests/warnings/test_asserting.py @@ -0,0 +1,439 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Tests for `asserting_no_warnings()` and `asserting_no_deprecations()`.""" + +import contextlib +import logging +import re +import subprocess +import sys +import threading +import warnings +from collections.abc import Iterator +from typing import Any + +import pytest + +from frequenz.core.warnings import ( + _asserting, + asserting_no_deprecations, + asserting_no_warnings, + ignoring_deprecations, +) + + +@contextlib.contextmanager +def _shown() -> Iterator[None]: + """Show every warning, since this suite otherwise turns them into errors.""" + with warnings.catch_warnings(action="always"): + yield + + +def test_asserting_no_warnings_passes_when_quiet() -> None: + """Test that a block raising no matching warning is fine.""" + with _shown(): + with asserting_no_warnings(category=DeprecationWarning): + warnings.warn("not a deprecation", UserWarning, stacklevel=1) + + +def test_asserting_no_warnings_reports_every_offender() -> None: + """Test that the failure names each warning and where it came from.""" + with _shown(): + with pytest.raises(AssertionError) as failure: + with asserting_no_warnings(category=DeprecationWarning): + warnings.warn("first", DeprecationWarning, stacklevel=1) + warnings.warn("second", FutureWarning, stacklevel=1) + text = str(failure.value) + assert "DeprecationWarning: first" in text + assert "second" not in text + assert f"({__file__}:" in text + + +def test_asserting_no_warnings_matches_the_message() -> None: + """Test that only the warnings matching the message are rejected.""" + with _shown(): + with asserting_no_warnings(category=UserWarning, message="only this one"): + warnings.warn("something else", UserWarning, stacklevel=1) + with pytest.raises(AssertionError, match="only this one, really"): + with asserting_no_warnings(category=UserWarning, message="only this one"): + warnings.warn("only this one, really", UserWarning, stacklevel=1) + + +def test_asserting_no_warnings_preserves_an_ignore_filter() -> None: + """Test that an ignored nonmatching warning stays ignored.""" + with warnings.catch_warnings( + record=True, action="ignore", category=UserWarning + ) as caught: + with asserting_no_warnings(category=UserWarning, message="target"): + warnings.warn("unrelated", UserWarning, stacklevel=1) + assert not caught + + +def test_asserting_no_warnings_preserves_a_default_filter() -> None: + """Test that a nonmatching warning keeps the ambient deduplication action.""" + + def warn() -> None: + warnings.warn("unrelated", UserWarning, stacklevel=1) + + with warnings.catch_warnings( + record=True, action="default", category=UserWarning + ) as caught: + with asserting_no_warnings(category=UserWarning, message="target"): + warn() + warn() + assert [str(warning.message) for warning in caught] == ["unrelated"] + + +def test_asserting_no_warnings_preserves_an_error_filter() -> None: + """Test that a nonmatching warning still raises under an ambient error filter.""" + with warnings.catch_warnings(action="error", category=UserWarning): + with pytest.raises(UserWarning, match="unrelated"): + with asserting_no_warnings(category=UserWarning, message="target"): + warnings.warn("unrelated", UserWarning, stacklevel=1) + + +def test_asserting_no_warnings_lets_other_warnings_through() -> None: + """Test that warnings the block doesn't watch are still shown afterwards.""" + with warnings.catch_warnings(record=True, action="always") as caught: + with asserting_no_warnings(category=DeprecationWarning): + warnings.warn("passed along", UserWarning, stacklevel=1) + assert [str(warning.message) for warning in caught] == ["passed along"] + + +def test_asserting_no_warnings_sees_already_shown_warnings() -> None: + """Test that a warning deduplicated by the ambient filters is still caught.""" + with warnings.catch_warnings(action="default"): + warnings.warn("said once", DeprecationWarning, stacklevel=1) + with pytest.raises(AssertionError, match="said once"): + with asserting_no_warnings(category=DeprecationWarning): + warnings.warn("said once", DeprecationWarning, stacklevel=1) + + +def test_asserting_no_warnings_does_not_swallow_exceptions() -> None: + """Test that an error in the block is what comes out.""" + with pytest.raises(ValueError, match="boom"): + with asserting_no_warnings(): + warnings.warn("ignored, the error wins", UserWarning, stacklevel=1) + raise ValueError("boom") + + +@pytest.mark.parametrize("kwargs", [{"category": int}, {"message": None}]) +def test_asserting_no_warnings_rejects_invalid_arguments( + kwargs: dict[str, Any], +) -> None: + """Test that the arguments are validated.""" + with pytest.raises(TypeError): + with asserting_no_warnings(**kwargs): + pass + + +def test_asserting_no_warnings_rejects_an_invalid_pattern() -> None: + """Test that a message that doesn't compile is reported as such.""" + with pytest.raises(re.error): + with asserting_no_warnings(message="("): + pass + + +def test_asserting_no_warnings_sees_warnings_from_other_threads() -> None: + """Test that the recording is process-wide, and when it is not. + + This pins down the limitation the docstring warns about, so a change in how + CPython scopes `catch_warnings` doesn't go unnoticed. + """ + isolated = bool(getattr(sys.flags, "context_aware_warnings", 0)) + ready = threading.Event() + done = threading.Event() + + def background() -> None: + ready.wait() + try: + warnings.warn("from another thread", UserWarning, stacklevel=1) + except UserWarning: + pass # An ambient "error" filter, which this test doesn't care about. + done.set() + + thread = threading.Thread(target=background) + thread.start() + leaked = False + try: + with asserting_no_warnings(category=UserWarning): + ready.set() + done.wait() + except AssertionError as error: + leaked = "from another thread" in str(error) + thread.join() + + assert leaked is not isolated + + +def test_asserting_no_warnings_replays_when_the_block_raises() -> None: + """Test that every warning survives an exception leaving the block. + + The matching ones too: the exception wins over the assertion, but dropping + them would leave them neither reported nor shown, so a deprecation raised + inside a `pytest.raises()` block would disappear without a trace. + """ + with warnings.catch_warnings(record=True, action="always") as caught: + with pytest.raises(ValueError, match="boom"): + with asserting_no_warnings(category=DeprecationWarning): + warnings.warn("important diagnostic", UserWarning, stacklevel=1) + warnings.warn("would have failed", DeprecationWarning, stacklevel=1) + raise ValueError("boom") + assert [str(warning.message) for warning in caught] == [ + "important diagnostic", + "would have failed", + ] + + +def test_asserting_no_warnings_replay_keeps_the_source() -> None: + """Test that a replayed warning still points at the object it was about.""" + source = [1, 2, 3] + with warnings.catch_warnings(record=True, action="always") as caught: + with asserting_no_warnings(category=DeprecationWarning): + warnings.warn("leaked", ResourceWarning, stacklevel=1, source=source) + assert len(caught) == 1 + assert caught[0].source is source + + +def test_replay_without_showwarnmsg(monkeypatch: pytest.MonkeyPatch) -> None: + """Test the replay path taken when `warnings._showwarnmsg()` is not there. + + Every interpreter this supports has it, so nothing would notice this branch + rotting until the one that drops it arrives. It is driven directly rather + than through a block, because taking the name away also takes the recording + of a `catch_warnings(record=True)` around it. What is lost on this path is + the `source` of a `ResourceWarning`: the public `showwarning()` has no + argument for it. + """ + seen: list[tuple[Any, ...]] = [] + + monkeypatch.delattr(warnings, "_showwarnmsg") + monkeypatch.setattr(warnings, "showwarning", lambda *args: seen.append(args)) + recorded = warnings.WarningMessage( + "leaked", ResourceWarning, "somewhere.py", 42, source=[1, 2, 3] + ) + + _asserting._replay(recorded) # pylint: disable=protected-access + + # Six arguments, and none of them is the source. + assert seen == [("leaked", ResourceWarning, "somewhere.py", 42, None, None)] + + +def test_asserting_no_warnings_yields_to_an_inner_ignore() -> None: + """Test that a warning the code under test silences for itself doesn't fail. + + `ignoring_warnings` adds its filter in front of the one installed here, so the + warning is never recorded. That is the intended reading, but it also means an + assertion around code that ignores its own deprecations is about what is left. + """ + with _shown(): + with asserting_no_deprecations(): + with ignoring_deprecations(): + warnings.warn("silenced on purpose", DeprecationWarning, stacklevel=1) + + +def test_asserting_no_warnings_replay_honours_a_custom_showwarning( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test that a replayed warning goes through a replaced `showwarning()`.""" + seen: list[str] = [] + + # The rest of the arguments a `showwarning()` gets are the category, the file + # name, the line number, the file and the line, and this one only needs to be + # told it was called. + def show(message: Warning | str, *_: Any) -> None: + seen.append(str(message)) + + with _shown(): + monkeypatch.setattr(warnings, "showwarning", show) + with asserting_no_warnings(category=DeprecationWarning): + warnings.warn("passed along", UserWarning, stacklevel=1) + assert seen == ["passed along"] + + +def test_asserting_no_warnings_replay_reaches_the_logs( + caplog: pytest.LogCaptureFixture, +) -> None: + """Test that a replayed warning still lands in the logs. + + `logging.captureWarnings()` is how an application usually ends up with a + replaced `showwarning()`, and losing a warning from the logs is the damage. + """ + with _shown(): + logging.captureWarnings(True) + try: + with asserting_no_warnings(category=DeprecationWarning): + warnings.warn("passed along", UserWarning, stacklevel=1) + finally: + logging.captureWarnings(False) + assert "passed along" in caplog.text + + +def test_asserting_no_warnings_fails_with_a_replaced_showwarning( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test that a matching warning fails the block when `showwarning()` is replaced. + + The recording only sees a warning while the default warnings output is the one + installed, so a handler taking it first would leave nothing to fail on. + """ + seen: list[str] = [] + + def show(message: Warning | str, *_: Any) -> None: + seen.append(str(message)) + + with _shown(): + monkeypatch.setattr(warnings, "showwarning", show) + with pytest.raises(AssertionError, match="must fail"): + with asserting_no_warnings(category=UserWarning): + warnings.warn("must fail", UserWarning, stacklevel=1) + assert not seen + # The handler is back afterwards, and the block didn't show what it rejected. + warnings.warn("after the block", UserWarning, stacklevel=1) + assert seen == ["after the block"] + + +def test_asserting_no_warnings_fails_under_logging_capture( + caplog: pytest.LogCaptureFixture, +) -> None: + """Test that a matching warning fails the block while the logs capture warnings.""" + with _shown(): + logging.captureWarnings(True) + try: + with pytest.raises(AssertionError, match="must fail"): + with asserting_no_warnings(category=UserWarning): + warnings.warn("must fail", UserWarning, stacklevel=1) + finally: + logging.captureWarnings(False) + assert "must fail" not in caplog.text + + +_CONTEXT_AWARE_SCRIPT = """ +import logging +import warnings + +from frequenz.core.warnings import asserting_no_warnings + +logged = [] + + +class Collect(logging.Handler): + def emit(self, record): + logged.append(record.getMessage()) + + +logging.getLogger("py.warnings").addHandler(Collect()) +warnings.simplefilter("always") +logging.captureWarnings(True) + +try: + with asserting_no_warnings(category=UserWarning): + warnings.warn("must fail", UserWarning, stacklevel=1) +except AssertionError: + pass +else: + raise SystemExit("the block passed, so the warning was never recorded") + +if logged: + raise SystemExit(f"the rejected warning was shown anyway: {logged}") + +warnings.warn("after the block", UserWarning, stacklevel=1) +if len(logged) != 1 or "after the block" not in logged[0]: + raise SystemExit(f"the logging capture was not restored: {logged}") + +print("the assertion fired and the logs are back") +""" + + +@pytest.mark.skipif( + sys.version_info < (3, 14), reason="context-aware warnings need Python 3.14" +) +def test_asserting_no_warnings_fails_with_context_aware_warnings() -> None: + """Test that the assertion holds with context-aware warnings enabled too. + + That mode can only be asked for at interpreter startup, and it is the one where + the recording used to see nothing at all once `showwarning()` was replaced, and + the only one where restoring the handler afterwards is this block's job, so it + runs a subprocess instead of leaving the case untested. + """ + result = subprocess.run( + [ + sys.executable, + "-X", + "context_aware_warnings=1", + "-c", + _CONTEXT_AWARE_SCRIPT, + ], + capture_output=True, + text=True, + check=False, + ) + assert result.returncode == 0, result.stderr + assert "the assertion fired" in result.stdout + + +def test_asserting_no_warnings_reports_and_replays_in_order() -> None: + """Test that every offender is reported and every bystander replayed, in order. + + The two are told apart in one pass, so the order of both has to survive it. + """ + with warnings.catch_warnings(record=True, action="always") as caught: + with pytest.raises(AssertionError) as failure: + with asserting_no_warnings(category=DeprecationWarning): + warnings.warn("bystander 1", UserWarning, stacklevel=1) + warnings.warn("offender 1", DeprecationWarning, stacklevel=1) + warnings.warn("bystander 2", UserWarning, stacklevel=1) + warnings.warn("offender 2", DeprecationWarning, stacklevel=1) + + assert [str(warning.message) for warning in caught] == [ + "bystander 1", + "bystander 2", + ] + listing = str(failure.value) + assert listing.index("offender 1") < listing.index("offender 2") + + +def test_asserting_no_warnings_reentry_keeps_what_was_caught() -> None: + """Test that a refused nested entry loses neither a warning nor the cleanup. + + Entering twice would replace the recording and the cleanup of the outer block, + so the warnings raised before it would be dropped, the assertion would pass, and + the recorder would still be installed afterwards. + """ + with warnings.catch_warnings(record=True, action="always") as caught: + block = asserting_no_warnings(category=DeprecationWarning) + with pytest.raises(AssertionError, match="must fail"): + with block: + warnings.warn("passed along", UserWarning, stacklevel=1) + warnings.warn("must fail", DeprecationWarning, stacklevel=1) + with pytest.raises( + RuntimeError, match="Cannot enter asserting_no_warnings" + ): + with block: + pass + warnings.warn("after the block", UserWarning, stacklevel=1) + assert [str(warning.message) for warning in caught] == [ + "passed along", + "after the block", + ] + + +def test_asserting_no_deprecations() -> None: + """Test the deprecation shortcut.""" + with _shown(): + with asserting_no_deprecations(): + warnings.warn("not a deprecation", UserWarning, stacklevel=1) + with pytest.raises(AssertionError, match="gone in v3"): + with asserting_no_deprecations(): + warnings.warn("gone in v3", DeprecationWarning, stacklevel=1) + + +def test_asserting_no_deprecations_matches_the_message() -> None: + """Test that the shortcut can reject one deprecation and let others be.""" + with _shown(): + with asserting_no_deprecations(message="ours"): + warnings.warn("somebody else's", DeprecationWarning, stacklevel=1) + with pytest.raises(AssertionError, match="ours is deprecated"): + with asserting_no_deprecations(message="ours"): + warnings.warn("ours is deprecated", DeprecationWarning, stacklevel=1) diff --git a/tests/warnings/test_blocks.py b/tests/warnings/test_blocks.py new file mode 100644 index 0000000..7e8fc5a --- /dev/null +++ b/tests/warnings/test_blocks.py @@ -0,0 +1,70 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Tests for what every context manager in this package has in common.""" + +from collections.abc import Callable +from typing import Any + +import pytest + +from frequenz.core.warnings import ( + asserting_no_deprecations, + asserting_no_warnings, + ignoring_deprecations, + ignoring_warnings, +) + + +@pytest.mark.parametrize( + "block", + [ + ignoring_warnings, + ignoring_deprecations, + asserting_no_warnings, + asserting_no_deprecations, + ], +) +def test_blocks_are_not_decorators(block: Callable[[], Any]) -> None: + """Test that none of these can be used as a decorator. + + Decorating would cover a whole function body, and for an `async` one it would + cover building the coroutine and nothing else, so an ignore would silence + nothing and an assertion could never fail. These are plain context manager + classes, so Python refuses on its own. + """ + + def convert(value: int) -> int: + return value * 2 + + with pytest.raises(TypeError, match="not callable"): + block()(convert) + # Without the call it would be the class itself doing the decorating, which its + # keyword-only arguments refuse. + with pytest.raises(TypeError): + block(convert) # type: ignore[call-arg] + + +@pytest.mark.parametrize( + "block", + [ + ignoring_warnings, + ignoring_deprecations, + asserting_no_warnings, + asserting_no_deprecations, + ], +) +def test_blocks_reject_reentry(block: Callable[[], Any]) -> None: + """Test that entering the same instance again is refused, and reusing it is not. + + Nesting one instance in itself would install its state twice and take it out + once, so it is refused before anything is touched. That only happens to an + instance kept in a variable, since a `with` statement makes a new one each time. + """ + instance = block() + with instance: + with pytest.raises(RuntimeError, match="Cannot enter .* twice"): + with instance: + pass + with instance: # Left, so it can be entered again. + pass diff --git a/tests/warnings/test_deprecated_aliases.py b/tests/warnings/test_deprecated_aliases.py new file mode 100644 index 0000000..297199a --- /dev/null +++ b/tests/warnings/test_deprecated_aliases.py @@ -0,0 +1,261 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Tests for `deprecated_aliases()`.""" + +import decimal +import fractions +from types import ModuleType +from typing import Any + +import pytest + +from frequenz.core.warnings import ( + _deprecated_aliases, + asserting_no_deprecations, + deprecated_aliases, +) +from tests.warnings import documented_aliases + + +def test_deprecated_aliases_warns_and_returns_the_real_object() -> None: + """Test that reaching an alias warns and yields the object from its new home.""" + module = ModuleType("old_home") + module.__getattr__ = deprecated_aliases( # type: ignore[method-assign] + module.__name__, {"Decimal": "decimal", "Fraction": "fractions"} + ) + + with pytest.warns(DeprecationWarning) as caught: + assert module.Decimal is decimal.Decimal + assert module.Fraction is fractions.Fraction + + assert [str(warning.message) for warning in caught] == [ + "old_home.Decimal is deprecated. Use decimal.Decimal instead.", + "old_home.Fraction is deprecated. Use fractions.Fraction instead.", + ] + # stacklevel=2, so the warning is attributed to this file, not to the helper. + assert all(warning.filename == __file__ for warning in caught) + + +def test_deprecated_aliases_rejects_unknown_names() -> None: + """Test that a name that is not an alias raises as a missing attribute would.""" + module = ModuleType("old_home_missing") + module.__getattr__ = deprecated_aliases( # type: ignore[method-assign] + module.__name__, {"Decimal": "decimal"} + ) + + with pytest.raises( + AttributeError, match="module 'old_home_missing' has no attribute 'Nope'" + ): + _ = module.Nope + + +@pytest.mark.parametrize( + "args, error", + [ + ((0, {"Decimal": "decimal"}), TypeError), + (("old_home_bad", {0: "decimal"}), TypeError), + (("old_home_bad", {"Decimal": 0}), TypeError), + (("old_home_bad", {"Decimal": ""}), ValueError), + (("old_home_bad", {"Decimal": ":Decimal"}), ValueError), + (("old_home_bad", {"Decimal": "decimal:"}), ValueError), + (("old_home_bad", {"Decimal": "a:b:c"}), ValueError), + ], + ids=[ + "module", + "name", + "target-type", + "target-empty", + "target-no-module", + "target-no-name", + "target-two-colons", + ], +) +def test_deprecated_aliases_rejects_a_bad_table( + args: tuple[Any, Any], error: type[Exception] +) -> None: + """Test that the alias table is checked when it is declared. + + Left to the lookup, a table like this fails wherever the alias happens to be + reached, with an error that says nothing about deprecated aliases: an + `AttributeError` about `str.partition` for a target that is not a string, say. + """ + with pytest.raises(error): + deprecated_aliases(*args) + + +def test_deprecated_aliases_copies_the_table() -> None: + """Test that changing the mapping afterwards doesn't get past the checks.""" + aliases: dict[str, str] = {"Decimal": "decimal"} + module = ModuleType("old_home_copied") + module.__getattr__ = deprecated_aliases( # type: ignore[method-assign] + module.__name__, aliases + ) + aliases["Fraction"] = "" + + with pytest.raises(AttributeError): + _ = module.Fraction + + +def test_deprecated_aliases_follows_renames() -> None: + """Test that an alias can point to a symbol with a different name.""" + module = ModuleType("old_home_renamed") + module.__getattr__ = deprecated_aliases( # type: ignore[method-assign] + module.__name__, {"Rational": "fractions:Fraction"} + ) + + with pytest.warns( + DeprecationWarning, + match=( + "^old_home_renamed.Rational is deprecated. " + "Use fractions.Fraction instead.$" + ), + ): + assert module.Rational is fractions.Fraction + + +def test_deprecated_aliases_customises_the_warning() -> None: + """Test the message, category and stacklevel overrides.""" + module = ModuleType("old_home_custom") + module.__getattr__ = deprecated_aliases( # type: ignore[method-assign] + module.__name__, + {"Decimal": "decimal"}, + message="{old} moved to {new} in v2", + category=FutureWarning, + stacklevel=1, + ) + + with pytest.warns( + FutureWarning, match="^old_home_custom.Decimal moved to decimal.Decimal in v2$" + ) as caught: + assert module.Decimal is decimal.Decimal + + # stacklevel=1, so the warning is attributed to the helper itself. + assert caught[0].filename == _deprecated_aliases.__file__ + + +@pytest.mark.parametrize( + "kwargs, error", + [ + ({"message": None}, TypeError), + ({"message": "{old} moved, see {'here': 1}"}, ValueError), + ({"message": "{old} moved to {where}"}, ValueError), + ({"category": int}, TypeError), + ({"stacklevel": "2"}, TypeError), + ({"stacklevel": 0}, ValueError), + ], + ids=[ + "message-type", + "message-stray-brace", + "message-unknown-field", + "category", + "stacklevel-type", + "stacklevel-range", + ], +) +def test_deprecated_aliases_rejects_bad_customisation( + kwargs: dict[str, Any], error: type[Exception] +) -> None: + """Test that the overrides are checked when the aliases are declared. + + A template is checked by formatting it once here, since a stray brace would + otherwise raise a `KeyError` from inside `warnings.warn()`, at the lookup. + """ + with pytest.raises(error): + deprecated_aliases("old_home_bad", {"Decimal": "decimal"}, **kwargs) + + +def test_deprecated_aliases_in_a_real_module() -> None: + """Test the documented shape in a package written the way the docstring says. + + The tests above assign the `__getattr__` onto a `ModuleType` they build + themselves, so none of them says anything about the `if TYPE_CHECKING:` and + `else:` pattern the documentation tells people to write. This one imports a + package written that way, where the `__getattr__` is reached through the + import system like a user's would be. + """ + with pytest.warns( + DeprecationWarning, + match=( + "^tests.warnings.documented_aliases.Decimal is deprecated. " + "Use decimal.Decimal instead.$" + ), + ): + assert documented_aliases.Decimal is decimal.Decimal + + with pytest.warns( + DeprecationWarning, + match=( + "^tests.warnings.documented_aliases.Rational is deprecated. " + "Use fractions.Fraction instead.$" + ), + ): + assert documented_aliases.Rational is fractions.Fraction + + # What the package defines itself is untouched: the `__getattr__` is only + # consulted for names the module doesn't have. + with asserting_no_deprecations(): + assert documented_aliases.kept() == "kept" + + with pytest.raises( + AttributeError, + match="module 'tests.warnings.documented_aliases' has no attribute 'Nope'", + ): + # The `type: ignore` is the point of the `else:` branch, not a nuisance: + # with the `__getattr__` at module level mypy would type this `Any` and + # say nothing, here and in every downstream import of a name that never + # existed. + _ = documented_aliases.Nope # type: ignore[attr-defined] + + +def test_deprecated_aliases_reach_star_imports() -> None: + """Test that the aliases come through a wildcard import, warning as they go. + + `__all__` is the only thing a wildcard import consults, and the names in it + that the module doesn't define are looked up one by one, so they go through + the `__getattr__` and warn like any other access. + """ + namespace: dict[str, Any] = {} + + with pytest.warns(DeprecationWarning) as caught: + # pylint: disable-next=exec-used + exec("from tests.warnings.documented_aliases import *", namespace) + + assert documented_aliases.__all__ == ["Decimal", "Rational", "kept"] + assert namespace["Decimal"] is decimal.Decimal + assert namespace["Rational"] is fractions.Fraction + assert namespace["kept"] is documented_aliases.kept + # Each name in a package's `__all__` is asked for twice, once by the import + # machinery finding out whether it names a submodule and once by the wildcard + # import itself, so every alias warns more than once here. Which messages come + # out is the part worth asserting; how many times CPython looks them up is not. + assert {str(warning.message) for warning in caught} == { + "tests.warnings.documented_aliases.Decimal is deprecated. " + "Use decimal.Decimal instead.", + "tests.warnings.documented_aliases.Rational is deprecated. " + "Use fractions.Fraction instead.", + } + + +def test_deprecated_aliases_refuses_a_module_target() -> None: + """Test that an alias landing on a module is refused instead of served. + + Serving it would half work: `from old import sub` would find it, while + `import old.sub` and `from old.sub import X` would still raise + `ModuleNotFoundError`, because the import system never asks a package's + `__getattr__`. A module that moved needs a real `__init__.py` at the old + path. + """ + module = ModuleType("old_home_package") + module.__getattr__ = deprecated_aliases( # type: ignore[method-assign] + module.__name__, {"path": "os"} # os.path is a module + ) + + with pytest.raises(TypeError, match="^the target of old_home_package.path is"): + _ = module.path + + # And the refusal wins over the warning, which would otherwise announce a move + # that can't be made. + with asserting_no_deprecations(): + with pytest.raises(TypeError): + _ = module.path diff --git a/tests/warnings/test_ignoring.py b/tests/warnings/test_ignoring.py new file mode 100644 index 0000000..c3828f7 --- /dev/null +++ b/tests/warnings/test_ignoring.py @@ -0,0 +1,340 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Tests for `ignoring_warnings()` and `ignoring_deprecations()`. + +Several scenarios compare the standard library's `catch_warnings` (which is expected +to repeat warnings, see https://github.com/python/cpython/issues/73858) with ours, and +check that ours never suppresses a warning the standard library would show. +""" + +import asyncio +import re +import warnings +from collections.abc import Callable +from types import ModuleType +from typing import Any + +import pytest + +from frequenz.core.warnings import _ignoring, ignoring_deprecations, ignoring_warnings + + +def _run(function: Callable[[], Any], times: int = 10) -> int: + """Call `function` repeatedly under "default" and count the warnings shown.""" + with warnings.catch_warnings(record=True, action="default") as caught: + for _ in range(times): + function() + return len(caught) + + +def test_filters_are_reachable() -> None: + """Test that this interpreter keeps its filters where we can reach them.""" + if _ignoring._live_filters() is None: # pylint: disable=protected-access + pytest.skip("The filters in effect are not reachable, using the fallback") + + +def test_matching_warnings_are_ignored() -> None: + """Test that the block ignores what it says and nothing else.""" + with warnings.catch_warnings(record=True, action="always") as caught: + with ignoring_warnings(category=DeprecationWarning): + warnings.warn("gone", DeprecationWarning, stacklevel=1) + warnings.warn("kept", UserWarning, stacklevel=1) + warnings.warn("also kept", DeprecationWarning, stacklevel=1) + assert [str(warning.message) for warning in caught] == ["kept", "also kept"] + + +def test_message_and_module_select_the_warnings() -> None: + """Test the message and module arguments.""" + with warnings.catch_warnings(record=True, action="always") as caught: + with ignoring_warnings(message="this is fine"): + warnings.warn("this is fine, actually", UserWarning, stacklevel=1) + warnings.warn("this is not", UserWarning, stacklevel=1) + with ignoring_warnings(module="no_such_module"): + warnings.warn("from this module", UserWarning, stacklevel=1) + assert [str(warning.message) for warning in caught] == [ + "this is not", + "from this module", + ] + + +def test_module_picks_the_warning_source( + make_module: Callable[[str], ModuleType], +) -> None: + """Test that the module argument silences the module it names, and only it. + + The case above only checks that a pattern matching nothing silences nothing, + which a filter that never matched at all would pass too. The pattern is matched + against the module's `__name__`, not against its file name. + """ + selected = make_module("selected") + other = make_module("other_module") + + with warnings.catch_warnings(record=True, action="always") as caught: + with ignoring_warnings(module="selected"): + selected.warn() + other.warn() + + assert [str(warning.message) for warning in caught] == ["warned in other_module"] + + +def test_stdlib_repeats_warnings(make_module: Callable[[str], ModuleType]) -> None: + """Test the standard library behaviour we are working around (control).""" + module = make_module("control") + block = lambda: warnings.catch_warnings( # noqa: E731 + action="ignore", category=DeprecationWarning + ) + # If this ever fails with a count of 1, cpython#73858 was fixed upstream: the + # standard library block stopped repeating warnings and this whole module may + # not be needed anymore on the versions that carry the fix. + assert _run(lambda: module.warn_then_block(block)) == 10 + + +def test_history_is_preserved(make_module: Callable[[str], ModuleType]) -> None: + """Test that a block after a warning doesn't make it repeat.""" + module = make_module("preserved") + block = lambda: ignoring_warnings(category=DeprecationWarning) # noqa: E731 + assert _run(lambda: module.warn_then_block(block)) == 1 + + +def test_warning_shown_inside_is_not_repeated_outside( + make_module: Callable[[str], ModuleType], +) -> None: + """Test that the history recorded inside the block is valid outside it.""" + module = make_module("inside") + + def step() -> None: + with ignoring_warnings(category=DeprecationWarning): + module.warn() + + assert _run(step) == 1 + + +def test_outer_error_filter_is_honoured() -> None: + """Test that a warning outside the ignored set still raises.""" + with warnings.catch_warnings(action="error"): + with ignoring_warnings(category=UserWarning): + warnings.warn("ignored", UserWarning, stacklevel=1) + with pytest.raises(DeprecationWarning): + warnings.warn("still an error", DeprecationWarning, stacklevel=1) + with pytest.raises(UserWarning): + warnings.warn("an error again", UserWarning, stacklevel=1) + + +def test_overlapping_blocks_leave_no_filter_behind() -> None: + """Test that blocks exited out of order don't strand each other's filters. + + Two threads or two asyncio tasks can enter and leave their blocks interleaved, + which is what this reproduces by driving the context managers by hand. + """ + live = _ignoring._live_filters() # pylint: disable=protected-access + assert live is not None + before = live[:] + + # Interleaving is the whole point here, so a `with` statement can't be used. + # pylint: disable=unnecessary-dunder-call + first = ignoring_warnings(category=UserWarning) + second = ignoring_warnings(category=DeprecationWarning) + first.__enter__() + second.__enter__() + first.__exit__(None, None, None) + second.__exit__(None, None, None) + + live = _ignoring._live_filters() # pylint: disable=protected-access + assert live is not None + assert live == before + + +def test_ignoring_warnings_reentry_leaves_the_block_working() -> None: + """Test that a refused nested entry doesn't disturb the block it was refused in. + + Entering twice would add a second filter and take only one of them out again, + leaving an ignore installed after the block. + """ + with warnings.catch_warnings(record=True, action="always") as caught: + live = _ignoring._live_filters() # pylint: disable=protected-access + assert live is not None + before = live[:] + + block = ignoring_warnings(category=UserWarning) + with block: + with pytest.raises(RuntimeError, match="Cannot enter ignoring_warnings"): + with block: + pass + warnings.warn("still ignored", UserWarning, stacklevel=1) + warnings.warn("shown", UserWarning, stacklevel=1) + + live = _ignoring._live_filters() # pylint: disable=protected-access + assert live is not None + assert live == before + assert [str(warning.message) for warning in caught] == ["shown"] + + +def test_replacing_the_filters_list_inside_strands_nothing() -> None: + """Test that a new filters list installed inside doesn't keep our filter. + + The copy carries our entry too and is the one in effect afterwards, so removing + it only from the list entered with would leave the ignore installed for good. + """ + original = warnings.filters + before = original[:] + try: + with warnings.catch_warnings(record=True, action="always") as caught: + with ignoring_warnings(category=UserWarning): + warnings.filters = list(warnings.filters) + warnings.warn("shown", UserWarning, stacklevel=1) + assert [str(warning.message) for warning in caught] == ["shown"] + assert list(original) == before + finally: + warnings.filters = original + + +def test_simplefilter_inside_keeps_its_filter() -> None: + """Test that a filter installed inside the block remains afterwards. + + It removes an *equal* entry before inserting its own, so on exit ours is not + there anymore and the equal one belongs to the caller. + """ + + class TestWarning(Warning): + """A warning category local to this test.""" + + live = _ignoring._live_filters() # pylint: disable=protected-access + assert live is not None + before = live[:] + + with ignoring_warnings(category=TestWarning): + warnings.simplefilter("ignore", TestWarning) + + live = _ignoring._live_filters() # pylint: disable=protected-access + assert live is not None + assert live == [("ignore", None, TestWarning, None, 0), *before] + + +@pytest.mark.parametrize( + "kwargs", + [ + {"category": "DeprecationWarning"}, + {"category": int}, + {"message": None}, + {"module": 0}, + ], +) +def test_invalid_arguments_are_rejected(kwargs: dict[str, Any]) -> None: + """Test that the checks the standard library does with asserts are done here.""" + with pytest.raises(TypeError): + with ignoring_warnings(**kwargs): + pass + + +@pytest.mark.parametrize("kwargs", [{"message": "("}, {"module": "[a-"}]) +def test_invalid_patterns_are_rejected(kwargs: dict[str, Any]) -> None: + """Test that a pattern that doesn't compile is reported as such.""" + with pytest.raises(re.error): + with ignoring_warnings(**kwargs): + pass + + +def test_exit_without_enter_does_nothing() -> None: + """Test that exiting a block that was never entered is harmless. + + The context manager protocol doesn't allow this, but `ExitStack.push()` and + hand-written cleanup code do, and getting a stray filter removed would be + worse than doing nothing. + """ + live = _ignoring._live_filters() # pylint: disable=protected-access + assert live is not None + before = live[:] + + ignoring_warnings(category=UserWarning).__exit__(None, None, None) + + live = _ignoring._live_filters() # pylint: disable=protected-access + assert live is not None + assert live == before + + +def test_fallback_still_ignores(monkeypatch: pytest.MonkeyPatch) -> None: + """Test the path taken when the filters in effect can't be reached.""" + monkeypatch.setattr( + _ignoring, "_live_filters", lambda: None, raising=True # noqa: ARG005 + ) + with warnings.catch_warnings(record=True, action="always") as caught: + with ignoring_warnings(category=UserWarning): + warnings.warn("gone", UserWarning, stacklevel=1) + warnings.warn("kept", UserWarning, stacklevel=1) + assert [str(warning.message) for warning in caught] == ["kept"] + + +async def test_ignoring_warnings_works_across_awaits() -> None: + """Test that a block held across an await keeps ignoring.""" + with warnings.catch_warnings(record=True, action="always") as caught: + with ignoring_warnings(category=UserWarning): + warnings.warn("before await", UserWarning, stacklevel=1) + await asyncio.sleep(0) + warnings.warn("after await", UserWarning, stacklevel=1) + assert not caught + + +async def test_ignoring_warnings_is_not_task_local() -> None: + """Test that a block held across an await also silences other tasks. + + This pins down the scope note, which is easy to read as a promise of task + isolation on Python 3.14. The sibling is created inside the outer + `catch_warnings` block, and a task inherits the filters list of the task that + started it, so in any mode the filter added here is the same one the sibling + is matched against. + """ + started = asyncio.Event() + warned = asyncio.Event() + + async def sibling() -> None: + await started.wait() + warnings.warn("from another task", UserWarning, stacklevel=1) + warned.set() + + with warnings.catch_warnings(record=True, action="default") as caught: + task = asyncio.create_task(sibling()) + with ignoring_warnings(category=UserWarning): + started.set() + await warned.wait() + await task + + assert not caught + + +def test_ignoring_deprecations_only_ignores_deprecations() -> None: + """Test that the shortcut silences deprecations and nothing else.""" + with warnings.catch_warnings(record=True, action="always") as caught: + with ignoring_deprecations(): + warnings.warn("gone", DeprecationWarning, stacklevel=1) + warnings.warn("kept", UserWarning, stacklevel=1) + warnings.warn("also kept", DeprecationWarning, stacklevel=1) + assert [str(warning.message) for warning in caught] == ["kept", "also kept"] + + +def test_ignoring_deprecations_preserves_history( + make_module: Callable[[str], ModuleType], +) -> None: + """Test that the shortcut doesn't make already shown warnings repeat.""" + module = make_module("deprecations") + + def step() -> None: + module.warn() + with ignoring_deprecations(): + module.warn("internal", DeprecationWarning) + + assert _run(step) == 1 + + +def test_ignoring_deprecations_matches_message_and_module() -> None: + """Test that the shortcut can narrow down to one deprecation.""" + with warnings.catch_warnings(record=True, action="always") as caught: + with ignoring_deprecations(message="ours"): + warnings.warn("ours is deprecated", DeprecationWarning, stacklevel=1) + warnings.warn("somebody else's", DeprecationWarning, stacklevel=1) + with ignoring_deprecations(module="no_such_module"): + warnings.warn("from this module", DeprecationWarning, stacklevel=1) + assert [str(warning.message) for warning in caught] == [ + "somebody else's", + "from this module", + ]