Skip to content

Testing Background and Periodic Tasks

Background and periodic tasks are awkward to test because nothing returns when they finish: a refresher loops forever, and a fire-and-forget email is still being sent when the request handler has already returned. Tests fall back on sleeping "long enough", which works on a developer laptop and fails on a busy CI runner. Measured on Python 3.14 with pytest 9, repeating each test 20 times on an idle machine and then with 8 busy processes pinned to the same core: a test that ran a 50 ms periodic loop for 0.52 s and expected 11 refreshes passed 20 of 20 idle and 5 of 20 loaded. The same loop with an injected fake sleep, checking 10 iterations exactly, passed 20 of 20 in both. For a background email sent through a thread, a test that slept 30 ms before checking passed 20 of 20 idle and 15 of 20 loaded; a test that awaited the service's own drain() passed 20 of 20 in both. When the background work used only asyncio.sleep, sleeping in the test passed even under load — timers on one loop fire in order — which is why these tests look reliable until the work touches a thread or the network. This guide makes both kinds deterministic.

Prerequisites

1. Split the loop from the work

A periodic task is a loop around one unit of work. Make the unit a separate method and test it directly; then test the loop's scheduling separately:

class Service:
    def __init__(self, interval=60.0, sleep=asyncio.sleep):
        self.interval, self.sleep = interval, sleep
        self.refreshes = 0

    async def refresh_once(self):              # the work: tested like any coroutine
        self.refreshes += 1

    async def refresh_forever(self):           # the schedule: tested with an injected sleep
        while True:
            await self.refresh_once()
            await self.sleep(self.interval)

Most assertions — what a refresh fetches, how it handles errors, what it stores — belong on refresh_once, which completes and returns. The loop needs only a few tests: that it calls the work, waits the configured interval, and survives a failing iteration.

Verify: refresh_once has its own tests, and no test of the work runs the infinite loop.

20 runs each, idle vs 8 busy processes on the same core A grid of 5 rows by 3 columns. 20 runs each, idle vs 8 busy processes on the same core test idle loaded periodic loop, real 50 ms interval, expect 11 in 0.52 s 20/20 5/20 periodic loop, injected fake sleep, expect exactly 10 20/20 20/20 background email via thread, test sleeps 30 ms 20/20 15/20 background email via thread, test awaits drain() 20/20 20/20 background work on asyncio.sleep only, test sleeps 20/20 20/20 Python 3.14, pytest 9, pytest-asyncio 1.4.

2. Test the schedule with an injected sleep

Real intervals make the test both slow and timing-dependent. Measured with a 50 ms interval and a 0.52 s observation window, expecting 11 refreshes: every idle run passed, and only 5 of 20 passed with the CPU busy, because each wake-up came late and fewer iterations fitted. Pass a fake sleep that records its argument and yields:

class FakeSleep:
    def __init__(self):
        self.calls: list[float] = []

    async def __call__(self, seconds: float):
        self.calls.append(seconds)
        await asyncio.sleep(0)                 # yield so the test can observe progress

async def test_refreshes_every_interval():
    fake = FakeSleep()
    service = Service(interval=60, sleep=fake)
    task = asyncio.create_task(service.refresh_forever())
    while len(fake.calls) < 10:
        await asyncio.sleep(0)
    task.cancel()
    assert service.refreshes == 10
    assert fake.calls == [60] * 10

Measured: 20 of 20 passed idle and loaded, and the test asserts the real 60-second interval without waiting for it. Loops that compute their next wake-up from a clock — to avoid drift — need the clock injected too; the same technique for rate limiters is in testing rate limiters deterministically.

Verify: periodic-task tests run in milliseconds and assert the configured interval exactly.

3. Give background work a way to be awaited

A request handler that starts a background task returns before the task finishes. Sleeping in the test and then checking is a guess about how long the work takes. Measured with an email sent through asyncio.to_thread taking 20 ms: a 30 ms sleep in the test passed 20 of 20 idle and 15 of 20 under load, when the thread was scheduled late. Track the tasks and expose a way to wait for them:

class Service:
    def __init__(self):
        self._background: set[asyncio.Task] = set()
        self.sent: list[str] = []

    async def handle_signup(self, email: str):
        task = asyncio.create_task(self.send_email(email))
        self._background.add(task)
        task.add_done_callback(self._background.discard)
        return {"ok": True}

    async def drain(self):
        while self._background:
            await asyncio.gather(*list(self._background), return_exceptions=True)

async def test_signup_sends_email():
    service = Service()
    await service.handle_signup("a@example.com")
    await service.drain()
    assert service.sent == ["a@example.com"]

Measured: 20 of 20 passed both idle and loaded. drain() is useful in production as well — graceful shutdown waits on it — and its while loop covers background tasks that start other background tasks.

Verify: no test sleeps to wait for background work; each awaits a drain or an event.

Passes out of 20 with a busy CPU 4 horizontal bars comparing periodic loop, real time with the others. Passes out of 20 with a busy CPU periodic loop, real time 5/20 background work, test sleeps 15/20 periodic loop, fake sleep 20/20 background work, await drain() 20/20 Every idle run passed; load exposed the timing assumptions.

4. Know why sleeping sometimes works

A test that sleeps to wait for background work can be reliable by accident. Measured with the email's work done by await asyncio.sleep(0.02) instead of a thread, the test's 30 ms sleep passed 20 of 20 even under load: both are timers on the same event loop, the 20 ms one was scheduled first and expires first, and load delays both equally. The guarantee disappears as soon as the work involves anything outside the loop — a thread, a subprocess, a socket — where scheduling depends on the operating system. The test then passes until the code changes or the CI runner gets busier, and the failure looks like a flaky test rather than a test that was always wrong.

async def send_email(self, to):
    await asyncio.to_thread(smtp_send, to)       # real work leaves the loop: timing no longer ordered
    self.sent.append(to)

Verify: tests do not rely on timer ordering between the test and background work; they await completion explicitly.

5. Test failure and shutdown paths too

Background work fails and gets cancelled, and both paths need tests. A failing iteration should not end a periodic loop; a shutdown should cancel it cleanly:

async def test_loop_survives_failing_refresh():
    fake = FakeSleep()
    service = Service(interval=60, sleep=fake)
    service.refresh_once = AsyncMock(side_effect=[RuntimeError("boom"), None, None])
    task = asyncio.create_task(service.refresh_forever_guarded())    # catches and logs per iteration
    while len(fake.calls) < 3:
        await asyncio.sleep(0)
    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task
    assert service.refresh_once.await_count == 3

The final await task inside pytest.raises(CancelledError) checks that the loop exits on cancellation instead of swallowing it. Cancellation and cleanup testing in general is covered in testing cancellation and cleanup paths, and leaked background tasks are caught by detecting leaked tasks in tests.

Verify: each periodic task has tests for a failing iteration and for cancellation, both with an injected sleep.

Making background work testable A flow of 5 stages. Making background work testable Split work method vs loop Inject sleep (and clock) into the loop Track spawned tasks in a set Await drain() instead of sleeping Paths failing iteration, cancellation Measured: 20/20 under load, against 5/20 and 15/20 for sleep-based tests.

Verification

Background and periodic tasks are tested deterministically when:

  • The work is tested directly, separate from the loop that schedules it.
  • Loops take an injected sleep, and tests assert exact intervals and counts.
  • Background tasks can be drained, and tests await them instead of sleeping.
  • Failure and cancellation are tested with the same fakes.

Diagnostic Hook: when a test of background or periodic work fails only on CI, run it repeatedly with a few CPU-burning processes pinned to the same core. A real-time periodic test fell from 20 of 20 to 5 of 20 passing that way here, which identifies the timing assumption.

Pitfalls & edge cases

  • Real intervals in tests. Measured: 5 of 20 passed under CPU load.
  • Sleeping to wait for background work. Measured: 15 of 20 when the work used a thread.
  • Trusting a sleep-based test that passes. Timer ordering hides the problem until the work leaves the loop.
  • Swallowing cancellation in the loop. The test's pytest.raises(CancelledError) catches it.

Frequently Asked Questions

How do I test an asyncio periodic task without waiting?

Inject the sleep function and pass a fake that records the interval and yields. The test checked 10 iterations of a 60 s interval in milliseconds, passing 20 of 20 under load.

How do I wait for a fire-and-forget task in a test?

Track spawned tasks in a set and await them through a drain() method. That passed 20 of 20 under load, where sleeping 30 ms passed 15 of 20.

Why does my sleep-based async test pass locally but fail on CI?

Real waits depend on scheduling. A 50 ms periodic loop expected 11 runs in 0.52 s: 20 of 20 idle, 5 of 20 with the CPU busy.

Is asyncio.sleep in a test ever safe for waiting?

Only by accident, when the background work is itself only asyncio timers on the same loop. Once it uses a thread or I/O, ordering is no longer guaranteed.