Created
September 30, 2021 23:06
-
-
Save maratori/118d1018893b367be8d35ccda996d3af to your computer and use it in GitHub Desktop.
Helper function for python tests to get rid of time.sleep
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import time | |
from typing import Callable, Iterable, TypeVar, Optional | |
T = TypeVar('T') | |
def wait_for( | |
fn: Callable[..., T], | |
message: str = "", | |
messageSupplier: Optional[Callable[[], str]] = None, | |
fnArgs: Optional[Iterable] = None, | |
fnKwArgs: Optional[dict] = None, | |
timeout: float = 4, | |
interval: float = 0.1, | |
) -> T: | |
""" | |
Wait for successful execution of function. | |
Successful means: | |
1. No exception was rised | |
2. Returned value is converted to True. | |
:param fn: Function to wait for | |
:param message: TimeoutError message (is used if messageSupplier is None) | |
:param messageSupplier: Function to create TimeoutError message | |
:param fnArgs: Positional arguments to pass into fn | |
:param fnKwArgs: Keyword arguments to pass into fn | |
:param timeout: Waiting timeout | |
:param interval: Interval between attempts to call fn | |
:return: Value returned by fn | |
""" | |
if fnArgs is None: | |
fnArgs = tuple() | |
if fnKwArgs is None: | |
fnKwArgs = dict() | |
lastError = None | |
endTime = time.time() + timeout | |
for i in range(int(timeout / interval) + 1): | |
try: | |
result = fn(*fnArgs, **fnKwArgs) | |
except Exception as e: | |
lastError = e | |
else: | |
lastError = None | |
if result: | |
return result | |
if time.time() > endTime: | |
break | |
time.sleep(interval) | |
msg = messageSupplier() if messageSupplier else message | |
raise TimeoutError(msg=msg) from lastError |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment