Testing Axum Services: Handlers, Databases, and Contracts

A Rust Axum Router is a Tower service, so a test can hand it a request value and read the response value without opening a socket. This lesson tests a small Storefront orders service at four levels: in-process handler tests, integration tests against PostgreSQL, contract tests, and a real server. A payment gateway is replaced with a fake.

Project setup

#[sqlx::test] connects with DATABASE_URL, creates a new database for each test, applies the files in migrations/, and passes the test a PgPool for that database. The database user needs permission to create databases.

Terminal

docker run --name pg -e POSTGRES_PASSWORD=pw -e POSTGRES_DB=storefront -p 5432:5432 -d postgres:16
export DATABASE_URL=postgres://postgres:pw@127.0.0.1:5432/storefront

Cargo.toml

TOML

# Cargo.toml
[package]
name = "storefront"
version = "0.1.0"
edition = "2021"

[dependencies]
async-trait = "0.1"
axum = "0.8"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
sqlx = { version = "0.8", features = ["runtime-tokio", "postgres", "migrate", "macros"] }
tokio = { version = "1", features = ["full"] }

[dev-dependencies]
http-body-util = "0.1"
reqwest = { version = "0.12", features = ["json"] }
tower = { version = "0.5", features = ["util"] }

Example 1: Test a handler with `oneshot`

tower::ServiceExt::oneshot calls a service once and returns the response future. The test builds a Request, sends it to the router, and reads the body with http_body_util::BodyExt. No port is bound, so the test cannot fail because a port is busy.

Rust

// tests/oneshot_basics.rs
use axum::{
    body::Body,
    extract::Json,
    http::{Request, StatusCode},
    routing::{get, post},
    Router,
};
use http_body_util::BodyExt;
use serde_json::{json, Value};
use tower::ServiceExt;

async fn double(Json(body): Json<Value>) -> Json<Value> {
    let n = body["n"].as_i64().unwrap_or(0);
    Json(json!({ "n": n * 2 }))
}

fn app() -> Router {
    Router::new()
        .route("/health", get(|| async { "ok" }))
        .route("/double", post(double))
}

fn json_request(body: &str) -> Request<Body> {
    Request::builder()
        .method("POST")
        .uri("/double")
        .header("content-type", "application/json")
        .body(Body::from(body.to_owned()))
        .unwrap()
}

#[tokio::test]
async fn health_returns_ok() {
    let request = Request::builder().uri("/health").body(Body::empty()).unwrap();
    let response = app().oneshot(request).await.unwrap();

    assert_eq!(response.status(), StatusCode::OK);
    let bytes = response.into_body().collect().await.unwrap().to_bytes();
    assert_eq!(&bytes[..], b"ok");
}

#[tokio::test]
async fn double_returns_json() {
    let response = app().oneshot(json_request(r#"{"n": 4}"#)).await.unwrap();

    assert_eq!(response.status(), StatusCode::OK);
    let bytes = response.into_body().collect().await.unwrap().to_bytes();
    let value: Value = serde_json::from_slice(&bytes).unwrap();
    assert_eq!(value["n"], 8);
}

#[tokio::test]
async fn wrong_content_type_is_415() {
    let request = Request::builder()
        .method("POST")
        .uri("/double")
        .body(Body::from(r#"{"n": 4}"#))
        .unwrap();
    let response = app().oneshot(request).await.unwrap();
    assert_eq!(response.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE);
}

#[tokio::test]
async fn malformed_json_is_400() {
    let response = app().oneshot(json_request("{not json")).await.unwrap();
    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
}

oneshot takes the router by value. Call app() again, or clone the router, for a second request in one test.
Axum rejects a Json request with a wrong content type (415), malformed JSON (400), or a body that does not match the type (422). Asserting these statuses makes a change to error behavior visible.
app() builds the router in a function. Example 2 applies the same pattern with real dependencies.

Example 2: The service under test

The service builds its router from a state struct of dependencies: a database pool and a payment gateway behind a trait. main passes real values and the tests pass test values. The trait exists at the one seam that reaches a remote system. The database is not hidden behind a trait, because tests run against a real one.

An order is inserted as pending, charged, and then set to paid or cancelled. The caller identity comes from an x-customer-id header here. Real authentication is in Authentication and authorization.

migrations/0001_init.sql

SQL

-- migrations/0001_init.sql
CREATE TABLE products (
    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    sku         TEXT    NOT NULL UNIQUE,
    name        TEXT    NOT NULL,
    price_cents INTEGER NOT NULL CHECK (price_cents > 0)
);

CREATE TABLE orders (
    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    customer_id BIGINT      NOT NULL,
    product_id  BIGINT      NOT NULL REFERENCES products (id),
    quantity    INTEGER     NOT NULL CHECK (quantity > 0),
    total_cents BIGINT      NOT NULL,
    status      TEXT        NOT NULL DEFAULT 'pending'
                CHECK (status IN ('pending', 'paid', 'cancelled')),
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

src/error.rs

Rust

// src/error.rs
use axum::{
    http::StatusCode,
    response::{IntoResponse, Response},
    Json,
};
use serde_json::json;

pub enum ApiError {
    Unauthorized,
    NotFound,
    Invalid(&'static str),
    PaymentDeclined,
    Internal,
}

impl From<sqlx::Error> for ApiError {
    fn from(err: sqlx::Error) -> Self {
        eprintln!("database error: {err}");
        ApiError::Internal
    }
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        let (status, message) = match self {
            ApiError::Unauthorized => (StatusCode::UNAUTHORIZED, "unauthorized"),
            ApiError::NotFound => (StatusCode::NOT_FOUND, "not found"),
            ApiError::Invalid(message) => (StatusCode::UNPROCESSABLE_ENTITY, message),
            ApiError::PaymentDeclined => (StatusCode::PAYMENT_REQUIRED, "payment declined"),
            ApiError::Internal => (StatusCode::INTERNAL_SERVER_ERROR, "internal error"),
        };
        (status, Json(json!({ "error": message }))).into_response()
    }
}

src/payments.rs

Rust

// src/payments.rs
use async_trait::async_trait;

#[derive(Debug)]
pub enum PaymentError {
    Declined,
}

/// The seam between the service and a payment provider.
#[async_trait]
pub trait PaymentGateway: Send + Sync {
    async fn charge(&self, order_id: i64, cents: i64) -> Result<String, PaymentError>;
}

/// Stand-in for a real provider client, used by `main`.
pub struct AlwaysApprove;

#[async_trait]
impl PaymentGateway for AlwaysApprove {
    async fn charge(&self, order_id: i64, _cents: i64) -> Result<String, PaymentError> {
        Ok(format!("ch_{order_id}"))
    }
}

src/lib.rs

Rust

// src/lib.rs
pub mod error;
pub mod payments;

use std::sync::Arc;

use axum::{
    extract::{Path, State},
    http::{HeaderMap, StatusCode},
    routing::{get, post},
    Json, Router,
};
use serde::{Deserialize, Serialize};
use sqlx::PgPool;

use error::ApiError;
use payments::{PaymentError, PaymentGateway};

#[derive(Clone)]
pub struct AppState {
    pub pool: PgPool,
    pub payments: Arc<dyn PaymentGateway>,
}

/// Builds the router from its dependencies. `main` and the tests both call this.
pub fn app(state: AppState) -> Router {
    Router::new()
        .route("/health", get(health))
        .route("/orders", post(create_order))
        .route("/orders/{id}", get(get_order))
        .with_state(state)
}

#[derive(Deserialize)]
pub struct NewOrder {
    product_id: i64,
    quantity: i32,
}

#[derive(Serialize, sqlx::FromRow)]
pub struct OrderDto {
    id: i64,
    customer_id: i64,
    status: String,
    total_cents: i64,
}

async fn health() -> &'static str {
    "ok"
}

/// Stand-in for real authentication: the caller id comes from a header.
fn caller_id(headers: &HeaderMap) -> Result<i64, ApiError> {
    headers
        .get("x-customer-id")
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.parse().ok())
        .ok_or(ApiError::Unauthorized)
}

async fn set_status(pool: &PgPool, order_id: i64, status: &str) -> Result<(), ApiError> {
    sqlx::query("UPDATE orders SET status = $2 WHERE id = $1")
        .bind(order_id)
        .bind(status)
        .execute(pool)
        .await?;
    Ok(())
}

async fn create_order(
    State(state): State<AppState>,
    headers: HeaderMap,
    Json(body): Json<NewOrder>,
) -> Result<(StatusCode, Json<OrderDto>), ApiError> {
    let customer_id = caller_id(&headers)?;
    if body.quantity <= 0 {
        return Err(ApiError::Invalid("quantity must be positive"));
    }

    let price: Option<i32> = sqlx::query_scalar("SELECT price_cents FROM products WHERE id = $1")
        .bind(body.product_id)
        .fetch_optional(&state.pool)
        .await?;
    let price = price.ok_or(ApiError::Invalid("unknown product"))?;
    let total_cents = i64::from(price) * i64::from(body.quantity);

    let order = sqlx::query_as::<_, OrderDto>(
        "INSERT INTO orders (customer_id, product_id, quantity, total_cents)
         VALUES ($1, $2, $3, $4)
         RETURNING id, customer_id, status, total_cents",
    )
    .bind(customer_id)
    .bind(body.product_id)
    .bind(body.quantity)
    .bind(total_cents)
    .fetch_one(&state.pool)
    .await?;

    match state.payments.charge(order.id, total_cents).await {
        Ok(_) => {
            set_status(&state.pool, order.id, "paid").await?;
            let paid = OrderDto {
                status: "paid".to_owned(),
                ..order
            };
            Ok((StatusCode::CREATED, Json(paid)))
        }
        Err(PaymentError::Declined) => {
            set_status(&state.pool, order.id, "cancelled").await?;
            Err(ApiError::PaymentDeclined)
        }
    }
}

async fn get_order(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(id): Path<i64>,
) -> Result<Json<OrderDto>, ApiError> {
    let customer_id = caller_id(&headers)?;
    let order = sqlx::query_as::<_, OrderDto>(
        "SELECT id, customer_id, status, total_cents
         FROM orders
         WHERE id = $1 AND customer_id = $2",
    )
    .bind(id)
    .bind(customer_id)
    .fetch_optional(&state.pool)
    .await?;
    order.map(Json).ok_or(ApiError::NotFound)
}

src/main.rs

Rust

// src/main.rs
use std::sync::Arc;

use sqlx::postgres::PgPoolOptions;
use storefront::{app, payments::AlwaysApprove, AppState};

#[tokio::main]
async fn main() {
    let url = std::env::var("DATABASE_URL").expect("DATABASE_URL must be set");
    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect(&url)
        .await
        .expect("database connection");
    sqlx::migrate!().run(&pool).await.expect("migrations");

    let state = AppState {
        pool,
        payments: Arc::new(AlwaysApprove),
    };

    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await.unwrap();
    axum::serve(listener, app(state)).await.unwrap();
}

Example 3: Integration tests and a fake gateway

A fake is a small working replacement that behaves like the real dependency in a simple way. It differs from a mock, which records calls and checks them against expectations. FakeGateway records charges and can decline once, which creates the failure that is hard to trigger in a real provider. The tests assert outcomes: the stored status, the response, and the recorded charges.

Files in tests/ compile as separate crates. Each declares mod common; to share helpers. The common module is shared by all later examples.

tests/common/mod.rs

Rust

// tests/common/mod.rs
#![allow(dead_code)]

use std::sync::{
    atomic::{AtomicBool, Ordering},
    Arc, Mutex,
};

use async_trait::async_trait;
use axum::{
    body::Body,
    http::{Request, StatusCode},
    Router,
};
use http_body_util::BodyExt;
use serde_json::Value;
use sqlx::PgPool;
use storefront::{
    app,
    payments::{PaymentError, PaymentGateway},
    AppState,
};
use tower::ServiceExt;

/// A payment gateway that records charges and can be told to decline once.
#[derive(Default)]
pub struct FakeGateway {
    charges: Mutex<Vec<(i64, i64)>>,
    decline: AtomicBool,
}

impl FakeGateway {
    pub fn decline_next(&self) {
        self.decline.store(true, Ordering::SeqCst);
    }

    pub fn charges(&self) -> Vec<(i64, i64)> {
        self.charges.lock().unwrap().clone()
    }
}

#[async_trait]
impl PaymentGateway for FakeGateway {
    async fn charge(&self, order_id: i64, cents: i64) -> Result<String, PaymentError> {
        if self.decline.swap(false, Ordering::SeqCst) {
            return Err(PaymentError::Declined);
        }
        self.charges.lock().unwrap().push((order_id, cents));
        Ok(format!("ch_{order_id}"))
    }
}

pub struct TestApp {
    pub router: Router,
    pub pool: PgPool,
    pub gateway: Arc<FakeGateway>,
    pub product_id: i64,
}

impl TestApp {
    /// Builds the router over a fresh database and inserts one product (250 cents).
    pub async fn new(pool: PgPool) -> Self {
        let product_id: i64 = sqlx::query_scalar(
            "INSERT INTO products (sku, name, price_cents) VALUES ('PEN-1', 'Pen', 250) RETURNING id",
        )
        .fetch_one(&pool)
        .await
        .unwrap();

        let gateway = Arc::new(FakeGateway::default());
        let state = AppState {
            pool: pool.clone(),
            payments: gateway.clone(),
        };
        TestApp {
            router: app(state),
            pool,
            gateway,
            product_id,
        }
    }

    /// Sends one request in process and returns the status and the JSON body.
    pub async fn send(&self, request: Request<Body>) -> (StatusCode, Value) {
        let response = self.router.clone().oneshot(request).await.unwrap();
        let status = response.status();
        let bytes = response.into_body().collect().await.unwrap().to_bytes();
        let body = if bytes.is_empty() {
            Value::Null
        } else {
            serde_json::from_slice(&bytes).unwrap()
        };
        (status, body)
    }

    pub async fn seed_order(&self, customer_id: i64) -> i64 {
        sqlx::query_scalar(
            "INSERT INTO orders (customer_id, product_id, quantity, total_cents, status)
             VALUES ($1, $2, 1, 250, 'paid') RETURNING id",
        )
        .bind(customer_id)
        .bind(self.product_id)
        .fetch_one(&self.pool)
        .await
        .unwrap()
    }

    pub async fn only_order(&self) -> (i64, String) {
        sqlx::query_as("SELECT id, status FROM orders")
            .fetch_one(&self.pool)
            .await
            .unwrap()
    }

    pub async fn order_count(&self) -> i64 {
        sqlx::query_scalar("SELECT count(*) FROM orders")
            .fetch_one(&self.pool)
            .await
            .unwrap()
    }
}

pub fn post_order(customer: Option<i64>, body: Value) -> Request<Body> {
    let mut builder = Request::builder()
        .method("POST")
        .uri("/orders")
        .header("content-type", "application/json");
    if let Some(id) = customer {
        builder = builder.header("x-customer-id", id.to_string());
    }
    builder.body(Body::from(body.to_string())).unwrap()
}

pub fn get_order(customer: Option<i64>, order_id: i64) -> Request<Body> {
    let mut builder = Request::builder().uri(format!("/orders/{order_id}"));
    if let Some(id) = customer {
        builder = builder.header("x-customer-id", id.to_string());
    }
    builder.body(Body::empty()).unwrap()
}

tests/orders.rs

Rust

// tests/orders.rs
mod common;

use axum::http::StatusCode;
use common::{post_order, TestApp};
use serde_json::json;
use sqlx::PgPool;

#[sqlx::test]
async fn paid_order_is_stored_and_charged(pool: PgPool) {
    let app = TestApp::new(pool).await;

    let request = post_order(Some(7), json!({ "product_id": app.product_id, "quantity": 2 }));
    let (status, body) = app.send(request).await;

    assert_eq!(status, StatusCode::CREATED);
    assert_eq!(body["status"], "paid");
    assert_eq!(body["total_cents"], 500);

    let order_id = body["id"].as_i64().unwrap();
    assert_eq!(app.only_order().await, (order_id, "paid".to_owned()));
    assert_eq!(app.gateway.charges(), vec![(order_id, 500)]);
}

#[sqlx::test]
async fn declined_payment_cancels_the_order(pool: PgPool) {
    let app = TestApp::new(pool).await;
    app.gateway.decline_next();

    let request = post_order(Some(7), json!({ "product_id": app.product_id, "quantity": 1 }));
    let (status, body) = app.send(request).await;

    assert_eq!(status, StatusCode::PAYMENT_REQUIRED);
    assert_eq!(body["error"], "payment declined");
    let (_, stored_status) = app.only_order().await;
    assert_eq!(stored_status, "cancelled");
    assert!(app.gateway.charges().is_empty());
}

#[sqlx::test]
async fn unknown_product_creates_nothing(pool: PgPool) {
    let app = TestApp::new(pool).await;

    let request = post_order(Some(7), json!({ "product_id": 999, "quantity": 1 }));
    let (status, _) = app.send(request).await;

    assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
    assert_eq!(app.order_count().await, 0);
}

Each test gets its own database with migrations applied, so tests run in parallel and do not depend on order.
The tests drive HTTP and then query the database. This finds missing columns, violated constraints, and transactions that never commit.
Use the PostgreSQL major version of production. SQLite is not a stand-in: types and locking differ.

Example 4: Authorization and error paths

Every route that needs a caller identity gets three tests: no credentials, bad credentials, and valid credentials for the wrong owner. The wrong-owner case finds authorization bugs. Returning 404 for another customer's order avoids confirming that the order exists.

tests/access.rs

Rust

// tests/access.rs
mod common;

use axum::http::StatusCode;
use common::{get_order, post_order, TestApp};
use serde_json::json;
use sqlx::PgPool;

#[sqlx::test]
async fn request_without_credentials_is_401(pool: PgPool) {
    let app = TestApp::new(pool).await;
    let order_id = app.seed_order(7).await;

    let (status, _) = app.send(get_order(None, order_id)).await;
    assert_eq!(status, StatusCode::UNAUTHORIZED);

    let request = post_order(None, json!({ "product_id": app.product_id, "quantity": 1 }));
    let (status, _) = app.send(request).await;
    assert_eq!(status, StatusCode::UNAUTHORIZED);
}

#[sqlx::test]
async fn customer_cannot_read_another_customers_order(pool: PgPool) {
    let app = TestApp::new(pool).await;
    let order_id = app.seed_order(7).await;

    let (status, body) = app.send(get_order(Some(8), order_id)).await;
    assert_eq!(status, StatusCode::NOT_FOUND);
    assert_eq!(body["error"], "not found");

    let (status, body) = app.send(get_order(Some(7), order_id)).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["id"], order_id);
}

#[sqlx::test]
async fn zero_quantity_is_rejected(pool: PgPool) {
    let app = TestApp::new(pool).await;

    let request = post_order(Some(7), json!({ "product_id": app.product_id, "quantity": 0 }));
    let (status, body) = app.send(request).await;

    assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
    assert_eq!(body["error"], "quantity must be positive");
}

Example 5: Contract tests

A contract is what an interface promises to its consumers: paths, methods, status codes, field names, types, and error shapes. A contract test fails when a change would break a consumer even if the service's own tests pass. The tests compare against serde_json::Value and not the handler's own struct. A renamed field then fails the test, because the test and the handler do not change together.

tests/contract.rs

Rust

// tests/contract.rs
mod common;

use axum::http::StatusCode;
use common::{get_order, TestApp};
use sqlx::PgPool;

#[sqlx::test]
async fn order_response_shape_is_stable(pool: PgPool) {
    let app = TestApp::new(pool).await;
    let order_id = app.seed_order(7).await;

    let (status, body) = app.send(get_order(Some(7), order_id)).await;
    assert_eq!(status, StatusCode::OK);

    let mut keys: Vec<&str> = body.as_object().unwrap().keys().map(String::as_str).collect();
    keys.sort();
    assert_eq!(keys, ["customer_id", "id", "status", "total_cents"]);
    assert!(body["id"].is_i64());
    assert!(body["customer_id"].is_i64());
    assert!(body["status"].is_string());
    assert!(body["total_cents"].is_i64());
}

#[sqlx::test]
async fn error_response_shape_is_stable(pool: PgPool) {
    let app = TestApp::new(pool).await;

    let (status, body) = app.send(get_order(Some(7), 12345)).await;
    assert_eq!(status, StatusCode::NOT_FOUND);

    let keys: Vec<&str> = body.as_object().unwrap().keys().map(String::as_str).collect();
    assert_eq!(keys, ["error"]);
    assert!(body["error"].is_string());
}

Other contract checks:

A snapshot of the generated OpenAPI document shows any change to paths or schemas in a diff (REST API design).
Consumer-driven contracts have each consumer publish the requests and responses it depends on, and the provider runs them in CI.
A schema compatibility check for protobuf or event schemas fails the build when a field is removed or renumbered (gRPC basics).

Example 6: A real server on a free port

Some behavior exists only over a real connection: keep-alive, graceful shutdown, and TLS. Binding to port 0 lets the operating system choose a free port, and local_addr returns it. Each such test is slower and harder to debug than an in-process test, so the number stays small. Timeout tests use Tokio's paused clock (Testing async code).

tests/e2e.rs

Rust

// tests/e2e.rs
mod common;

use common::TestApp;
use sqlx::PgPool;

#[sqlx::test]
async fn real_server_answers_health(pool: PgPool) {
    let app = TestApp::new(pool).await;

    // Port 0 lets the operating system choose a free port.
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let router = app.router.clone();
    tokio::spawn(async move {
        axum::serve(listener, router).await.unwrap();
    });

    let response = reqwest::get(format!("http://{addr}/health")).await.unwrap();
    assert_eq!(response.status(), 200);
    assert_eq!(response.text().await.unwrap(), "ok");
}

Complete example

File tree:

Text

storefront/
  Cargo.toml
  migrations/0001_init.sql
  src/
    main.rs
    lib.rs
    error.rs
    payments.rs
  tests/
    common/mod.rs
    oneshot_basics.rs
    orders.rs
    access.rs
    contract.rs
    e2e.rs

Cargo.toml

TOML

# Cargo.toml
[package]
name = "storefront"
version = "0.1.0"
edition = "2021"

[dependencies]
async-trait = "0.1"
axum = "0.8"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
sqlx = { version = "0.8", features = ["runtime-tokio", "postgres", "migrate", "macros"] }
tokio = { version = "1", features = ["full"] }

[dev-dependencies]
http-body-util = "0.1"
reqwest = { version = "0.12", features = ["json"] }
tower = { version = "0.5", features = ["util"] }

migrations/0001_init.sql

SQL

-- migrations/0001_init.sql
CREATE TABLE products (
    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    sku         TEXT    NOT NULL UNIQUE,
    name        TEXT    NOT NULL,
    price_cents INTEGER NOT NULL CHECK (price_cents > 0)
);

CREATE TABLE orders (
    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    customer_id BIGINT      NOT NULL,
    product_id  BIGINT      NOT NULL REFERENCES products (id),
    quantity    INTEGER     NOT NULL CHECK (quantity > 0),
    total_cents BIGINT      NOT NULL,
    status      TEXT        NOT NULL DEFAULT 'pending'
                CHECK (status IN ('pending', 'paid', 'cancelled')),
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

src/main.rs

Rust

// src/main.rs
use std::sync::Arc;

use sqlx::postgres::PgPoolOptions;
use storefront::{app, payments::AlwaysApprove, AppState};

#[tokio::main]
async fn main() {
    let url = std::env::var("DATABASE_URL").expect("DATABASE_URL must be set");
    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect(&url)
        .await
        .expect("database connection");
    sqlx::migrate!().run(&pool).await.expect("migrations");

    let state = AppState {
        pool,
        payments: Arc::new(AlwaysApprove),
    };

    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await.unwrap();
    axum::serve(listener, app(state)).await.unwrap();
}

src/lib.rs

Rust

// src/lib.rs
pub mod error;
pub mod payments;

use std::sync::Arc;

use axum::{
    extract::{Path, State},
    http::{HeaderMap, StatusCode},
    routing::{get, post},
    Json, Router,
};
use serde::{Deserialize, Serialize};
use sqlx::PgPool;

use error::ApiError;
use payments::{PaymentError, PaymentGateway};

#[derive(Clone)]
pub struct AppState {
    pub pool: PgPool,
    pub payments: Arc<dyn PaymentGateway>,
}

/// Builds the router from its dependencies. `main` and the tests both call this.
pub fn app(state: AppState) -> Router {
    Router::new()
        .route("/health", get(health))
        .route("/orders", post(create_order))
        .route("/orders/{id}", get(get_order))
        .with_state(state)
}

#[derive(Deserialize)]
pub struct NewOrder {
    product_id: i64,
    quantity: i32,
}

#[derive(Serialize, sqlx::FromRow)]
pub struct OrderDto {
    id: i64,
    customer_id: i64,
    status: String,
    total_cents: i64,
}

async fn health() -> &'static str {
    "ok"
}

/// Stand-in for real authentication: the caller id comes from a header.
fn caller_id(headers: &HeaderMap) -> Result<i64, ApiError> {
    headers
        .get("x-customer-id")
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.parse().ok())
        .ok_or(ApiError::Unauthorized)
}

async fn set_status(pool: &PgPool, order_id: i64, status: &str) -> Result<(), ApiError> {
    sqlx::query("UPDATE orders SET status = $2 WHERE id = $1")
        .bind(order_id)
        .bind(status)
        .execute(pool)
        .await?;
    Ok(())
}

async fn create_order(
    State(state): State<AppState>,
    headers: HeaderMap,
    Json(body): Json<NewOrder>,
) -> Result<(StatusCode, Json<OrderDto>), ApiError> {
    let customer_id = caller_id(&headers)?;
    if body.quantity <= 0 {
        return Err(ApiError::Invalid("quantity must be positive"));
    }

    let price: Option<i32> = sqlx::query_scalar("SELECT price_cents FROM products WHERE id = $1")
        .bind(body.product_id)
        .fetch_optional(&state.pool)
        .await?;
    let price = price.ok_or(ApiError::Invalid("unknown product"))?;
    let total_cents = i64::from(price) * i64::from(body.quantity);

    let order = sqlx::query_as::<_, OrderDto>(
        "INSERT INTO orders (customer_id, product_id, quantity, total_cents)
         VALUES ($1, $2, $3, $4)
         RETURNING id, customer_id, status, total_cents",
    )
    .bind(customer_id)
    .bind(body.product_id)
    .bind(body.quantity)
    .bind(total_cents)
    .fetch_one(&state.pool)
    .await?;

    match state.payments.charge(order.id, total_cents).await {
        Ok(_) => {
            set_status(&state.pool, order.id, "paid").await?;
            let paid = OrderDto {
                status: "paid".to_owned(),
                ..order
            };
            Ok((StatusCode::CREATED, Json(paid)))
        }
        Err(PaymentError::Declined) => {
            set_status(&state.pool, order.id, "cancelled").await?;
            Err(ApiError::PaymentDeclined)
        }
    }
}

async fn get_order(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(id): Path<i64>,
) -> Result<Json<OrderDto>, ApiError> {
    let customer_id = caller_id(&headers)?;
    let order = sqlx::query_as::<_, OrderDto>(
        "SELECT id, customer_id, status, total_cents
         FROM orders
         WHERE id = $1 AND customer_id = $2",
    )
    .bind(id)
    .bind(customer_id)
    .fetch_optional(&state.pool)
    .await?;
    order.map(Json).ok_or(ApiError::NotFound)
}

src/error.rs

Rust

// src/error.rs
use axum::{
    http::StatusCode,
    response::{IntoResponse, Response},
    Json,
};
use serde_json::json;

pub enum ApiError {
    Unauthorized,
    NotFound,
    Invalid(&'static str),
    PaymentDeclined,
    Internal,
}

impl From<sqlx::Error> for ApiError {
    fn from(err: sqlx::Error) -> Self {
        eprintln!("database error: {err}");
        ApiError::Internal
    }
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        let (status, message) = match self {
            ApiError::Unauthorized => (StatusCode::UNAUTHORIZED, "unauthorized"),
            ApiError::NotFound => (StatusCode::NOT_FOUND, "not found"),
            ApiError::Invalid(message) => (StatusCode::UNPROCESSABLE_ENTITY, message),
            ApiError::PaymentDeclined => (StatusCode::PAYMENT_REQUIRED, "payment declined"),
            ApiError::Internal => (StatusCode::INTERNAL_SERVER_ERROR, "internal error"),
        };
        (status, Json(json!({ "error": message }))).into_response()
    }
}

src/payments.rs

Rust

// src/payments.rs
use async_trait::async_trait;

#[derive(Debug)]
pub enum PaymentError {
    Declined,
}

/// The seam between the service and a payment provider.
#[async_trait]
pub trait PaymentGateway: Send + Sync {
    async fn charge(&self, order_id: i64, cents: i64) -> Result<String, PaymentError>;
}

/// Stand-in for a real provider client, used by `main`.
pub struct AlwaysApprove;

#[async_trait]
impl PaymentGateway for AlwaysApprove {
    async fn charge(&self, order_id: i64, _cents: i64) -> Result<String, PaymentError> {
        Ok(format!("ch_{order_id}"))
    }
}

tests/common/mod.rs

Rust

// tests/common/mod.rs
#![allow(dead_code)]

use std::sync::{
    atomic::{AtomicBool, Ordering},
    Arc, Mutex,
};

use async_trait::async_trait;
use axum::{
    body::Body,
    http::{Request, StatusCode},
    Router,
};
use http_body_util::BodyExt;
use serde_json::Value;
use sqlx::PgPool;
use storefront::{
    app,
    payments::{PaymentError, PaymentGateway},
    AppState,
};
use tower::ServiceExt;

/// A payment gateway that records charges and can be told to decline once.
#[derive(Default)]
pub struct FakeGateway {
    charges: Mutex<Vec<(i64, i64)>>,
    decline: AtomicBool,
}

impl FakeGateway {
    pub fn decline_next(&self) {
        self.decline.store(true, Ordering::SeqCst);
    }

    pub fn charges(&self) -> Vec<(i64, i64)> {
        self.charges.lock().unwrap().clone()
    }
}

#[async_trait]
impl PaymentGateway for FakeGateway {
    async fn charge(&self, order_id: i64, cents: i64) -> Result<String, PaymentError> {
        if self.decline.swap(false, Ordering::SeqCst) {
            return Err(PaymentError::Declined);
        }
        self.charges.lock().unwrap().push((order_id, cents));
        Ok(format!("ch_{order_id}"))
    }
}

pub struct TestApp {
    pub router: Router,
    pub pool: PgPool,
    pub gateway: Arc<FakeGateway>,
    pub product_id: i64,
}

impl TestApp {
    /// Builds the router over a fresh database and inserts one product (250 cents).
    pub async fn new(pool: PgPool) -> Self {
        let product_id: i64 = sqlx::query_scalar(
            "INSERT INTO products (sku, name, price_cents) VALUES ('PEN-1', 'Pen', 250) RETURNING id",
        )
        .fetch_one(&pool)
        .await
        .unwrap();

        let gateway = Arc::new(FakeGateway::default());
        let state = AppState {
            pool: pool.clone(),
            payments: gateway.clone(),
        };
        TestApp {
            router: app(state),
            pool,
            gateway,
            product_id,
        }
    }

    /// Sends one request in process and returns the status and the JSON body.
    pub async fn send(&self, request: Request<Body>) -> (StatusCode, Value) {
        let response = self.router.clone().oneshot(request).await.unwrap();
        let status = response.status();
        let bytes = response.into_body().collect().await.unwrap().to_bytes();
        let body = if bytes.is_empty() {
            Value::Null
        } else {
            serde_json::from_slice(&bytes).unwrap()
        };
        (status, body)
    }

    pub async fn seed_order(&self, customer_id: i64) -> i64 {
        sqlx::query_scalar(
            "INSERT INTO orders (customer_id, product_id, quantity, total_cents, status)
             VALUES ($1, $2, 1, 250, 'paid') RETURNING id",
        )
        .bind(customer_id)
        .bind(self.product_id)
        .fetch_one(&self.pool)
        .await
        .unwrap()
    }

    pub async fn only_order(&self) -> (i64, String) {
        sqlx::query_as("SELECT id, status FROM orders")
            .fetch_one(&self.pool)
            .await
            .unwrap()
    }

    pub async fn order_count(&self) -> i64 {
        sqlx::query_scalar("SELECT count(*) FROM orders")
            .fetch_one(&self.pool)
            .await
            .unwrap()
    }
}

pub fn post_order(customer: Option<i64>, body: Value) -> Request<Body> {
    let mut builder = Request::builder()
        .method("POST")
        .uri("/orders")
        .header("content-type", "application/json");
    if let Some(id) = customer {
        builder = builder.header("x-customer-id", id.to_string());
    }
    builder.body(Body::from(body.to_string())).unwrap()
}

pub fn get_order(customer: Option<i64>, order_id: i64) -> Request<Body> {
    let mut builder = Request::builder().uri(format!("/orders/{order_id}"));
    if let Some(id) = customer {
        builder = builder.header("x-customer-id", id.to_string());
    }
    builder.body(Body::empty()).unwrap()
}

tests/oneshot_basics.rs

Rust

// tests/oneshot_basics.rs
use axum::{
    body::Body,
    extract::Json,
    http::{Request, StatusCode},
    routing::{get, post},
    Router,
};
use http_body_util::BodyExt;
use serde_json::{json, Value};
use tower::ServiceExt;

async fn double(Json(body): Json<Value>) -> Json<Value> {
    let n = body["n"].as_i64().unwrap_or(0);
    Json(json!({ "n": n * 2 }))
}

fn app() -> Router {
    Router::new()
        .route("/health", get(|| async { "ok" }))
        .route("/double", post(double))
}

fn json_request(body: &str) -> Request<Body> {
    Request::builder()
        .method("POST")
        .uri("/double")
        .header("content-type", "application/json")
        .body(Body::from(body.to_owned()))
        .unwrap()
}

#[tokio::test]
async fn health_returns_ok() {
    let request = Request::builder().uri("/health").body(Body::empty()).unwrap();
    let response = app().oneshot(request).await.unwrap();

    assert_eq!(response.status(), StatusCode::OK);
    let bytes = response.into_body().collect().await.unwrap().to_bytes();
    assert_eq!(&bytes[..], b"ok");
}

#[tokio::test]
async fn double_returns_json() {
    let response = app().oneshot(json_request(r#"{"n": 4}"#)).await.unwrap();

    assert_eq!(response.status(), StatusCode::OK);
    let bytes = response.into_body().collect().await.unwrap().to_bytes();
    let value: Value = serde_json::from_slice(&bytes).unwrap();
    assert_eq!(value["n"], 8);
}

#[tokio::test]
async fn wrong_content_type_is_415() {
    let request = Request::builder()
        .method("POST")
        .uri("/double")
        .body(Body::from(r#"{"n": 4}"#))
        .unwrap();
    let response = app().oneshot(request).await.unwrap();
    assert_eq!(response.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE);
}

#[tokio::test]
async fn malformed_json_is_400() {
    let response = app().oneshot(json_request("{not json")).await.unwrap();
    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
}

tests/orders.rs

Rust

// tests/orders.rs
mod common;

use axum::http::StatusCode;
use common::{post_order, TestApp};
use serde_json::json;
use sqlx::PgPool;

#[sqlx::test]
async fn paid_order_is_stored_and_charged(pool: PgPool) {
    let app = TestApp::new(pool).await;

    let request = post_order(Some(7), json!({ "product_id": app.product_id, "quantity": 2 }));
    let (status, body) = app.send(request).await;

    assert_eq!(status, StatusCode::CREATED);
    assert_eq!(body["status"], "paid");
    assert_eq!(body["total_cents"], 500);

    let order_id = body["id"].as_i64().unwrap();
    assert_eq!(app.only_order().await, (order_id, "paid".to_owned()));
    assert_eq!(app.gateway.charges(), vec![(order_id, 500)]);
}

#[sqlx::test]
async fn declined_payment_cancels_the_order(pool: PgPool) {
    let app = TestApp::new(pool).await;
    app.gateway.decline_next();

    let request = post_order(Some(7), json!({ "product_id": app.product_id, "quantity": 1 }));
    let (status, body) = app.send(request).await;

    assert_eq!(status, StatusCode::PAYMENT_REQUIRED);
    assert_eq!(body["error"], "payment declined");
    let (_, stored_status) = app.only_order().await;
    assert_eq!(stored_status, "cancelled");
    assert!(app.gateway.charges().is_empty());
}

#[sqlx::test]
async fn unknown_product_creates_nothing(pool: PgPool) {
    let app = TestApp::new(pool).await;

    let request = post_order(Some(7), json!({ "product_id": 999, "quantity": 1 }));
    let (status, _) = app.send(request).await;

    assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
    assert_eq!(app.order_count().await, 0);
}

tests/access.rs

Rust

// tests/access.rs
mod common;

use axum::http::StatusCode;
use common::{get_order, post_order, TestApp};
use serde_json::json;
use sqlx::PgPool;

#[sqlx::test]
async fn request_without_credentials_is_401(pool: PgPool) {
    let app = TestApp::new(pool).await;
    let order_id = app.seed_order(7).await;

    let (status, _) = app.send(get_order(None, order_id)).await;
    assert_eq!(status, StatusCode::UNAUTHORIZED);

    let request = post_order(None, json!({ "product_id": app.product_id, "quantity": 1 }));
    let (status, _) = app.send(request).await;
    assert_eq!(status, StatusCode::UNAUTHORIZED);
}

#[sqlx::test]
async fn customer_cannot_read_another_customers_order(pool: PgPool) {
    let app = TestApp::new(pool).await;
    let order_id = app.seed_order(7).await;

    let (status, body) = app.send(get_order(Some(8), order_id)).await;
    assert_eq!(status, StatusCode::NOT_FOUND);
    assert_eq!(body["error"], "not found");

    let (status, body) = app.send(get_order(Some(7), order_id)).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["id"], order_id);
}

#[sqlx::test]
async fn zero_quantity_is_rejected(pool: PgPool) {
    let app = TestApp::new(pool).await;

    let request = post_order(Some(7), json!({ "product_id": app.product_id, "quantity": 0 }));
    let (status, body) = app.send(request).await;

    assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
    assert_eq!(body["error"], "quantity must be positive");
}

tests/contract.rs

Rust

// tests/contract.rs
mod common;

use axum::http::StatusCode;
use common::{get_order, TestApp};
use sqlx::PgPool;

#[sqlx::test]
async fn order_response_shape_is_stable(pool: PgPool) {
    let app = TestApp::new(pool).await;
    let order_id = app.seed_order(7).await;

    let (status, body) = app.send(get_order(Some(7), order_id)).await;
    assert_eq!(status, StatusCode::OK);

    let mut keys: Vec<&str> = body.as_object().unwrap().keys().map(String::as_str).collect();
    keys.sort();
    assert_eq!(keys, ["customer_id", "id", "status", "total_cents"]);
    assert!(body["id"].is_i64());
    assert!(body["customer_id"].is_i64());
    assert!(body["status"].is_string());
    assert!(body["total_cents"].is_i64());
}

#[sqlx::test]
async fn error_response_shape_is_stable(pool: PgPool) {
    let app = TestApp::new(pool).await;

    let (status, body) = app.send(get_order(Some(7), 12345)).await;
    assert_eq!(status, StatusCode::NOT_FOUND);

    let keys: Vec<&str> = body.as_object().unwrap().keys().map(String::as_str).collect();
    assert_eq!(keys, ["error"]);
    assert!(body["error"].is_string());
}

tests/e2e.rs

Rust

// tests/e2e.rs
mod common;

use common::TestApp;
use sqlx::PgPool;

#[sqlx::test]
async fn real_server_answers_health(pool: PgPool) {
    let app = TestApp::new(pool).await;

    // Port 0 lets the operating system choose a free port.
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let router = app.router.clone();
    tokio::spawn(async move {
        axum::serve(listener, router).await.unwrap();
    });

    let response = reqwest::get(format!("http://{addr}/health")).await.unwrap();
    assert_eq!(response.status(), 200);
    assert_eq!(response.text().await.unwrap(), "ok");
}

Start PostgreSQL and run the tests:

Terminal

docker run --name pg -e POSTGRES_PASSWORD=pw -e POSTGRES_DB=storefront -p 5432:5432 -d postgres:16
export DATABASE_URL=postgres://postgres:pw@127.0.0.1:5432/storefront

Terminal

cargo test

cargo test also prints result lines for the library and binary targets (0 tests each) and the doc-tests. The lines for the five test files are below, with the elapsed time omitted:

Output

test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out   (access)
test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out   (contract)
test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out   (e2e)
test result: ok. 4 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out   (oneshot_basics)
test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out   (orders)

The service also runs on its own. Start it in one terminal. It applies the migration at startup.

Terminal

cargo run

In a second terminal, insert a product and send requests:

Terminal

docker exec pg psql -U postgres -d storefront \
  -c "INSERT INTO products (sku, name, price_cents) VALUES ('PEN-1', 'Pen', 250)"

req() { curl -s -w ' [%{http_code}]\n' "$@"; }
J='content-type: application/json'

req localhost:3000/health
req -X POST localhost:3000/orders -H "$J" -H 'x-customer-id: 7' -d '{"product_id":1,"quantity":2}'
req localhost:3000/orders/1 -H 'x-customer-id: 7'
req localhost:3000/orders/1 -H 'x-customer-id: 8'
req localhost:3000/orders/1

Output

ok [200]
{"id":1,"customer_id":7,"status":"paid","total_cents":500} [201]
{"id":1,"customer_id":7,"status":"paid","total_cents":500} [200]
{"error":"not found"} [404]
{"error":"unauthorized"} [401]

Summary

Router implements tower::Service, so ServiceExt::oneshot tests a handler without networking.
Read response bodies with http_body_util::BodyExt::collect.
Build the router in a function that takes its dependencies. Put a trait only at a remote seam.
#[sqlx::test] creates a database per test, applies migrations, and passes a PgPool.
Run integration tests on the production database engine and major version.
A fake replaces a remote dependency and exposes switches that cause failures.
Test the three authorization cases: no credentials, bad credentials, and the wrong owner.
Contract tests pin JSON shapes and compare against serde_json::Value.
Bind to 127.0.0.1:0 for real-server tests and read the address from the listener.

Project setup

Example 1: Test a handler with oneshot

Example 2: The service under test

Example 3: Integration tests and a fake gateway

Example 4: Authorization and error paths

Example 5: Contract tests

Example 6: A real server on a free port

Complete example

Summary

Example 1: Test a handler with `oneshot`