Skip to main content

Testing a Kinesis service

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

The service under test charges card payments. A payment arrives on card-payments, and its partition key is the account to charge. The handler replies with a receipt on card-receipts. A payment without an amount 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-payments.review, the dead-letter stream. The app is built by service(broker), so a test chooses the broker and main uses the production one. The tests need the testing feature of ruststream-kinesis in [dev-dependencies].

A keyed payment​

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

async fn a_receipt_names_the_account_of_the_key(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<KinesisBroker>()
        .message(&Payment {
            id: 1,
            amount_cents: 2_500,
        })
        .with_headers(keyed("acct-7"))
        .publish()
        .await?;

    tb.broker::<KinesisBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::ack());
    tb.broker::<KinesisBroker>()
        .published::<Receipt>("card-receipts")
        .assert_called_once()
        .with(&Receipt {
            payment: 1,
            account: "acct-7".to_owned(),
        });
    tb.shutdown().await?;
    Ok(())
}

TestApp::start connects KinesisBroker to an in-process Kinesis. No AWS endpoint is called, and no credentials are needed. The test addresses the broker by its production type, tb.broker::<KinesisBroker>(). The partition key of a record travels as the partition-key header (PARTITION_KEY_HEADER), so the test sets it with with_headers. The broker turns that header into the record's partition key, and the delivery reports it back under the same header. 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 an amount​

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

    tb.broker::<KinesisBroker>()
        .message(&Payment {
            id: 2,
            amount_cents: 0,
        })
        .with_headers(keyed("acct-7"))
        .publish()
        .await?;

    tb.broker::<KinesisBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::drop());
    tb.broker::<KinesisBroker>()
        .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(KinesisBroker::new())).await?;
    let held = Payment {
        id: 3,
        amount_cents: 250_000,
    };

    tb.broker::<KinesisBroker>()
        .message(&held)
        .with_headers(keyed("acct-7"))
        .publish()
        .await?;
    tb.broker::<KinesisBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::retry_after(HOLD));

    // The delay is a copy of the record, published back to the stream after `HOLD`.
    tb.advance(HOLD).await?;
    tb.broker::<KinesisBroker>()
        .published::<Payment>("card-payments")
        .assert_called(2)
        .with(&held)
        .with_header(RETRY_COUNT_HEADER, "1")
        .with_header(PARTITION_KEY_HEADER, "acct-7");

    // The third delivery is the last one: it goes to the review stream as it arrived.
    tb.advance(HOLD).await?;
    assert_eq!(
        tb.broker::<KinesisBroker>()
            .subscriber("card-payments")
            .outcomes(),
        [Outcome::Nack, Outcome::Nack, Outcome::Nack],
    );
    tb.broker::<KinesisBroker>()
        .published::<Payment>("card-payments.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. Kinesis cannot hold a record back, so retry_after publishes a copy of the record to its stream when the delay ends. The copy carries the retry count and the record's partition key in its headers, and published shows it on card-payments next to the test's own record. The third delivery is the last one the mount site allows, so the record goes to card-payments.review as it arrived. Each held delivery is recorded as Outcome::Nack.

What the in-process Kinesis models​

The in-process Kinesis writes and reads records the way the service does for the broker. The same partition-key rules apply, user headers ride the same envelope, and a delivery carries the same headers and context as one from a shard:

  • A broker reads a stream through one subscription, which reads every record of it and opens at the tip; a second subscription to the stream is refused, as the service's broker refuses it.
  • start_at and the handler's seek handle read the stream's log again from the position they name (Replay and seeking).
  • retry with no cap is answered as Kinesis answers it, unsupported, and the record does not come back.
  • A deferred retry comes back as a counted copy, and a spent record goes to the dead-letter stream.
  • A record the service refuses is refused: a stream name over 128 characters or outside [a-zA-Z0-9_.-], a partition key that is empty or over 256 characters, a record over 10 MiB, the most a stream can be configured to take. A test that publishes one fails with the service's error.
  • A publisher or a seek handle used after shutdown reports KinesisError::NotConnected.

Every stream has one shard there, so every record is ordered against every other, and every stream a service names exists. Several shards, leases and checkpoints, retention and resharding need a stack.

The same test on a stack​

src/main.rs
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_keyed_payment_gets_a_receipt_on_a_stack() -> Result<(), Box<dyn Error>> {
    let Some(endpoint) = support::endpoint() else {
        return Ok(());
    };
    // The service reads `card-payments` and creates it; the receipts are only published.
    support::ensure_stream(&support::config(&endpoint).await, "card-receipts", 1).await;
    let broker = KinesisBroker::new()
        .endpoint(endpoint)
        .region("us-east-1")
        .test_credentials();
    let tb = TestApp::start_live(service(broker)).await?;
    a_receipt_names_the_account_of_the_key(tb).await
}

TestApp::start_live connects the same app to a running Kinesis, here LocalStack. A publish waits up to ten seconds for the reaction to settle; start_live_within sets a longer deadline. The test reads the endpoint from KINESIS_TEST_ENDPOINT with support::endpoint, a helper of the examples, and skips when it is unset. The service creates card-payments when its subscription opens; the test creates card-receipts, which only receives. The subscription opens at the tip, so records an earlier run left in the stream are not read. The assertions are the in-process ones, unchanged.

cargo test
KINESIS_TEST_ENDPOINT=http://localhost:4566 cargo test -- --test-threads=1