Testing

TestApp::start connects FileBroker to a log in memory and StdioBroker to a pipe in memory, with the production app unchanged.

A test hands the harness the app main runs and addresses the broker by its production type, tb.broker::<FileBroker>() or tb.broker::<StdioBroker>(). The testing feature in [dev-dependencies] adds the in-process mode. No file is opened and no pipe is attached, so tests run side by side in one process.

A stream file in process is a log in memory, new and empty for each broker value and shared by the broker's clones. It keeps what the file keeps: each stream key numbered from one, a live subscription that starts at the tip, a replay that completes at the end of the log, positions and seeking with the file's refusals, and the end-of-stream mark. A test publishes into the log, and published::<T>(key) reads what the service appended, retry copies and dead letters included.

A pipe in process is standard input the test feeds and standard output the service writes. A message the test publishes is a line arriving under its key. A line the service writes is recorded under its key, and published::<T>(key) reads it the way the next stage would. It reaches the service's own subscriptions only under loopback, as on a real pipe.

Both refuse what the real transport refuses: a stream key the client does not take, an empty line on a pipe, a beacon interval that is not a multiple of 1024. A header crosses the same envelope a file or a line carries. A publish through the harness returns once every handler it woke has settled, so the assertions read finished state; tb.advance(..) on a paused clock brings a retry_after copy back. What only the real transport shows, such as the bytes of a .ss file or what a file keeps across runs, is for a live test. The guide Testing a file or pipeline service tests a whole service this way.

Example

The dependency line:

[dev-dependencies]
ruststream-sea-file = { version = "0.7", features = ["testing"] }

A file service answers an order with a receipt.

Macros
Manual

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

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

    tb.broker::<FileBroker>()
        .subscriber("orders")
        .assert_called_once()
        .with(&Order { id: 7 })
        .settled(HandlerOutcome::ack());
    tb.broker::<FileBroker>()
        .published::<Receipt>("receipts")
        .assert_called_once()
        .with(&Receipt { order: 7 });
    tb.shutdown().await?;
    Ok(())
}

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

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

    tb.broker::<FileBroker>()
        .subscriber("orders")
        .assert_called_once()
        .with(&Order { id: 7 })
        .settled(HandlerOutcome::ack());
    tb.broker::<FileBroker>()
        .published::<Receipt>("receipts")
        .assert_called_once()
        .with(&Receipt { order: 7 });
    tb.shutdown().await?;
    Ok(())
}

A pipeline stage answers a jobs line with a results line, which the test reads off the stage's standard output.

Macros
Manual

examples/sea_testing.rs
#[tokio::test]
async fn a_job_line_gets_a_result_line() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(stage()).await?;

    // A line on the stage's standard input.
    tb.broker::<StdioBroker>()
        .message(&Job { id: 7 })
        .to("jobs")
        .publish()
        .await?;

    // A line the stage wrote to its standard output.
    tb.broker::<StdioBroker>()
        .published::<Done>("results")
        .assert_called_once()
        .with(&Done { job: 7 });
    tb.shutdown().await?;
    Ok(())
}

examples/manual/sea_testing.rs
#[tokio::test]
async fn a_job_line_gets_a_result_line() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(stage()).await?;

    // A line on the stage's standard input.
    tb.broker::<StdioBroker>()
        .message(&Job { id: 7 })
        .to("jobs")
        .publish()
        .await?;

    // A line the stage wrote to its standard output.
    tb.broker::<StdioBroker>()
        .published::<Done>("results")
        .assert_called_once()
        .with(&Done { job: 7 });
    tb.shutdown().await?;
    Ok(())
}

Example​

Example