Skip to main content

Testing an MQTT service

A test runs the service's own app with TestApp, against an in-process MQTT server or against Mosquitto.

The service under test watches freezers. Each freezer publishes its temperature to freezers/<id>/temperature, and the monitors share the readings in a share group. Every reading goes into a temperature log. A freezer warmer than -18 °C raises an alert on freezers/alerts. When the log does not answer, the handler asks for the reading again in ten seconds with retry_after. After three deliveries a reading goes to freezers/review, the dead-letter topic.

The handler watch publishes the alert through an Alerts slot. 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-rumqttc in [dev-dependencies].

A warm freezer​

src/main.rs
#[tokio::test]
async fn a_warm_freezer_raises_an_alert() -> Result<(), Box<dyn Error + Send + Sync>> {
    let broker = MqttBroker::new("mqtt://localhost:1883", "cold-chain");
    let tb = TestApp::start(service(broker)).await?;
    a_warm_reading_raises_one_alert(tb).await
}

async fn a_warm_reading_raises_one_alert(
    tb: TestApp<()>,
) -> Result<(), Box<dyn Error + Send + Sync>> {
    tb.broker::<MqttBroker>()
        .message(&reading("f1", -12.5))
        .to("freezers/f1/temperature")
        .publish()
        .await?;

    tb.broker::<MqttBroker>()
        .subscriber("freezers/+/temperature")
        .assert_called_once()
        .settled(HandlerOutcome::ack());
    tb.broker::<MqttBroker>()
        .published::<Alert>("freezers/alerts")
        .assert_called_once()
        .with(&Alert {
            freezer: "f1".to_owned(),
            celsius: -12.5,
        });
    tb.shutdown().await?;
    Ok(())
}

TestApp::start connects MqttBroker to an in-process MQTT server. The URL is never dialled. The test addresses the broker by its production type, tb.broker::<MqttBroker>(). A Reading names no topic, so the test names one with .to("freezers/f1/temperature"), as a freezer would. The handler's filter matches it the way Mosquitto would. publish returns once the handler and its alert have settled. So the next lines assert on finished state: how the subscription settled and what reached freezers/alerts. The subscription is addressed by its filter, without the share group. The body takes the started app as an argument, so the live test below runs it too.

A cold freezer​

src/main.rs
#[tokio::test]
async fn a_cold_freezer_raises_no_alert() -> Result<(), Box<dyn Error + Send + Sync>> {
    let broker = MqttBroker::new("mqtt://localhost:1883", "cold-chain");
    let tb = TestApp::start(service(broker)).await?;

    tb.broker::<MqttBroker>()
        .message(&reading("f1", -21.0))
        .to("freezers/f1/temperature")
        .publish()
        .await?;

    tb.broker::<MqttBroker>()
        .subscriber("freezers/+/temperature")
        .assert_called_once()
        .settled(HandlerOutcome::ack());
    tb.out::<Alerts>().assert_not_called();
    tb.shutdown().await?;
    Ok(())
}

settled checks the outcome the handler returned. tb.out::<Alerts>() reads what the handler published through its Alerts slot, and assert_not_called checks that it published nothing.

A delay, without waiting for it​

src/main.rs
#[tokio::test(start_paused = true)]
async fn a_reading_the_log_keeps_refusing_goes_to_review()
-> Result<(), Box<dyn Error + Send + Sync>> {
    let broker = MqttBroker::new("mqtt://localhost:1883", "cold-chain");
    let tb = TestApp::start(service(broker)).await?;
    let refused = reading("f7", -20.0);

    tb.broker::<MqttBroker>()
        .message(&refused)
        .to("freezers/f7/temperature")
        .publish()
        .await?;
    tb.broker::<MqttBroker>()
        .subscriber("freezers/+/temperature")
        .assert_called_once()
        .settled(HandlerOutcome::retry_after(RETRY));

    // The delay is a copy of the reading, published to the retry topic once it has passed.
    tb.advance(RETRY).await?;
    tb.broker::<MqttBroker>()
        .published::<Reading>("freezers/retry/temperature")
        .assert_called_once()
        .with(&refused)
        .with_header(RETRY_COUNT_HEADER, "1");

    // The third delivery is the last one: it goes to review as it arrived.
    tb.advance(RETRY).await?;
    assert_eq!(
        tb.broker::<MqttBroker>()
            .subscriber("freezers/+/temperature")
            .outcomes(),
        [Outcome::Nack, Outcome::Nack, Outcome::Nack],
    );
    tb.broker::<MqttBroker>()
        .published::<Reading>("freezers/review")
        .assert_called_once()
        .with(&refused);
    tb.shutdown().await?;
    Ok(())
}

The test starts on a paused clock, and advance moves it forward. Ten seconds of RETRY cost nothing. retry_after publishes a copy of the reading to freezers/retry/temperature when the delay ends, and the filter reads it again. The copy carries the retry count in a header, and published shows it. The third delivery is the last one the mount site allows, so the reading goes to freezers/review as it arrived. Each delayed delivery is recorded as Outcome::Nack.

What the in-process server models​

The in-process server reads the broker's own settings, and a publish goes through the same validation and the same mapping onto MQTT 5 properties as on the wire. It models the server one connection talks to:

  • A publish reaches every subscription whose filter matches its topic, with +, # and the rule for topics that start with $.
  • A share group takes one copy between its members, and overlapping filters each receive their own.
  • A retained message reaches a later subscription outside a share group, and an empty retained payload clears it.
  • A message larger than max_packet_size never arrives, and a topic or header the client refuses fails the publish.
  • An acknowledgement after shutdown is refused.

Each broker connected in process is a server of its own. A persistent session and its resumption, the last will, the receive-maximum window, the QoS 1 and 2 handshakes and several connections on one server need a real broker.

The same test on Mosquitto​

src/main.rs
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_warm_freezer_raises_an_alert_on_mosquitto()
-> Result<(), Box<dyn Error + Send + Sync>> {
    let Some(url) = live::url("MQTT_TEST_URL") else {
        return Ok(());
    };
    let broker = MqttBroker::new(url, live::client_id("cold-chain"));
    let tb = TestApp::start_live(service(broker)).await?;
    a_warm_reading_raises_one_alert(tb).await
}

TestApp::start_live connects the same app to a running server. The test reads the address from MQTT_TEST_URL with live::url, a helper of the example's own, and returns at once when it is unset. It connects with a client id of its own, so it shares the server with other runs. The assertions are the in-process ones, unchanged.

cargo test
MQTT_TEST_URL=mqtt://localhost:1883 cargo test -- --test-threads=1

RUSTSTREAM_REQUIRE_LIVE=1 turns the skip into a failure, for a CI job that starts a server and must not pass without it.