Skip to main content

Testing a NATS service

A test runs the service's own app with TestApp, against an in-process NATS or against a server.

The service under test charges card payments. A payment arrives on card.payments, which the stream CARD_PAYMENTS stores, and a durable consumer reads it. The handler replies with a receipt on card.receipts. A payment without a card is dropped. A payment over the bank's instant limit is held: the handler asks for it again in a minute with retry_after. After three deliveries a payment goes to card.review, the dead-letter subject. The app is built by service(broker), so a test chooses the broker's address and main uses the production one. The tests need the testing feature of ruststream-nats in [dev-dependencies].

A payment and its receipt​

src/main.rs
#[tokio::test]
async fn a_payment_gets_a_receipt() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(service(NatsBroker::new("nats://localhost:4222"))).await?;
    a_charged_payment_gets_its_receipt(tb).await
}

async fn a_charged_payment_gets_its_receipt(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<NatsBroker>()
        .message(&payment(1, 2_500))
        .publish()
        .await?;

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

TestApp::start connects NatsBroker to an in-process NATS. The address is never dialled, and no stream has to be created first. The test addresses the broker by its production type, tb.broker::<NatsBroker>(). message(..).publish() publishes the payment to the subject its type declares, and the stream stores it. 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 card.receipts. The body takes the started app as an argument, so the live test below runs it too.

A payment without a card​

src/main.rs
#[tokio::test]
async fn a_payment_without_a_card_is_dropped() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(service(NatsBroker::new("nats://localhost:4222"))).await?;
    let no_card = Payment {
        card: String::new(),
        ..payment(2, 2_500)
    };

    tb.broker::<NatsBroker>()
        .message(&no_card)
        .publish()
        .await?;

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

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

A delay, without waiting for it​

src/main.rs
#[tokio::test(start_paused = true)]
async fn a_payment_the_bank_keeps_holding_goes_to_review() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(service(NatsBroker::new("nats://localhost:4222"))).await?;
    let held = payment(3, 250_000);

    tb.broker::<NatsBroker>().message(&held).publish().await?;
    tb.broker::<NatsBroker>()
        .subscriber("card.payments")
        .assert_called_once()
        .settled(HandlerOutcome::retry_after(HOLD));

    // The consumer holds the payment and delivers it again once `HOLD` has passed.
    tb.advance(HOLD).await?;
    tb.broker::<NatsBroker>()
        .subscriber("card.payments")
        .assert_called(2);

    // The third delivery is the last one: it goes to the review subject as it arrived.
    tb.advance(HOLD).await?;
    assert_eq!(
        tb.broker::<NatsBroker>()
            .subscriber("card.payments")
            .outcomes(),
        [Outcome::Nack, Outcome::Nack, Outcome::Nack],
    );
    tb.broker::<NatsBroker>()
        .published::<Payment>("card.review")
        .assert_called_once()
        .with(&held);
    tb.shutdown().await?;
    Ok(())
}

The test starts on a paused clock, and advance moves it forward. A minute of HOLD costs nothing. The consumer holds the payment as a server does and delivers it again when the delay ends. The third delivery is the last one the mount site allows, so the payment goes to card.review as it arrived. Each held delivery is recorded as Outcome::Nack.

What the in-process NATS models​

The in-process NATS reads what the client would read: the addresses, and the connection options that change what the server delivers (no_echo, the inbox prefix). It routes and stores as a server does:

  • Subjects match with * and >, and a queue group hands each message to one member.
  • A stream keeps sequences, its duplicate window and the expectations a publish states.
  • A consumer starts where its deliver policy says and carries the metadata of the JetStream context.
  • An acknowledgement, a negative one with or without a delay, and a termination act as on a server; a delivery left unsettled comes back after ack_wait.
  • A request reaches the app's responder, and a request nobody reads gets "no responders" at once.

It refuses what the client or the server refuses: a subject with whitespace, a header with no NATS form, a message over 1 MiB, a JetStream publish no stream takes. A stream there stores what the app's consumers read, and a subscription finds every stream it names. A stream's own subject list and limits, max_ack_pending, authentication and TLS need a server.

The same test on a server​

src/main.rs
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_payment_gets_a_receipt_on_a_server() -> Result<(), Box<dyn Error>> {
    let Some(url) = live::url("NATS_TEST_URL") else {
        return Ok(());
    };
    // The stream is the server's state: the test creates it, replacing what an interrupted
    // run left behind, and deletes it afterwards.
    let admin = NatsBroker::new(url.as_str()).connect().await?;
    let jetstream = admin.jetstream();
    let _ = jetstream.delete_stream("CARD_PAYMENTS").await;
    jetstream
        .create_stream(StreamConfig {
            name: "CARD_PAYMENTS".to_owned(),
            subjects: vec!["card.payments".to_owned()],
            ..StreamConfig::default()
        })
        .await?;

    let tb = TestApp::start_live(service(NatsBroker::new(url.as_str()))).await;
    let outcome = match tb {
        Ok(tb) => a_charged_payment_gets_its_receipt(tb).await,
        Err(err) => Err(err.into()),
    };

    jetstream.delete_stream("CARD_PAYMENTS").await?;
    admin.shutdown().await?;
    outcome
}

TestApp::start_live connects the same app to a running NATS server. The test reads the address from NATS_TEST_URL with live::url, a helper of the example's own, and skips when it is unset. A stream is the server's state, so the test creates CARD_PAYMENTS on a connection of its own first and deletes it afterwards. Deleting the stream also deletes its consumer, so the next run starts clean. The assertions are the in-process ones, unchanged.

cargo test
NATS_TEST_URL=nats://localhost:4222 cargo test -- --test-threads=1