Testing

TestApp::start runs the production app on LapinBroker in process, with exchanges, bindings, requeues, quorum counts and reply-to.

A test hands the harness the app main runs, unchanged, and addresses the broker by its production type, tb.broker::<LapinBroker>(). 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. It reads the broker's: declare_topology decides whether a subscription declares its queue, as on the server. A publish is framed by the same conversion a publish to the server goes through.

The transport keeps RabbitMQ's semantics. The default exchange and the direct, topic, fanout and headers exchanges route through the bindings the descriptors declare. Consumers of one queue compete for its messages, and a requeue puts a message back in its queue. A queue keeps what reaches it from its first subscription on, until an auto-delete queue loses its last consumer. A consumer that closes returns what it took and left unsettled. A quorum queue counts deliveries and dead-letters a spent one through its route. A waiting queue releases a delayed copy, and direct reply-to gives every request an address of its own.

The transport refuses what the server refuses, with the same error. That covers a name over 255 bytes, a header the frame cannot carry, and a body over the server's default max_message_size of 16 MiB. It covers the declarations the server refuses: a transient queue that is not exclusive, a binding on the default exchange, a queue declared again with other settings, a quorum queue with a priority. A delivery settled after its connection closed reports the closed channel.

A publish through the harness returns once every handler it woke has settled, a reply or a slot publish included. The assertions that follow read finished state. What only a server has is for a live test: the prefetch window, publisher confirms, the atomicity of a server transaction, a plugin exchange's routing, a queue two connections share, and a topology the service finds without declaring it.

Example

The dependency line:

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

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

Macros
Manual

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

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

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

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

The handler sends a receipt to the topic exchange receipts, and the queue bound to it under receipt.* hands the receipt to its own handler.

Macros
Manual

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

    // `receipts` is a topic exchange, and `receipts.audit` is bound to it under `receipt.*`.
    tb.broker::<LapinBroker>()
        .subscriber("receipts.audit")
        .assert_called_once()
        .with(&Receipt { amount: 100 })
        .settled(HandlerOutcome::ack());
    Ok(())
}

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

    // `receipts` is a topic exchange, and `receipts.audit` is bound to it under `receipt.*`.
    tb.broker::<LapinBroker>()
        .subscriber("receipts.audit")
        .assert_called_once()
        .with(&Receipt { amount: 100 })
        .settled(HandlerOutcome::ack());
    Ok(())
}

tb.out reads what a slot published, with the properties each publish carried.

Macros
Manual

examples/lapin_testing_in_process.rs
#[tokio::test]
async fn a_large_payment_is_receipted_first() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    tb.broker::<LapinBroker>()
        .message(&Payment { amount: 5_000 })
        .publish()
        .await?;

    // The step's priority, over the policy's default of 3.
    tb.out::<Receipts>()
        .assert_called_once()
        .with_options(&LapinPublishOptions {
            priority: Some(9),
            ..LapinPublishOptions::default()
        });
    Ok(())
}

examples/manual/lapin_testing_in_process.rs
#[tokio::test]
async fn a_large_payment_is_receipted_first() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    tb.broker::<LapinBroker>()
        .message(&Payment { amount: 5_000 })
        .publish()
        .await?;

    // The step's priority, over the policy's default of 3.
    tb.out::<Receipts>()
        .assert_called_once()
        .with_options(&LapinPublishOptions {
            priority: Some(9),
            ..LapinPublishOptions::default()
        });
    Ok(())
}

Example​

Example