Перейти к основному содержимому
Нормативная версия документации - английская

Эта страница переведена с английского языковой моделью. При любом расхождении верен английский оригинал.

Тестирование сервиса на MQTT

Тест запускает собственное приложение сервиса через TestApp на внутрипроцессном сервере MQTT или на Mosquitto.

Тестируемый сервис следит за морозильными камерами. Каждая камера публикует свою температуру в freezers/<id>/temperature, а мониторы делят показания в группе совместной подписки. Каждое показание записывается в журнал температур. Камера теплее -18 °C поднимает тревогу в freezers/alerts. Когда журнал температур не отвечает, обработчик через retry_after просит доставить показание снова через десять секунд. После трёх доставок показание уходит в freezers/review, тему dead-letter.

Обработчик watch публикует тревогу через слот Alerts. Приложение строит service(broker), поэтому тест выбирает брокер сам, а main берёт рабочий. Тестам нужна фича testing крейта ruststream-rumqttc в [dev-dependencies].

Тёплая камера​

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 подключает MqttBroker к внутрипроцессному серверу MQTT. Соединение по URL не устанавливается. Тест обращается к брокеру по его рабочему типу: tb.broker::<MqttBroker>(). Reading не называет темы, поэтому тест называет её сам через .to("freezers/f1/temperature"), как это сделала бы камера. Фильтр обработчика сопоставляет её так же, как Mosquitto. publish возвращает управление, когда обработчик и его тревога завершены. Поэтому следующие строки проверяют уже готовое состояние: как подписка завершила доставку и что дошло до freezers/alerts. К подписке тест обращается по её фильтру, без группы совместной подписки. Тело теста принимает запущенное приложение аргументом, поэтому его же выполняет и тест на Mosquitto ниже.

Холодная камера​

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 проверяет исход, который вернул обработчик. tb.out::<Alerts>() читает то, что обработчик опубликовал через слот Alerts, а assert_not_called проверяет, что он ничего не опубликовал.

Задержка без ожидания​

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(())
}

Тест начинается с остановленными часами, а advance переводит их вперёд. Десять секунд RETRY проходят мгновенно. Когда задержка истекает, retry_after публикует копию показания в freezers/retry/temperature, и фильтр читает её снова. В копии есть заголовок с числом попыток, и published его показывает. Третья доставка - последняя, которую разрешает точка монтирования, поэтому показание уходит в freezers/review в том виде, в каком пришло. Каждая отложенная доставка записывается как Outcome::Nack.

Что моделирует внутрипроцессный сервер​

Внутрипроцессный сервер читает собственные настройки брокера, а публикация проходит ту же проверку и то же отображение на свойства MQTT 5, что и при передаче по сети. Он моделирует сервер, с которым работает одно соединение:

  • Публикация доходит до каждой подписки, фильтр которой подходит к её теме, с учётом +, # и правила для тем, которые начинаются с $.
  • Группа совместной подписки получает одну копию на всех участников, а пересекающиеся фильтры получают каждый свою.
  • Сохранённое сообщение доходит до более поздней подписки вне группы совместной подписки, а пустая сохранённая полезная нагрузка его стирает.
  • Сообщение больше max_packet_size не доходит, а тема или заголовок, которые отклоняет клиент, приводят к ошибке публикации.
  • Подтверждение после остановки отклоняется.

Каждый брокер, подключённый внутри процесса, - это отдельный сервер. Постоянной сессии и её возобновлению, последней воле, окну Receive Maximum, обменам подтверждениями при QoS 1 и 2 и нескольким соединениям с одним сервером нужен настоящий брокер.

Тот же тест на 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 подключает то же приложение к работающему серверу. Тест читает адрес из MQTT_TEST_URL через live::url, собственную вспомогательную функцию примера, и сразу завершается, если переменная не задана. Он подключается со своим идентификатором клиента, поэтому делит сервер с другими запусками. Проверки те же, что внутри процесса, без изменений.

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

RUSTSTREAM_REQUIRE_LIVE=1 превращает пропуск в ошибку. Это нужно задаче CI, которая запускает сервер и не должна проходить без него.