Skip to content

Controlling Time in asyncio Tests

A circuit breaker opens after three failures and allows a trial request after 30 seconds. The test for the half-open transition either sleeps for 30 seconds — and so runs only in the nightly job — or shortens the window to 50 milliseconds, which tests a configuration nobody deploys and fails on a loaded CI runner when 50 ms turns into 70. The same dilemma appears for every time-shaped behaviour: exponential backoff that reaches minutes, idle-connection reaping, token-bucket refill, heartbeat intervals, request deadlines. The way out is to stop letting the test depend on the wall clock at all. asyncio's scheduler only ever asks the loop what time it is, so a loop that answers with a virtual clock — and advances that clock instantly whenever nothing is ready to run — executes real production timings, in exact order, in microseconds. This guide builds that loop, uses it to assert exact backoff schedules and long breaker windows, and marks the boundary where virtual time stops being trustworthy.

Prerequisites

Waiting for the next timer 2 columns contrasting production loop, virtual-time loop. Waiting for the next timer production loop time() is time.monotonic() next timer due in 30 s selector blocks for 30 s timer runs at +30 s real virtual-time loop time() is a counter next timer due in 30 s nothing ready: clock += 30 timer runs at once The scheduler, the timer heap and task ordering are identical on both.

1. Make the code read the loop's clock

Virtual time can only control what reads the loop's clock. asyncio.sleep(), asyncio.timeout(), call_later() and loop.time() all do. Direct calls to time.time(), time.monotonic() or datetime.now() bypass it, so a breaker that stores time.monotonic() when it opens cannot be tested this way. Route time through the loop, or inject a clock.

import asyncio


class CircuitBreaker:
    def __init__(self, threshold: int = 3, reset_after: float = 30.0) -> None:
        self.threshold = threshold
        self.reset_after = reset_after
        self.failures = 0
        self.opened_at: float | None = None

    @staticmethod
    def _now() -> float:
        return asyncio.get_running_loop().time()      # not time.monotonic()

    def allow(self) -> bool:
        if self.opened_at is None:
            return True
        return self._now() - self.opened_at >= self.reset_after   # half-open after the window

    def record_failure(self) -> None:
        self.failures += 1
        if self.failures >= self.threshold:
            self.opened_at = self._now()

    def record_success(self) -> None:
        self.failures, self.opened_at = 0, None

loop.time() is monotonic in production — the default loop uses time.monotonic() — so this change costs nothing outside tests. For code you cannot change, inject a clock: Callable[[], float] parameter defaulting to time.monotonic and pass the loop's clock in tests.

Verify: search the module for time.time(, time.monotonic( and datetime.now(; every remaining use is either wall-clock display (timestamps in logs) or a documented injection point.

2. Build a virtual-time event loop

The loop needs two changes. time() returns a counter instead of the system clock. And when the scheduler asks the selector to wait timeout seconds for I/O because the next timer is that far away, the selector checks for I/O without waiting and, if nothing is ready, advances the counter by timeout. The scheduler then finds the timer due and runs it immediately.

# virtual_clock.py
import asyncio
import selectors


class _JumpingSelector(selectors.DefaultSelector):
    """When nothing is ready, jump the loop's clock to the next timer instead of sleeping."""

    def __init__(self) -> None:
        super().__init__()
        self.loop: "VirtualClockLoop | None" = None

    def select(self, timeout=None):
        if timeout is None or timeout <= 0:
            return super().select(timeout)            # no timers pending, or work is ready
        ready = super().select(0)                     # real I/O still gets served
        if not ready:
            self.loop.advance(timeout)                # time passes instantly
        return ready


class VirtualClockLoop(asyncio.SelectorEventLoop):
    def __init__(self) -> None:
        selector = _JumpingSelector()
        super().__init__(selector)
        selector.loop = self
        self._now = 0.0

    def time(self) -> float:
        return self._now

    def advance(self, seconds: float) -> None:
        self._now += seconds


def run_virtual(coro):
    """Run a coroutine to completion on a fresh virtual-time loop."""
    with asyncio.Runner(loop_factory=VirtualClockLoop) as runner:
        return runner.run(coro)

This works because the scheduler already computes the exact distance to the next timer before calling the selector; the loop simply stops pretending that distance takes real time. Everything that is not a timer — ready callbacks, completed futures, task steps — runs exactly as it would on the production loop, in the same order.

Verify: run_virtual(asyncio.sleep(3600)) returns in well under a millisecond, and a coroutine that reads asyncio.get_running_loop().time() after await asyncio.sleep(3600) sees 3600.0.

3. Assert exact backoff schedules

With real time, a backoff test can only assert "it took at least roughly this long". With virtual time, it can record the loop clock at each attempt and assert the schedule exactly — which catches off-by-one exponents, missing caps and jitter applied in the wrong place.

import asyncio
import random

from virtual_clock import run_virtual


async def retry(fn, attempts: int = 5, base: float = 1.0, cap: float = 60.0,
                rng: random.Random | None = None):
    for attempt in range(attempts):
        try:
            return await fn()
        except ConnectionError:
            if attempt == attempts - 1:
                raise
            delay = min(cap, base * 2 ** attempt)
            if rng is not None:
                delay = rng.uniform(0, delay)          # full jitter
            await asyncio.sleep(delay)


def test_exponential_schedule_is_exact() -> None:
    seen: list[float] = []

    async def flaky():
        seen.append(asyncio.get_running_loop().time())
        if len(seen) < 5:
            raise ConnectionError()
        return "ok"

    assert run_virtual(retry(flaky)) == "ok"
    assert seen == [0.0, 1.0, 3.0, 7.0, 15.0]


def test_jitter_stays_within_bounds() -> None:
    seen: list[float] = []

    async def always_fails():
        seen.append(asyncio.get_running_loop().time())
        raise ConnectionError()

    async def scenario():
        try:
            await retry(always_fails, attempts=8, cap=10.0, rng=random.Random(42))
        except ConnectionError:
            pass

    run_virtual(scenario())
    gaps = [b - a for a, b in zip(seen, seen[1:])]
    assert all(0 <= gap <= min(10.0, 2 ** i) for i, gap in enumerate(gaps))

Seeding the jitter source with random.Random(42) keeps the test deterministic while still exercising the jitter code path. The first test would take 15 real seconds; the second, with an 8-attempt schedule capped at 10 seconds, would take close to a minute.

Verify: change the exponent to 2 ** (attempt + 1) and the exact-schedule assertion fails with the precise wrong list [0.0, 2.0, 6.0, 14.0, 30.0], showing where the bug is rather than just that "it was too slow".

Attempt times on the virtual clock 5 bars comparing attempt 1 with the others. Attempt times on the virtual clock attempt 1 0.0 s attempt 2 1.0 s attempt 3 3.0 s attempt 4 7.0 s attempt 5 15.0 s Values from the step 3 test: delay = base × 2^attempt, recorded with loop.time(). Fifteen virtual seconds, asserted exactly, in under a millisecond of real time.

4. Test timeouts and long windows at production values

Timeouts and breaker windows can now be tested with the values that ship. asyncio.timeout() schedules its deadline on the loop's clock, so a 60-second timeout fires at virtual 60.0; a 30-second breaker window elapses after virtual sleeps that total 30 seconds.

import asyncio

import pytest

from virtual_clock import run_virtual


def test_request_timeout_fires_at_sixty_seconds() -> None:
    async def scenario() -> float:
        loop = asyncio.get_running_loop()
        with pytest.raises(TimeoutError):
            async with asyncio.timeout(60):
                await asyncio.Event().wait()           # a dependency that never answers
        return loop.time()

    assert run_virtual(scenario()) == 60.0


def test_breaker_half_opens_after_thirty_seconds() -> None:
    async def scenario() -> None:
        breaker = CircuitBreaker(threshold=3, reset_after=30.0)
        for _ in range(3):
            breaker.record_failure()
        assert not breaker.allow()                     # open
        await asyncio.sleep(29.5)
        assert not breaker.allow()                     # still open just before the window
        await asyncio.sleep(0.5)
        assert breaker.allow()                         # half-open exactly at 30 s

    run_virtual(scenario())

The boundary assertions — still open at 29.5 seconds, allowed at 30 — are the ones that catch > versus >= mistakes and window arithmetic bugs, and they are impossible to write reliably against a real clock. Choose boundary values that are exactly representable in binary floating point, such as halves and quarters, so sums of virtual sleeps land exactly on the threshold.

Verify: both tests finish in milliseconds. Change >= to > in allow() and the half-open assertion fails at exactly 30.0.

5. Know where virtual time stops being real

Virtual time is exact for code whose waiting is all timers. It becomes misleading as soon as a coroutine waits on something that takes real time: a thread, a subprocess, a socket to another process. When the loop has a timer pending and the real work has not finished yet, the selector sees no ready I/O and jumps the clock — so timeouts fire immediately, before the real work had any chance.

import asyncio
import time

from virtual_clock import run_virtual


def test_real_thread_versus_virtual_timeout() -> None:
    async def scenario() -> str:
        try:
            async with asyncio.timeout(5):                        # generous in real terms
                return await asyncio.to_thread(time.sleep, 0.05)  # 50 ms of real work
        except TimeoutError:
            return "timed out"

    assert run_virtual(scenario()) == "timed out"                 # the clock jumped past 5 s

The same effect makes the runner's own shutdown noisy: asyncio.Runner waits for the default executor with a timeout, and in virtual time that timeout expires at once, producing a "did not finish joining its threads" warning if a thread was used. Keep virtual time for pure async logic — schedulers, retries, breakers, rate limiters, heartbeat bookkeeping — and test code that crosses into threads, processes or sockets with real time and short, explicit values. Integration behaviour such as handling SIGTERM in asyncio services belongs in that second group.

Verify: the test above passes, documenting the limitation where the next engineer will find it; the same scenario on a normal loop returns None from time.sleep without timing out.

Is virtual time valid for this test? A decision on What does the code wait on with 3 outcomes. Is virtual time valid for this test? What does the code wait on? timers only virtual time exact and instant threads, processes, sockets real time short explicit values calls time.monotonic() change the code first read loop.time() A timeout that fires immediately is the sign a test is in the wrong branch.

Verification

Time-controlled tests are trustworthy when:

  • Production values are used: timeouts, windows and backoff bases in tests match configuration defaults rather than shortened copies.
  • Schedules are asserted exactly: tests record loop.time() at each event and compare against the full expected sequence.
  • Boundaries are covered: each time threshold has an assertion just before and exactly at the boundary.
  • The suite stays fast: time-heavy tests finish in milliseconds; per-test duration reports show no second-long unit tests.
  • Scope is respected: no virtual-time test waits on threads, subprocesses or external sockets.

Pitfalls & edge cases

  • Wall-clock reads in the code under test. A single time.monotonic() in a limiter makes it immune to virtual time, and the test silently measures nothing. Grep for clock calls, or inject the clock.
  • Floating-point accumulation. Summing many 0.1 sleeps does not land exactly on a round threshold. Use exactly representable values at boundaries, or compare with a small tolerance.
  • Busy loops never advance time. Code that polls with await asyncio.sleep(0) in a tight loop always has ready work, so the clock never jumps and the test spins. That is usually a production bug too; replace polling with an event or a real interval.
  • Mixing with pytest-asyncio's loop. A test marked @pytest.mark.asyncio already runs on the plugin's loop, and run_virtual cannot be called from inside a running loop. Write virtual-time tests as plain synchronous tests that call run_virtual.
  • Windows event loops. The default Windows loop is the proactor loop, which this selector-based approach does not cover. Run virtual-time tests on the selector loop, which is available on every platform.

Frequently Asked Questions

How do I test asyncio.sleep without actually waiting?

Run the coroutine on an event loop whose time method returns a virtual counter and whose selector advances that counter to the next scheduled timer whenever no I/O is ready. asyncio.sleep, asyncio.timeout and call_later all use the loop's clock, so long delays complete instantly and in the correct order.

Can I test a 30-second circuit breaker window in a unit test?

Yes, if the breaker reads time from the event loop with loop.time() instead of time.monotonic(). On a virtual-time loop the test can sleep 29.5 virtual seconds, assert the breaker is still open, sleep another half second, and assert it allows a trial request, all in a few milliseconds.

Why does my timeout fire immediately on a virtual-time loop?

The coroutine is waiting on something that takes real time, such as a thread, subprocess or external socket. While that real work is pending, the loop sees no ready I/O and jumps the virtual clock to the next timer, which is the timeout. Use virtual time only for code whose waits are timers, and real time for integration tests.

Should I patch time.monotonic in async tests instead?

Patching time functions globally is fragile, because the event loop, libraries and the test runner also call them, and the patch does not make sleeps return early. Reading time through loop.time() or an injected clock and running on a virtual-time loop controls both the clock and the waiting consistently.