From ed10ae71bfa7fe75635f4877a971e62829deb8cf Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Sun, 20 Sep 2026 16:33:22 +0000 Subject: [PATCH 01/14] Update the Ubuntu version we test on The platforms section claims 20.04, but the runners moved to 24.04, so we update the README to report what is currently actually supported. Signed-off-by: Leandro Lucarella --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 249b9f9..eff8306 100644 --- a/README.md +++ b/README.md @@ -20,7 +20,7 @@ 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 +- **Operating System:** Ubuntu Linux 24.04 - **Architectures:** amd64, arm64 ## Installation From c9967cc52f2557fbe9835c96e7b0fb806aef0dd6 Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Sat, 19 Sep 2026 11:13:47 +0000 Subject: [PATCH 02/14] Add `ignoring_warnings()` MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Silencing a warning around a call is normally done with a `warnings.catch_warnings` block, but entering and leaving one bumps CPython's internal filters version, which invalidates the `__warningregistry__` of every loaded module. Every warning already shown in the program is then shown again, on every call, including warnings emitted by unrelated code (python/cpython#73858, open since 2017). For a converter that silences a deprecation once per message, that turns a warning shown once into one shown for every message that goes by. This new utilty adds the filter to the list the interpreter consults and takes it out again on exit, so the version never moves and no registry ever goes stale. That is sound only for "ignore", which is all this offers: the registry is consulted before the filters, and an ignore can only take warnings away, so a warning recorded as shown on either side of the block is still correctly recorded on the other. `simplefilter()` and `filterwarnings()` bump the version on purpose, because they can also add warnings back. Repairing the registries afterwards is not affordable: it has to scan `sys.modules` and copy every registry it finds, measured at 82 µs per block with 282 modules loaded and ~130 µs with ~370 in a dispatch client field test, against 1.3 µs for the standard library block and 1.9 µs for this. A test suite went from 2.8 s to 12 s on it. Two hazards are handled on exit: 1. Code in the block can replace `warnings.filters` with a copy, which carries our entry. If the lists differ, the entry is removed from both: leaving it in the live one keeps the ignore active, and leaving it in the original lets a `catch_warnings` block restore it. 2. Only our entry is touched. An entry/exit snapshot would make blocks overlapping across threads (A enters, B enters, A exits, B exits) restore each other's stale copies, leaving an ignore installed for good, as the new test shows. So changes made inside the block aren't undone, and the docstring says this is no `catch_warnings` replacement. Re-entering an instance inside its own block raises `RuntimeError` before touching anything, as it would add two filters and remove one. Supporting it not worth it. Removal is a lookup plus a delete by index, so a thread mutating the list concurrently can make it delete the wrong filter. A lock would cost every block to protect only programs changing filters from several threads, but threads are still poorly supported in Python, so we skip it and warn about it in the docs for now. Nor is the ignore per task or thread: the filters are shared, so a block held across an `await` also ignores other tasks' warnings, and an outer `catch_warnings` uses the same state. Python 3.14's `-X context_aware_warnings` doesn't help either: 1. only a `catch_warnings` block creates a new context; 2. a task inherits the filters of the task that started it; 3. outside such a block, the process-wide filters apply. Isolation would need our own context, possible only from 3.14 with the flag on: with it off, the C code reads the context but the Python API writes the module global, so a `catch_warnings` inside records nothing. The scope note explains this, and a test pins down the sibling task case. As the entry is inserted directly, the arguments `warnings.filterwarnings()` checks are checked here too, with real exceptions (it uses `assert`s up to 3.12). A bad pattern raises `re.error`. If the filters in effect can't be reached, it falls back to a standard library block, the old behavior. Signed-off-by: Leandro Lucarella --- src/frequenz/core/warnings/__init__.py | 54 +++++ src/frequenz/core/warnings/_ignoring.py | 271 +++++++++++++++++++++ tests/__init__.py | 8 + tests/warnings/__init__.py | 4 + tests/warnings/conftest.py | 59 +++++ tests/warnings/test_blocks.py | 55 +++++ tests/warnings/test_ignoring.py | 302 ++++++++++++++++++++++++ 7 files changed, 753 insertions(+) create mode 100644 src/frequenz/core/warnings/__init__.py create mode 100644 src/frequenz/core/warnings/_ignoring.py create mode 100644 tests/__init__.py create mode 100644 tests/warnings/__init__.py create mode 100644 tests/warnings/conftest.py create mode 100644 tests/warnings/test_blocks.py create mode 100644 tests/warnings/test_ignoring.py diff --git a/src/frequenz/core/warnings/__init__.py b/src/frequenz/core/warnings/__init__.py new file mode 100644 index 0000000..30d2fc6 --- /dev/null +++ b/src/frequenz/core/warnings/__init__.py @@ -0,0 +1,54 @@ +# 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. + +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 ._ignoring import ignoring_warnings + +__all__ = [ + "ignoring_warnings", +] diff --git a/src/frequenz/core/warnings/_ignoring.py b/src/frequenz/core/warnings/_ignoring.py new file mode 100644 index 0000000..89aac72 --- /dev/null +++ b/src/frequenz/core/warnings/_ignoring.py @@ -0,0 +1,271 @@ +# 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) 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/test_blocks.py b/tests/warnings/test_blocks.py new file mode 100644 index 0000000..1e3b387 --- /dev/null +++ b/tests/warnings/test_blocks.py @@ -0,0 +1,55 @@ +# 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 ignoring_warnings + + +@pytest.mark.parametrize( + "block", + [ignoring_warnings], +) +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], +) +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_ignoring.py b/tests/warnings/test_ignoring.py new file mode 100644 index 0000000..f2e79bb --- /dev/null +++ b/tests/warnings/test_ignoring.py @@ -0,0 +1,302 @@ +# 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_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 From 7afa197a6446f85d73ea1dfea812e89ee3450d7e Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Sat, 19 Sep 2026 16:12:04 +0000 Subject: [PATCH 03/14] Add `ignoring_deprecations()` Silencing the deprecations a library raises on itself is the case that motivated the module, and the one that will appear at dozens of call sites, so it gets a name instead of having to use `ignoring_warnings(DeprecationWarning)` every time. Signed-off-by: Leandro Lucarella --- src/frequenz/core/warnings/__init__.py | 7 +- src/frequenz/core/warnings/_ignoring.py | 95 +++++++++++++++++++++++++ tests/warnings/test_blocks.py | 6 +- tests/warnings/test_ignoring.py | 40 ++++++++++- 4 files changed, 142 insertions(+), 6 deletions(-) diff --git a/src/frequenz/core/warnings/__init__.py b/src/frequenz/core/warnings/__init__.py index 30d2fc6..52fece4 100644 --- a/src/frequenz/core/warnings/__init__.py +++ b/src/frequenz/core/warnings/__init__.py @@ -5,7 +5,9 @@ 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. +[`warnings.catch_warnings`][] has, and +[`ignoring_deprecations`][.ignoring_deprecations] for the common case of a library +having to touch a symbol it deprecated itself. The documented way of silencing a warning locally is a [`warnings.catch_warnings`][] block, but merely entering and leaving one invalidates @@ -47,8 +49,9 @@ def convert(value: str) -> int: `UserWarning` ten times. """ -from ._ignoring import ignoring_warnings +from ._ignoring import ignoring_deprecations, ignoring_warnings __all__ = [ + "ignoring_deprecations", "ignoring_warnings", ] diff --git a/src/frequenz/core/warnings/_ignoring.py b/src/frequenz/core/warnings/_ignoring.py index 89aac72..e0c3b1b 100644 --- a/src/frequenz/core/warnings/_ignoring.py +++ b/src/frequenz/core/warnings/_ignoring.py @@ -269,3 +269,98 @@ def __exit__( 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/warnings/test_blocks.py b/tests/warnings/test_blocks.py index 1e3b387..fabeb77 100644 --- a/tests/warnings/test_blocks.py +++ b/tests/warnings/test_blocks.py @@ -8,12 +8,12 @@ import pytest -from frequenz.core.warnings import ignoring_warnings +from frequenz.core.warnings import ignoring_deprecations, ignoring_warnings @pytest.mark.parametrize( "block", - [ignoring_warnings], + [ignoring_warnings, ignoring_deprecations], ) def test_blocks_are_not_decorators(block: Callable[[], Any]) -> None: """Test that none of these can be used as a decorator. @@ -37,7 +37,7 @@ def convert(value: int) -> int: @pytest.mark.parametrize( "block", - [ignoring_warnings], + [ignoring_warnings, ignoring_deprecations], ) def test_blocks_reject_reentry(block: Callable[[], Any]) -> None: """Test that entering the same instance again is refused, and reusing it is not. diff --git a/tests/warnings/test_ignoring.py b/tests/warnings/test_ignoring.py index f2e79bb..c3828f7 100644 --- a/tests/warnings/test_ignoring.py +++ b/tests/warnings/test_ignoring.py @@ -17,7 +17,7 @@ import pytest -from frequenz.core.warnings import _ignoring, ignoring_warnings +from frequenz.core.warnings import _ignoring, ignoring_deprecations, ignoring_warnings def _run(function: Callable[[], Any], times: int = 10) -> int: @@ -300,3 +300,41 @@ async def sibling() -> None: 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", + ] From 69364c45d57b6496fc947c3fd773e1cd450eda13 Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Sun, 20 Sep 2026 16:01:38 +0000 Subject: [PATCH 04/14] Add helpers for deprecating and aliasing symbols PEP 702 rejected deprecating modules, attributes and constants, and rejected a `Deprecated[T, message]` modifier along with them, so no type checker reports a use of one of these names, and a type alias even launders a deprecation that does exist. This commit introduces `deprecated_aliases()` to overcome this, allowing to define a table of symbols in a module that are deprecated aliases for a symbol new location. A module `__getattr__` is needed for this because the symbol's new location can't be marked as deprecated, it would deprecate it for everybody, including the users of its new home. Serving the object itself, rather than a wrapper, is what keeps `isinstance` working through both import paths, and looking it up lazily keeps the old module from importing the new one just to hold a reference. The alias table is checked and copied when it is declared, rather than left for the lookup to trip over. The usage is a bit tricky/hacky, so it is properly documented. The module's `__getattr__` assignment must be in a `else:` branch of the `if TYPE_CHECKING:` block that declares the aliases, never at module level. Not doing so is problematic because mypy treats a module with a `__getattr__` as `dict[str, Any]`, so all symbols in the module suddenly lose all type annotations. So the assignment must stay out of mypy's eyes. Signed-off-by: Leandro Lucarella --- src/frequenz/core/warnings/__init__.py | 5 + .../core/warnings/_deprecated_aliases.py | 126 ++++++++++++++++++ tests/warnings/test_deprecated_aliases.py | 81 +++++++++++ 3 files changed, 212 insertions(+) create mode 100644 src/frequenz/core/warnings/_deprecated_aliases.py create mode 100644 tests/warnings/test_deprecated_aliases.py diff --git a/src/frequenz/core/warnings/__init__.py b/src/frequenz/core/warnings/__init__.py index 52fece4..ac53608 100644 --- a/src/frequenz/core/warnings/__init__.py +++ b/src/frequenz/core/warnings/__init__.py @@ -9,6 +9,9 @@ [`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. + 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 @@ -49,9 +52,11 @@ def convert(value: str) -> int: `UserWarning` ten times. """ +from ._deprecated_aliases import deprecated_aliases from ._ignoring import ignoring_deprecations, ignoring_warnings __all__ = [ + "deprecated_aliases", "ignoring_deprecations", "ignoring_warnings", ] diff --git a/src/frequenz/core/warnings/_deprecated_aliases.py b/src/frequenz/core/warnings/_deprecated_aliases.py new file mode 100644 index 0000000..38dba94 --- /dev/null +++ b/src/frequenz/core/warnings/_deprecated_aliases.py @@ -0,0 +1,126 @@ +# 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 typing import Any + + +def _checked_aliases(module: str, aliases: Mapping[str, str]) -> dict[str, str]: + """Check a set of deprecated aliases and take a snapshot of it. + + 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: + A copy of the mapping, so a later change to it can't get past these checks. + + Raises: + TypeError: If `module` is not a string, or a name or target is not one. + ValueError: If a target is empty. + """ + if not isinstance(module, str): + raise TypeError(f"module must be a str, got {module!r}") + checked = dict(aliases) + for name, target in checked.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}") + if not target: + raise ValueError(f"the target of {name!r} is empty") + return checked + + +def deprecated_aliases( # noqa: DOC502 + module: str, aliases: Mapping[str, str] +) -> 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. + + 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 name to the fully qualified name of the + module that now owns it. It is copied, so changing it afterwards has no + effect. + + Returns: + A function suitable for use as the module's `__getattr__`. + + Raises: + TypeError: If `module` is not a string, or a name or target is not one. + ValueError: If a target is empty. + """ + aliases = _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. + """ + target_module = aliases.get(name) + if target_module is None: + raise AttributeError(f"module {module!r} has no attribute {name!r}") + warnings.warn( + f"{module}.{name} is deprecated. Use {target_module}.{name} instead.", + DeprecationWarning, + stacklevel=2, + ) + return getattr(importlib.import_module(target_module), name) + + return module_getattr diff --git a/tests/warnings/test_deprecated_aliases.py b/tests/warnings/test_deprecated_aliases.py new file mode 100644 index 0000000..dcce6ba --- /dev/null +++ b/tests/warnings/test_deprecated_aliases.py @@ -0,0 +1,81 @@ +# 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 + + +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), + ], + ids=["module", "name", "target-type", "target-empty"], +) +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 From 0972d477d6f36c6833922563133b516c88e3b522 Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Sun, 20 Sep 2026 16:03:33 +0000 Subject: [PATCH 05/14] Let deprecated aliases point to a renamed symbol A symbol that moves to another module might also be renamed on the way, and so far the alias could only keep the name it had. The target string now takes an optional `:name` suffix, as an entry point does: `"fractions:Fraction"` next to the plain `"decimal"`. The suffix is checked along with the rest of the table, since `partition()` stops at the first colon: `"a:b:c"` would otherwise pass as module `a` and name `b:c`, and the deprecation would tell the user to reach for `a.b:c`. An empty module or an empty name after the colon is refused there too. Taking the string apart once, when the table is declared rather than on every lookup, is what makes that possible. Signed-off-by: Leandro Lucarella --- .../core/warnings/_deprecated_aliases.py | 77 +++++++++++++++---- tests/warnings/test_deprecated_aliases.py | 30 +++++++- 2 files changed, 89 insertions(+), 18 deletions(-) diff --git a/src/frequenz/core/warnings/_deprecated_aliases.py b/src/frequenz/core/warnings/_deprecated_aliases.py index 38dba94..0b9a1e4 100644 --- a/src/frequenz/core/warnings/_deprecated_aliases.py +++ b/src/frequenz/core/warnings/_deprecated_aliases.py @@ -12,8 +12,10 @@ from typing import Any -def _checked_aliases(module: str, aliases: Mapping[str, str]) -> dict[str, str]: - """Check a set of deprecated aliases and take a snapshot of it. +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, @@ -24,22 +26,37 @@ def _checked_aliases(module: str, aliases: Mapping[str, str]) -> dict[str, str]: aliases: The mapping to check. Returns: - A copy of the mapping, so a later change to it can't get past these checks. + 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. + 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(aliases) - for name, target in checked.items(): + 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}") - if not target: - raise ValueError(f"the target of {name!r} is empty") + 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 @@ -81,6 +98,28 @@ def deprecated_aliases( # noqa: DOC502 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", + }, + ) + ``` + 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; @@ -88,18 +127,20 @@ def deprecated_aliases( # noqa: DOC502 Args: module: The fully qualified name of the module defining the aliases. - aliases: A mapping of each deprecated name to the fully qualified name of the - module that now owns it. It is copied, so changing it afterwards has no - effect. + 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. Returns: A function suitable for use as the module's `__getattr__`. Raises: TypeError: If `module` is not a string, or a name or target is not one. - ValueError: If a target is empty. + ValueError: If a target is empty, carries more than one `:`, or has + nothing after it. """ - aliases = _checked_aliases(module, aliases) + targets = _checked_aliases(module, aliases) def module_getattr(name: str) -> Any: """Return a deprecated alias, warning about its new location. @@ -113,14 +154,16 @@ def module_getattr(name: str) -> Any: Raises: AttributeError: If the name is not one of the deprecated aliases. """ - target_module = aliases.get(name) - if target_module is None: + target = targets.get(name) + if target is None: raise AttributeError(f"module {module!r} has no attribute {name!r}") + target_module, target_name = target warnings.warn( - f"{module}.{name} is deprecated. Use {target_module}.{name} instead.", + f"{module}.{name} is deprecated. " + f"Use {target_module}.{target_name} instead.", DeprecationWarning, stacklevel=2, ) - return getattr(importlib.import_module(target_module), name) + return getattr(importlib.import_module(target_module), target_name) return module_getattr diff --git a/tests/warnings/test_deprecated_aliases.py b/tests/warnings/test_deprecated_aliases.py index dcce6ba..aa75ab3 100644 --- a/tests/warnings/test_deprecated_aliases.py +++ b/tests/warnings/test_deprecated_aliases.py @@ -52,8 +52,19 @@ def test_deprecated_aliases_rejects_unknown_names() -> None: (("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", ], - ids=["module", "name", "target-type", "target-empty"], ) def test_deprecated_aliases_rejects_a_bad_table( args: tuple[Any, Any], error: type[Exception] @@ -79,3 +90,20 @@ def test_deprecated_aliases_copies_the_table() -> None: 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 From 7ae5c007b42070cd7e9a0f682b6fb2b4cd0ebdf2 Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Sun, 20 Sep 2026 16:06:22 +0000 Subject: [PATCH 06/14] Let deprecated aliases customise the warning The default message says where the symbol went and nothing else, which is enough for a plain move but not when the deprecation comes with a removal version, a link to a migration guide, or a category other than `DeprecationWarning` (`FutureWarning` for something aimed at end users, a library's own category so it can be filtered on its own). So `message` is a template formatted with the old and new fully qualified names, `category` is the warning class, and `stacklevel` is there for the rare case of the returned `__getattr__` being wrapped. All three are checked where the alias table already is, when the aliases are declared. The template is checked by formatting it once with empty names, which is the only way to catch a stray brace: a message carrying something like `{'a': 1}` otherwise raises a `KeyError` from inside `warnings.warn()`, at the lookup, pointing at neither the template nor the call that set it. Signed-off-by: Leandro Lucarella --- .../core/warnings/_deprecated_aliases.py | 71 +++++++++++++++++-- tests/warnings/test_deprecated_aliases.py | 53 +++++++++++++- 2 files changed, 116 insertions(+), 8 deletions(-) diff --git a/src/frequenz/core/warnings/_deprecated_aliases.py b/src/frequenz/core/warnings/_deprecated_aliases.py index 0b9a1e4..3af0f6e 100644 --- a/src/frequenz/core/warnings/_deprecated_aliases.py +++ b/src/frequenz/core/warnings/_deprecated_aliases.py @@ -61,7 +61,12 @@ def _checked_aliases( def deprecated_aliases( # noqa: DOC502 - module: str, aliases: Mapping[str, str] + 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. @@ -120,6 +125,30 @@ def deprecated_aliases( # noqa: DOC502 ) ``` + 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.", + ) + ``` + 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; @@ -131,15 +160,42 @@ def deprecated_aliases( # noqa: DOC502 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 `module` is not a string, or a name or target is not one. + TypeError: If an argument has the wrong type, including a name or target + that is not a string. ValueError: If a target is empty, carries more than one `:`, or has - nothing after it. + 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: @@ -159,10 +215,11 @@ def module_getattr(name: str) -> Any: raise AttributeError(f"module {module!r} has no attribute {name!r}") target_module, target_name = target warnings.warn( - f"{module}.{name} is deprecated. " - f"Use {target_module}.{target_name} instead.", - DeprecationWarning, - stacklevel=2, + message.format( + old=f"{module}.{name}", new=f"{target_module}.{target_name}" + ), + category, + stacklevel=stacklevel, ) return getattr(importlib.import_module(target_module), target_name) diff --git a/tests/warnings/test_deprecated_aliases.py b/tests/warnings/test_deprecated_aliases.py index aa75ab3..9d29f77 100644 --- a/tests/warnings/test_deprecated_aliases.py +++ b/tests/warnings/test_deprecated_aliases.py @@ -10,7 +10,7 @@ import pytest -from frequenz.core.warnings import deprecated_aliases +from frequenz.core.warnings import _deprecated_aliases, deprecated_aliases def test_deprecated_aliases_warns_and_returns_the_real_object() -> None: @@ -107,3 +107,54 @@ def test_deprecated_aliases_follows_renames() -> None: ), ): 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) From a64a1e41f93ecad13862c60193591b9ab2d1d50f Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Tue, 22 Sep 2026 16:12:17 +0000 Subject: [PATCH 07/14] Add `asserting_no_warnings()` The usual way to assert that a call doesn't warn is an `"error"` filter, which is also what pytest documents since `pytest.warns(None)` was removed in pytest 8. But it makes `warnings.warn()` raise inside the code under test, where a broad `except` can swallow it or turn it into an unrelated failure, and the failure points at the `warn()` call instead of the call the test is about. This records instead. Matching warnings are reported together, each with its message and `filename:lineno`. The `"always"` filter added inside matches on message as well as category, so a warning already shown elsewhere is still caught, while other warnings keep whatever the surrounding filters say (ignored, deduplicated or raised). Recording needs `showwarning` reset to the default, and `catch_warnings(record=True)` only does that in process-wide mode. With context-aware warnings (3.14, default in free-threaded builds), a handler installed by the application, like `logging.captureWarnings()`, keeps taking every warning and nothing is recorded, so the assertion passes silently (python/cpython#151149, which also breaks `pytest.warns`). So the reset is done here in every mode, and undone before recording ends so the replay reaches the application's handler. A handler the code under test installs inside the block still wins. On exit, every warning not turned into a failure is replayed, since recording is the only reason it wasn't shown. If an exception is propagating, it takes precedence over the assertion and the matching warnings are replayed too, so a deprecation inside `pytest.raises()` isn't lost. Replay goes through `_showwarnmsg()`: the public `showwarning()` drops the `source` of a `ResourceWarning`, and the default implementation under it ignores an application's `showwarning()`, sending logged warnings to stderr. Unlike `ignoring_warnings()`, this resets the deduplication history. It has to: a warning already shown is rejected by `__warningregistry__` before the filters run, and only bumping the filters version, as `catch_warnings()` does, invalidates it. For tests, this cost is reasonable, but the docstring says to keep it out of hot paths and long-running processes. Other choices: - No `module` argument: a recorded `WarningMessage` has a filename, not the module name the filter matched, so the report couldn't reproduce the filter's decision. A comment explains this. - Recorded warnings are sorted in a single pass, since an `"always"` filter in a loop can record tens of thousands of them. - Re-entering an instance raises `RuntimeError`, as in `ignoring_warnings()`. Otherwise the inner entry would replace the outer recording, and the assertion would silently pass. Known limits, documented in the docstring: - Without context-aware warnings recording is process-wide, so a matching warning from another thread, or from another task across an `await`, fails the block (a test covers this). With them, new threads escape it, but tasks started inside still land here. - An `ignoring_warnings()` inside the code under test wins over this filter, so the assertion covers only what the code didn't silence. Signed-off-by: Leandro Lucarella --- src/frequenz/core/warnings/__init__.py | 6 +- src/frequenz/core/warnings/_asserting.py | 260 +++++++++++++++ tests/warnings/test_asserting.py | 401 +++++++++++++++++++++++ tests/warnings/test_blocks.py | 18 +- 4 files changed, 681 insertions(+), 4 deletions(-) create mode 100644 src/frequenz/core/warnings/_asserting.py create mode 100644 tests/warnings/test_asserting.py diff --git a/src/frequenz/core/warnings/__init__.py b/src/frequenz/core/warnings/__init__.py index ac53608..a62b064 100644 --- a/src/frequenz/core/warnings/__init__.py +++ b/src/frequenz/core/warnings/__init__.py @@ -10,7 +10,9 @@ 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. +path of a symbol that moved to another module working, warning whoever uses it, and +[`asserting_no_warnings`][.asserting_no_warnings], 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 @@ -52,10 +54,12 @@ def convert(value: str) -> int: `UserWarning` ten times. """ +from ._asserting import asserting_no_warnings from ._deprecated_aliases import deprecated_aliases from ._ignoring import ignoring_deprecations, ignoring_warnings __all__ = [ + "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..f89ebf4 --- /dev/null +++ b/src/frequenz/core/warnings/_asserting.py @@ -0,0 +1,260 @@ +# 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 around code using + [`ignoring_deprecations`][..ignoring_deprecations] internally, this block + 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}") diff --git a/tests/warnings/test_asserting.py b/tests/warnings/test_asserting.py new file mode 100644 index 0000000..e9b4714 --- /dev/null +++ b/tests/warnings/test_asserting.py @@ -0,0 +1,401 @@ +# 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_warnings + + +@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_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", + ] diff --git a/tests/warnings/test_blocks.py b/tests/warnings/test_blocks.py index fabeb77..e786b1f 100644 --- a/tests/warnings/test_blocks.py +++ b/tests/warnings/test_blocks.py @@ -8,12 +8,20 @@ import pytest -from frequenz.core.warnings import ignoring_deprecations, ignoring_warnings +from frequenz.core.warnings import ( + asserting_no_warnings, + ignoring_deprecations, + ignoring_warnings, +) @pytest.mark.parametrize( "block", - [ignoring_warnings, ignoring_deprecations], + [ + ignoring_warnings, + ignoring_deprecations, + asserting_no_warnings, + ], ) def test_blocks_are_not_decorators(block: Callable[[], Any]) -> None: """Test that none of these can be used as a decorator. @@ -37,7 +45,11 @@ def convert(value: int) -> int: @pytest.mark.parametrize( "block", - [ignoring_warnings, ignoring_deprecations], + [ + ignoring_warnings, + ignoring_deprecations, + asserting_no_warnings, + ], ) def test_blocks_reject_reentry(block: Callable[[], Any]) -> None: """Test that entering the same instance again is refused, and reusing it is not. From 0942ecffcbea49135bac85aa6a9cb8cb65af6700 Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Tue, 22 Sep 2026 16:12:31 +0000 Subject: [PATCH 08/14] Add `asserting_no_deprecations()` Checking that the replacement for a deprecated symbol doesn't itself go through the deprecated one is the case these assertions will be written for most of the time, so it gets a name instead of `asserting_no_warnings(category=DeprecationWarning)` spelled out at every call site, the same way `ignoring_deprecations()` shortens the other block. `message` is forwarded, and is the only way to narrow this one down: the block wraps the code being tested, so unlike an `ignoring_deprecations()` block it can't be made more precise by making it shorter. That matters when the code under test legitimately deprecates something else, or reaches a third-party deprecation there is nothing to be done about. Signed-off-by: Leandro Lucarella --- src/frequenz/core/warnings/__init__.py | 8 ++-- src/frequenz/core/warnings/_asserting.py | 55 ++++++++++++++++++++++-- tests/warnings/test_asserting.py | 40 ++++++++++++++++- tests/warnings/test_blocks.py | 3 ++ 4 files changed, 99 insertions(+), 7 deletions(-) diff --git a/src/frequenz/core/warnings/__init__.py b/src/frequenz/core/warnings/__init__.py index a62b064..6ad19a1 100644 --- a/src/frequenz/core/warnings/__init__.py +++ b/src/frequenz/core/warnings/__init__.py @@ -11,8 +11,9 @@ 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], to check in a test that a piece of -code doesn't warn. +[`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 @@ -54,11 +55,12 @@ def convert(value: str) -> int: `UserWarning` ten times. """ -from ._asserting import asserting_no_warnings +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", diff --git a/src/frequenz/core/warnings/_asserting.py b/src/frequenz/core/warnings/_asserting.py index f89ebf4..1bd93f3 100644 --- a/src/frequenz/core/warnings/_asserting.py +++ b/src/frequenz/core/warnings/_asserting.py @@ -94,9 +94,10 @@ class asserting_no_warnings(contextlib.AbstractContextManager[None]): 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 around code using - [`ignoring_deprecations`][..ignoring_deprecations] internally, this block - only checks the deprecations the code didn't silence, not all of them. + 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 @@ -258,3 +259,51 @@ def __exit__( 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/tests/warnings/test_asserting.py b/tests/warnings/test_asserting.py index e9b4714..f6a508c 100644 --- a/tests/warnings/test_asserting.py +++ b/tests/warnings/test_asserting.py @@ -15,7 +15,12 @@ import pytest -from frequenz.core.warnings import _asserting, asserting_no_warnings +from frequenz.core.warnings import ( + _asserting, + asserting_no_deprecations, + asserting_no_warnings, + ignoring_deprecations, +) @contextlib.contextmanager @@ -215,6 +220,19 @@ def test_replay_without_showwarnmsg(monkeypatch: pytest.MonkeyPatch) -> None: 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: @@ -399,3 +417,23 @@ def test_asserting_no_warnings_reentry_keeps_what_was_caught() -> None: "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 index e786b1f..7e8fc5a 100644 --- a/tests/warnings/test_blocks.py +++ b/tests/warnings/test_blocks.py @@ -9,6 +9,7 @@ import pytest from frequenz.core.warnings import ( + asserting_no_deprecations, asserting_no_warnings, ignoring_deprecations, ignoring_warnings, @@ -21,6 +22,7 @@ ignoring_warnings, ignoring_deprecations, asserting_no_warnings, + asserting_no_deprecations, ], ) def test_blocks_are_not_decorators(block: Callable[[], Any]) -> None: @@ -49,6 +51,7 @@ def convert(value: int) -> int: ignoring_warnings, ignoring_deprecations, asserting_no_warnings, + asserting_no_deprecations, ], ) def test_blocks_reject_reentry(block: Callable[[], Any]) -> None: From a228111c5ce2f64667ffa18017208c14e3865038 Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Mon, 21 Sep 2026 12:42:38 +0000 Subject: [PATCH 09/14] Test deprecated_alias in a real package Every `deprecated_aliases()` test so far builds a `ModuleType` and assigns the returned function onto it. That covers the function but not the `if TYPE_CHECKING:` and `else:` structure the docstring tells people to write, or a module the import system actually loaded, nor `__all__`. The pattern could stop working and the tests would stay green. Add `tests/warnings/documented_aliases/` for the new tests. The new tests check what the documentation promises: the alias is the very same object as its target, reaching it warns with the name of its new home, a name the package defines itself is untouched, an unknown name still raises `AttributeError`, and `__all__` with a wildcard import pulls the aliases through, warning on the way. The wildcard test asserts the set of messages and not their number, because a package is asked for each name in `__all__` twice, once by the import machinery finding out whether it names a submodule and once by the wildcard import itself. That is CPython's business, not this module's. The `type: ignore[attr-defined]` on the unknown name is the whole point of the `else:` branch rather than a wart: with the `__getattr__` at module level mypy would type that access `Any` and say nothing at all. The package sits next to the tests that import it and is reached as `tests.warnings.documented_aliases`, which is the name both pytest and mypy know it by, since `tests/` and `tests/warnings/` are packages. That is also why the deprecation messages it raises carry that whole path. Signed-off-by: Leandro Lucarella --- tests/warnings/documented_aliases/__init__.py | 48 +++++++++++ tests/warnings/test_deprecated_aliases.py | 79 ++++++++++++++++++- 2 files changed, 126 insertions(+), 1 deletion(-) create mode 100644 tests/warnings/documented_aliases/__init__.py 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_deprecated_aliases.py b/tests/warnings/test_deprecated_aliases.py index 9d29f77..c1cc2bf 100644 --- a/tests/warnings/test_deprecated_aliases.py +++ b/tests/warnings/test_deprecated_aliases.py @@ -10,7 +10,12 @@ import pytest -from frequenz.core.warnings import _deprecated_aliases, deprecated_aliases +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: @@ -158,3 +163,75 @@ def test_deprecated_aliases_rejects_bad_customisation( """ 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.", + } From 2b59774288c65b044c179f30472fa4b0c7d7a7e7 Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Mon, 21 Sep 2026 12:46:53 +0000 Subject: [PATCH 10/14] Refuse to alias a module `deprecated_aliases()` can keep a symbol reachable from its old module, and nothing more. A submodule that moved is out of its reach, because the import system never consults a package's `__getattr__`: with a table mapping `sub` to `newpkg:sub`, `import old.sub` and `from old.sub import X` both raise `ModuleNotFoundError`, and only `from old import sub` goes through the alias, and then only when the new package imports the submodule in its own `__init__.py`. Serving it anyway is the worst of the three outcomes. A user who writes the one form that works ships code that breaks the moment somebody else writes either of the other two, and the error they get says nothing about the alias table that promised the move. So resolving an alias onto a module now raises `TypeError` naming the alias and saying what to do instead, which is to keep a real `__init__.py` at the old path with the alias table in its body. The check is at resolution and not at declaration, unlike everything else checked here. Telling a module from a name at declaration time means asking `importlib.util.find_spec()`, which imports the target's parent package to answer, and the laziness that buys is the reason the helper exists. Resolution costs nothing and fires the first time anything reaches the alias, which in these repositories is their own deprecation test. The target is resolved before the warning is raised now, for the same reason: an alias that can't work should say so rather than first announce a move that didn't happen, and under an `"error"` filter that warning would raise and hide the `TypeError` entirely. Signed-off-by: Leandro Lucarella --- .../core/warnings/_deprecated_aliases.py | 28 +++++++++++++++++-- tests/warnings/test_deprecated_aliases.py | 24 ++++++++++++++++ 2 files changed, 50 insertions(+), 2 deletions(-) diff --git a/src/frequenz/core/warnings/_deprecated_aliases.py b/src/frequenz/core/warnings/_deprecated_aliases.py index 3af0f6e..8881822 100644 --- a/src/frequenz/core/warnings/_deprecated_aliases.py +++ b/src/frequenz/core/warnings/_deprecated_aliases.py @@ -9,6 +9,7 @@ import importlib import warnings from collections.abc import Callable, Mapping +from types import ModuleType from typing import Any @@ -149,6 +150,12 @@ def deprecated_aliases( # noqa: DOC502 ) ``` + 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; @@ -174,7 +181,8 @@ def deprecated_aliases( # noqa: DOC502 Raises: TypeError: If an argument has the wrong type, including a name or target - that is not a string. + 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. @@ -209,11 +217,27 @@ def module_getattr(name: str) -> Any: 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}" @@ -221,6 +245,6 @@ def module_getattr(name: str) -> Any: category, stacklevel=stacklevel, ) - return getattr(importlib.import_module(target_module), target_name) + return value return module_getattr diff --git a/tests/warnings/test_deprecated_aliases.py b/tests/warnings/test_deprecated_aliases.py index c1cc2bf..297199a 100644 --- a/tests/warnings/test_deprecated_aliases.py +++ b/tests/warnings/test_deprecated_aliases.py @@ -235,3 +235,27 @@ def test_deprecated_aliases_reach_star_imports() -> None: "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 From 307634ff03f455ff7e8e438105425144e399521d Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Sun, 20 Sep 2026 16:33:40 +0000 Subject: [PATCH 11/14] Test on every Python version this supports `requires-python` says `>= 3.11, < 4`, but the matrix stopped at 3.12, so 3.13 and 3.14 were shipped untested. The new `warnings` module makes that expensive: it reaches for `warnings._get_filters()`, `_showwarning_orig` and `_showwarnmsg()`, private names that only exist or only matter on those versions, and its one test for context-aware warnings carries a `skipif(sys.version_info < (3, 14))`, so it never ran anywhere. The fallbacks around those names are silent by design, which is what makes the gap worth closing rather than noting: if `_showwarning_orig` goes away, `asserting_no_warnings()` stops recording under `logging.captureWarnings()` and every assertion passes, and nothing says so. The README claimed 3.11 alone under the platforms it is tested on, which was true of the matrix and not of `requires-python`; it now lists the four. Signed-off-by: Leandro Lucarella --- .github/workflows/ci.yaml | 4 ++++ README.md | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) 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 eff8306..85b884b 100644 --- a/README.md +++ b/README.md @@ -19,7 +19,7 @@ library with utilities that are frequently needed across different projects. The following platforms are officially supported (tested): -- **Python:** 3.11 +- **Python:** 3.11, 3.12, 3.13, 3.14 - **Operating System:** Ubuntu Linux 24.04 - **Architectures:** amd64, arm64 From 4da142c5603c668d2530163a4c3b34c256fecc90 Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Mon, 21 Sep 2026 12:34:52 +0000 Subject: [PATCH 12/14] tests: Error on warnings, except for deprecations The `addopts` line predates the current `frequenz-repo-config` template, which has no `addopts` at all and says the same thing through `filterwarnings`, an `"error"` default with `"once"` for the two deprecation categories. Expressing it as a list also makes room for the per-warning exceptions a repository eventually needs, which a command line string can only carry as more `-W` flags. The policy itself is unchanged and worth restating: deprecations must never be errors, here or downstream, so they are shown rather than raised. `"once"` is per message and category for the whole run, where the old `"default"` was per location, which is why the test module builds its warning messages out of the module name. The `-vv` goes with it, separately: the template dropped it, and repo-config carried a `migrate.py` step to take it out of existing repositories, which this one never ran. Signed-off-by: Leandro Lucarella --- pyproject.toml | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) 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" From 63cefa12f7b015fafc8c6c0a6e8f91bb717f861d Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Tue, 22 Sep 2026 16:13:37 +0000 Subject: [PATCH 13/14] Add the warnings module to the README The README lists what the library provides and carries a short example per module, `enum` and `typing` included, so a new module that appears only in the generated API reference is invisible from the landing page. The examples are the three entry points, each showing the thing that makes it worth having rather than its full shape: the deduplication history that survives an `ignoring_deprecations()` block, an alias table with a rename in it, and an assertion that a call doesn't warn. Signed-off-by: Leandro Lucarella --- README.md | 77 ++++++++++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 74 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 85b884b..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 @@ -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: From 0f6e269dfbd09eb761e26a6c019402363bc79036 Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Tue, 22 Sep 2026 16:13:52 +0000 Subject: [PATCH 14/14] Update release notes Signed-off-by: Leandro Lucarella --- RELEASE_NOTES.md | 16 +++++----------- 1 file changed, 5 insertions(+), 11 deletions(-) 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.