Skip to main content

Testing a service on SQL

A test runs the service's own app with TestApp against a real database: SQLite in memory, or a database of the test's own on a server.

A queue's behaviour lives in its SQL: the claim, the settlement, the service's own Publish. So the harness has no model of a database. Its in-process mode runs the service's statements against the database the test hands the service, and keeps the books the harness waits on.

The service under test makes thumbnails. A small image is made at once. The resizer takes a large image only on a second try, so the first delivery of a large image is put off for a minute with retry_after. The app is built by service(url), so a test chooses the database and main uses the production one. The tests need the testing feature of ruststream-sqlx in [dev-dependencies].

A small image​

src/main.rs
/// A database in memory that every connection naming it shares, alive while one of them is
/// open: the test holds that one beside the pool it reads the table through. The service
/// creates the table when it starts.
async fn database(
    name: &str,
) -> Result<(String, SqlitePool, SqliteConnection), Box<dyn Error>> {
    let url = format!("sqlite:file:{name}?mode=memory&cache=shared");
    let options: SqliteConnectOptions = url.parse()?;
    let keeper = SqliteConnection::connect_with(&options).await?;
    let pool = SqlitePoolOptions::new()
        .min_connections(1)
        .idle_timeout(None)
        .connect_with(options)
        .await?;
    Ok((url, pool, keeper))
}

/// The scenario, on whichever database the app runs.
async fn a_small_image_is_made_at_once(broker: BrokerHandle<'_>) -> Result<(), Box<dyn Error>> {
    let image = Image {
        path: "cat.png".to_owned(),
        large: false,
    };
    broker.message(&image).publish().await?;

    broker
        .subscriber("thumbnails")
        .assert_called_once()
        .with(&image)
        .settled(HandlerOutcome::ack());
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn in_process() -> Result<(), Box<dyn Error>> {
    let (url, pool, _keeper) = database("guide_testing").await?;
    let tb = TestApp::start(service(&url)).await?;
    a_small_image_is_made_at_once(tb.broker::<SqlxBroker<Sqlite>>()).await?;

    let left: i64 = sqlx::query_scalar("SELECT count(*) FROM thumbnail_jobs")
        .fetch_one(&pool)
        .await?;
    assert_eq!(left, 0, "the acknowledgement deleted the row");
    Ok(())
}

database creates a SQLite database in memory that every connection naming it shares. It lives while one of those connections is open, so the test holds one beside the pool it reads the table through. The service creates the table when it starts, as in production. TestApp::start connects the production app to it in process, and the test addresses the broker by its production type, tb.broker::<SqlxBroker<Sqlite>>().

The image goes in through the route and the service's own Publish, as in production. publish returns once the handler it woke has settled, so the lines after it read finished state: how thumbnail settled the row, and what the test's own query finds in the table. A row written with the service's own SQL before the app starts, a seed or a migration, is counted when its subscription opens, so settle waits for it as for a published message. A row written that way while the test runs is counted only once a claim takes it, so settle does not wait for it; such a test runs on the real clock and lets the poll interval pass, as Enqueueing jobs does. The scenario takes the broker's handle as an argument, so the Postgres test below runs it too.

A table without a payload column hands its handler the row itself (Rows). A test of such a handler checks the row with with_value and received_values in place of with and received.

A delay, without waiting for it​

src/main.rs
#[tokio::test]
async fn a_large_image_is_made_a_minute_later() -> Result<(), Box<dyn Error>> {
    let (url, _pool, _keeper) = database("guide_testing_clock").await?;
    // The queue reads "now" from this clock, and the database calls run off it. A paused
    // clock takes the single-threaded runtime `#[tokio::test]` starts.
    tokio::time::pause();
    let tb = TestApp::start(service(&url)).await?;
    tb.broker::<SqlxBroker<Sqlite>>()
        .message(&Image {
            path: "panorama.png".to_owned(),
            large: true,
        })
        .publish()
        .await?;
    tb.advance(Duration::from_secs(61)).await?;

    // Put off for a minute, then made.
    let outcomes = tb
        .broker::<SqlxBroker<Sqlite>>()
        .subscriber("thumbnails")
        .assert_called(2)
        .outcomes();
    assert_eq!(outcomes, [Outcome::Nack, Outcome::Ack]);
    Ok(())
}

In process the queue reads "now" from the tokio clock, anchored to the system clock when the broker connected. The test pauses that clock, and advance moves it: the large image's row comes due a minute later, and the second delivery makes the thumbnail. The minute costs nothing. The database calls run off the paused clock, so it does not jump while the database answers. A lease runs out the same way when the clock passes it. A table on the database's clock reads the database's time, which the test clock does not move, and a clock of the service's own stays in force; tests of those run on the real clock.

The same test on Postgres​

src/main.rs
// Skips without `POSTGRES_TEST_URL`; fails the skip under `RUSTSTREAM_REQUIRE_LIVE`.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn on_postgres() -> Result<(), Box<dyn Error>> {
    // A database of the test's own on the server, dropped at the end.
    let Some(db) = live::postgres::database().await else {
        return Ok(());
    };
    let tb = TestApp::start_live(service_on_postgres(&db.url())).await?;
    a_small_image_is_made_at_once(tb.broker::<SqlxBroker<Postgres>>()).await?;

    tb.shutdown().await?;
    db.finish().await;
    Ok(())
}

A service that runs on Postgres in production is tested on Postgres too: SQLite runs other SQL, and has no row locks, no notifications and no isolation levels to choose. TestApp::start_live connects the app the ordinary way and waits on the real clock. The test takes a database of its own on the server, where the service creates its table, so nothing an earlier run left is read; live::postgres::database is a helper of the repository's tests, and #[sqlx::test] gives a test the same, a database with the migrations applied. The test reads the server's address from POSTGRES_TEST_URL and skips without it, so cargo test passes on a machine without a server; RUSTSTREAM_REQUIRE_LIVE=1 turns that skip into a failure where the server must be. The scenario is the in-process one, unchanged.

cargo test
POSTGRES_TEST_URL=postgres://user:password@localhost:5432/postgres cargo test

The outbox in tests​

A test build leaves the outbox off unless RUSTSTREAM_SQLX_OUTBOX=on turns it on, so the service's own tests run its handlers untracked. A test of the tracking itself runs with the switch set and skips without it (Switch).