Testing

TestApp::start connects KafkaBroker to an in-process cluster with topics, groups, offsets and transactions.

A test hands the harness the app main runs, unchanged, and addresses the broker by its production type, tb.broker::<KafkaBroker>(). The testing feature in [dev-dependencies] adds the in-process mode, and turns on the core's harness with it.

The cluster has no settings of its own. It reads what librdkafka would read: the broker's config and producer_config, each descriptor's options, and librdkafka's defaults for the rest. A property name or value librdkafka refuses fails TestApp::start as it fails the connect.

The cluster keeps Kafka's semantics. A topic comes into being with one partition when a record is produced to it or a subscription names it. Each consumer group reads a record once, and every group reads its own copy. The subscription's commit mode decides what is committed, and a seek moves the partitions the subscription reads. A read_committed reader, librdkafka's default, sees a transaction's records when it commits and never after an abort. The cluster refuses what a cluster refuses: a partition the topic lacks, an illegal topic name, a record over message.max.bytes.

A publish through the harness returns once every handler it woke has settled, a reply or a committed transaction included. The assertions that follow read finished state. What only a cluster has, such as a topic with several partitions or the timing of a real group join, is for a live test.

Example

The dependency line:

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

The first test publishes a payment and reads how the handler settled it.

Macros

examples/kafka_testing.rs

#[tokio::test]
async fn a_payment_is_accepted() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    tb.broker::<KafkaBroker>()
        .message(&Payment { amount: 100 })
        .publish()
        .await?;

    tb.broker::<KafkaBroker>()
        .subscriber("payments")
        .assert_called_once()
        .with(&Payment { amount: 100 })
        .settled(HandlerOutcome::ack());
    Ok(())
}

Manual

examples/manual/kafka_testing.rs

#[tokio::test]
async fn a_payment_is_accepted() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    tb.broker::<KafkaBroker>()
        .message(&Payment { amount: 100 })
        .publish()
        .await?;

    tb.broker::<KafkaBroker>()
        .subscriber("payments")
        .assert_called_once()
        .with(&Payment { amount: 100 })
        .settled(HandlerOutcome::ack());
    Ok(())
}

A refund publishes its lines inside one transaction, and the handler commits or aborts it. The topic shows the lines of the committed refund and none of the aborted one.

Macros

examples/kafka_testing.rs

#[tokio::test]
async fn an_aborted_refund_publishes_no_line() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let refund = Refund {
        order_id: 1,
        lines: 3,
        settle: false,
    };
    tb.broker::<KafkaBroker>()
        .message(&refund)
        .publish()
        .await?;

    // A `read_committed` reader, librdkafka's default, never sees an aborted transaction.
    tb.broker::<KafkaBroker>()
        .published::<RefundLine>("refund-lines")
        .assert_not_called();
    Ok(())
}

#[tokio::test]
async fn a_committed_refund_publishes_every_line() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let refund = Refund {
        order_id: 2,
        lines: 3,
        settle: true,
    };
    tb.broker::<KafkaBroker>()
        .message(&refund)
        .publish()
        .await?;

    tb.broker::<KafkaBroker>()
        .published::<RefundLine>("refund-lines")
        .assert_called(3);
    Ok(())
}

Manual

examples/manual/kafka_testing.rs

#[tokio::test]
async fn an_aborted_refund_publishes_no_line() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let refund = Refund {
        order_id: 1,
        lines: 3,
        settle: false,
    };
    tb.broker::<KafkaBroker>()
        .message(&refund)
        .publish()
        .await?;

    // A `read_committed` reader, librdkafka's default, never sees an aborted transaction.
    tb.broker::<KafkaBroker>()
        .published::<RefundLine>("refund-lines")
        .assert_not_called();
    Ok(())
}

#[tokio::test]
async fn a_committed_refund_publishes_every_line() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let refund = Refund {
        order_id: 2,
        lines: 3,
        settle: true,
    };
    tb.broker::<KafkaBroker>()
        .message(&refund)
        .publish()
        .await?;

    tb.broker::<KafkaBroker>()
        .published::<RefundLine>("refund-lines")
        .assert_called(3);
    Ok(())
}