tests!: retry_until(), deprecate retry() (#372)

Fixes #368, as retry() in its current form is broke and won't work within a
`with` block. The function is deprecated and will be removed in 0.13.x. 0.12.x
will warn when using it.

retry_until() is now available and will either raise or return False if raise=False
is passed.

See also:
- tmux-python/tmuxp#620
- tmux-python/tmuxp#704 (comment)

Author: tony

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

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
 ```

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"""

libtmux/test.py

@@ -4,25 +4,34 @@
 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__))
 example_dir = os.path.abspath(os.path.join(current_dir, "..", "examples"))
 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.

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