From e3e88a1c570f16c8a55a6b5c08345771ed536ec9 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Wed, 29 Jul 2026 18:41:55 -0500 Subject: [PATCH] retry_until(feat): Forward args to the predicate why: A zero-argument predicate has nowhere to put the object being waited on, so a wait written inside a loop must close over the loop variable. Ruff's B023 rejects that closure, and B023's documented fix (a default-argument binding) makes mypy report "Cannot infer type of lambda", leaving `# type: ignore[misc]` as the only way out. Refs #726. what: - Add a keyword-only `args` tuple to retry_until(), unpacked into every call of the predicate - Add two overloads: the zero-argument predicate keeps strict checking when `args` is absent, and `args` is required on the forwarding overload so it cannot be selected by accident - Document `args` and add a doctest that waits on each pane of a loop - Cover the loop case and argument forwarding in tests/test/test_retry.py --- CHANGES | 6 +++++ src/libtmux/test/retry.py | 48 +++++++++++++++++++++++++++++++++++++-- tests/test/test_retry.py | 37 ++++++++++++++++++++++++++++++ 3 files changed, 89 insertions(+), 2 deletions(-) diff --git a/CHANGES b/CHANGES index 7a691b0cc..9606caf4c 100644 --- a/CHANGES +++ b/CHANGES @@ -45,6 +45,12 @@ $ uvx --from 'libtmux' --prerelease allow python _Notes on the upcoming release will go here._ +### Fixes + +- {func}`~libtmux.test.retry.retry_until` accepts an `args` tuple forwarded to + the predicate, so waiting on a loop variable no longer needs a closure that + ruff's `B023` rejects and mypy cannot infer (#726) + ### Documentation #### Cleaner `from_env` examples (#719) diff --git a/src/libtmux/test/retry.py b/src/libtmux/test/retry.py index bad491171..3ddc38bf7 100644 --- a/src/libtmux/test/retry.py +++ b/src/libtmux/test/retry.py @@ -19,10 +19,32 @@ pass +@t.overload def retry_until( fun: Callable[[], bool], + seconds: float = ..., + *, + interval: float = ..., + raises: bool | None = ..., +) -> bool: ... + + +@t.overload +def retry_until( + fun: Callable[..., bool], + seconds: float = ..., + *, + args: tuple[t.Any, ...], + interval: float = ..., + raises: bool | None = ..., +) -> bool: ... + + +def retry_until( + fun: Callable[..., bool], seconds: float = RETRY_TIMEOUT_SECONDS, *, + args: tuple[t.Any, ...] = (), interval: float = RETRY_INTERVAL_SECONDS, raises: bool | None = True, ) -> bool: @@ -33,10 +55,15 @@ def retry_until( ---------- fun : callable A function that will be called repeatedly until it returns ``True`` or - the specified time passes. + the specified time passes. Called with ``args`` unpacked, so it accepts + no arguments unless ``args`` is given. seconds : float Seconds to retry. Defaults to ``8``, which is configurable via ``RETRY_TIMEOUT_SECONDS`` environment variables. + args : tuple + Positional arguments passed to ``fun`` on every call. Pass the subject + being waited on here instead of closing over it, which keeps predicates + written inside a loop free of a loop-variable binding. interval : float Time in seconds to wait between calls. Defaults to ``0.05`` and is configurable via ``RETRY_INTERVAL_SECONDS`` environment variable. @@ -55,10 +82,27 @@ def retry_until( In pytest: >>> assert retry_until(fn, raises=False) + + Waiting on something from a loop takes ``args`` rather than a closure: + + >>> window = session.new_window(window_name='retry_until_args') + >>> panes = [window.active_pane, window.split()] + >>> for pane in panes: + ... pane.send_keys('echo ready') + + Compare whole lines. ``capture_pane`` returns the command tmux echoed onto + the pane, so ``'ready' in line`` matches that echo and is already true + before the shell has run anything: + + >>> for pane in panes: + ... assert retry_until( + ... lambda p: any(line.strip() == 'ready' for line in p.capture_pane()), + ... args=(pane,), + ... ) """ ini = time.time() - while not fun(): + while not fun(*args): end = time.time() if end - ini >= seconds: if raises: diff --git a/tests/test/test_retry.py b/tests/test/test_retry.py index c2a0f9255..dd7a8aac5 100644 --- a/tests/test/test_retry.py +++ b/tests/test/test_retry.py @@ -2,6 +2,7 @@ from __future__ import annotations +import typing as t from time import sleep, time import pytest @@ -9,6 +10,9 @@ from libtmux import exc from libtmux.test.retry import retry_until +if t.TYPE_CHECKING: + from libtmux.session import Session + def test_retry_three_times() -> None: """Test retry_until().""" @@ -103,3 +107,36 @@ def call_me_three_times() -> bool: end = time() assert 0.9 <= (end - ini) <= 1.1 # Allow for small timing variations + + +def test_retry_until_forwards_args_from_loop(session: Session) -> None: + """Waiting on a loop variable needs no closure, so no B023 suppression.""" + window = session.new_window(window_name="retry_until_args") + active_pane = window.active_pane + assert active_pane is not None + panes = [active_pane, window.split()] + + for pane in panes: + pane.send_keys("echo ready") + + for pane in panes: + # Whole lines, not a substring: capture_pane returns the command tmux + # echoed onto the pane, so ``"ready" in ...`` is true before the shell + # has run and the wait would prove nothing. + assert retry_until( + lambda p: any(line.strip() == "ready" for line in p.capture_pane()), + 2, + args=(pane,), + ) + + +def test_retry_until_args_reach_predicate() -> None: + """Every retry passes the same ``args`` to the predicate.""" + seen: list[tuple[str, int]] = [] + + def ready(name: str, threshold: int) -> bool: + seen.append((name, threshold)) + return len(seen) >= threshold + + assert retry_until(ready, 1, args=("pane", 3), interval=0) + assert seen == [("pane", 3)] * 3