Skip to main content

Testing a Redis service

A test runs the service's own app with TestApp, against an in-process Redis or against a server.

The service under test charges card payments. A payment arrives on the stream card-payments, and the handler replies with a receipt on card-receipts. A payment without an amount is dropped. Each account may spend 1,000.00 a day: the handler adds the amount to the key spent:<account> through the broker's connection pool, and holds a payment over the limit for a minute with retry_after. After three deliveries a payment goes to card-payments.review, the dead-letter stream. A refund arrives on the reliable list card-refunds and gives the amount back. The app is built by service(url), so a test chooses the server and main uses the production one. The handler calls the pool through fred's command traits, so the service lists fred among its dependencies.

The tests need the testing feature of ruststream-fred and a Tokio runtime:

Cargo.toml
[dev-dependencies]
ruststream-fred = { version = "0.7", features = ["testing"] }
tokio = { version = "1", features = ["macros", "rt", "rt-multi-thread"] }

A payment and its receipt​

src/main.rs
/// The body both modes run; only the start call differs.
async fn a_payment_is_answered_with_a_receipt(
    tb: TestApp<()>,
    account: &str,
) -> Result<(), Box<dyn Error>> {
    tb.broker::<RedisBroker>()
        .message(&payment(1, account, 2500))
        .publish()
        .await?;

    tb.broker::<RedisBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::ack());
    tb.broker::<RedisBroker>()
        .published::<Receipt>("card-receipts")
        .assert_called_once()
        .with(&Receipt { payment_id: 1 });
    tb.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    a_payment_is_answered_with_a_receipt(TestApp::start(app()).await?, "acct-1").await
}

TestApp::start connects RedisBroker to a Redis modelled inside the test process. The URL is never dialled. The test addresses the broker by its production type, tb.broker::<RedisBroker>(). A test's message is written to its name the way the subscription reading that name expects it: LPUSH for a list, PUBLISH for a channel or a pattern, XADD for a stream or a name no subscription reads. publish returns once the handler and its reply have settled. So the next lines assert on finished state: how the subscriber settled, and what reached card-receipts. The body takes the started app as an argument, so the live test below runs it too.

A payment without an amount​

src/main.rs
#[tokio::test]
async fn a_payment_without_an_amount_is_dropped() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;

    tb.broker::<RedisBroker>()
        .message(&payment(2, "acct-2", 0))
        .publish()
        .await?;

    tb.broker::<RedisBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::drop());
    tb.broker::<RedisBroker>()
        .published::<Receipt>("card-receipts")
        .assert_not_called();
    tb.shutdown().await?;
    Ok(())
}

settled checks the outcome the handler returned. assert_not_called checks that no receipt was published.

The handler's own commands, and a delay​

src/main.rs
#[tokio::test(start_paused = true)]
async fn a_held_payment_passes_once_a_refund_frees_room() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let redis = tb.broker::<RedisBroker>();

    redis
        .message(&payment(3, "acct-3", 80_000))
        .publish()
        .await?;
    redis
        .message(&payment(4, "acct-3", 30_000))
        .publish()
        .await?;
    redis
        .subscriber("card-payments")
        .assert_called(2)
        .settled(HandlerOutcome::retry_after(HOLD));

    // The refund is a list entry; the handler gives the room back through the pool.
    redis
        .message(&Refund {
            account: "acct-3".to_owned(),
            amount_cents: 50_000,
        })
        .publish()
        .await?;
    tb.advance(HOLD).await?;

    assert_eq!(
        redis.subscriber("card-payments").outcomes(),
        [Outcome::Ack, Outcome::Nack, Outcome::Ack],
    );
    redis
        .published::<Receipt>("card-receipts")
        .assert_called(2)
        .with(&Receipt { payment_id: 4 });
    tb.shutdown().await?;
    Ok(())
}

The handler's INCRBY and DECRBY run against the in-process Redis, as every command of the crate does. The second payment takes the account over its limit, so it is held. The refund is a list entry, and its handler gives the room back.

The test starts on a paused clock, and advance moves it forward. A minute of HOLD costs nothing. The in-process Redis reads the test's clock too, so entry ids, idle times and the scores of a delay set stand still until the clock moves. outcomes lists how each delivery settled, in order; the held one is recorded as Outcome::Nack.

The cap​

src/main.rs
#[tokio::test(start_paused = true)]
async fn a_payment_held_three_times_goes_to_review() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;

    tb.broker::<RedisBroker>()
        .message(&payment(5, "acct-5", 150_000))
        .publish()
        .await?;
    tb.advance(HOLD).await?;
    tb.advance(HOLD).await?;

    tb.broker::<RedisBroker>()
        .subscriber("card-payments")
        .assert_called(3);
    tb.broker::<RedisBroker>()
        .published::<Payment>("card-payments.review")
        .assert_called_once()
        .with(&payment(5, "acct-5", 150_000))
        .with_header(RETRY_COUNT_HEADER, "3");
    tb.shutdown().await?;
    Ok(())
}

A payment above the daily limit is held on every delivery. Each hold publishes a copy back to the stream with the retry count raised, and published reads every write to a name back, those copies included. The third delivery is the last one the mount site allows, so the payment goes to card-payments.review as it arrived.

What the in-process Redis models​

The in-process Redis takes the commands the crate sends and answers them as Redis 8.4 does, so a claiming subscription mounts. It reads every setting off the broker: the URL, the pool size, the default group and a cluster topology. It keeps:

  • key types, so a write of the wrong type fails with WRONGTYPE;
  • consumer groups with their cursor, pending entries and delivery counts, XAUTOCLAIM, XREADGROUP .. CLAIM and a seek;
  • lists, with a reliable list's processing list and its recovery set;
  • Pub/Sub channels, patterns and sharded channels, which keep nothing for a subscriber that is not there;
  • sorted sets, key expiry and MULTI / EXEC;
  • on a cluster topology, the refusal of a command or a transaction that spans hash slots.

A handler's own commands, through the pool or a pipelined round, run against it too. It runs no scripts: EVAL, EVALSHA, FCALL and a command it does not know fail with an error naming them. No handshake runs, so authentication and TLS are not exercised, and neither is a failover or a resharding. Those need a server.

The same test against a server​

src/main.rs
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_a_server() -> Result<(), Box<dyn Error>> {
    let Some(url) = live::url("REDIS_TEST_URL") else {
        return Ok(());
    };
    // A server keeps what earlier runs left, so each run spends from an account of its own.
    let run = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos();
    let account = format!("acct-{run}");
    a_payment_is_answered_with_a_receipt(TestApp::start_live(service(&url)).await?, &account)
        .await
}

TestApp::start_live connects the same app to a running Redis. A publish then waits on the real clock until every subscription has handled it, ten seconds at most. The test reads the address from REDIS_TEST_URL with live::url, a helper of the example's own, and skips when it is unset. A server keeps what earlier runs left, so the test spends from an account of its own. The assertions are the in-process ones, unchanged.

cargo test
docker run --name redis -p 6379:6379 redis:8.4
REDIS_TEST_URL=redis://localhost:6379 cargo test -- --test-threads=1