jenkinsapi.utils.retry: basic abstract retry handler

Author: cdleonard

jenkinsapi/utils/retry.py

@@ -0,0 +1,65 @@
+import time
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+
+
+class RetryConfig(ABC):
+    """
+    Base class for retry configuration
+
+    Usage::
+
+        retry_check = retry_config.begin()
+        while True:
+            result = try_something()
+            if result:
+                return result
+            retry_check.check_retry()
+
+    All state is stored in the `RetryCheck` instance so `RetryConfig` can be
+    used in multiple contexts simultaneously.
+    """
+
+    @abstractmethod
+    def begin(self) -> "RetryState": ...
+
+
+class RetryState(ABC):
+    """
+    Base class for limited retry checks
+    """
+
+    @abstractmethod
+    def check_retry(self) -> None:
+        """Sleep or raise `TimeoutError`"""
+        pass
+
+
+@dataclass
+class SimpleRetryConfig(RetryConfig):
+    sleep_period: float = 1
+    timeout: float = 5
+
+    def begin(self) -> "SimpleRetryState":
+        return SimpleRetryState(self)
+
+
+@dataclass
+class SimpleRetryState(RetryState):
+    """Basic implementation of RetryCheck with fixed sleep and timeout."""
+
+    config: SimpleRetryConfig
+    start_time: float
+
+    def get_current_time(self) -> float:
+        return time.monotonic()
+
+    def __init__(self, config: SimpleRetryConfig):
+        self.config = config
+        self.start_time = self.get_current_time()
+
+    def check_retry(self) -> None:
+        curr_time = self.get_current_time()
+        if curr_time - self.start_time > self.config.timeout:
+            raise TimeoutError("Retry timed out")
+        time.sleep(self.config.sleep_period)

jenkinsapi_tests/unittests/test_retry.py

@@ -0,0 +1,62 @@
+from contextlib import ExitStack
+from unittest import mock
+
+import pytest
+
+from jenkinsapi.utils.retry import RetryConfig, RetryState, SimpleRetryConfig
+
+
+def validate_retry_check(
+    retry_config: RetryConfig,
+    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_state = retry_config.begin()
+        assert isinstance(retry_state, RetryState)
+        while True:
+            attempt_index += 1
+            if attempt_index >= pass_index:
+                success = True
+                break
+            retry_state.check_retry()
+    if expected_pass:
+        assert success
+    else:
+        assert success is False
+    assert mock_sleep.call_count == expected_sleep_count
+
+
+def test_simple_retry_check():
+    retry_config = SimpleRetryConfig(sleep_period=1, timeout=5)
+    validate_retry_check(
+        retry_config,
+        pass_index=3,
+        expected_sleep_count=2,
+        expected_pass=True,
+    )
+
+
+def test_simple_retry_check_fail():
+    retry_config = SimpleRetryConfig(sleep_period=1, timeout=5)
+    validate_retry_check(
+        retry_config,
+        pass_index=10,
+        expected_sleep_count=5,
+        expected_pass=False,
+    )
+
+
+def test_repr():
+    retry_config = SimpleRetryConfig(sleep_period=1, timeout=5)
+    assert repr(retry_config) == "SimpleRetryConfig(sleep_period=1, timeout=5)"