Function wait_until

tests/utils/wait.py:13–26

(fn, sleep_s=0.2, on_timeout=None, timeout_s=DEFAULT_TIMEOUT)

def wait_until(fn, sleep_s=0.2, on_timeout=None, timeout_s=DEFAULT_TIMEOUT):
    end = time.time() + timeout_s

    while time.time() < end:
        value = fn()
        if value is not None and value is not False:
            return value
        time.sleep(sleep_s)
    extra_info = None
    if on_timeout is not None:
        extra_info = on_timeout()
    if extra_info is None:
        extra_info = ""
    raise TimeoutException(f"Wait timeouted after {timeout_s} seconds\n{extra_info}")


TERMINAL_STATES = ["failed", "cancelled", "finished"]