Class poll

src/py/reactpy/reactpy/testing/common.py:33–98

Wait until the result of an sync or async function meets some condition

class poll(Generic[_R]):  # noqa: N801
    """Wait until the result of an sync or async function meets some condition"""

    def __init__(
        self,
        function: Callable[_P, Awaitable[_R] | _R],
        *args: _P.args,
        **kwargs: _P.kwargs,
    ) -> None:
        coro: Callable[_P, Awaitable[_R]]
        if not inspect.iscoroutinefunction(function):

            async def coro(*args: _P.args, **kwargs: _P.kwargs) -> _R:
                return cast(_R, function(*args, **kwargs))

        else:
            coro = cast(Callable[_P, Awaitable[_R]], function)
        self._func = coro
        self._args = args
        self._kwargs = kwargs

    async def until(
        self,
        condition: Callable[[_R], bool],
        timeout: float = REACTPY_TESTING_DEFAULT_TIMEOUT.current,
        delay: float = _DEFAULT_POLL_DELAY,
        description: str = "condition to be true",
    ) -> None:
        """Check that the coroutines result meets a condition within the timeout"""
        started_at = time.time()
        while True:
            await asyncio.sleep(delay)
            result = await self._func(*self._args, **self._kwargs)
            if condition(result):
                break
            elif (time.time() - started_at) > timeout:  # nocov
                msg = f"Expected {description} after {timeout} seconds - last value was {result!r}"
                raise asyncio.TimeoutError(msg)

    async def until_is(
        self,
        right: _R,
        timeout: float = REACTPY_TESTING_DEFAULT_TIMEOUT.current,
        delay: float = _DEFAULT_POLL_DELAY,
    ) -> None:
        """Wait until the result is identical to the given value"""
        return await self.until(
            lambda left: left is right,
            timeout,
            delay,
            f"value to be identical to {right!r}",
        )

    async def until_equals(
        self,
        right: _R,
        timeout: float = REACTPY_TESTING_DEFAULT_TIMEOUT.current,
        delay: float = _DEFAULT_POLL_DELAY,