Function _wait_for

python/test_commands_and_elicitation.py:28–40

Poll predicate until True or timeout. Replaces brittle ``asyncio.sleep`` waits. Used in unit tests where we dispatch an event and need to wait for the consumer coroutine to invoke a handler and (sometimes) for the handler to issue an RPC that our mock captures. Polling at 5ms means fast

(predicate: Callable[[], bool], timeout: float = 2.0)

async def _wait_for(predicate: Callable[[], bool], timeout: float = 2.0) -> None:
    """Poll predicate until True or timeout. Replaces brittle ``asyncio.sleep`` waits.

    Used in unit tests where we dispatch an event and need to wait for the consumer
    coroutine to invoke a handler and (sometimes) for the handler to issue an RPC
    that our mock captures. Polling at 5ms means fast machines exit quickly while
    slow machines still get up to ``timeout`` seconds before the test fails.
    """
    deadline = asyncio.get_event_loop().time() + timeout
    while not predicate():
        if asyncio.get_event_loop().time() >= deadline:
            raise AssertionError(f"Condition not met within {timeout}s")
        await asyncio.sleep(0.005)


# ============================================================================