feat: adds waiting functionality

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "stackit-core"
-version = "0.0.1a"
+version = "0.0.1a1"
 authors = [
   "STACKIT Developer Tools <[email protected]>",
 ]

src/stackit/core/configuration.py

@@ -2,22 +2,22 @@
 
 
 class EnvironmentVariables:
-    SERVICE_ACCOUNT_EMAIL_ENV = "STACKIT_SERVICE_ACCOUNT_EMAIL"
-    SERVICE_ACCOUNT_TOKEN_ENV = "STACKIT_SERVICE_ACCOUNT_TOKEN"  # noqa: S105 false positive
-    SERVICE_ACCOUNT_KEY_PATH_ENV = "STACKIT_SERVICE_ACCOUNT_KEY_PATH"
-    PRIVATE_KEY_PATH_ENV = "STACKIT_PRIVATE_KEY_PATH"
-    TOKEN_BASEURL_ENV = "STACKIT_TOKEN_BASEURL"  # noqa: S105 false positive
-    CREDENTIALS_PATH_ENV = "STACKIT_CREDENTIALS_PATH"
-    REGION_ENV = "STACKIT_REGION"
+    _SERVICE_ACCOUNT_EMAIL_ENV = "STACKIT_SERVICE_ACCOUNT_EMAIL"
+    _SERVICE_ACCOUNT_TOKEN_ENV = "STACKIT_SERVICE_ACCOUNT_TOKEN"  # noqa: S105 false positive
+    _SERVICE_ACCOUNT_KEY_PATH_ENV = "STACKIT_SERVICE_ACCOUNT_KEY_PATH"
+    _PRIVATE_KEY_PATH_ENV = "STACKIT_PRIVATE_KEY_PATH"
+    _TOKEN_BASEURL_ENV = "STACKIT_TOKEN_BASEURL"  # noqa: S105 false positive
+    _CREDENTIALS_PATH_ENV = "STACKIT_CREDENTIALS_PATH"
+    _REGION_ENV = "STACKIT_REGION"
 
     def __init__(self):
-        self.account_email = os.environ.get(self.SERVICE_ACCOUNT_EMAIL_ENV)
-        self.service_account_token = os.environ.get(self.SERVICE_ACCOUNT_TOKEN_ENV)
-        self.account_key_path = os.environ.get(self.SERVICE_ACCOUNT_KEY_PATH_ENV)
-        self.private_key_path = os.environ.get(self.PRIVATE_KEY_PATH_ENV)
-        self.token_baseurl = os.environ.get(self.TOKEN_BASEURL_ENV)
-        self.credentials_path = os.environ.get(self.CREDENTIALS_PATH_ENV)
-        self.region = os.environ.get(self.REGION_ENV)
+        self.account_email = os.environ.get(self._SERVICE_ACCOUNT_EMAIL_ENV)
+        self.service_account_token = os.environ.get(self._SERVICE_ACCOUNT_TOKEN_ENV)
+        self.account_key_path = os.environ.get(self._SERVICE_ACCOUNT_KEY_PATH_ENV)
+        self.private_key_path = os.environ.get(self._PRIVATE_KEY_PATH_ENV)
+        self.token_baseurl = os.environ.get(self._TOKEN_BASEURL_ENV)
+        self.credentials_path = os.environ.get(self._CREDENTIALS_PATH_ENV)
+        self.region = os.environ.get(self._REGION_ENV)
 
 
 class Configuration:

src/stackit/core/wait.py

@@ -0,0 +1,61 @@
+from dataclasses import dataclass, field
+import signal
+import time
+from http import HTTPStatus
+from typing import Any, Callable, List, Tuple, Union
+
+
+@dataclass
+class WaitConfig:
+    sleep_before_wait: int = 0
+    throttle: int = 5
+    timeout: int = 1800
+    temp_error_retry_limit: int = 5
+    retry_http_error_status_codes: List[int] = field(
+        default_factory=lambda: [HTTPStatus.BAD_GATEWAY, HTTPStatus.GATEWAY_TIMEOUT]
+    )
+
+
+class Wait:
+
+    def __init__(
+        self,
+        check_function: Callable[[None], Tuple[bool, Union[Exception, None], Union[int, None], Any]],
+        config: Union[WaitConfig, None] = None,
+    ) -> None:
+        self._config = config if config else WaitConfig()
+        if self._config.throttle == 0:
+            raise ValueError("throttle can't be 0")
+        self._check_function = check_function
+
+    @staticmethod
+    def _timeout_handler(signum, frame):
+        raise TimeoutError("Wait has timed out")
+
+    def wait(self) -> Any:
+        time.sleep(self._config.sleep_before_wait)
+
+        retry_temp_error_counter = 0
+
+        signal.signal(signal.SIGALRM, Wait._timeout_handler)
+        signal.alarm(self._config.timeout)
+
+        while True:
+
+            done, error, code, result = self._check_function()
+            if error:
+                retry_temp_error_counter = self._handle_error(retry_temp_error_counter, error, code)
+
+            if done:
+                return result
+            time.sleep(self._config.throttle)
+
+    def _handle_error(self, retry_temp_error_counter: int, error, code: int):
+
+        if code in self._config.retry_http_error_status_codes:
+            retry_temp_error_counter += 1
+            if retry_temp_error_counter == self._config.temp_error_retry_limit:
+                raise error
+            return retry_temp_error_counter
+        else:
+            raise error

tests/core/test_auth.py

@@ -1,9 +1,8 @@
+import json
 from pathlib import Path, PurePath
+from unittest.mock import Mock, mock_open, patch
 
 import pytest
-import json
-from unittest.mock import patch, mock_open, Mock
-
 from requests.auth import HTTPBasicAuth
 
 from stackit.core.auth_methods.key_auth import KeyAuth

tests/core/test_wait.py

@@ -0,0 +1,81 @@
+import time
+from typing import Any, Callable, List, Tuple, Union
+
+import pytest
+
+from stackit.core.wait import Wait, WaitConfig
+
+
+def timeout_check_function() -> Tuple[bool, Union[Exception, None], Union[int, None], Any]:
+    time.sleep(9999)
+    return True, None, None, None
+
+
+def create_check_function(
+    error_codes: Union[List[int], None], tries_to_success: int, correct_return: str
+) -> Callable[[None], Tuple[bool, Union[Exception, None], Union[int, None], Any]]:
+    error_code_counter = 0
+    tries_to_success_counter = 0
+
+    def check_function() -> Tuple[bool, Union[Exception, None], Union[int, None], Any]:
+        nonlocal error_code_counter
+        nonlocal tries_to_success_counter
+
+        if error_codes and error_code_counter < len(error_codes):
+            code = error_codes[error_code_counter]
+            error_code_counter += 1
+            return False, Exception("Some exception"), code, None
+        elif tries_to_success_counter < tries_to_success:
+            tries_to_success_counter += 1
+            return False, None, 200, None
+        else:
+            return True, None, 200, correct_return
+
+    return check_function
+
+
+class TestWait:
+    def test_timeout_throws_timeouterror(self):
+        wait = Wait(timeout_check_function, WaitConfig(timeout=1))
+
+        with pytest.raises(TimeoutError, match="Wait has timed out"):
+            wait.wait()
+
+    def test_throttle_0_throws_error(self):
+        with pytest.raises(ValueError, match="throttle can't be 0"):
+            _ = Wait(lambda: (True, None, None, None), WaitConfig(throttle=0))
+
+    @pytest.mark.parametrize(
+        "check_function",
+        [
+            create_check_function([400], 3, "Shouldn't be returned"),
+        ],
+    )
+    def test_throws_for_no_retry_status_code(self, check_function):
+        wait = Wait(check_function)
+        with pytest.raises(Exception, match="Some exception"):
+            wait.wait()
+
+    @pytest.mark.parametrize(
+        "correct_return,error_retry_limit,check_function",
+        [
+            (
+                "This was a triumph.",
+                0,
+                create_check_function(None, 0, "This was a triumph."),
+            ),
+            (
+                "I'm making a note here: HUGE SUCCESS.",
+                3,
+                create_check_function([502, 504], 3, "I'm making a note here: HUGE SUCCESS."),
+            ),
+        ],
+    )
+    def test_return_is_correct(
+        self,
+        correct_return: str,
+        error_retry_limit: int,
+        check_function: Callable[[None], Tuple[bool, Union[Exception, None], Union[int, None], Any]],
+    ):
+        wait = Wait(check_function, WaitConfig(error_retry_limit))
+        assert wait.wait() == correct_return