Skip to main content

Testing

TestApp::start runs the service against a real database, SQLite in memory for a start, with the harness waiting on every publish and a pausable clock.

A test hands the harness the app main runs, unchanged, and addresses the broker by its production type, tb.broker::<SqlxBroker<Sqlite>>(). The testing feature in [dev-dependencies] adds the in-process mode and turns on the core's harness with it. The crate registers SqlxBroker<Postgres>, SqlxBroker<MySql>, SqlxBroker<Sqlite> and SqlxBroker<Any>; a broker on a dialect of the service's own is registered by the service.

The in-process mode connects to the database the test's pool reaches, as connect does, because the service's SQL is part of what the test checks. SQLite needs no server: a database in memory named in a shared-cache URL serves every connection that names it, and lives while one of them is open, so the test holds one beside the pool. A test on Postgres, MySQL or MariaDB takes a database of its own on a server, with the service's migrations applied.

The harness counts what reaches a queue and waits for it. A test's input goes in through the route and the service's Publish, as in production, and so do the service's own replies and slots. A publish through the harness returns once every handler it woke has settled, so the assertions that follow read finished state. A row is counted for the queue it lands in, its table and group, whatever name the subscription reads it by. A message published before its subscription opened waits in the table and is delivered once it opens. A row the service wrote with its own SQL before the test, a seed or a migration, is counted when its subscription opens, so settle waits for it as for a published message. A row written with the service's own SQL while the test runs is counted only once a claim takes it, so settle does not wait for it: such a test moves the clock past the poll interval with tb.advance.

The queue reads "now" from a clock the test controls. After tokio::time::pause(), tb.advance(d) moves the queue's time by d, and a lease runs out or a delayed retry comes back when the clock passes it. The database calls run off that clock, so a paused clock does not jump while the database answers. 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. Such a test, and one that needs what only a server has, runs live.

A test build leaves the outbox off unless the switch turns it on.

Example​

The dependency line:

[dev-dependencies]
ruststream-sqlx = { version = "0.7", features = ["inbox", "sqlite", "chrono", "testing"] }

The first test publishes an image and reads how the handler settled it: the first attempt puts it off for a minute.

examples/sqlx_testing.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.
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?;
    sqlx::raw_sql(SCHEMA).execute(&pool).await?;
    Ok((url, pool, keeper))
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_image_is_put_off_once() -> Result<(), Box<dyn Error>> {
    let (url, _pool, _keeper) = database("sqlx_testing").await?;
    let tb = TestApp::start(service(&url)).await?;
    let image = Image {
        path: "cat.png".to_owned(),
    };
    // Through the route and the service's own `Publish`, as in production.
    tb.broker::<SqlxBroker<Sqlite>>()
        .message(&image)
        .publish()
        .await?;

    tb.broker::<SqlxBroker<Sqlite>>()
        .subscriber("thumbnails")
        .assert_called_once()
        .with(&image)
        .settled(HandlerOutcome::retry_after(Duration::from_secs(60)));
    Ok(())
}

The second pauses the clock and moves it past the minute: the row comes back, and the second attempt makes the thumbnail.

examples/sqlx_testing.rs
#[tokio::test]
async fn a_minute_later_the_thumbnail_is_made() -> Result<(), Box<dyn Error>> {
    let (url, _pool, _keeper) = database("sqlx_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: "cat.png".to_owned(),
        })
        .publish()
        .await?;
    tb.advance(Duration::from_secs(61)).await?;

    // The delayed retry, then the acknowledgement once its minute had passed.
    let outcomes = tb
        .broker::<SqlxBroker<Sqlite>>()
        .subscriber("thumbnails")
        .assert_called(2)
        .outcomes();
    assert_eq!(outcomes, [Outcome::Nack, Outcome::Ack]);
    Ok(())
}