Introduce timeout helper

This should help us stop copy-pasting the calculations to let us timeout functions

Related to saltstack#27089 Saltcloud virtualbox provider

salt/utils/timeout.py

@@ -0,0 +1,45 @@
+import time
+
+BLUR_FACTOR = 0.95
+"""
+To give us some leeway when making time-calculations
+"""
+
+
+def wait_for(func, timeout=10, step=1, default=None, func_args=(), func_kwargs=None):
+    """
+    Call `func` at regular intervals and Waits until the given function returns a truthy value
+    within the given timeout and returns that value.
+
+    @param func:
+    @type func: function
+    @param timeout:
+    @type timeout: int | float
+    @param step: Interval at which we should check for the value
+    @type step: int | float
+    @param default: Value that should be returned should `func` not return a truthy value
+    @type default:
+    @param func_args: *args for `func`
+    @type func_args: list | tuple
+    @param func_kwargs: **kwargs for `func`
+    @type func_kwargs: dict
+    @return: `default` or result of `func`
+    """
+    if func_kwargs is None:
+        func_kwargs = dict()
+    max_time = time.time() + timeout
+    # Time moves forward so we might not reenter the loop if we step too long
+    step = min(step or 1, timeout) * BLUR_FACTOR
+
+    ret = default
+    while time.time() <= max_time:
+        call_ret = func(*func_args, **func_kwargs)
+        if call_ret:
+            ret = call_ret
+            break
+        else:
+            time.sleep(step)
+
+            # Don't allow cases of over-stepping the timeout
+            step = min(step, max_time - time.time()) * BLUR_FACTOR
+    return ret

tests/utils/test_timeout.py

@@ -0,0 +1,93 @@
+import unittest
+import logging
+
+import time
+
+from utils.timeout import wait_for
+
+log = logging.getLogger(__name__)
+
+
+def return_something_after(seconds, something=True):
+    start = time.time()
+    end = start + seconds
+    log.debug("Will return %s at %s", something, end)
+
+    def actual():
+        t = time.time()
+        condition = t >= end
+        log.debug("Return something at %s ? %s", t, condition)
+        if condition:
+            return something
+        else:
+            return False
+
+    return actual
+
+
+def return_args_after(seconds):
+    start = time.time()
+    end = start + seconds
+
+    def actual(*args):
+        if time.time() >= end:
+            return args
+        else:
+            return False
+
+    return actual
+
+
+def return_kwargs_after(seconds):
+    start = time.time()
+    end = start + seconds
+
+    def actual(**kwargs):
+        if time.time() >= end:
+            return kwargs
+        else:
+            return False
+
+    return actual
+
+
+class WaitForTests(unittest.TestCase):
+    def setUp(self):
+        self.true_after_1s = return_something_after(1)
+        self.self_after_1s = return_something_after(1, something=self)
+
+    def test_wait_for_true(self):
+        ret = wait_for(self.true_after_1s, timeout=2, step=0.5)
+        self.assertTrue(ret)
+
+    def test_wait_for_self(self):
+        ret = wait_for(self.self_after_1s, timeout=2, step=0.5)
+        self.assertEqual(ret, self)
+
+    def test_wait_for_too_long(self):
+        ret = wait_for(self.true_after_1s, timeout=0.5, step=0.1, default=False)
+        self.assertFalse(ret)
+
+    def test_wait_for_with_big_step(self):
+        ret = wait_for(self.true_after_1s, timeout=1.5, step=2)
+        self.assertTrue(ret)
+
+    def test_wait_for_custom_args(self):
+        args_after_1s = return_args_after(1)
+        args = ("one", "two")
+        ret = wait_for(args_after_1s, timeout=2, step=0.5, func_args=args)
+        self.assertEqual(ret, args)
+
+    def test_wait_for_custom_kwargs(self):
+        kwargs_after_1s = return_kwargs_after(1)
+        kwargs = {"one": 1, "two": 2}
+        ret = wait_for(kwargs_after_1s, timeout=2, step=0.5, func_kwargs=kwargs)
+        self.assertEqual(ret, kwargs)
+
+    def test_return_false(self):
+        ret = self.true_after_1s()
+        self.assertFalse(ret)
+
+
+if __name__ == '__main__':
+    unittest.main()