Rust Async Testing with tokio-test: Patterns for Complex Async Code
Testing async Rust code with #[tokio::test] handles the basics — but complex async patterns (state machines, channels, time-dependent logic, concurrent operations) need more. This guide covers advanced async testing patterns beyond the basics.
The Foundation: #[tokio::test]
The baseline for async Rust tests:
[dev-dependencies]
tokio = { version = "1", features = ["full", "test-util"] }
tokio-test = "0.4"#[tokio::test]
async fn test_basic_async() {
let result = my_async_function().await;
assert_eq!(result, 42);
}Testing Time-Dependent Code with tokio::time::pause
Real-time sleeps make tests slow and flaky. Use tokio::time::pause() to control time:
use tokio::time::{sleep, Duration, pause, advance};
// The function being tested uses tokio sleep
pub async fn rate_limited_operation() -> Result<(), String> {
for attempt in 0..3 {
match try_operation().await {
Ok(result) => return Ok(result),
Err(_) if attempt < 2 => {
let delay = Duration::from_secs(2u64.pow(attempt)); // Exponential backoff
sleep(delay).await;
}
Err(e) => return Err(e),
}
}
unreachable!()
}#[tokio::test]
async fn test_rate_limited_operation_retries_with_backoff() {
// Pause tokio's clock — time doesn't advance automatically
pause();
let operation_calls = std::sync::Arc::new(std::sync::atomic::AtomicU32::new(0));
let calls_clone = operation_calls.clone();
let handle = tokio::spawn(async move {
rate_limited_operation_with_counter(&calls_clone).await
});
// First attempt fails immediately
assert_eq!(operation_calls.load(Ordering::SeqCst), 1);
// Advance 2 seconds — first retry delay
advance(Duration::from_secs(2)).await;
assert_eq!(operation_calls.load(Ordering::SeqCst), 2);
// Advance 4 more seconds — second retry delay
advance(Duration::from_secs(4)).await;
assert_eq!(operation_calls.load(Ordering::SeqCst), 3);
// Total elapsed: 6 seconds in real time: 0 seconds
handle.await.unwrap().unwrap();
}Testing Channels
Tokio channels are common in async code. Test them directly:
use tokio::sync::{mpsc, oneshot};
// Producer-consumer pattern
pub async fn process_items(
rx: &mut mpsc::Receiver<String>,
processed: &mut Vec<String>,
) {
while let Some(item) = rx.recv().await {
processed.push(item.to_uppercase());
}
}
#[tokio::test]
async fn test_items_processed_in_order() {
let (tx, mut rx) = mpsc::channel(10);
tx.send("hello".to_string()).await.unwrap();
tx.send("world".to_string()).await.unwrap();
drop(tx); // Close the channel
let mut processed = Vec::new();
process_items(&mut rx, &mut processed).await;
assert_eq!(processed, vec!["HELLO", "WORLD"]);
}
#[tokio::test]
async fn test_channel_backpressure() {
let (tx, mut rx) = mpsc::channel(2); // Buffer of 2
tx.send("first".to_string()).await.unwrap();
tx.send("second".to_string()).await.unwrap();
// Channel is full — this should fail immediately
let result = tx.try_send("third".to_string());
assert!(result.is_err());
// Drain one item
rx.recv().await;
// Now there's space
tx.try_send("third".to_string()).unwrap();
}
#[tokio::test]
async fn test_oneshot_channel() {
let (tx, rx) = oneshot::channel::<String>();
tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(10)).await;
tx.send("response".to_string()).unwrap();
});
let response = rx.await.unwrap();
assert_eq!(response, "response");
}Testing Timeouts
use tokio::time::{timeout, Duration};
pub async fn fetch_with_timeout(url: &str, secs: u64) -> Result<String, String> {
timeout(Duration::from_secs(secs), fetch(url))
.await
.map_err(|_| format!("Timeout after {}s", secs))?
.map_err(|e| e.to_string())
}
#[tokio::test]
async fn test_timeout_returns_error_when_exceeded() {
pause();
// Start a slow operation
let handle = tokio::spawn(async {
fetch_with_timeout("http://slow-server.example", 5).await
});
// Advance past the timeout
advance(Duration::from_secs(6)).await;
let result = handle.await.unwrap();
assert!(result.is_err());
assert!(result.unwrap_err().contains("Timeout"));
}
#[tokio::test]
async fn test_completes_before_timeout() {
let result = fetch_with_timeout("http://fast-server.example", 30).await;
assert!(result.is_ok());
}Testing Concurrent Operations
use tokio::task::JoinSet;
pub async fn process_batch(items: Vec<String>) -> Vec<Result<String, String>> {
let mut set = JoinSet::new();
for item in items {
set.spawn(async move { process_single(&item).await });
}
let mut results = Vec::new();
while let Some(res) = set.join_next().await {
results.push(res.unwrap());
}
results
}
#[tokio::test]
async fn test_batch_processes_all_items() {
let items: Vec<String> = (0..10).map(|i| format!("item-{}", i)).collect();
let results = process_batch(items).await;
assert_eq!(results.len(), 10);
assert!(results.iter().all(|r| r.is_ok()));
}
#[tokio::test]
async fn test_batch_handles_partial_failures() {
let items = vec![
"good-item".to_string(),
"bad-item".to_string(), // Causes error
"good-item-2".to_string(),
];
let results = process_batch(items).await;
assert_eq!(results.len(), 3);
let success_count = results.iter().filter(|r| r.is_ok()).count();
assert_eq!(success_count, 2);
}Testing Async State Machines
State machines are common in async code. Test state transitions:
#[derive(Debug, PartialEq, Clone)]
pub enum ConnectionState {
Disconnected,
Connecting,
Connected,
Error(String),
}
pub struct Connection {
state: ConnectionState,
tx: mpsc::Sender<ConnectionState>,
}
impl Connection {
pub async fn connect(&mut self) -> Result<(), String> {
self.set_state(ConnectionState::Connecting).await;
match tcp_connect().await {
Ok(_) => {
self.set_state(ConnectionState::Connected).await;
Ok(())
}
Err(e) => {
let err = e.to_string();
self.set_state(ConnectionState::Error(err.clone())).await;
Err(err)
}
}
}
}
#[tokio::test]
async fn test_connection_state_transitions_on_success() {
let (tx, mut rx) = mpsc::channel(10);
let mut conn = Connection::new_test(tx);
let result = conn.connect().await;
assert!(result.is_ok());
// Verify state transitions in order
let states: Vec<ConnectionState> = {
let mut v = Vec::new();
while let Ok(s) = rx.try_recv() {
v.push(s);
}
v
};
assert_eq!(states, vec![
ConnectionState::Connecting,
ConnectionState::Connected,
]);
}
#[tokio::test]
async fn test_connection_state_transitions_on_failure() {
let (tx, mut rx) = mpsc::channel(10);
let mut conn = Connection::new_test_with_failing_tcp(tx);
let result = conn.connect().await;
assert!(result.is_err());
let states: Vec<ConnectionState> = {
let mut v = Vec::new();
while let Ok(s) = rx.try_recv() { v.push(s); }
v
};
assert!(matches!(states.last(), Some(ConnectionState::Error(_))));
}Using tokio_test::io for Stream Testing
tokio_test::io::Builder lets you create mock async I/O:
use tokio_test::io::Builder;
#[tokio::test]
async fn test_reads_data_correctly() {
let mock = Builder::new()
.read(b"Hello")
.read(b", World!")
.build();
let mut reader = tokio::io::BufReader::new(mock);
let mut buf = String::new();
use tokio::io::AsyncReadExt;
reader.read_to_string(&mut buf).await.unwrap();
assert_eq!(buf, "Hello, World!");
}
#[tokio::test]
async fn test_handles_connection_reset() {
use std::io::ErrorKind;
let mock = Builder::new()
.read(b"partial data")
.read_error(std::io::Error::new(ErrorKind::ConnectionReset, "reset"))
.build();
let result = read_until_complete(mock).await;
assert!(result.is_err());
}Testing Retry Logic
pub async fn fetch_with_retry(url: &str, max_attempts: u32) -> Result<String, String> {
let mut last_error = String::new();
for attempt in 0..max_attempts {
match fetch(url).await {
Ok(data) => return Ok(data),
Err(e) => {
last_error = e;
if attempt < max_attempts - 1 {
sleep(Duration::from_millis(100 * 2u64.pow(attempt))).await;
}
}
}
}
Err(format!("Failed after {} attempts: {}", max_attempts, last_error))
}
#[tokio::test]
async fn test_retry_succeeds_on_third_attempt() {
let attempt_count = Arc::new(AtomicU32::new(0));
let count = attempt_count.clone();
// Mock that fails twice then succeeds
let result = fetch_with_retry_and_counter(&count, 3).await;
assert!(result.is_ok());
assert_eq!(attempt_count.load(Ordering::SeqCst), 3);
}
#[tokio::test]
async fn test_retry_gives_up_after_max_attempts() {
let attempt_count = Arc::new(AtomicU32::new(0));
let count = attempt_count.clone();
let result = fetch_always_failing_with_counter(&count, 3).await;
assert!(result.is_err());
assert_eq!(attempt_count.load(Ordering::SeqCst), 3);
assert!(result.unwrap_err().contains("3 attempts"));
}CI Configuration
- name: Run async tests
run: cargo test -- --test-threads=4
env:
TOKIO_WORKER_THREADS: "1" # Single worker for deterministic ordering in testsConclusion
Advanced async testing in Rust requires three tools working together: #[tokio::test] for the test runtime, tokio::time::pause()/advance() for deterministic time control, and mock I/O from tokio_test for stream testing. With these, you can test retry logic without sleeping, channel backpressure without race conditions, and state machines with explicit transition verification — all running in milliseconds rather than seconds.
