Testing Async Code

An async test needs a runtime, and a test that depends on real time or real scheduling order fails now and then. This lesson shows how to run async tests, how to replace the clock and the network with controlled substitutes, and how to test the awkward parts: timeouts, servers, and shutdown.

Tool	Use
`#[tokio::test]`	runs an `async fn` as a test on its own runtime
`start_paused`, `time::advance`	virtual time; sleeps finish instantly
`tokio::io::duplex`	an in-memory connected pair of streams
`tokio_test::io::Builder`	a scripted mock of a stream
`tokio_test::task::spawn` and macros	poll a future by hand and assert `Pending` or `Ready`
port `0`	a real TCP server without a fixed port

The examples need tokio with the test-util feature (cargo add tokio --features full,test-util), tokio-test, tokio-util (for CancellationToken), tracing, and tracing-subscriber. Every code block below is a test module. Run the tests with cargo test; the output block under each shows the cargo test report.

Example 1: The smallest async test

#[tokio::test] replaces #[test]. The macro builds a runtime, and runs the body to completion with block_on. The test passes when the function returns and fails when it panics, as with a normal test.

Rustcargo test

async fn double(n: u32) -> u32 {
    tokio::task::yield_now().await;
    n * 2
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn doubles_a_number() {
        assert_eq!(double(21).await, 42);
    }
}

Output

running 1 test
test tests::doubles_a_number ... ok

test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

Without the macro, an async fn marked #[test] does not compile, because the test harness needs a function that returns () or a Result, and an async fn returns a future.

Example 2: Runtime options

#[tokio::test] accepts the same options as #[tokio::main]. The default is the current_thread runtime: one thread, which runs the test body and every spawned task. Other options: flavor = "multi_thread" and worker_threads = N for parallel scheduling, and start_paused = true for virtual time.

Rustcargo test

#[cfg(test)]
mod tests {
    use tokio::runtime::{Handle, RuntimeFlavor};

    #[tokio::test]
    async fn default_is_current_thread() {
        assert_eq!(Handle::current().runtime_flavor(), RuntimeFlavor::CurrentThread);
        assert_eq!(Handle::current().metrics().num_workers(), 1);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 3)]
    async fn multi_thread_with_three_workers() {
        assert_eq!(Handle::current().runtime_flavor(), RuntimeFlavor::MultiThread);
        assert_eq!(Handle::current().metrics().num_workers(), 3);
    }
}

Output

running 2 tests
test tests::default_is_current_thread ... ok
test tests::multi_thread_with_three_workers ... ok

test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

Every test gets its own runtime, so tasks spawned by one test cannot leak into another. When a test returns, its runtime is dropped and unfinished spawned tasks are dropped with it.

The current_thread default is usually the better fit for tests because task interleaving is repeatable. A multi_thread test is needed only for code that must be exercised with real parallelism, such as code that uses block_in_place, which panics on current_thread.

Example 3: Tests that return `Result`

A test may return Result<(), E> where E: Debug. An Err fails the test and prints the error, which allows ? in the body instead of unwrap.

Rustcargo test

#[cfg(test)]
mod tests {
    use std::num::ParseIntError;

    async fn parse(text: &str) -> Result<i32, ParseIntError> {
        text.parse()
    }

    #[tokio::test]
    async fn parses_with_question_mark() -> Result<(), ParseIntError> {
        let a = parse("40").await?;
        let b = parse("2").await?;
        assert_eq!(a + b, 42);
        Ok(())
    }
}

Output

running 1 test
test tests::parses_with_question_mark ... ok

test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

Example 4: Expecting panics, including in spawned tasks

#[should_panic] works on async tests. A panic inside a spawned task does not fail the test by itself: it surfaces as a JoinError on the task's handle, so a test that must catch it awaits the handle.

Rustcargo test

#[cfg(test)]
mod tests {
    #[tokio::test]
    #[should_panic(expected = "index out of bounds")]
    async fn panics_in_the_test_body() {
        let items = vec![1, 2, 3];
        let index = items.len();
        let _ = items[index];
    }

    #[tokio::test]
    async fn spawned_panic_is_reported_on_the_handle() {
        let handle = tokio::spawn(async {
            if true {
                panic!("worker crashed");
            }
        });
        let error = handle.await.unwrap_err();
        assert!(error.is_panic());
    }
}

Output

running 2 tests
test tests::panics_in_the_test_body - should panic ... ok
test tests::spawned_panic_is_reported_on_the_handle ... ok

test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

A test that spawns a task and never awaits the handle passes even when the task panics. Awaiting handles, or using JoinSet, is how a test observes task failures.

Example 5: Paused time

start_paused = true starts the runtime with its clock frozen. When every task is waiting on a timer and nothing else can run, the runtime jumps the clock to the earliest timer. A ten-second sleep completes at once in real time, and Instant::now() shows exactly ten seconds later (Time).

Rustcargo test

#[cfg(test)]
mod tests {
    use std::time::Duration;
    use tokio::time::{Instant, sleep};

    #[tokio::test(start_paused = true)]
    async fn ten_seconds_pass_instantly() {
        let start = Instant::now();
        sleep(Duration::from_secs(10)).await;
        assert!(start.elapsed() >= Duration::from_secs(10));
        assert!(start.elapsed() < Duration::from_secs(11));
    }

    #[tokio::test(start_paused = true)]
    async fn sleeps_run_in_deadline_order() {
        let order = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
        let mut handles = Vec::new();
        for (name, secs) in [("slow", 30), ("fast", 5), ("medium", 10)] {
            let order = order.clone();
            handles.push(tokio::spawn(async move {
                sleep(Duration::from_secs(secs)).await;
                order.lock().unwrap().push(name);
            }));
        }
        for handle in handles {
            handle.await.unwrap();
        }
        assert_eq!(*order.lock().unwrap(), ["fast", "medium", "slow"]);
    }
}

Output

running 2 tests
test tests::sleeps_run_in_deadline_order ... ok
test tests::ten_seconds_pass_instantly ... ok

test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

The order assertion holds on every run because the clock, not the scheduler, decides when each sleep ends. The same test on the real clock would take 30 seconds and depend on timer resolution.

Auto-advance has one condition: the runtime must have no other work to do. A task that is blocked in spawn_blocking or waiting on real I/O keeps the clock from advancing while it runs.

Example 6: `advance` and stepping the clock by hand

tokio::time::advance(duration) moves the paused clock forward and lets the timers that became due fire. Use it when a test must observe the state between two moments, for example "not done at 9 seconds, done at 10".

Rustcargo test

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::sync::atomic::{AtomicU32, Ordering};
    use std::time::Duration;
    use tokio::sync::{mpsc, oneshot};
    use tokio::time::{advance, interval, sleep};

    #[tokio::test(start_paused = true)]
    async fn state_between_two_moments() {
        let done = Arc::new(AtomicU32::new(0));
        let flag = done.clone();
        let (started_tx, started_rx) = oneshot::channel();
        let handle = tokio::spawn(async move {
            let timer = sleep(Duration::from_secs(10)); // the deadline is fixed here
            started_tx.send(()).unwrap();
            timer.await;
            flag.store(1, Ordering::SeqCst);
        });
        started_rx.await.unwrap();

        advance(Duration::from_secs(9)).await;
        assert_eq!(done.load(Ordering::SeqCst), 0);

        advance(Duration::from_secs(1)).await;
        handle.await.unwrap();
        assert_eq!(done.load(Ordering::SeqCst), 1);
    }

    #[tokio::test(start_paused = true)]
    async fn counts_ticks() {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let mut timer = interval(Duration::from_secs(1));
        tokio::spawn(async move {
            let mut ticks = 0;
            loop {
                timer.tick().await; // the first tick completes immediately
                ticks += 1;
                if tx.send(ticks).is_err() {
                    break;
                }
            }
        });
        assert_eq!(rx.recv().await, Some(1));

        advance(Duration::from_secs(3)).await;
        // Missed ticks are delivered in a burst, the default `MissedTickBehavior`.
        for expected in 2..=4 {
            assert_eq!(rx.recv().await, Some(expected));
        }
    }
}

Output

running 2 tests
test tests::counts_ticks ... ok
test tests::state_between_two_moments ... ok

test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

The documentation for advance states that a jump completes every timer inside the interval at the same moment, and that a sleep is not guaranteed to finish on the first poll after the jump. Tests therefore wait on a channel, a oneshot signal, or a JoinHandle instead of assuming another task has already run. When the test only needs "wait until this timer has fired", awaiting a sleep and relying on auto-advance is more reliable than advance.

Example 7: Testing a timeout

Code that uses timeout is tested with a slow dependency that sleeps for longer than the limit. Under paused time the test runs instantly and there is no risk of a slow machine turning a fast dependency into a timeout.

Rustcargo test

use std::time::Duration;
use tokio::time::{error::Elapsed, sleep, timeout};

async fn lookup(delay: Duration) -> &'static str {
    sleep(delay).await;
    "found"
}

async fn lookup_with_limit(delay: Duration) -> Result<&'static str, Elapsed> {
    timeout(Duration::from_millis(200), lookup(delay)).await
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test(start_paused = true)]
    async fn fast_lookup_succeeds() {
        assert_eq!(lookup_with_limit(Duration::from_millis(199)).await, Ok("found"));
    }

    #[tokio::test(start_paused = true)]
    async fn slow_lookup_times_out() {
        assert!(lookup_with_limit(Duration::from_millis(201)).await.is_err());
    }

    #[tokio::test(start_paused = true)]
    async fn boundary_is_visible_because_time_is_exact() {
        let start = tokio::time::Instant::now();
        let _ = lookup_with_limit(Duration::from_secs(60)).await;
        assert!(start.elapsed() < Duration::from_millis(250));
    }
}

Output

running 3 tests
test tests::boundary_is_visible_because_time_is_exact ... ok
test tests::fast_lookup_succeeds ... ok
test tests::slow_lookup_times_out ... ok

test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

The last test shows that timeout stopped waiting at the limit and did not wait for the 60-second dependency: the inner future was dropped when the deadline passed.

Example 8: In-memory streams with `duplex`

tokio::io::duplex(max_buf_size) returns two connected DuplexStream values. What is written to one side is read from the other. Code that is generic over AsyncRead + AsyncWrite runs against it without a socket (AsyncRead and AsyncWrite).

Rustcargo test

use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader};

/// Reads lines and answers each with its length.
async fn length_service<S>(stream: S) -> std::io::Result<()>
where
    S: AsyncRead + AsyncWrite + Unpin,
{
    let mut reader = BufReader::new(stream);
    let mut line = String::new();
    while reader.read_line(&mut line).await? > 0 {
        let reply = format!("{}\n", line.trim_end().len());
        reader.get_mut().write_all(reply.as_bytes()).await?;
        line.clear();
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::io::AsyncReadExt;

    #[tokio::test]
    async fn answers_each_line() {
        let (client, server) = tokio::io::duplex(1024);
        let service = tokio::spawn(length_service(server));

        let (mut read_half, mut write_half) = tokio::io::split(client);
        write_half.write_all(b"hello\nhi\n").await.unwrap();
        write_half.shutdown().await.unwrap(); // the service sees EOF and returns

        let mut replies = String::new();
        read_half.read_to_string(&mut replies).await.unwrap();
        assert_eq!(replies, "5\n2\n");
        service.await.unwrap().unwrap();
    }
}

Output

running 1 test
test tests::answers_each_line ... ok

test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

If the buffer size given to duplex is smaller than the data written, a write waits until the other side reads. That makes duplex useful for testing backpressure as well.

Example 9: Scripted mocks with `tokio_test::io::Builder`

tokio_test::io::Builder builds a Mock that implements AsyncRead and AsyncWrite and follows a script: the next read returns these bytes, the next write must be exactly these bytes. Any deviation panics with a message naming the expectation, so the test checks the wire format byte for byte.

Rustcargo test

use tokio::io::{AsyncReadExt, AsyncWriteExt};

/// Reads a name, writes a greeting.
async fn greet<S>(stream: &mut S) -> std::io::Result<()>
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
    let mut name = [0u8; 4];
    stream.read_exact(&mut name).await?;
    stream.write_all(b"hello ").await?;
    stream.write_all(&name).await?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio_test::io::Builder;

    #[tokio::test]
    async fn greets_by_name() {
        let mut mock = Builder::new().read(b"Asha").write(b"hello ").write(b"Asha").build();
        greet(&mut mock).await.unwrap();
    }

    #[tokio::test]
    async fn handles_a_read_split_in_two() {
        // The peer delivers the four bytes as two separate reads.
        let mut mock = Builder::new().read(b"As").read(b"ha").write(b"hello ").write(b"Asha").build();
        greet(&mut mock).await.unwrap();
    }

    #[tokio::test]
    async fn read_error_is_propagated() {
        let broken = std::io::Error::new(std::io::ErrorKind::ConnectionReset, "reset");
        let mut mock = Builder::new().read_error(broken).build();
        let error = greet(&mut mock).await.unwrap_err();
        assert_eq!(error.kind(), std::io::ErrorKind::ConnectionReset);
    }
}

Output

running 3 tests
test tests::greets_by_name ... ok
test tests::handles_a_read_split_in_two ... ok
test tests::read_error_is_propagated ... ok

test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

The second test checks something a socket rarely does on a fast machine, a read that returns fewer bytes than requested. Scripted partial reads are the reliable way to test code that must handle them. Builder::wait(duration) adds a delay between steps, and write_error scripts a failing write.

Example 10: Polling by hand with `tokio_test::task::spawn`

tokio_test::task::spawn(future) wraps a future in a mock task. poll() polls it once with a waker the test controls, and is_woken() reports whether the waker was called since the last poll. The macros assert_pending! and assert_ready! check the result. These tests do not need a runtime for futures that use no timers or I/O, and they verify the waker contract from Polling, Wakers, and a Tiny Executor: a pending future must arrange to be woken.

Rustcargo test

#[cfg(test)]
mod tests {
    use tokio::sync::mpsc;
    use tokio_test::{assert_pending, assert_ready, assert_ready_eq, task};

    #[test]
    fn recv_is_pending_until_a_message_arrives() {
        let (tx, mut rx) = mpsc::channel::<u32>(4);
        let mut recv = task::spawn(rx.recv());

        assert_pending!(recv.poll());
        assert!(!recv.is_woken());

        tx.try_send(7).unwrap();
        assert!(recv.is_woken()); // the send woke the task
        assert_ready_eq!(recv.poll(), Some(7));
    }

    #[test]
    fn recv_completes_with_none_when_all_senders_drop() {
        let (tx, mut rx) = mpsc::channel::<u32>(4);
        let mut recv = task::spawn(rx.recv());
        assert_pending!(recv.poll());

        drop(tx);
        assert!(recv.is_woken());
        let value = assert_ready!(recv.poll());
        assert_eq!(value, None);
    }

    #[test]
    fn send_on_a_full_channel_is_pending_until_space_frees() {
        let (tx, mut rx) = mpsc::channel::<u32>(1);
        tx.try_send(1).unwrap();

        let mut send = task::spawn(tx.send(2));
        assert_pending!(send.poll());

        assert_eq!(rx.try_recv().unwrap(), 1); // frees one slot
        assert!(send.is_woken());
        assert_ready!(send.poll()).unwrap();
        assert_eq!(rx.try_recv().unwrap(), 2);
    }
}

Output

running 3 tests
test tests::recv_completes_with_none_when_all_senders_drop ... ok
test tests::recv_is_pending_until_a_message_arrives ... ok
test tests::send_on_a_full_channel_is_pending_until_space_frees ... ok

test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

The tests are plain #[test] functions. Spawn supplies the Context, so no executor thread is involved, and every step is a single poll call. That precision is what makes these tests deterministic: the test decides exactly when the future runs.

Example 11: Timers under a mock task

A sleep needs a runtime with a timer, so timer tests combine tokio_test::task::spawn with #[tokio::test(start_paused = true)]. The test checks that the future is pending, advances the clock by hand, checks that the waker fired, and polls again.

Rustcargo test

#[cfg(test)]
mod tests {
    use std::time::Duration;
    use tokio_test::{assert_pending, assert_ready, task};

    #[tokio::test(start_paused = true)]
    async fn sleep_wakes_its_task_at_the_deadline() {
        let mut sleeping = task::spawn(tokio::time::sleep(Duration::from_secs(5)));
        assert_pending!(sleeping.poll());

        tokio::time::advance(Duration::from_secs(4)).await;
        assert!(!sleeping.is_woken());
        assert_pending!(sleeping.poll());

        tokio::time::advance(Duration::from_secs(1)).await;
        assert!(sleeping.is_woken());
        assert_ready!(sleeping.poll());
    }
}

Output

running 1 test
test tests::sleep_wakes_its_task_at_the_deadline ... ok

test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

Example 12: Testing channel behavior

Channel tests state the guarantees a design depends on: capacity, ordering, and what closing does. try_send and try_recv return at once, so a test observes a full or empty channel without waiting.

Rustcargo test

#[cfg(test)]
mod tests {
    use tokio::sync::mpsc::{self, error::TrySendError};

    #[tokio::test]
    async fn bounded_channel_rejects_when_full() {
        let (tx, mut rx) = mpsc::channel::<u32>(2);
        tx.try_send(1).unwrap();
        tx.try_send(2).unwrap();
        assert!(matches!(tx.try_send(3), Err(TrySendError::Full(3))));

        assert_eq!(rx.recv().await, Some(1));
        tx.try_send(3).unwrap();
        assert_eq!(rx.recv().await, Some(2));
        assert_eq!(rx.recv().await, Some(3));
    }

    #[tokio::test]
    async fn messages_from_one_sender_keep_their_order() {
        let (tx, mut rx) = mpsc::channel::<u32>(100);
        let producer = tokio::spawn(async move {
            for n in 0..50 {
                tx.send(n).await.unwrap();
            }
        });
        producer.await.unwrap();

        let mut received = Vec::new();
        while let Some(n) = rx.recv().await {
            received.push(n);
        }
        assert_eq!(received, (0..50).collect::<Vec<_>>());
    }

    #[tokio::test]
    async fn recv_returns_none_after_the_last_sender_drops() {
        let (tx, mut rx) = mpsc::channel::<u32>(1);
        let second = tx.clone();
        drop(tx);
        second.send(9).await.unwrap();
        drop(second);
        assert_eq!(rx.recv().await, Some(9));
        assert_eq!(rx.recv().await, None);
    }
}

Output

running 3 tests
test tests::bounded_channel_rejects_when_full ... ok
test tests::messages_from_one_sender_keep_their_order ... ok
test tests::recv_returns_none_after_the_last_sender_drops ... ok

test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

Example 13: A server on port 0

A test that binds a fixed port fails when the port is busy or when two tests run at once. Binding port 0 asks the operating system for any free port, and local_addr() reports which one it chose. The server runs as a spawned task, and the test connects to it as a client.

Rustcargo test

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;

async fn serve(listener: TcpListener) {
    loop {
        let (mut socket, _) = listener.accept().await.unwrap();
        tokio::spawn(async move {
            let mut buffer = [0u8; 64];
            let count = socket.read(&mut buffer).await.unwrap();
            let shout = buffer[..count].to_ascii_uppercase();
            socket.write_all(&shout).await.unwrap();
        });
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::net::TcpStream;

    async fn start() -> (std::net::SocketAddr, tokio::task::JoinHandle<()>) {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        (address, tokio::spawn(serve(listener)))
    }

    #[tokio::test]
    async fn uppercases_a_message() {
        let (address, server) = start().await;
        let mut client = TcpStream::connect(address).await.unwrap();
        client.write_all(b"ping").await.unwrap();

        let mut reply = [0u8; 4];
        client.read_exact(&mut reply).await.unwrap();
        assert_eq!(&reply, b"PING");
        server.abort();
    }

    #[tokio::test]
    async fn two_clients_are_served_independently() {
        let (address, server) = start().await;
        let mut first = TcpStream::connect(address).await.unwrap();
        let mut second = TcpStream::connect(address).await.unwrap();
        second.write_all(b"two").await.unwrap();
        first.write_all(b"one").await.unwrap();

        let mut buffer = [0u8; 3];
        second.read_exact(&mut buffer).await.unwrap();
        assert_eq!(&buffer, b"TWO");
        first.read_exact(&mut buffer).await.unwrap();
        assert_eq!(&buffer, b"ONE");
        server.abort();
    }
}

Output

running 2 tests
test tests::two_clients_are_served_independently ... ok
test tests::uppercases_a_message ... ok

test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

The listener is bound before the client connects, so the client cannot connect too early, and no sleep is needed to wait for the server to start. The test stops the accept loop with abort at the end; the runtime would drop it anyway when the test returns. TCP Clients and Servers explains the accept loop.

Example 14: Testing shutdown

A shutdown test checks two things: the task stops after the signal, and it finishes the work it had already accepted. CancellationToken from tokio-util is the signal here (Graceful Shutdown). A timeout around the join makes a task that never stops fail the test instead of hanging it, and paused time makes the timeout instant.

Rustcargo test

use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;

async fn worker(mut jobs: mpsc::Receiver<u32>, token: CancellationToken, done: Arc<AtomicU32>) {
    loop {
        tokio::select! {
            _ = token.cancelled() => break,
            job = jobs.recv() => match job {
                Some(_) => { done.fetch_add(1, Ordering::SeqCst); }
                None => break,
            },
        }
    }
    // Drain what is already queued before exiting.
    while let Ok(_) = jobs.try_recv() {
        done.fetch_add(1, Ordering::SeqCst);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::Duration;

    #[tokio::test(start_paused = true)]
    async fn worker_stops_after_cancel_and_drains_the_queue() {
        let (tx, rx) = mpsc::channel(16);
        let token = CancellationToken::new();
        let done = Arc::new(AtomicU32::new(0));
        let handle = tokio::spawn(worker(rx, token.clone(), done.clone()));

        for job in 0..5 {
            tx.send(job).await.unwrap();
        }
        token.cancel();

        let joined = tokio::time::timeout(Duration::from_secs(5), handle).await;
        assert!(joined.is_ok(), "worker did not stop within 5 seconds");
        assert_eq!(done.load(Ordering::SeqCst), 5);
    }

    #[tokio::test(start_paused = true)]
    async fn worker_stops_when_the_channel_closes() {
        let (tx, rx) = mpsc::channel::<u32>(1);
        let handle = tokio::spawn(worker(rx, CancellationToken::new(), Arc::default()));
        drop(tx);
        assert!(tokio::time::timeout(Duration::from_secs(5), handle).await.is_ok());
    }
}

Output

running 2 tests
test tests::worker_stops_after_cancel_and_drains_the_queue ... ok
test tests::worker_stops_when_the_channel_closes ... ok

test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

The assertion message ("worker did not stop within 5 seconds") turns a hang into a readable failure. Without the timeout, a worker that ignored the token would leave the test running until the test runner's own limit.

Example 15: Asserting on logs

A test can capture tracing output in a buffer and assert on it (Tracing and Observability explains the setup). set_default installs the subscriber for the current thread until its guard drops, which covers a current_thread test runtime because the test body and its tasks run on that one thread.

Rustcargo test

use std::io;
use std::sync::{Arc, Mutex};
use tracing_subscriber::fmt::MakeWriter;

#[derive(Clone, Default)]
struct Buffer(Arc<Mutex<Vec<u8>>>);

impl io::Write for Buffer {
    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
        self.0.lock().unwrap().extend_from_slice(bytes);
        Ok(bytes.len())
    }
    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

impl<'a> MakeWriter<'a> for Buffer {
    type Writer = Buffer;
    fn make_writer(&'a self) -> Buffer {
        self.clone()
    }
}

#[tracing::instrument]
async fn charge(order: u32, cents: u32) -> Result<(), String> {
    if cents == 0 {
        tracing::warn!("nothing to charge");
        return Err("empty order".to_string());
    }
    tracing::info!("charged");
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    fn capture() -> (Buffer, tracing::subscriber::DefaultGuard) {
        let buffer = Buffer::default();
        let subscriber = tracing_subscriber::fmt()
            .with_writer(buffer.clone())
            .with_ansi(false)
            .without_time()
            .with_target(false)
            .finish();
        (buffer, tracing::subscriber::set_default(subscriber))
    }

    fn text(buffer: &Buffer) -> String {
        String::from_utf8(buffer.0.lock().unwrap().clone()).unwrap()
    }

    #[tokio::test]
    async fn logs_a_warning_for_an_empty_order() {
        let (buffer, _guard) = capture();
        assert!(charge(7, 0).await.is_err());
        let output = text(&buffer);
        assert!(output.contains("WARN charge{order=7 cents=0}: nothing to charge"), "{output}");
    }

    #[tokio::test]
    async fn logs_success_without_warnings() {
        let (buffer, _guard) = capture();
        charge(8, 500).await.unwrap();
        let output = text(&buffer);
        assert!(output.contains("INFO charge{order=8 cents=500}: charged"), "{output}");
        assert!(!output.contains("WARN"));
    }
}

Output

running 2 tests
test tests::logs_a_warning_for_an_empty_order ... ok
test tests::logs_success_without_warnings ... ok

test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

Example 16: Wait for a signal, not for time

Many flaky tests use sleep as a synchronization tool: "start the task, sleep 100 ms, assume it is ready". On a loaded machine the task may not be ready after 100 ms, and on a fast machine the test wastes time. A signal from the task (a oneshot, a Notify, a channel message) makes the test wait exactly as long as needed.

Rustcargo test

#[cfg(test)]
mod tests {
    use tokio::sync::{oneshot, watch};

    #[tokio::test]
    async fn wait_for_readiness_with_a_oneshot() {
        let (ready_tx, ready_rx) = oneshot::channel();
        let server = tokio::spawn(async move {
            // ... bind, load configuration, warm caches ...
            ready_tx.send("listening").unwrap();
            std::future::pending::<()>().await;
        });

        assert_eq!(ready_rx.await.unwrap(), "listening");
        server.abort();
    }

    #[tokio::test]
    async fn wait_for_a_state_with_a_watch_channel() {
        let (state_tx, mut state_rx) = watch::channel(0u32);
        tokio::spawn(async move {
            for step in 1..=3 {
                tokio::task::yield_now().await;
                state_tx.send(step).unwrap();
            }
        });

        // Wait until the worker reaches step 3, however long that takes.
        let reached = state_rx.wait_for(|step| *step >= 3).await.unwrap();
        assert_eq!(*reached, 3);
    }
}

Output

running 2 tests
test tests::wait_for_a_state_with_a_watch_channel ... ok
test tests::wait_for_readiness_with_a_oneshot ... ok

test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

watch::Receiver::wait_for returns as soon as the stored value satisfies the closure, including when it already does. A test that waits on the condition is correct on any machine.

Example 17: `start_paused` needs `current_thread`

Paused time exists only on the current_thread runtime. Combining start_paused = true with flavor = "multi_thread" is rejected when the macro expands, before the program runs.

Rust✗ does not compile

#[tokio::test(flavor = "multi_thread", start_paused = true)]
async fn not_allowed() {}

fn main() {}

Output

error: The `start_paused` option requires the `current_thread` runtime flavor. Use `#[tokio::test(flavor = "current_thread")]`
 --> src/main.rs:1:55
  |
1 | #[tokio::test(flavor = "multi_thread", start_paused = true)]
  |                                                       ^^^^

Example 18: Causes of flaky tests

Cause	Symptom	Deterministic replacement
`sleep` used to wait for another task	fails under load, wastes time otherwise	wait on a `oneshot`, `Notify`, channel, or `watch::wait_for`
real clock with short timeouts	timeout fires on a slow machine	`start_paused = true`
fixed port	"address already in use" when tests overlap	bind `127.0.0.1:0` and read `local_addr()`
assertion on the order of events from different tasks	passes on `current_thread`, fails on `multi_thread`	assert on a sorted result or on a set, or synchronize the tasks
test task never awaited	panic or failure inside the task goes unnoticed	await the `JoinHandle` and check the result
hang instead of failure	test runs until the CI limit	wrap the wait in `timeout` under paused time
state shared between tests (files, env vars, global subscriber)	passes alone, fails in a full run	unique temp paths, per-test state, `set_default` instead of a global `init`
real network calls	fails when the network is slow or down	`duplex`, `tokio_test::io::Builder`, local server on port 0

cargo test runs test functions on several threads by default. The tests in this lesson share nothing, so that is safe. Tests that share a process-wide resource, such as a global tracing subscriber installed with init, need -- --test-threads=1 or a per-test set_default.

Example 19: Concurrency bugs and `loom`

Some bugs depend on one particular interleaving of atomic operations and lock acquisitions across threads, and a normal test can run for years without hitting it. loom is a separate model-checking crate that re-runs a test body under every possible interleaving of its threads and its loom::sync types, and reports the interleaving that fails. It tests the code of synchronization primitives, such as a hand-written lock-free queue or a waker-handling future, by compiling the code against loom's replacements for the std types. Tokio uses it for its own internals. It is described here without code because it needs the code under test to be written against loom types and does not apply to ordinary application tests.

The unstable runtime option unhandled_panic = "shutdown_runtime" (requires --cfg tokio_unstable) makes a panic in any spawned task shut down the runtime, so a test cannot miss a panic in a task it forgot to await. Its documentation is in the tokio::test macro.

Interview points

#[tokio::test] builds a runtime per test; the default is current_thread. Options: flavor, worker_threads, start_paused.
start_paused = true freezes the clock and auto-advances it when the runtime is idle. It works only on current_thread, and time::advance needs the test-util feature.
advance completes due timers together and does not guarantee that other tasks ran; yield after it, or use sleep and auto-advance.
A panic in a spawned task fails a test only if the test awaits the handle.
duplex gives two connected in-memory streams; tokio_test::io::Builder scripts exact reads and writes and can inject errors and partial reads.
tokio_test::task::spawn with assert_pending!, assert_ready!, and is_woken() tests the waker behavior of a future without a runtime thread.
Tests bind port 0 and read local_addr(); they never depend on a fixed port.
Replace sleep-based synchronization with signals; wrap waits in timeout so hangs become failures.
loom explores thread interleavings for low-level synchronization code. It does not replace normal async tests.

Summary

Use #[tokio::test(start_paused = true)] for anything involving time; keep real time out of unit tests.
Test I/O code against duplex or scripted mocks; test servers over real sockets bound to port 0.
Poll futures by hand with tokio_test::task::spawn to assert Pending, Ready, and wake-ups.
Await every spawned handle whose outcome matters.
Test shutdown with a token or channel close, and bound the wait with timeout.
Capture tracing output in a buffer with set_default to assert on logs.
Flaky tests come from sleeps as synchronization, real clocks, fixed ports, and shared global state.

Example 1: The smallest async test

Example 2: Runtime options

Example 3: Tests that return Result

Example 4: Expecting panics, including in spawned tasks

Example 5: Paused time

Example 6: advance and stepping the clock by hand

Example 7: Testing a timeout

Example 8: In-memory streams with duplex

Example 9: Scripted mocks with tokio_test::io::Builder

Example 10: Polling by hand with tokio_test::task::spawn

Example 11: Timers under a mock task

Example 12: Testing channel behavior

Example 13: A server on port 0

Example 14: Testing shutdown

Example 15: Asserting on logs

Example 16: Wait for a signal, not for time

Example 17: start_paused needs current_thread

Example 18: Causes of flaky tests

Example 19: Concurrency bugs and loom

Interview points

Summary

Example 3: Tests that return `Result`

Example 6: `advance` and stepping the clock by hand

Example 8: In-memory streams with `duplex`

Example 9: Scripted mocks with `tokio_test::io::Builder`

Example 10: Polling by hand with `tokio_test::task::spawn`

Example 17: `start_paused` needs `current_thread`

Example 19: Concurrency bugs and `loom`