Skip to content

Instantly share code, notes, and snippets.

@maratori
Created September 30, 2021 23:06
Show Gist options
  • Save maratori/118d1018893b367be8d35ccda996d3af to your computer and use it in GitHub Desktop.
Save maratori/118d1018893b367be8d35ccda996d3af to your computer and use it in GitHub Desktop.
Helper function for python tests to get rid of time.sleep
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