feat: add waiting helpers

Author: paulschwoerer

sipgate_e2e_test_utils/waiting.py

@@ -0,0 +1,44 @@
+import asyncio
+from datetime import timedelta
+from typing import Callable, Any
+
+
+async def wait_for_condition(
+        condition: Callable[[], bool], attempts: int = 5, interval: timedelta = timedelta(milliseconds=100)) -> None:
+    if attempts < 1:
+        raise ValueError('needs at least one attempt')
+
+    if interval < timedelta(milliseconds=0):
+        raise ValueError('interval cannot be negative')
+
+    attempt = 1
+    while condition() is False:
+        attempt += 1
+
+        if attempt > attempts:
+            raise TimeoutError('timed out waiting for condition')
+
+        await asyncio.sleep(interval.total_seconds())
+
+
+async def wait_for_assertions(
+        asserter: Callable[[], Any], attempts: int = 5, interval: timedelta = timedelta(milliseconds=100)) -> None:
+    if attempts < 1:
+        raise ValueError('needs at least one attempt')
+
+    if interval < timedelta(milliseconds=0):
+        raise ValueError('interval cannot be negative')
+
+    attempt = 1
+    while True:
+        try:
+            asserter()
+        except AssertionError as e:
+            attempt += 1
+
+            if attempt > attempts:
+                raise e
+
+            await asyncio.sleep(interval.total_seconds())
+        else:
+            return

tests/test_waiting.py

@@ -0,0 +1,38 @@
+import unittest
+
+from sipgate_e2e_test_utils.waiting import wait_for_condition, wait_for_assertions
+
+
+class WaitingTest(unittest.IsolatedAsyncioTestCase):
+    async def test_completes_when_condition_is_met(self) -> None:
+        await wait_for_condition(lambda: True)
+
+    async def test_throws_after_num_attempts_when_condition_is_not_met(self) -> None:
+        attempts_done = 0
+
+        def waiter() -> bool:
+            nonlocal attempts_done
+            attempts_done += 1
+            return False
+
+        with self.assertRaises(TimeoutError):
+            await wait_for_condition(waiter, attempts=3)
+
+        self.assertEqual(3, attempts_done)
+
+    async def test_completes_when_assertion_succeeds(self) -> None:
+        await wait_for_assertions(lambda: self.assertEqual(1, 1))
+
+    async def test_throws_after_num_attempts_when_assertion_fails(self) -> None:
+        attempts_done = 0
+
+        def asserter() -> None:
+            nonlocal attempts_done
+            attempts_done += 1
+
+            self.assertEqual(1, 2)
+
+        with self.assertRaises(AssertionError):
+            await wait_for_assertions(asserter, attempts=3)
+
+        self.assertEqual(3, attempts_done)