Testing¶

The testing feature ships RedisTestBroker, an in-process transport that routes by exact stream key with no server. Its connected form implements ruststream::testing::TestableBroker, so the same transport drives the TestApp harness and the conformance suite. It reproduces routing, ack/nack, and headers. 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.6", 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 hands it to TestApp. Publishing through the harness handle drives the reaction to quiescence, so the assertions need no waiting.

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 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(RedisTestBroker::new(), |b| {
        b.include(process_payment);
    });

// The harness runs the app's real startup and drives every publish to quiescence, so the
// assertions below need no waiting.
let tb = TestApp::start(app).await?;

// The valid payment is saved; the invalid one (amount == 0) is dropped.
tb.broker::<RedisTestBroker>()
    .publish("payments", &payment(1, 100))
    .await?;
tb.broker::<RedisTestBroker>()
    .publish("payments", &payment(2, 0))
    .await?;

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);

tb.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 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(RedisTestBroker::new(), |b| {
                b.include(process_payment);
            });

        let tb = TestApp::start(app).await.expect("startup failed");

        tb.broker::<RedisTestBroker>()
            .publish("payments", &payment(1, 100))
            .await
            .expect("publish valid");
        tb.broker::<RedisTestBroker>()
            .publish("payments", &payment(2, 0))
            .await
            .expect("publish invalid");

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

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

Transport-specific examples¶

Redis StreamRedis ListPub/Sub

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

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

let tb = TestApp::start(app).await?;
tb.broker::<RedisTestBroker>()
    .publish("events", &payment(1, 100))
    .await?;

tb.broker::<RedisTestBroker>()
    .subscriber("events")
    .assert_called_once()
    .settled(HandlerResult::Ack);

tb.shutdown().await?;

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

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

let tb = TestApp::start(app).await?;
tb.broker::<RedisTestBroker>()
    .publish("jobs", &payment(1, 100))
    .await?;

tb.broker::<RedisTestBroker>()
    .subscriber("jobs")
    .assert_called_once()
    .settled(HandlerResult::Ack);

tb.shutdown().await?;

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

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

let tb = TestApp::start(app).await?;
tb.broker::<RedisTestBroker>()
    .publish("notifications", &payment(1, 100))
    .await?;

tb.broker::<RedisTestBroker>()
    .subscriber("notifications")
    .assert_called_once()
    .settled(HandlerResult::Ack);

tb.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;