Live

TestApp::start_live runs the same test body against a Redis server, for scripts, failover, authentication and TLS.

It connects the broker through its ordinary connect, so the client dials the server, logs in and runs the handshake. 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. The wait lasts ten seconds at most, and start_live_within sets another bound.

The live mode is for what the in-process Redis leaves out: Lua scripts and functions, a cluster under resharding, a sentinel failover, authentication and TLS. A server keeps what earlier runs left in it, so a run reads keys no other test writes.

In the example, live::url returns the server's address from REDIS_TEST_URL, and the test returns early without one, so cargo test passes on a machine with no server. A job that stood a server up sets RUSTSTREAM_REQUIRE_LIVE=1, and a missing address then fails the test instead of skipping it. The repository's stand is docker-compose.test.yml, which just brokers-up starts. just test-brokers starts it, runs every suite against it with RUSTSTREAM_REQUIRE_LIVE=1, and stops it. Each topology has a variable of its own:

Variable	Server	Stand address
`REDIS_TEST_URL`	a standalone Redis 7	`redis://127.0.0.1:6379`
`REDIS_84_TEST_URL`	a standalone Redis 8.4, for `claiming`	`redis://127.0.0.1:6386`
`REDIS_AUTH_TEST_URL`	a password-protected Redis 7 with an ACL user	`redis://127.0.0.1:6385`
`REDIS_CLUSTER_TEST_URL`	a seed node of a cluster	`127.0.0.1:7000`
`REDIS_SENTINEL_TEST_URL`	a sentinel watching the primary `mymaster`	`127.0.0.1:26379`

Example

The two tests share one body: one in process, one against the server at REDIS_TEST_URL.

Macros
Manual

examples/redis_testing_live.rs
/// The body both modes run; only the start call differs.
async fn accepts_a_payment(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<RedisBroker>()
        .message(&Payment { amount: 100 })
        .publish()
        .await?;

    tb.broker::<RedisBroker>()
        .subscriber("testing-live.payments")
        .assert_called_once()
        .with(&Payment { amount: 100 })
        .settled(HandlerOutcome::ack());
    tb.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    accepts_a_payment(TestApp::start(app("redis://localhost:6379")).await?).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_the_server() -> Result<(), Box<dyn Error>> {
    let Some(url) = live::url("REDIS_TEST_URL") else {
        return Ok(());
    };
    accepts_a_payment(TestApp::start_live(app(&url)).await?).await
}

examples/manual/redis_testing_live.rs
/// The body both modes run; only the start call differs.
async fn accepts_a_payment(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<RedisBroker>()
        .message(&Payment { amount: 100 })
        .publish()
        .await?;

    tb.broker::<RedisBroker>()
        .subscriber("testing-live.payments")
        .assert_called_once()
        .with(&Payment { amount: 100 })
        .settled(HandlerOutcome::ack());
    tb.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    accepts_a_payment(TestApp::start(app("redis://localhost:6379")).await?).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_the_server() -> Result<(), Box<dyn Error>> {
    let Some(url) = live::url("REDIS_TEST_URL") else {
        return Ok(());
    };
    accepts_a_payment(TestApp::start_live(app(&url)).await?).await
}

Example​

Example