Testing

TestApp::start runs the production app on an in-process MQTT server: wildcards, share groups, retained messages and QoS, with no broker.

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

The in-process server has no settings of its own. It reads the broker's, and checks them as the connect does: a keep-alive under five seconds or an mqtts:// URL without tls_ca fails TestApp::start. A publish goes through the same validation and the same mapping onto MQTT 5 properties as on the wire.

It routes as a server routes. A publish reaches every subscription whose filter matches its topic, + and # included, and a filter starting with a wildcard skips the topics that start with $. A share group hands each message to one of its members. Two overlapping filters each receive their own copy, as a server sends them. A delivery arrives at the lower of the publish and the subscription QoS, and a QoS 0 delivery has nothing to acknowledge. A retained message reaches every later subscription outside a share group, and an empty retained payload clears it. A message over the broker's max_packet_size never arrives, and a publish or an acknowledgement after shutdown is refused.

A publish through the harness returns once every handler it woke has settled, a reply included. The assertions that follow read finished state. What only a server has is for a live test: the login and TLS, a persistent session, the last will, the receive window, a lost connection, and several connections on one server.

Example

The dependency line:

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

The test publishes a hot reading to one device's topic, and reads how the handler settled it and the alert it published.

Macros
Manual

examples/mqtt_testing.rs
#[tokio::test]
async fn a_hot_reading_raises_an_alert() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let reading = Telemetry {
        device: "dev42".to_owned(),
        temperature: 31.5,
    };
    tb.broker::<MqttBroker>()
        .message(&reading)
        .to("devices/dev42/telemetry")
        .publish()
        .await?;

    tb.broker::<MqttBroker>()
        .subscriber("devices/+/telemetry")
        .assert_called_once()
        .with(&reading)
        .settled(HandlerOutcome::ack());
    tb.broker::<MqttBroker>()
        .published::<Alert>("alerts")
        .assert_called_once()
        .with(&Alert {
            device: "dev42".to_owned(),
        });
    Ok(())
}

examples/manual/mqtt_testing.rs
#[tokio::test]
async fn a_hot_reading_raises_an_alert() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let reading = Telemetry {
        device: "dev42".to_owned(),
        temperature: 31.5,
    };
    tb.broker::<MqttBroker>()
        .message(&reading)
        .to("devices/dev42/telemetry")
        .publish()
        .await?;

    tb.broker::<MqttBroker>()
        .subscriber("devices/+/telemetry")
        .assert_called_once()
        .with(&reading)
        .settled(HandlerOutcome::ack());
    tb.broker::<MqttBroker>()
        .published::<Alert>("alerts")
        .assert_called_once()
        .with(&Alert {
            device: "dev42".to_owned(),
        });
    Ok(())
}

The filter devices/+/telemetry matches one level in place of the +, so a topic one level deeper reaches no handler.

Macros
Manual

examples/mqtt_testing.rs
#[tokio::test]
async fn the_filter_matches_one_level_for_each_plus() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let engine = Telemetry {
        device: "dev42-engine".to_owned(),
        temperature: 21.5,
    };
    let cabin = Telemetry {
        device: "dev42".to_owned(),
        temperature: 21.5,
    };
    // `+` stands for one level, so the first topic is outside the filter.
    tb.broker::<MqttBroker>()
        .message(&engine)
        .to("devices/dev42/engine/telemetry")
        .publish()
        .await?;
    tb.broker::<MqttBroker>()
        .message(&cabin)
        .to("devices/dev42/telemetry")
        .publish()
        .await?;

    tb.broker::<MqttBroker>()
        .subscriber("devices/+/telemetry")
        .assert_called_once()
        .with(&cabin);
    Ok(())
}

examples/manual/mqtt_testing.rs
#[tokio::test]
async fn the_filter_matches_one_level_for_each_plus() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let engine = Telemetry {
        device: "dev42-engine".to_owned(),
        temperature: 21.5,
    };
    let cabin = Telemetry {
        device: "dev42".to_owned(),
        temperature: 21.5,
    };
    // `+` stands for one level, so the first topic is outside the filter.
    tb.broker::<MqttBroker>()
        .message(&engine)
        .to("devices/dev42/engine/telemetry")
        .publish()
        .await?;
    tb.broker::<MqttBroker>()
        .message(&cabin)
        .to("devices/dev42/telemetry")
        .publish()
        .await?;

    tb.broker::<MqttBroker>()
        .subscriber("devices/+/telemetry")
        .assert_called_once()
        .with(&cabin);
    Ok(())
}

Example​

Example