Testing

TestApp::start connects KinesisBroker in process: one retained log per stream, positions and seeking, and the service's refusals.

A test hands the harness the app main runs, unchanged, and addresses the broker by its production type, tb.broker::<KinesisBroker>(). The testing feature in [dev-dependencies] adds the in-process mode, and turns on the core's harness with it. The in-process transport has no settings of its own. The descriptors and the publish policies of the app are the production ones, and the AWS configuration is never read.

A record is written the way the publisher writes it for the service: the partition key resolved in the same order, the headers in the same envelope. It is read back by the same decoding, so a delivery carries the position, the three headers and the context a record from a shard carries. Each stream is one retained log with one shard, so every record of a stream is ordered against every other. A subscription opens at the tip, and start_at and a seek read the retained log again from the position they name. A plain retry is a requeue the stream cannot perform, so the record does not come back; a retry_after copy does, once tb.advance() passes its delay under tokio::time::pause().

The transport refuses what the service refuses. A stream name the service does not accept, a partition key outside 1 to 256 characters, and a record over 10 MiB, the most a stream can be configured to take, are refused with the service's error. A record a handler publishes there gets the error back, as it would from the service. A record the test itself publishes there fails the test at the publish. A publisher or a seek handle used after shutdown reports KinesisError::NotConnected.

A publish through the harness returns once every handler it woke has settled, the records those handlers published included. The assertions that follow read finished state. What only the service has is for a live test: a stream's shards and the hashing of keys to them, the existence of a stream, leases and checkpoints, resharding, retention, throttling, credentials and endpoints.

Example

The dependency line:

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

The first test publishes a payment and reads how the handler settled it and what it booked on the ledger.

Macros
Manual

examples/kinesis_testing.rs
#[tokio::test]
async fn a_payment_is_booked() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    tb.broker::<KinesisBroker>()
        .message(&payment("acc-1"))
        .publish()
        .await?;

    tb.broker::<KinesisBroker>()
        .subscriber("payments")
        .assert_called_once()
        .with(&payment("acc-1"))
        .settled(HandlerOutcome::ack());
    tb.broker::<KinesisBroker>()
        .published::<Entry>("ledger")
        .assert_called_once()
        .with(&Entry { amount: 100 })
        .with_header(PARTITION_KEY_HEADER, "acc-1");
    Ok(())
}

examples/manual/kinesis_testing.rs
#[tokio::test]
async fn a_payment_is_booked() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    tb.broker::<KinesisBroker>()
        .message(&payment("acc-1"))
        .publish()
        .await?;

    tb.broker::<KinesisBroker>()
        .subscriber("payments")
        .assert_called_once()
        .with(&payment("acc-1"))
        .settled(HandlerOutcome::ack());
    tb.broker::<KinesisBroker>()
        .published::<Entry>("ledger")
        .assert_called_once()
        .with(&Entry { amount: 100 })
        .with_header(PARTITION_KEY_HEADER, "acc-1");
    Ok(())
}

A payment whose account is too long to be a partition key cannot be booked: the transport refuses the entry, as the service would, and the handler settles the payment as a retry.

Macros
Manual

examples/kinesis_testing.rs
#[tokio::test]
async fn an_entry_the_service_refuses_is_refused() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    // A partition key is 1 to 256 characters, so the entry for this account is refused.
    let unbookable = payment(&"a".repeat(300));
    tb.broker::<KinesisBroker>()
        .message(&unbookable)
        .publish()
        .await?;

    tb.broker::<KinesisBroker>()
        .subscriber("payments")
        .assert_called_once()
        .settled(HandlerOutcome::retry());
    tb.broker::<KinesisBroker>()
        .published::<Entry>("ledger")
        .assert_not_called();
    Ok(())
}

examples/manual/kinesis_testing.rs
#[tokio::test]
async fn an_entry_the_service_refuses_is_refused() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    // A partition key is 1 to 256 characters, so the entry for this account is refused.
    let unbookable = payment(&"a".repeat(300));
    tb.broker::<KinesisBroker>()
        .message(&unbookable)
        .publish()
        .await?;

    tb.broker::<KinesisBroker>()
        .subscriber("payments")
        .assert_called_once()
        .settled(HandlerOutcome::retry());
    tb.broker::<KinesisBroker>()
        .published::<Entry>("ledger")
        .assert_not_called();
    Ok(())
}

Example​

Example