diff --git a/CHANGES b/CHANGES index da08e4fe8..21ef7fc33 100644 --- a/CHANGES +++ b/CHANGES @@ -8,7 +8,7 @@ To install the unreleased libtmux version, see [developmental releases](https:// $ pip install --user --upgrade --pre libtmux ``` -## libtmux current (unreleased) +## libtmux 0.12.x (unreleased) - _Insert changes/features/fixes for next release here_ @@ -22,6 +22,11 @@ $ pip install --user --upgrade --pre libtmux - Try out sphinx-autoapi for its table of contents generation ({issue}`367`) +### Testing + +- `retry()`: Add deprecation warning. This will be removed in 0.13.x ({issue}`368`, {issue}`372`) +- New function `retry_until()`: Polls a callback function for a set period of time until it returns `True` or times out. By default it will raise {exc}`libtmux.exc.WaitTimeout`, with `raises=False` it will return `False`. Thank you @categulario! ({issue}`368`, {issue}`372`) + ## libtmux 0.11.0 (2022-03-10) ### Compatibility diff --git a/docs/api.md b/docs/api.md index 11ab33040..8fcdc2499 100644 --- a/docs/api.md +++ b/docs/api.md @@ -170,6 +170,10 @@ versions. .. automethod:: libtmux.test.retry ``` +```{eval-rst} +.. automethod:: libtmux.test.retry_until +``` + ```{eval-rst} .. automethod:: libtmux.test.get_test_session_name ``` diff --git a/libtmux/exc.py b/libtmux/exc.py index 541bcbca2..e76c41263 100644 --- a/libtmux/exc.py +++ b/libtmux/exc.py @@ -49,3 +49,8 @@ class InvalidOption(OptionError): class AmbiguousOption(OptionError): """Option that could potentially match more than one.""" + + +class WaitTimeout(LibTmuxException): + + """Function timed out without meeting condition""" diff --git a/libtmux/test.py b/libtmux/test.py index 73f10eb83..ad5494e2b 100644 --- a/libtmux/test.py +++ b/libtmux/test.py @@ -4,11 +4,16 @@ import os import tempfile import time +import warnings +from typing import Callable, Optional + +from .exc import WaitTimeout logger = logging.getLogger(__name__) TEST_SESSION_PREFIX = "libtmux_" RETRY_TIMEOUT_SECONDS = int(os.getenv("RETRY_TIMEOUT_SECONDS", 8)) +RETRY_INTERVAL_SECONDS = float(os.getenv("RETRY_INTERVAL_SECONDS", 0.05)) namer = tempfile._RandomNameSequence() current_dir = os.path.abspath(os.path.dirname(__file__)) @@ -16,13 +21,17 @@ fixtures_dir = os.path.realpath(os.path.join(current_dir, "fixtures")) -def retry(seconds=RETRY_TIMEOUT_SECONDS): +def retry(seconds: Optional[float] = RETRY_TIMEOUT_SECONDS) -> bool: """ Retry a block of code until a time limit or ``break``. + .. deprecated:: 0.12.0 + `retry` doesn't work, it will be removed in libtmux 0.13.0, it is replaced by + `retry_until`, more info: https://github.com/tmux-python/libtmux/issues/368. + Parameters ---------- - seconds : int + seconds : float Seconds to retry, defaults to ``RETRY_TIMEOUT_SECONDS``, which is configurable via environmental variables. @@ -40,9 +49,64 @@ def retry(seconds=RETRY_TIMEOUT_SECONDS): ... if p.current_path == pane_path: ... break """ + warnings.warn( + "retry() is being deprecated and will soon be replaced by retry_until()", + DeprecationWarning, + ) return (lambda: time.time() < time.time() + seconds)() +def retry_until( + fun: Callable, + seconds: float = RETRY_TIMEOUT_SECONDS, + *, + interval: Optional[float] = RETRY_INTERVAL_SECONDS, + raises: Optional[bool] = True, +) -> bool: + """ + Retry a function until a condition meets or the specified time passes. + + Parameters + ---------- + fun : callable + A function that will be called repeatedly until it returns ``True`` or + the specified time passes. + seconds : float + Seconds to retry. Defaults to ``8``, which is configurable via + ``RETRY_TIMEOUT_SECONDS`` environment variables. + interval : float + Time in seconds to wait between calls. Defaults to ``0.05`` and is + configurable via ``RETRY_INTERVAL_SECONDS`` environment variable. + raises : bool + Wether or not to raise an exception on timeout. Defaults to ``True``. + + Examples + -------- + + >>> def f(): + ... p = w.attached_pane + ... p.server._update_panes() + ... return p.current_path == pane_path + ... + ... retry(f) + + In pytest: + + >>> assert retry(f, raises=False) + """ + ini = time.time() + + while not fun(): + end = time.time() + if end - ini >= seconds: + if raises: + raise WaitTimeout() + else: + return False + time.sleep(interval) + return True + + def get_test_session_name(server, prefix=TEST_SESSION_PREFIX): """ Faker to create a session name that doesn't exist. diff --git a/tests/test_test.py b/tests/test_test.py new file mode 100644 index 000000000..e98bd8771 --- /dev/null +++ b/tests/test_test.py @@ -0,0 +1,87 @@ +from time import time + +import pytest + +from libtmux.test import WaitTimeout, retry_until + + +def test_retry_three_times(): + ini = time() + value = 0 + + def call_me_three_times(): + nonlocal value + + if value == 2: + return True + + value += 1 + + return False + + retry_until(call_me_three_times, 1) + + end = time() + + assert abs((end - ini) - 0.1) < 0.01 + + +def test_function_times_out(): + ini = time() + + def never_true(): + return False + + with pytest.raises(WaitTimeout): + retry_until(never_true, 1) + + end = time() + + assert abs((end - ini) - 1.0) < 0.01 + + +def test_function_times_out_no_rise(): + ini = time() + + def never_true(): + return False + + retry_until(never_true, 1, raises=False) + + end = time() + + assert abs((end - ini) - 1.0) < 0.01 + + +def test_function_times_out_no_raise_assert(): + ini = time() + + def never_true(): + return False + + assert not retry_until(never_true, 1, raises=False) + + end = time() + + assert abs((end - ini) - 1.0) < 0.01 + + +def test_retry_three_times_no_raise_assert(): + ini = time() + value = 0 + + def call_me_three_times(): + nonlocal value + + if value == 2: + return True + + value += 1 + + return False + + assert retry_until(call_me_three_times, 1, raises=False) + + end = time() + + assert abs((end - ini) - 0.1) < 0.01