Add testing mode

ref #59

CHANGELOG.md

@@ -23,6 +23,9 @@ You can find our backwards-compatibility policy [here](https://github.com/hynek/
 - `stamina.Attempt` now has a `next_wait` attribute that contains the time the *next* backoff will wait, if the *current* attempt fails (sans jitter).
   [#72](https://github.com/hynek/stamina/pull/72)
 
+- It is now possible to switch *stamina* into a testing mode using `stamina.set_testing()`.
+  It disables backoffs and caps the number of retries.
+
 
 ## [24.2.0](https://github.com/hynek/stamina/compare/24.1.0...24.2.0) - 2024-01-31
 

README.md

@@ -23,7 +23,7 @@ It is the result of years of copy-pasting the same configuration over and over a
 - Automatic **async** support – including [Trio](https://trio.readthedocs.io/).
 - Preserve **type hints** of the decorated callable.
 - Flexible **instrumentation** with [Prometheus](https://github.com/prometheus/client_python), [*structlog*](https://www.structlog.org/), and standard library's `logging` support out-of-the-box.
-- Easy _global_ deactivation for testing.
+- Dedicated support for **testing** that allows to _globally_ deactivate retries, or to limit the number of retries and to remove backoffs.
 
 For example:
 

docs/api.md

@@ -51,6 +51,16 @@
 .. autofunction:: is_active
 ```
 
+## Testing
+
+::: {seealso}
+{doc}`testing`
+:::
+
+```{eval-rst}
+.. autofunction:: set_testing
+.. autofunction:: is_testing
+```
 
 ## Instrumentation
 

docs/index.md

@@ -17,6 +17,7 @@ Check out {doc}`motivation` if you need more convincing, or jump into our {doc}`
 
 motivation
 tutorial
+testing
 instrumentation
 api
 ```

docs/testing.md

@@ -0,0 +1,41 @@
+# Testing
+
+Testing code that has retry logic can be tricky, therefore *stamina* helps you with dedicated testing helpers.
+
+The easiest way is to disable retries globally using {func}`stamina.set_active`:
+
+```python
+import pytest
+import stamina
+
+@pytest.fixture(autouse=True, scope="session")
+def deactivate_retries():
+    stamina.set_active(False)
+```
+
+This is a great approach when you're only using the decorator-based API.
+
+---
+
+When you need more control, you're going to use the iterator-based APIs around {func}`stamina.retry_context`.
+
+Here, it can make sense to actually trigger retries and test what happens.
+However, you don't want the backoff and you probably don't want to go to the full number of attempts.
+
+For this use-case *stamina* comes with a dedicated testing mode that disables backoff and caps retries -- by default to a single attempt: {func}`stamina.set_testing`.
+
+Therefore this script will only print "trying 1" and "trying 2" very quickly and raise a `ValueError`:
+
+```python
+import stamina
+
+stamina.set_testing(True)  # no backoff, 1 attempt
+stamina.set_testing(True, attempts=2)  # no backoff, 2 attempts
+
+for attempt in stamina.retry_context(on=ValueError):
+    with attempt:
+        print("trying", attempt.num)
+        raise ValueError("nope")
+
+stamina.set_testing(False)
+```

src/stamina/__init__.py

@@ -3,7 +3,7 @@
 # SPDX-License-Identifier: MIT
 
 from . import instrumentation
-from ._config import is_active, set_active
+from ._config import is_active, is_testing, set_active, set_testing
 from ._core import (
     AsyncRetryingCaller,
     Attempt,
@@ -22,10 +22,12 @@
     "BoundRetryingCaller",
     "instrumentation",
     "is_active",
+    "is_testing",
     "retry_context",
     "retry",
     "RetryingCaller",
     "set_active",
+    "set_testing",
 ]
 
 

src/stamina/_config.py

@@ -12,17 +12,39 @@
 from .typing import RetryHook
 
 
+class _Testing:
+    """
+    Test mode specification.
+
+    Strictly private.
+    """
+
+    __slots__ = ("attempts",)
+
+    attempts: int
+
+    def __init__(self, attempts: int) -> None:
+        self.attempts = attempts
+
+
 class _Config:
     """
     Global stamina configuration.
 
     Strictly private.
     """
 
-    __slots__ = ("lock", "_is_active", "_on_retry", "_get_on_retry")
+    __slots__ = (
+        "lock",
+        "_is_active",
+        "_on_retry",
+        "_get_on_retry",
+        "_testing",
+    )
 
     lock: Lock
     _is_active: bool
+    _testing: _Testing | None
     _on_retry: (
         tuple[RetryHook, ...] | tuple[RetryHook | RetryHookFactory, ...] | None
     )
@@ -31,6 +53,7 @@ class _Config:
     def __init__(self, lock: Lock) -> None:
         self.lock = lock
         self._is_active = True
+        self._testing = None
 
         # Prepare delayed initialization.
         self._on_retry = None
@@ -45,6 +68,15 @@ def is_active(self, value: bool) -> None:
         with self.lock:
             self._is_active = value
 
+    @property
+    def testing(self) -> _Testing | None:
+        return self._testing
+
+    @testing.setter
+    def testing(self, value: _Testing | None) -> None:
+        with self.lock:
+            self._testing = value
+
     @property
     def on_retry(self) -> tuple[RetryHook, ...]:
         return self._get_on_retry()
@@ -94,3 +126,26 @@ def set_active(active: bool) -> None:
     Is idempotent and can be called repeatedly with the same value.
     """
     CONFIG.is_active = bool(active)
+
+
+def is_testing() -> bool:
+    """
+    Check whether test mode is enabled.
+
+    .. versionadded:: 24.3.0
+    """
+    return CONFIG.testing is not None
+
+
+def set_testing(testing: bool, *, attempts: int = 1) -> None:
+    """
+    Activate or deactivate test mode.
+
+    In testing mode, backoffs are disabled, and attempts are capped to
+    *attempts*.
+
+    Is idempotent and can be called repeatedly with the same values.
+
+    .. versionadded:: 24.3.0
+    """
+    CONFIG.testing = _Testing(attempts) if testing else None

src/stamina/_core.py

@@ -25,7 +25,7 @@
 
 import tenacity as _t
 
-from ._config import CONFIG, _Config
+from ._config import CONFIG, _Config, _Testing
 from .instrumentation._data import RetryDetails, guess_name
 
 
@@ -456,6 +456,19 @@ def with_name(
         """
         return replace(self, _name=name, _args=args, _kw=kw)
 
+    def _apply_maybe_test_mode_to_tenacity_kw(
+        self, testing: _Testing | None
+    ) -> dict[str, object]:
+        if testing is None:
+            return self._t_kw
+
+        t_kw = self._t_kw.copy()
+
+        del t_kw["wait"]
+        t_kw["stop"] = _t.stop_after_attempt(testing.attempts)
+
+        return t_kw
+
     def __iter__(self) -> Iterator[Attempt]:
         if not CONFIG.is_active:
             for r in _t.Retrying(reraise=True, stop=_STOP_NO_RETRY):
@@ -467,7 +480,7 @@ def __iter__(self) -> Iterator[Attempt]:
             before_sleep=_make_before_sleep(
                 self._name, CONFIG, self._args, self._kw
             ),
-            **self._t_kw,
+            **self._apply_maybe_test_mode_to_tenacity_kw(CONFIG.testing),
         ):
             yield Attempt(r)
 
@@ -478,7 +491,7 @@ def __aiter__(self) -> AsyncIterator[Attempt]:
                 before_sleep=_make_before_sleep(
                     self._name, CONFIG, self._args, self._kw
                 ),
-                **self._t_kw,
+                **self._apply_maybe_test_mode_to_tenacity_kw(CONFIG.testing),
             )
 
         self._t_a_retrying = self._t_a_retrying.__aiter__()

tests/test_async.py

@@ -212,6 +212,28 @@ async def test_next_wait():
             assert pytest.approx(0.001) == attempt.next_wait
 
 
+async def test_testing_mode():
+    """
+    Testing mode can be set and reset.
+    """
+    stamina.set_testing(True, attempts=3)
+
+    assert stamina.is_testing()
+
+    with pytest.raises(ValueError):  # noqa: PT012
+        async for attempt in stamina.retry_context(on=ValueError):
+            assert 0.0 == attempt.next_wait
+
+            with attempt:
+                raise ValueError
+
+    assert 3 == attempt.num
+
+    stamina.set_testing(False)
+
+    assert not stamina.is_testing()
+
+
 async def test_retry_blocks_can_be_disabled():
     """
     Async context retries respect the config.

tests/test_sync.py

@@ -184,6 +184,28 @@ def test_next_wait():
             assert pytest.approx(0.001) == attempt.next_wait
 
 
+def test_testing_mode():
+    """
+    Testing mode can be set and reset.
+    """
+    stamina.set_testing(True, attempts=3)
+
+    assert stamina.is_testing()
+
+    with pytest.raises(ValueError):  # noqa: PT012
+        for attempt in stamina.retry_context(on=ValueError):
+            assert 0.0 == attempt.next_wait
+
+            with attempt:
+                raise ValueError
+
+    assert 3 == attempt.num
+
+    stamina.set_testing(False)
+
+    assert not stamina.is_testing()
+
+
 class TestMakeStop:
     def test_never(self):
         """