Struct TestApp 

pub struct TestApp<State> { /* private fields */ }

A test harness around the service’s production app.

Takes the app main builds, unchanged. start connects each of its brokers through the broker’s InProcess transition instead of connect, so the same connected forms carry the routes with no server; start_live connects them through connect, against a running stand. Either way the harness drives input through the broker, records what handlers saw and what the app published, and exposes per-broker assertions addressed by the production broker type, so one test body runs in both modes and only the start call differs.

Build one with start (runs the app’s real on_startup) or with_state (injects a mirror state for non-broker dependencies). Drive input with broker / broker_named and assert on what happened.

Examples

use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, HandlerOutcome, RustStream};
use ruststream::testing::TestApp;
use ruststream::{Outgoing, subscriber};
use serde::{Deserialize, Serialize};

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

#[subscriber("orders")]
async fn handle(order: &Order) -> HandlerOutcome {
    let _ = order;
    HandlerOutcome::ack()
}

let app = RustStream::new(AppInfo::new("svc", "0.1.0"))
    .with_broker(MemoryBroker::new(), |b| { b.include(handle); });
let tb = TestApp::start(app).await?;

tb.broker::<MemoryBroker>()
    .message(&Order { id: 1 })
    .to("orders")
    .publish()
    .await?;
tb.broker::<MemoryBroker>()
    .subscriber("orders")
    .assert_called_once()
    .with(&Order { id: 1 })
    .settled(HandlerOutcome::ack());

Implementations

impl<State: Send + Sync + 'static> TestApp<State>

pub async fn start<A>(app: A) -> Result<Self, TestError>

where A: App<State = State>,

Starts the harness in the service’s own startup order: the app’s real on_startup, then each broker’s connect, then the subscriptions, then after_startup. Each broker connects through its InProcess transition, which performs no I/O; the rest of startup is the production one, so an on_startup that fails leaves no broker connected.

Errors

Returns TestError::NoTransport if a broker of the app has no in-process mode, TestError::Startup if a lifecycle hook fails, TestError::Connect if a broker fails to connect in process, or TestError::Subscribe if a subscription fails to open. The app’s publish pipeline is whatever it was built with: an app-wide publish_layer is already baked into the mounted routes, so the harness drives the same stack production would.

pub async fn start_live<A>(app: A) -> Result<Self, TestError>

where A: App<State = State>,

Starts the harness against running brokers: each broker of the app connects through its ordinary connect, and the rest of startup is the production one, in its order: the app’s real on_startup runs before any broker connects.

The test body is the one an in-process test runs; only this call differs. What differs underneath is how long things take. A publish, settle and advance wait on the broker’s real clock, up to a deadline of ten seconds (start_live_within sets another), for the subscriptions to handle what was published to them and what fell due.

Errors

Returns TestError::PausedClock on a paused clock, TestError::Startup if a lifecycle hook fails, TestError::Connect if a broker fails to connect, or TestError::Subscribe if a subscription fails to open.

Examples

use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, HandlerOutcome, RustStream};
use ruststream::testing::TestApp;
use ruststream::{Outgoing, subscriber};
use serde::{Deserialize, Serialize};

#[derive(Outgoing, Serialize, Deserialize)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
}

#[subscriber("orders")]
async fn accept(order: &Order) -> HandlerOutcome {
    let _ = order.id;
    HandlerOutcome::ack()
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(accept);
    })
}

let tb = TestApp::start_live(app()).await?;
tb.broker::<MemoryBroker>().message(&Order { id: 1 }).publish().await?;
tb.broker::<MemoryBroker>().subscriber("orders").assert_called_once();
tb.shutdown().await?;

pub async fn start_live_within<A>( app: A, deadline: Duration, ) -> Result<Self, TestError>

where A: App<State = State>,

start_live with deadline as the longest a publish, a settle or an advance waits for the reaction to settle.

Errors

As start_live.

Examples

use std::time::Duration;

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

// A stand far away: give the reaction half a minute before the test gives up on it.
let app = RustStream::new(AppInfo::new("orders", "0.1.0"))
    .register_broker(MemoryBroker::new());
let tb = TestApp::start_live_within(app, Duration::from_secs(30)).await?;
tb.shutdown().await?;

pub async fn with_state<A, F>(app: A, build: F) -> Result<Self, TestError>

where A: App<State = State>, F: FnOnce(&TestBrokers<'_>) -> State,

Starts the harness with an injected mirror state, instead of running the app’s on_startup. build receives the brokers so it can wire the mirror state’s publishers onto the same transports (tb.broker::<MemoryBroker>().publisher()) and supply fakes for non-broker dependencies. Each broker connects in process first, so the mirror state’s publishers pair against connected brokers.

Errors

Returns TestError::NoTransport if a broker of the app has no in-process mode, TestError::Connect if a broker fails to connect in process, or TestError::Subscribe if a subscription fails to open.

pub fn broker<B>(&self) -> BrokerHandle<'_>

where B: Broker + 'static, B::Connected: DefaultPublish, <<B::Connected as DefaultPublish>::Policy as PublishPolicy<B::Connected>>::Live: Publisher,

Addresses the unique broker of type B, the production type the app was built on.

Live, a test’s input reaches the broker through the connected form’s DefaultPublish policy, which is what the bound asks for.

Panics

Panics if no broker of type B is registered, or more than one is (address by label with broker_named instead).

pub fn broker_named(&self, label: &str) -> BrokerHandle<'_>

Addresses the broker registered under label (see with_broker_labeled).

Panics

Panics if no broker carries label.

pub fn out<M: OutSlot>(&self) -> PublishedAssertions<()>

Asserts on what was published through the Out slot marked M: exactly the messages the handler sent through that injected publisher, with their destinations and headers, across all brokers.

The untyped assertions (assert_called_once, with_raw, …) apply directly; decode the payloads with decoded_as for the typed with form.

Publishes made outside the handler task (a spawned sibling task, a settled owned transaction’s buffer) are not attributed to the slot.

Examples

use ruststream::memory::{MemoryBroker, MemoryPublish};
use ruststream::runtime::{AppInfo, HandlerOutcome, Out, RustStream};
use ruststream::testing::TestApp;
use ruststream::{Deserialized, OutSlot, Outgoing, Publisher, Serialized, subscriber};

#[derive(Deserialized)]
struct Chunk<'a>(&'a [u8]);

// The frame on the way out: a serialized type carries its own bytes, so what the
// handler republishes is byte-for-byte what it received.
#[derive(Outgoing, Serialized)]
struct Frame(Vec<u8>);

#[derive(OutSlot)]
#[publishes(Frame)]
struct Encoded;

#[subscriber("chunks")]
async fn transcode(
    chunk: &Chunk<'_>,
    Out(out): Out<impl Publisher, Encoded>,
) -> HandlerOutcome {
    let frame = Frame(chunk.0.to_vec());
    if out.message(&frame).to("encoded").publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

let app = RustStream::new(AppInfo::new("svc", "0.1.0"))
    .with_broker(MemoryBroker::new(), |b| {
        b.include(transcode).out(Encoded, MemoryPublish).build();
    });
let tb = TestApp::start(app).await?;
tb.broker::<MemoryBroker>()
    .message(&Frame(b"frame".to_vec()))
    .to("chunks")
    .publish()
    .await?;
tb.out::<Encoded>().assert_called_once().with_raw(b"frame");

pub async fn publish<T: Serialize + Sync>( &self, name: &str, value: &T, ) -> Result<(), TestError>

Publishes to the only registered broker, a convenience for single-broker apps.

Errors

Returns TestError::Ambiguous when more than one broker is registered (use broker / broker_named), or any error from BrokerHandle::publish.

pub fn message<'a, T>( &'a self, value: &'a T, ) -> PublishBuilder<InjectSink<'a>, MessageBody<'a, T>, UnnamedCodec, HeadersUnset, T::Form>

where T: OutgoingDestination,

Starts a typed injection on the only registered broker, a convenience for single-broker apps: tb.message(&order).to("orders").publish().await?.

The scoped BrokerHandle::message with the broker chosen for you. An app registering more than one broker has no single target, so the publish reports TestError::Ambiguous and the test addresses a broker with broker / broker_named instead.

Examples

use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, HandlerOutcome, RustStream};
use ruststream::testing::TestApp;
use ruststream::{Outgoing, subscriber};
use serde::{Deserialize, Serialize};

#[derive(Outgoing, Serialize, Deserialize)]
#[outgoing(name = "orders")]
struct Order {
    id: u32,
}

#[subscriber("orders")]
async fn handle(order: &Order) -> HandlerOutcome {
    let _ = order.id;
    HandlerOutcome::ack()
}

let app = RustStream::new(AppInfo::new("svc", "0.1.0"))
    .with_broker(MemoryBroker::new(), |b| { b.include(handle); });
let tb = TestApp::start(app).await?;

tb.message(&Order { id: 7 }).publish().await?;

pub async fn settle(&self) -> Result<(), TestError>

Drives any in-flight reaction to a standstill (handlers run, their publishes cascade) without publishing anything new. A publish through the harness calls this for you; use it after a reaction the test started some other way.

In process it waits for the transports to report nothing in flight. Live, where nothing in this process can read the broker’s queues, it waits until every subscription has handled what the harness saw published to it and every redelivery that fell due, and no handler is running, up to the deadline the harness was started with.

Errors

Returns TestError::NotQuiescent if an in-process reaction does not settle within the step budget, or TestError::NotSettled naming the subscription a live reaction left unsettled at the deadline.

pub async fn advance(&self, by: Duration) -> Result<(), TestError>

Lets by pass, fires every nack_after / retry_after redelivery now due, and drives the resulting reaction to a standstill. Use it to test delayed redeliveries: publish records the immediate NackAfter settlement and returns; advance then delivers the message again. A broker’s native delayed redelivery and the runtime’s deferred re-publish both arrive this way, in either mode; the copy the runtime publishes for a zero delay does not wait for it.

In process it moves the paused clock by by, so an in-process test runs on a paused clock (#[tokio::test(start_paused = true)] or tokio::time::pause; on a running clock tokio::time::advance panics). Live it lets by of real time pass, then settles, waiting for every redelivery that fell due by then.

Errors

Returns TestError::NotQuiescent if an in-process reaction does not settle within the step budget, or TestError::NotSettled naming the subscription a live reaction left unsettled at the deadline.

pub async fn drain(&self)

Waits for the post-settle continuations of everything settled so far to finish, for tests that assert on their side effects. Synchronous handler effects need only settle.

A settlement registers its continuations - the outcome’s and_after and the context’s after(..) / after_settle(..) hooks alike - before it releases the delivery, so an injection that has returned has already put its continuations here for this to find.

pub fn run_result(&self) -> Result<(), RustStreamError>

The result the real run would return: Ok while the service is healthy, or RustStreamError::Dispatch once a fail-fast failure tore it down.

Errors

Returns RustStreamError::Dispatch when a handler panic (or a fail-fast decode failure) triggered shutdown.

pub fn assert_running(&self)

Asserts the service is still running (no fail-fast shutdown was triggered).

Panics

Panics if a fail-fast failure has torn the service down.

pub fn assert_shut_down(&self)

Asserts a fail-fast failure has shut the service down.

Panics

Panics if the service is still running.

pub async fn shutdown(self) -> Result<(), RustStreamError>

Shuts the harness down: stops the dispatch loops, drains in-flight handlers and post-settle continuations (bounded by the app’s shutdown timeout), shuts each broker down, and returns run_result.

Errors

Returns RustStreamError::Dispatch when a fail-fast failure tore the service down.

Trait Implementations

impl<State> Debug for TestApp<State>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.