Skip to main content

Testing a file or pipeline service

A test runs the service's own app with TestApp, on an in-process transport or on a real stream file.

The service under test is a checkout. An order arrives under orders, and the handler replies with a receipt under receipts. An order without an amount is dropped. An order over the fraud-check limit is held: the handler asks for it again in a minute with retry_after. After three deliveries an order goes to orders.review, the dead-letter key. The app is built by service(file), so a test chooses the file and main uses the production one. The service is the example file's own, examples/guide/testing.rs. The tests need the testing feature of ruststream-sea-file in [dev-dependencies].

An order and its receipt​

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

async fn an_order_gets_a_receipt_in(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<FileBroker>()
        .message(&Order {
            id: 1,
            amount_cents: 2_500,
        })
        .publish()
        .await?;

    tb.broker::<FileBroker>()
        .subscriber("orders")
        .assert_called_once()
        .settled(HandlerOutcome::ack());
    tb.broker::<FileBroker>()
        .published::<Receipt>("receipts")
        .assert_called_once()
        .with(&Receipt {
            order: 1,
            amount_cents: 2_500,
        });
    tb.shutdown().await?;
    Ok(())
}

TestApp::start connects FileBroker to a log in memory. The path is never opened, so tests run side by side in one process. The test addresses the broker by its production type, tb.broker::<FileBroker>(). publish returns once the handler and its reply have settled. So the next lines assert on finished state: how the subscriber settled and what reached receipts. The body takes the started app as an argument, so the live test below runs it too.

An order without an amount​

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

    tb.broker::<FileBroker>()
        .message(&Order {
            id: 2,
            amount_cents: 0,
        })
        .publish()
        .await?;

    tb.broker::<FileBroker>()
        .subscriber("orders")
        .assert_called_once()
        .settled(HandlerOutcome::drop());
    tb.broker::<FileBroker>()
        .published::<Receipt>("receipts")
        .assert_not_called();
    tb.shutdown().await?;
    Ok(())
}

settled checks the outcome the handler returned. assert_not_called checks that no receipt was appended.

A delay, without waiting for it​

src/main.rs
#[tokio::test(start_paused = true)]
async fn an_order_held_three_times_goes_to_review() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let large = Order {
        id: 3,
        amount_cents: 250_000,
    };

    tb.broker::<FileBroker>().message(&large).publish().await?;
    tb.broker::<FileBroker>()
        .subscriber("orders")
        .assert_called_once()
        .settled(HandlerOutcome::retry_after(HOLD));

    // The delay ends in a copy of the order, appended under `orders`.
    tb.advance(HOLD).await?;
    tb.broker::<FileBroker>()
        .published::<Order>("orders")
        .assert_called(2)
        .with(&large)
        .with_header(RETRY_COUNT_HEADER, "1");

    // The third delivery is the last one: it is written under `orders.review` as it arrived.
    tb.advance(HOLD).await?;
    assert_eq!(
        tb.broker::<FileBroker>().subscriber("orders").outcomes(),
        [Outcome::Nack, Outcome::Nack, Outcome::Nack],
    );
    tb.broker::<FileBroker>()
        .published::<Order>("orders.review")
        .assert_called_once()
        .with(&large);
    tb.shutdown().await?;
    Ok(())
}

The test starts on a paused clock, and advance moves it forward. A minute of HOLD costs nothing. retry_after appends a copy of the order under orders when the delay ends. The copy carries the retry count in a header, and published shows it under orders next to the test's own order. The third delivery is the last one the mount site allows, so the order goes to orders.review as it arrived. Each held delivery is recorded as Outcome::Nack.

What the in-process transports model​

The in-process stream file is a log in memory, new for each broker value and shared by its clones. It keeps what a file keeps:

  • Each stream key is numbered from one, and a live subscription starts at the tip.
  • A replay reads what the log holds and completes at its end.
  • Positions and seeks resolve against the log with the file's refusals.
  • end_with_eos ends every subscription at shutdown.

A log the service reads at startup is recorded on a clone of its broker before TestApp::start, as the Resuming and seeking tests do.

The in-process pipe is standard input the test feeds and standard output the service writes. tb.broker::<StdioBroker>().message(..) is a line arriving under its key, and published::<T>(key) reads a line the stage wrote. A stage reads its own output only under loopback, as on a real pipe. Both transports refuse what the real ones refuse: a stream key the client does not take, an empty line on a pipe, a beacon interval that is not a multiple of 1024.

The bytes of a .ss file, what a file keeps across runs, and the line format on a real pipe show only on the real transport.

The same test on a stream file​

src/main.rs
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_order_gets_a_receipt_on_a_stream_file() -> Result<(), Box<dyn Error>> {
    // A file of this run's own, so nothing an earlier run appended is read.
    let path = env::temp_dir().join(format!("checkout-{}.ss", process::id()));
    let file = FileBroker::new(path.to_string_lossy());
    let tb = TestApp::start_live(service(file)).await?;
    an_order_gets_a_receipt_in(tb).await?;
    fs::remove_file(&path)?;
    Ok(())
}

TestApp::start_live runs the same app on a real stream file in the temp directory. The transport is the local file system, so the live test needs no server and runs with the others in cargo test. The file is named after the process, so nothing an earlier run appended is read, and the test removes it at the end. The assertions are the in-process ones, unchanged. A pipeline stage is tested live as a child process with piped standard input and output, as Shell pipelines shows.