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

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

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

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

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

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

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

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

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

TestApp::start подключает NatsBroker к NATS внутри процесса. Соединение по адресу не устанавливается, и стрим не нужно создавать заранее. Тест обращается к брокеру по его рабочему типу: tb.broker::<NatsBroker>(). message(..).publish() публикует платёж в субъект, который объявляет его тип, и стрим сохраняет его. 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(NatsBroker::new("nats://localhost:4222"))).await?;
    let no_card = Payment {
        card: String::new(),
        ..payment(2, 2_500)
    };

    tb.broker::<NatsBroker>()
        .message(&no_card)
        .publish()
        .await?;

    tb.broker::<NatsBroker>()
        .subscriber("card.payments")
        .assert_called_once()
        .settled(HandlerOutcome::drop());
    tb.broker::<NatsBroker>()
        .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(NatsBroker::new("nats://localhost:4222"))).await?;
    let held = payment(3, 250_000);

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

    // The consumer holds the payment and delivers it again once `HOLD` has passed.
    tb.advance(HOLD).await?;
    tb.broker::<NatsBroker>()
        .subscriber("card.payments")
        .assert_called(2);

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

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

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

NATS внутри процесса читает то же, что прочитал бы клиент: адреса и те параметры соединения, которые меняют то, что доставляет сервер (no_echo, префикс инбокса). Маршрутизирует и хранит он так же, как сервер:

  • Субъекты сопоставляются с шаблонами * и >, а очередь отдаёт каждое сообщение одному участнику.
  • Стрим ведёт порядковые номера и окно дубликатов и проверяет ожидания, которые задаёт публикация.
  • Консьюмер начинает с того места, которое задаёт его политика доставки, и заполняет метаданные контекста JetStream.
  • Подтверждение, отрицательное подтверждение с задержкой и без неё и прекращение доставки действуют так же, как на сервере. Доставка без исхода возвращается после ack_wait.
  • Запрос доходит до отвечающей стороны приложения, а на запрос, который никто не читает, сразу приходит «no responders».

Он отклоняет то же, что отклоняют клиент или сервер: субъект с пробельными символами, заголовок, у которого нет формы в NATS, сообщение больше 1 МиБ и публикацию JetStream, которую не принимает ни один стрим. Стрим внутри процесса хранит то, что читают консьюмеры приложения, а подписка находит каждый стрим, который называет. Собственный список субъектов стрима и его пределы, max_ack_pending, аутентификация и TLS требуют сервера.

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

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("NATS_TEST_URL") else {
        return Ok(());
    };
    // The stream is the server's state: the test creates it, replacing what an interrupted
    // run left behind, and deletes it afterwards.
    let admin = NatsBroker::new(url.as_str()).connect().await?;
    let jetstream = admin.jetstream();
    let _ = jetstream.delete_stream("CARD_PAYMENTS").await;
    jetstream
        .create_stream(StreamConfig {
            name: "CARD_PAYMENTS".to_owned(),
            subjects: vec!["card.payments".to_owned()],
            ..StreamConfig::default()
        })
        .await?;

    let tb = TestApp::start_live(service(NatsBroker::new(url.as_str()))).await;
    let outcome = match tb {
        Ok(tb) => a_charged_payment_gets_its_receipt(tb).await,
        Err(err) => Err(err.into()),
    };

    jetstream.delete_stream("CARD_PAYMENTS").await?;
    admin.shutdown().await?;
    outcome
}

TestApp::start_live подключает то же приложение к работающему серверу NATS. Тест берёт адрес из NATS_TEST_URL через live::url, вспомогательную функцию самого примера, и пропускается, если переменная не задана. Стрим - это состояние сервера, поэтому тест сначала создаёт CARD_PAYMENTS через собственное соединение, а в конце удаляет его. Вместе со стримом удаляется и его консьюмер, поэтому следующий запуск начинается с чистого состояния. Проверки те же, что и внутри процесса, без изменений.

cargo test
NATS_TEST_URL=nats://localhost:4222 cargo test -- --test-threads=1