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

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

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

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

Тестируемый сервис списывает платежи по картам. Платёж приходит в очередь card-payments, и обработчик отвечает квитанцией в card-receipts. Платёж без суммы отбрасывается. Платёж сверх мгновенного лимита банка откладывается: обработчик просит обработать его снова через минуту с retry_after. После трёх доставок платёж уходит в card-payments.review. Приложение строит service(broker), поэтому тест выбирает брокер сам, а main берёт рабочий. Тестам нужна фича testing крейта ruststream-lapin в [dev-dependencies].

Платёж и квитанция​

src/main.rs
#[tokio::test]
async fn a_payment_gets_a_receipt() -> Result<(), Box<dyn Error>> {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    let tb = TestApp::start(service(broker)).await?;
    a_receipt_follows_a_payment(tb).await
}

async fn a_receipt_follows_a_payment(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<LapinBroker>()
        .message(&Payment {
            id: 1,
            amount_cents: 2_500,
        })
        .publish()
        .await?;

    tb.broker::<LapinBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::ack());
    tb.broker::<LapinBroker>()
        .published::<Receipt>("card-receipts")
        .assert_called_once()
        .with(&Receipt {
            payment: 1,
            amount_cents: 2_500,
        });
    tb.shutdown().await?;
    Ok(())
}

TestApp::start подключает LapinBroker к RabbitMQ внутри процесса. URI разбирается так же, как при настоящем соединении, но соединение не устанавливается. Тест обращается к брокеру по его рабочему типу: tb.broker::<LapinBroker>(). Payment объявляет свой адрес назначения, card-payments, поэтому тест публикует его без to(..). publish возвращает управление, когда обработчик и его ответ завершены. Поэтому следующие строки проверяют уже готовое состояние: как подписчик подтвердил доставку и что дошло до card-receipts. Тело теста принимает запущенное приложение аргументом, поэтому его же выполняет и тест на сервере ниже.

Отброшенный платёж​

src/main.rs
#[tokio::test]
async fn a_payment_without_an_amount_is_dropped() -> Result<(), Box<dyn Error>> {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    let tb = TestApp::start(service(broker)).await?;

    tb.broker::<LapinBroker>()
        .message(&Payment {
            id: 2,
            amount_cents: 0,
        })
        .publish()
        .await?;

    tb.broker::<LapinBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::drop());
    tb.broker::<LapinBroker>()
        .published::<Receipt>("card-receipts")
        .assert_not_called();
    tb.shutdown().await?;
    Ok(())
}

settled проверяет исход, который вернул обработчик. assert_not_called проверяет, что квитанция не опубликована.

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

src/main.rs
#[tokio::test(start_paused = true)]
async fn a_payment_the_bank_keeps_holding_goes_to_review() -> Result<(), Box<dyn Error>> {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    let tb = TestApp::start(service(broker)).await?;
    let held = Payment {
        id: 3,
        amount_cents: 250_000,
    };

    tb.broker::<LapinBroker>().message(&held).publish().await?;
    tb.broker::<LapinBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::retry_after(HOLD));

    // The service holds the copy and publishes it back to the queue after `HOLD`.
    tb.advance(HOLD).await?;
    tb.broker::<LapinBroker>()
        .published::<Payment>("card-payments")
        .assert_called(2)
        .with(&held)
        .with_header(RETRY_COUNT_HEADER, "1");

    // The third delivery is the last one: it goes to the review queue as it arrived.
    tb.advance(HOLD).await?;
    assert_eq!(
        tb.broker::<LapinBroker>()
            .subscriber("card-payments")
            .outcomes(),
        [Outcome::Nack, Outcome::Nack, Outcome::Nack],
    );
    tb.broker::<LapinBroker>()
        .published::<Payment>("card-payments.review")
        .assert_called_once()
        .with(&held);
    tb.shutdown().await?;
    Ok(())
}

Тест начинается с остановленными часами, а advance сдвигает их вперёд. Минута HOLD не стоит ничего. У очереди нет delay, поэтому копию держит сервис и по окончании задержки публикует её обратно в card-payments. Копия хранит число попыток в заголовке, и published показывает её рядом с собственным сообщением теста. Третья доставка - последняя, которую допускает точка монтирования, поэтому платёж уходит в card-payments.review в том виде, в каком пришёл. Каждая отложенная доставка записывается как Outcome::Nack.

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

RabbitMQ внутри процесса читает настройки брокера и дескрипторы маршрутов - те же, что видит сервер. declare_topology, как и на сервере, решает, объявляет ли подписка свою очередь. Он моделирует именно RabbitMQ, а не очередь вообще:

  • Обменники по умолчанию, direct, topic, fanout и headers маршрутизируют по привязкам, которые объявляют дескрипторы.
  • Конкурирующие потребители делят одну очередь, а сообщение, возвращённое в очередь, снова проходит через неё.
  • Очередь хранит то, что в неё пришло, пока её никто не читает, а закрывшийся потребитель возвращает то, что взял и не подтвердил.
  • Кворумная очередь считает доставки и переносит исчерпавшее их сообщение в очередь недоставленных сообщений по своему маршруту.
  • Очередь ожидания выпускает отложенную копию, а direct reply-to даёт каждому запросу собственный адрес.
  • Транзакция остаётся невидимой до фиксации.

Он отвергает то, что отвергает сервер: имя или заголовок, которые протокол не может передать, тело больше 16 МиБ, неустойчивую очередь без признака эксклюзивности, привязку к обменнику по умолчанию, повторное объявление очереди с другими настройками, кворумную очередь с приоритетом. Доставка, подтверждённая после закрытия её соединения, возвращает ошибку закрытого канала.

То, что есть только у сервера, требует сервера: окно предвыборки, подтверждения издателя, атомарность транзакции канала, обменники из плагинов, очередь, которую делят два соединения, и очередь, которую сервис ожидает найти без объявления.

Тот же тест на сервере​

src/main.rs
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_payment_gets_a_receipt_on_a_server() -> Result<(), Box<dyn Error>> {
    let Some(url) = live::url("AMQP_TEST_URL") else {
        return Ok(());
    };
    let tb = TestApp::start_live(service(LapinBroker::new(url).declare_topology(true))).await?;
    a_receipt_follows_a_payment(tb).await
}

TestApp::start_live подключает то же приложение к работающему RabbitMQ. Тест читает URI из AMQP_TEST_URL через live::url, собственную вспомогательную функцию примера, и пропускается, если переменная не задана. Брокер объявляет свою топологию, поэтому тест работает на пустом сервере. Проверки те же, что и внутри процесса, без изменений. На сервере RabbitMQ сам выбирает, какой потребитель очереди получит сообщение, поэтому тест на сервере держит одну подписку на очередь.

cargo test
AMQP_TEST_URL=amqp://127.0.0.1:5672 cargo test -- --test-threads=1

Для запуска на сервере достаточно RabbitMQ из руководства Начало работы.