Skip to content

Testing

The testing feature ships RedisTestBroker, an in-process transport that routes by exact stream key with no server and implements ruststream::testing::TestableBroker, so the same type drives the TestApp harness and the conformance suite. It reproduces routing, ack/nack, and headers, and passes the framework's conformance suite. It does not simulate consumer-group cursors, XAUTOCLAIM redelivery, trimming, or dead-letter routing - exercise those against a real Redis server (see the crate's integration_fred tests and docker-compose.test.yml).

[dev-dependencies]
ruststream-fred = { version = "0.5", features = ["testing"] }

Unit-testing a handler

Because a #[subscriber] handler is wired through a RustStream app, the most realistic in-process test builds the same app around a RedisTestBroker and drives publishes by injecting messages onto the broker's bus. The service runs until the test signals shutdown.

Business-logic test

A real handler validates input, persists valid messages through a repository connector, and drops invalid ones. The handler has no knowledge of the test harness.

crates/ruststream-fred/examples/fred_testing.rs
/// A repository connector. In production this would wrap a real database client;
/// the test uses the same connector with an in-memory store so the handler stays test-agnostic.
#[derive(Clone, Default)]
struct PaymentRepository {
    payments: Arc<Mutex<Vec<Payment>>>,
}

impl PaymentRepository {
    async fn save(&self, payment: Payment) {
        self.payments.lock().await.push(payment);
    }

    async fn count(&self) -> usize {
        self.payments.lock().await.len()
    }

    async fn contains(&self, id: u64) -> bool {
        self.payments.lock().await.iter().any(|p| p.id == id)
    }
}
crates/ruststream-fred/examples/fred_testing.rs
/// A real production handler: validate the message, persist it, or drop it on validation failure.
#[subscriber(
    RedisStream::new("payments")
        .group("workers")
)]
async fn process_payment(
    payment: &Payment,
    ctx: &mut Context<'_, (), PaymentRepository>,
) -> HandlerResult {
    if payment.amount == 0 {
        // Invalid message: do not requeue, drop it.
        return HandlerResult::drop();
    }

    // The handler names its app state as the third `Context` generic; `ctx.state()` borrows the
    // typed `PaymentRepository` directly, with no lookup or downcast.
    ctx.state().save(payment.clone()).await;

    HandlerResult::ack()
}

The test publishes a valid payment and an invalid payment, then asserts that only the valid one was saved:

crates/ruststream-fred/examples/fred_testing.rs
let broker = RedisTestBroker::new();
let repository = PaymentRepository::default();
let repository_for_app = repository.clone();

let app = RustStream::new(AppInfo::new("test", "0.1.0"))
    // The startup hook produces the typed app state; the test keeps its own clone (the inner
    // store is shared via `Arc`) to assert on it afterwards.
    .on_startup(move |()| async move { Ok::<_, std::convert::Infallible>(repository_for_app) })
    .with_broker(broker.clone(), |b| {
        b.include(process_payment);
    });

// `start` resolves once subscriptions are open, so the injects below cannot race startup.
let running = app.start().await?;

// Valid payment is saved.
broker.inject(OutgoingMessage::new(
    "payments",
    br#"{"id":1,"user_id":42,"amount":100}"#,
));
// Invalid payment (amount == 0) is dropped.
broker.inject(OutgoingMessage::new(
    "payments",
    br#"{"id":2,"user_id":42,"amount":0}"#,
));

// Wait until the valid payment is persisted: the handler runs on another task, so
// yielding between checks is enough to let it progress.
tokio::time::timeout(Duration::from_secs(2), async {
    while !repository.contains(1).await {
        tokio::task::yield_now().await;
    }
})
.await
.expect("valid payment was not saved in time");

assert!(repository.contains(1).await, "valid payment was not saved");
assert!(
    !repository.contains(2).await,
    "invalid payment should have been dropped"
);
assert_eq!(repository.count().await, 1);

running.shutdown().await?;

In your own crate you usually copy the test body into a #[tokio::test] inside a #[cfg(test)] module:

crates/ruststream-fred/examples/fred_testing.rs
#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn valid_payment_is_saved_and_invalid_is_dropped() {
        let broker = RedisTestBroker::new();
        let repository = PaymentRepository::default();
        let repository_for_app = repository.clone();

        let app = RustStream::new(AppInfo::new("test", "0.1.0"))
            .on_startup(
                move |()| async move { Ok::<_, std::convert::Infallible>(repository_for_app) },
            )
            .with_broker(broker.clone(), |b| {
                b.include(process_payment);
            });

        // `start` resolves once subscriptions are open, so the injects below cannot race startup.
        let running = app.start().await.expect("startup failed");

        broker.inject(OutgoingMessage::new(
            "payments",
            br#"{"id":1,"user_id":42,"amount":100}"#,
        ));
        broker.inject(OutgoingMessage::new(
            "payments",
            br#"{"id":2,"user_id":42,"amount":0}"#,
        ));

        tokio::time::timeout(Duration::from_secs(2), async {
            while !repository.contains(1).await {
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("valid payment was not saved in time");

        assert!(repository.contains(1).await);
        assert!(!repository.contains(2).await);
        assert_eq!(repository.count().await, 1);

        running.shutdown().await.expect("graceful shutdown failed");
    }
}

Transport-specific examples

#[subscriber(
    RedisStream::new("events")
        .group("workers")
)]
async fn handle_stream_event(payment: &Payment) -> HandlerResult {
    println!("stream event {}", payment.id);
    HandlerResult::Ack
}

let broker = RedisTestBroker::new();

let app = RustStream::new(AppInfo::new("test", "0.1.0")).with_broker(broker.clone(), |b| {
    b.include(handle_stream_event);
});

// `start` resolves once subscriptions are open, so the injects below cannot race startup.
let running = app.start().await?;
broker.inject(OutgoingMessage::new(
    "events",
    br#"{"id":1,"user_id":42,"amount":100}"#,
));
running.shutdown().await?;

#[subscriber(
    RedisList::new("jobs")
        .reliable()
)]
async fn handle_list_job(payment: &Payment) -> HandlerResult {
    println!("list job {}", payment.id);
    HandlerResult::Ack
}

let broker = RedisTestBroker::new();

let app = RustStream::new(AppInfo::new("test", "0.1.0")).with_broker(broker.clone(), |b| {
    b.include(handle_list_job);
});

// `start` resolves once subscriptions are open, so the injects below cannot race startup.
let running = app.start().await?;
broker.inject(OutgoingMessage::new(
    "jobs",
    br#"{"id":1,"user_id":42,"amount":100}"#,
));
running.shutdown().await?;

#[subscriber(RedisPubSub::new("notifications"))]
async fn handle_pubsub_notification(payment: &Payment) -> HandlerResult {
    println!("pubsub notification {}", payment.id);
    HandlerResult::Ack
}

let broker = RedisTestBroker::new();

let app = RustStream::new(AppInfo::new("test", "0.1.0")).with_broker(broker.clone(), |b| {
    b.include(handle_pubsub_notification);
});

// `start` resolves once subscriptions are open, so the injects below cannot race startup.
let running = app.start().await?;
broker.inject(OutgoingMessage::new(
    "notifications",
    br#"{"id":1,"user_id":42,"amount":100}"#,
));
running.shutdown().await?;

Conformance suite

Run the framework's full conformance suite against the stub broker:

crates/ruststream-fred/examples/fred_testing.rs
// The framework's conformance suite exercises routing, ack/nack, headers,
// and requeue against the in-process test broker - no Redis server required.
harness::run_suite(RedisTestBroker::new).await;