jenkinsapi.utils.retry: basic abstract retry handler

Author: cdleonard

jenkinsapi/utils/retry.py

@@ -0,0 +1,59 @@
+from abc import ABC, abstractmethod
+import time
+from typing import Optional
+
+
+class Retry(ABC):
+    """
+    Base class for limited retry checks
+
+    Usage::
+
+        retry.begin()
+        while True:
+            result = try_something()
+            if result:
+                return result
+            retry.check()
+    """
+
+    def begin(self) -> None:
+        pass
+
+    @abstractmethod
+    def check(self) -> None:
+        """Sleep or raise `TimeoutError`"""
+
+
+class SimpleRetry(Retry):
+    """Basic implementation of RetryCheck with fixed sleep and timeout."""
+
+    def get_current_time(self) -> float:
+        return time.monotonic()
+
+    start_time: Optional[float] = None
+
+    def __init__(self, sleep_period: float = 1, timeout: float = 5):
+        self.sleep_period = sleep_period
+        self.timeout = timeout
+
+    def begin(self) -> None:
+        self.start_time = self.get_current_time()
+
+    def check(self) -> None:
+        start_time = self.start_time
+        if start_time is None:
+            raise RuntimeError(
+                "Retry has not been started. Call begin() first."
+            )
+        curr_time = self.get_current_time()
+        if curr_time - start_time > self.timeout:
+            raise TimeoutError("Retry timed out")
+        time.sleep(self.sleep_period)
+
+    def __repr__(self) -> str:
+        return (
+            f"{self.__class__.__name__}("
+            f"sleep_period={self.sleep_period}, "
+            f"timeout={self.timeout})"
+        )

jenkinsapi_tests/unittests/test_retry.py

@@ -0,0 +1,61 @@
+from contextlib import ExitStack
+from unittest import mock
+
+import pytest
+
+from jenkinsapi.utils.retry import Retry, SimpleRetry
+
+
+def validate_retry_check(
+    retry: Retry,
+    pass_index: int,
+    expected_sleep_count: int,
+    expected_pass: bool = True,
+) -> None:
+    """Check if the retry check works as expected."""
+    attempt_index = 0
+    success = False
+    with ExitStack() as exit_stack:
+        exit_stack.enter_context(
+            mock.patch("time.monotonic", side_effect=range(100, 1000))
+        )
+        mock_sleep = exit_stack.enter_context(mock.patch("time.sleep"))
+        if not expected_pass:
+            exit_stack.enter_context(pytest.raises(TimeoutError))
+        retry.begin()
+        while True:
+            attempt_index += 1
+            if attempt_index >= pass_index:
+                success = True
+                break
+            retry.check()
+    if expected_pass:
+        assert success
+    else:
+        assert success is False
+    assert mock_sleep.call_count == expected_sleep_count
+
+
+def test_simple_retry_check():
+    retry = SimpleRetry(sleep_period=1, timeout=5)
+    validate_retry_check(
+        retry,
+        pass_index=3,
+        expected_sleep_count=2,
+        expected_pass=True,
+    )
+
+
+def test_simple_retry_check_fail():
+    retry = SimpleRetry(sleep_period=1, timeout=5)
+    validate_retry_check(
+        retry,
+        pass_index=10,
+        expected_sleep_count=5,
+        expected_pass=False,
+    )
+
+
+def test_repr():
+    retry_check = SimpleRetry(sleep_period=1, timeout=5)
+    assert repr(retry_check) == "SimpleRetry(sleep_period=1, timeout=5)"