Testing¶

RustStream services are tested at two levels:

In-process unit tests drive your real handlers, middleware, and codecs with the TestApp harness - no server, no docker, no network (the in-process broker's connect is I/O-free). This is the default path and it covers handler logic end to end: decode, dispatch, the outcome (ack / nack / drop / panic / decode failure), and any messages the handler publishes.
Integration tests run against a real broker, gated behind an environment variable, and cover the semantics only a real server has (durable consumers, redelivery timers, partitions).

What the harness does and does not model

The harness drives a broker's in-process transport: publishing fans a message out to the subscribers whose subject matches, runs your handler through the real dispatch path, and records the outcome and any downstream publishes. It does not model JetStream durable cursors, ack_wait redelivery, max_ack_pending, retention, Kafka offsets or consumer groups, or RabbitMQ exchanges and dead-letter routing. Those are real-broker concerns; test them in the integration suite.

MemoryBroker is a real broker (local in-process queues), not a test double - the harness drives it through the same dispatch path the production runtime uses.

Unit-testing a service with `TestApp`¶

TestApp takes a built RustStream application, connects its brokers (I/O-free for the in-process bus), mounts the handlers, and records every delivery. You publish input, and the publish drives the whole reaction (the handler, its downstream publishes, any cross-broker cascade) to a standstill before it returns - then you assert.

The handler under test (in a real service it lives in your handler module and the test imports it):

tests/doc_testing_memory.rs

use ruststream::subscriber;
use serde::{Deserialize, Serialize};

#[derive(Debug, Deserialize, Serialize, PartialEq)]
struct Order {
    id: u64,
    quantity: u32,
}

#[derive(Debug, Deserialize, Serialize, PartialEq)]
struct Confirmation {
    id: u64,
    accepted: bool,
}

#[subscriber("orders", publish("confirmations"))]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation {
        id: order.id,
        accepted: order.quantity > 0,
    }
}

The test:

tests/doc_testing_memory.rs

use ruststream::memory::{MemoryBroker, MemoryPublish};
use ruststream::runtime::{AppInfo, HandlerResult, RustStream, TypedPublisher};
use ruststream::testing::TestApp;

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn confirms_valid_orders() {
    // The app under test: production wiring, in-memory broker.
    let app = RustStream::new(AppInfo::new("orders-test", "0.0.0")).with_broker(
        MemoryBroker::new(),
        |b| {
            let replies = TypedPublisher::new(MemoryPublish);
            b.include(confirm).publisher(replies);
        },
    );

    // Start the harness (no connect, no server) and publish an order; the publish drives the
    // handler to a standstill before it returns.
    let tb = TestApp::start(app).await.expect("start harness");
    tb.broker::<MemoryBroker>()
        .publish("orders", &Order { id: 1, quantity: 2 })
        .await
        .expect("publish");

    // The handler decoded the order, acked, and published a confirmation.
    tb.broker::<MemoryBroker>()
        .subscriber("orders")
        .assert_called_once()
        .with(&Order { id: 1, quantity: 2 })
        .settled(HandlerResult::Ack);
    tb.broker::<MemoryBroker>()
        .published::<Confirmation>("confirmations")
        .assert_called_once()
        .with(&Confirmation {
            id: 1,
            accepted: true,
        });
}

This test runs in this repository's CI

The code above is embedded from tests/doc_testing_memory.rs, which cargo test --all-features runs on every change - the example cannot silently rot.

Enable the testing feature in your dev-dependencies:

[dev-dependencies]
ruststream = { version = "0.6", features = ["testing", "memory", "macros", "json"] }

Addressing brokers¶

tb.broker::<MemoryBroker>() addresses the broker by type; tb.broker_named("ingress") addresses it by the label from with_broker_labeled when a service mounts several brokers and their subjects collide. The unscoped tb.publish(name, &value) is a convenience for single-broker apps and returns TestError::Ambiguous when more than one broker is registered.

Asserting on a handler¶

tb.broker::<B>().subscriber(name) returns a fluent builder over what that handler received:

Method	Asserts
`assert_called_once()` / `assert_called(n)` / `assert_not_called()`	the delivery count
`with(&value)`	the most recent delivery decodes to `value` (with the default codec)
`with_raw(bytes)`	the most recent raw payload
`settled(HandlerResult::Ack)`	how it settled
`assert_outcome(Outcome::Drop)`	the classified outcome (ack / nack / drop / decode-failure / panic)
`panicked()`	the handler panicked on the last delivery
`assert_last_failed_to_decode()`	the payload failed to decode

tb.broker::<B>().published::<T>(name) asserts on what the handler published downstream, read from the broker's publish log: .assert_called_once().with(&Receipt { id: 1 }).

Beyond the assertions, the messages themselves are retrievable for custom checks: subscriber(name).received::<T>() / .received_raw() returns what the handler received, and published::<T>(name).decoded() / .messages() returns every message published to the channel - both in order.

The decoding helpers (with, received, decoded) use the default codec. If a handler or publisher was mounted with a different codec (include_with / with_broker_codec), pass it explicitly with the _with / with_codec variants - subscriber(name).with_codec(&CborCodec, &expected), .received_with(&CborCodec), published::<T>(name).with_codec(&CborCodec, &expected), .decoded_with(&CborCodec) - while with_raw / received_raw / messages stay codec-free.

Failure policy, panic, and shutdown¶

The harness runs dispatch under the application's real FailurePolicy, so a negative test is a first-class path. Under the default panic = fail_fast, a handler panic tears the service down just as in production:

tests/testing_harness.rs

// The panicking delivery still drives to quiescence (the message is dropped, unsettled).
tb.broker::<MemoryBroker>()
    .publish("orders", &Order { id: 0 })
    .await
    .unwrap();

tb.broker::<MemoryBroker>()
    .subscriber("orders")
    .assert_called_once()
    .panicked();
tb.assert_shut_down();
assert!(matches!(
    tb.run_result(),
    Err(ruststream::runtime::RustStreamError::Dispatch(_))
));
// A publish after the fail-fast shutdown is rejected.
assert!(matches!(
    tb.broker::<MemoryBroker>()
        .publish("orders", &Order { id: 1 })
        .await,
    Err(TestError::ShutDown)
));

Under on_failure(panic = skip) the panic is acked and consumption continues, so tb.assert_running() holds. run_result() returns what the real run would: Ok while healthy, an error once a fail-fast failure shut the service down.

Panic catching needs unwinding

The harness rides the runtime's catch_unwind, so a deliberate panic does not kill the test thread. A build compiled with panic = "abort" cannot catch handler panics.

Delayed redelivery (`retry_after`)¶

A handler that returns retry_after(delay) schedules a delayed redelivery. publish records the immediate NackAfter settlement and returns; the redelivery is driven separately by advancing a paused clock:

tests/testing_harness.rs

#[subscriber("delayed")]
async fn delayed_retry(order: &Order, ctx: &mut Context<'_, (), Counter>) -> HandlerResult {
    let _ = order;
    if ctx.state().seen.fetch_add(1, Ordering::SeqCst) == 0 {
        HandlerResult::retry_after(std::time::Duration::from_secs(30))
    } else {
        HandlerResult::Ack
    }
}

#[tokio::test(start_paused = true)]
async fn retry_after_redelivers_after_advancing_time() {
    let seen = Arc::new(AtomicU32::new(0));
    let state_seen = Arc::clone(&seen);
    let app = RustStream::new(AppInfo::new("svc", "0.1.0"))
        .on_startup(move |()| {
            let seen = state_seen;
            async move { Ok::<_, std::convert::Infallible>(Counter { seen }) }
        })
        .with_broker(MemoryBroker::new(), |b| b.include(delayed_retry));
    let tb = TestApp::start(app).await.unwrap();

    // The publish records the immediate NackAfter settlement and returns; the redelivery is pending.
    tb.publish("delayed", &Order { id: 1 }).await.unwrap();
    tb.broker::<MemoryBroker>()
        .subscriber("delayed")
        .assert_called_once()
        .settled(HandlerResult::NackAfter {
            delay: std::time::Duration::from_secs(30),
        });
    assert_eq!(seen.load(Ordering::SeqCst), 1);

    // Advancing past the delay fires the redelivery and drives it to settle.
    tb.advance(std::time::Duration::from_secs(30))
        .await
        .unwrap();
    tb.broker::<MemoryBroker>()
        .subscriber("delayed")
        .assert_called(2)
        .settled(HandlerResult::Ack);
    assert_eq!(seen.load(Ordering::SeqCst), 2);
}

Integration tests against a real broker¶

Behaviour that depends on real broker semantics belongs in a separate suite gated behind an environment variable, so the default cargo test stays fast and offline:

tests/integration_nats.rs

fn test_url() -> Option<String> {
    std::env::var("NATS_TEST_URL").ok()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn durable_consumer_resumes_after_restart() {
    let Some(url) = test_url() else {
        eprintln!("skipping: set NATS_TEST_URL to run");
        return;
    };
    // connect NatsBroker::new(url), drive the real JetStream consumer ...
}

Run it explicitly against a live server:

docker run -d -p 4222:4222 nats:latest -js
NATS_TEST_URL=nats://127.0.0.1:4222 cargo test --test integration_nats

This mirrors faststream's with_real=True split: handler logic on the in-memory path, broker semantics on the real one. Keep both suites over the same handler modules so the production code has a single source of truth.

Writing a broker crate?

The machinery that makes TestApp work against a broker - the in-process transport and the TestableBroker contract - is the broker author's side of this story. It lives in Broker authors: test support and Conformance.