Live

TestApp::start_live runs the same test body against a running RabbitMQ, one subscription per queue, for what only a server has.

It connects the broker through its ordinary connect, so every subscription consumes from the server. The body is the one an in-process test runs; only the start call differs. A publish, settle and advance wait on the real clock until every subscription has handled what was published to it. The wait lasts ten seconds at most, and start_live_within sets another bound.

The live mode is for what the in-process transport leaves to the server: the prefetch window, publisher confirms, the atomicity of a server transaction, a plugin exchange's routing, and a topology the service finds without declaring it. A live test keeps one subscription per queue. The server picks which consumer of a queue takes a message, so a test with two subscriptions of one queue stops at the first publish and names the queue.

A server keeps what earlier runs left in it. In the example the service declares its queue with declare_topology, and the queue is exclusive: it lives as long as the test's connection.

live::url is the repository's test helper: it returns the server's address from AMQP_TEST_URL, and the test returns early without one, so cargo test passes on a machine with no server. A job that stood a server up sets RUSTSTREAM_REQUIRE_LIVE=1, and a missing address then fails the test instead of skipping it. The repository's stand is docker-compose.test.yml, which just brokers-up starts.

Example

The two tests share one body: one in process, one against the server at AMQP_TEST_URL.

Macros
Manual

examples/lapin_testing_live.rs
/// The body both modes run; only the start call differs.
async fn accepts_a_payment(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<LapinBroker>()
        .message(&Payment { amount: 100 })
        .publish()
        .await?;

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

#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    accepts_a_payment(TestApp::start(app("amqp://rabbit:5672")).await?).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_the_stand() -> Result<(), Box<dyn Error>> {
    let Some(uri) = live::url("AMQP_TEST_URL") else {
        return Ok(());
    };
    accepts_a_payment(TestApp::start_live(app(&uri)).await?).await
}

examples/manual/lapin_testing_live.rs
/// The body both modes run; only the start call differs.
async fn accepts_a_payment(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<LapinBroker>()
        .message(&Payment { amount: 100 })
        .publish()
        .await?;

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

#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    accepts_a_payment(TestApp::start(app("amqp://rabbit:5672")).await?).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_the_stand() -> Result<(), Box<dyn Error>> {
    let Some(uri) = live::url("AMQP_TEST_URL") else {
        return Ok(());
    };
    accepts_a_payment(TestApp::start_live(app(&uri)).await?).await
}

Example​

Example