Live

TestApp::start_live runs the same test body on the real clock, against a database of the test's own on Postgres, MySQL or MariaDB.

It connects the broker through its ordinary connect, and the queue reads the clock production reads. The body is the one an in-process test runs; only the start call differs. A publish, settle and advance wait on the real clock until every subscription has handled what was published to it, so tb.advance(d) lets d of real time pass. start_live refuses a paused clock.

The live mode is for what the in-process mode leaves to the clock and to the server: a table on the database's clock, a clock of the service's own, the row lock form, isolation levels, listen_notify, and the MySQL and MariaDB specifics. Each test takes a database of its own on the stand and applies the service's migrations, so nothing an earlier run left is read.

The repository's live tests read the stand's address from POSTGRES_TEST_URL, MYSQL_TEST_URL or MARIADB_TEST_URL, and return early without one, so cargo test passes on a machine with no stand. A job that stood the stand up sets RUSTSTREAM_REQUIRE_LIVE=1, and a missing address then fails the test instead of skipping it. The stand is docker-compose.test.yml, which just brokers-up starts. The examples of the Postgres and MySQL pages carry such tests.

Example

The two tests share one body: one in process on a paused clock, one on the real clock, both over SQLite in memory.

Macros
Manual

examples/sqlx_testing_live.rs
/// The body both modes run; only the start call differs.
async fn the_thumbnail_is_made_a_second_later(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<SqlxBroker<Sqlite>>()
        .message(&Image {
            path: "cat.png".to_owned(),
        })
        .publish()
        .await?;
    // In process the paused clock jumps; live, two seconds pass.
    tb.advance(Duration::from_secs(2)).await?;

    let outcomes = tb
        .broker::<SqlxBroker<Sqlite>>()
        .subscriber("thumbnails")
        .assert_called(2)
        .outcomes();
    assert_eq!(outcomes, [Outcome::Nack, Outcome::Ack]);
    tb.shutdown().await?;
    Ok(())
}

// A paused clock takes the single-threaded runtime `#[tokio::test]` starts.
#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    let (url, _pool, _keeper) = database("sqlx_testing_live_in_process").await?;
    tokio::time::pause();
    the_thumbnail_is_made_a_second_later(TestApp::start(service(&url)).await?).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn on_the_real_clock() -> Result<(), Box<dyn Error>> {
    let (url, _pool, _keeper) = database("sqlx_testing_live_real_clock").await?;
    the_thumbnail_is_made_a_second_later(TestApp::start_live(service(&url)).await?).await
}

examples/manual/sqlx_testing_live.rs
/// The body both modes run; only the start call differs.
async fn the_thumbnail_is_made_a_second_later(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<SqlxBroker<Sqlite>>()
        .message(&Image {
            path: "cat.png".to_owned(),
        })
        .publish()
        .await?;
    // In process the paused clock jumps; live, two seconds pass.
    tb.advance(Duration::from_secs(2)).await?;

    let outcomes = tb
        .broker::<SqlxBroker<Sqlite>>()
        .subscriber("thumbnails")
        .assert_called(2)
        .outcomes();
    assert_eq!(outcomes, [Outcome::Nack, Outcome::Ack]);
    tb.shutdown().await?;
    Ok(())
}

// A paused clock takes the single-threaded runtime `#[tokio::test]` starts.
#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    let (url, _pool, _keeper) = database("manual_sqlx_testing_live_in_process").await?;
    tokio::time::pause();
    the_thumbnail_is_made_a_second_later(TestApp::start(service(&url)).await?).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn on_the_real_clock() -> Result<(), Box<dyn Error>> {
    let (url, _pool, _keeper) = database("manual_sqlx_testing_live_real_clock").await?;
    the_thumbnail_is_made_a_second_later(TestApp::start_live(service(&url)).await?).await
}

Example​

Example