Check that the coroutines result meets a condition within the timeout
(
self,
condition: Callable[[_R], bool],
timeout: float = REACTPY_TESTING_DEFAULT_TIMEOUT.current,
delay: float = _DEFAULT_POLL_DELAY,
description: str = "condition to be true",
)
| 52 | self._kwargs = kwargs |
| 53 | |
| 54 | async def until( |
| 55 | self, |
| 56 | condition: Callable[[_R], bool], |
| 57 | timeout: float = REACTPY_TESTING_DEFAULT_TIMEOUT.current, |
| 58 | delay: float = _DEFAULT_POLL_DELAY, |
| 59 | description: str = "condition to be true", |
| 60 | ) -> None: |
| 61 | """Check that the coroutines result meets a condition within the timeout""" |
| 62 | started_at = time.time() |
| 63 | while True: |
| 64 | await asyncio.sleep(delay) |
| 65 | result = await self._func(*self._args, **self._kwargs) |
| 66 | if condition(result): |
| 67 | break |
| 68 | elif (time.time() - started_at) > timeout: # nocov |
| 69 | msg = f"Expected {description} after {timeout} seconds - last value was {result!r}" |
| 70 | raise asyncio.TimeoutError(msg) |
| 71 | |
| 72 | async def until_is( |
| 73 | self, |
no outgoing calls