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

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

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

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

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

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

src/main.rs
#[tokio::test]
async fn a_payment_gets_a_receipt() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(service(AmqpBroker::new(URL))).await?;
    a_charged_payment_gets_its_receipt(tb, 1).await
}

async fn a_charged_payment_gets_its_receipt(
    tb: TestApp<()>,
    id: u64,
) -> Result<(), Box<dyn Error>> {
    tb.broker::<AmqpBroker>()
        .message(&payment(id, 2_500))
        .publish()
        .await?;

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

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

Платёж без карты​

src/main.rs
#[tokio::test]
async fn a_payment_without_a_card_is_dropped() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(service(AmqpBroker::new(URL))).await?;

    tb.broker::<AmqpBroker>()
        .message(&Payment {
            id: 2,
            card: String::new(),
            amount_cents: 2_500,
        })
        .publish()
        .await?;

    tb.broker::<AmqpBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::drop());
    tb.broker::<AmqpBroker>()
        .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 tb = TestApp::start(service(AmqpBroker::new(URL))).await?;
    let held = payment(3, 250_000);

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

    // The delay is a copy of the payment, published back to its queue after `HOLD`.
    tb.advance(HOLD).await?;
    tb.broker::<AmqpBroker>()
        .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 address as it arrived.
    tb.advance(HOLD).await?;
    assert_eq!(
        tb.broker::<AmqpBroker>()
            .subscriber("card-payments")
            .outcomes(),
        [Outcome::Nack, Outcome::Nack, Outcome::Nack],
    );
    tb.broker::<AmqpBroker>()
        .published::<Payment>("card-payments.review")
        .assert_called_once()
        .with(&held);
    tb.shutdown().await?;
    Ok(())
}

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

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

Брокер внутри процесса читает URL брокера так же, как connect. URL, который клиент не может открыть, не даёт тесту запуститься, и amqps:// там тоже требует фичи rustls или native-tls. Публикация превращается в сообщение AMQP и обратно, поэтому обработчик читает те же заголовки, что и при доставке с работающего брокера. Брокер внутри процесса моделирует маршрутизацию брокера:

  • Потребители адреса queue конкурируют, и каждое сообщение доставляется одному из них; каждая подписка topic получает копию.
  • Адрес raw доставляется точно в том виде, в каком написан, одному потребителю.
  • Подписка на очередь создаёт свою очередь, и очередь хранит то, что приходит, пока к ней не присоединён ни один потребитель.
  • retry учитывает попытку в delivery-count, а неурегулированная доставка возвращается в очередь, когда её подписка завершается.
  • Подписка at-most-once получает доставки уже урегулированными, и ack возвращает AckError::Unsupported.
  • Транзакция ничего не публикует до фиксации, а запрос завершается по тайм-ауту, когда никто не отвечает.

Для кредита связи, политики недоставленных сообщений и предела доставок брокера, долговечности и адресов, которые брокер создаёт при публикации, нужен работающий брокер. Внутри процесса сообщение в адрес, на который не открыта ни одна подписка на очередь, отбрасывается.

Тот же тест на брокере​

src/main.rs
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_payment_gets_a_receipt_from_a_broker() -> Result<(), Box<dyn Error>> {
    let Some(url) = live::url("AMQP_TEST_URL") else {
        return Ok(());
    };
    let tb = TestApp::start_live(service(AmqpBroker::new(url))).await?;
    a_charged_payment_gets_its_receipt(tb, u64::from(process::id())).await
}

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

cargo test
AMQP_TEST_URL=amqp://artemis:artemis@localhost:5672 cargo test -- --test-threads=1