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

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

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

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

Тестируемый сервис - оформление заказа. Заказ приходит в checkout-orders, confirm отвечает на него подтверждением в checkout-confirmations, а notify его читает. Заказ без суммы отбрасывается. Заказ, которого нет на складе, ждёт час через retry_after и возвращается на повторную проверку.

Приложение собирает service(broker), поэтому тест выбирает брокер сам, а main использует промышленный. Тестам нужна фича testing крейта ruststream-sqs-sns в [dev-dependencies].

Заказ и его подтверждение​

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

async fn a_confirmation_follows_the_order(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<SqsBroker>()
        .message(&order(1, 2_500, false))
        .publish()
        .await?;

    tb.broker::<SqsBroker>()
        .subscriber("checkout-orders")
        .assert_called_once()
        .with(&order(1, 2_500, false))
        .settled(HandlerOutcome::ack());
    tb.broker::<SqsBroker>()
        .subscriber("checkout-confirmations")
        .assert_called_once()
        .with(&Confirmation {
            order: 1,
            amount_cents: 2_500,
        });
    tb.shutdown().await?;
    Ok(())
}

TestApp::start подключает SqsBroker к SQS внутри процесса. Никакая конечная точка не вызывается, и учётные данные не нужны. Тест обращается к брокеру по его промышленному типу, tb.broker::<SqsBroker>().

publish возвращается, когда обработчики, до которых дошла публикация, завершили работу, включая ответ и его читателя. Поэтому следующие строки проверяют уже готовое состояние: как confirm завершил заказ и что получил notify. Тело принимает запущенное приложение аргументом, поэтому его выполняет и тест на стенде ниже.

Пустой заказ​

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

    tb.broker::<SqsBroker>()
        .message(&order(2, 0, false))
        .publish()
        .await?;

    tb.broker::<SqsBroker>()
        .subscriber("checkout-orders")
        .assert_called_once()
        .settled(HandlerOutcome::drop());
    tb.broker::<SqsBroker>()
        .published::<Confirmation>("checkout-confirmations")
        .assert_not_called();
    tb.shutdown().await?;
    Ok(())
}

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

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

src/main.rs
#[tokio::test(start_paused = true)]
async fn a_back_ordered_order_comes_back_after_the_restock() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(service(SqsBroker::new())).await?;
    let waiting = order(3, 2_500, true);

    tb.broker::<SqsBroker>().message(&waiting).publish().await?;
    tb.advance(RESTOCK).await?;

    tb.broker::<SqsBroker>()
        .subscriber("checkout-orders")
        .assert_called(2)
        .with(&waiting)
        .settled(HandlerOutcome::retry_after(RESTOCK));
    // The same message came back; nothing was published again to bring it.
    tb.broker::<SqsBroker>()
        .published::<Order>("checkout-orders")
        .assert_called_once();
    tb.shutdown().await?;
    Ok(())
}

Тест стартует на остановленных часах, и advance переводит их вперёд. Час RESTOCK ничего не стоит. retry_after делает сообщение невидимым на время задержки, как это делает SQS, и очередь выдаёт его повторно, когда часы её пройдут. published показывает один заказ в checkout-orders: повторная доставка - это то же сообщение, а не копия.

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

SQS внутри процесса - это аккаунт внутри теста, и в нём существует каждая очередь и каждая тема, которую называет сервис. Он моделирует то, от чего зависит логика сервиса:

  • Очередь отдаёт каждое сообщение одной из своих подписок, а полученное сообщение остаётся невидимым, пока его не удалят или не истечёт его тайм-аут видимости.
  • Счётчик получений растёт с каждым получением, а политика перенаправления переносит в DLQ сообщение, у которого кончились получения.
  • Пакет содержит не больше десяти сообщений, которые возвращает одно получение.
  • Группа FIFO ждёт, пока одно из её сообщений в обработке, а идентификатор дедупликации запоминается на пять минут.
  • С фичей sns тема доставляет копию в каждую очередь, подписанную через subscribe_queue_to_topic.

То, что отвергают SQS или SNS, отвергается и там: больше десяти заголовков, имя заголовка, которого SQS не принимает, пустое тело или значение заголовка, сообщение больше предельного размера, имя очереди длиннее 80 символов, DLQ другого вида, maxReceiveCount больше 1000, видимость, которая закончилась бы позже двенадцати часов после получения, и очередь FIFO на стандартной теме.

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

Тот же тест на запущенном стенде​

src/main.rs
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_order_is_confirmed_on_a_live_stack() -> Result<(), Box<dyn Error>> {
    let Some(endpoint) = live::endpoint("SQS_TEST_ENDPOINT") else {
        return Ok(());
    };
    delete_queues(&endpoint).await?;
    let tb = TestApp::start_live(service(live::broker(&endpoint))).await?;
    a_confirmation_follows_the_order(tb).await
}

TestApp::start_live подключает то же приложение к запущенному SQS. Тест читает конечную точку из SQS_TEST_ENDPOINT через live::endpoint, вспомогательную функцию самого примера, и пропускается, если переменная не задана. Сначала он удаляет очереди сервиса, чтобы не учитывать то, что в них оставил прошлый прогон; при старте сервис создаёт их заново. Проверки те же, что внутри процесса, без изменений.

cargo test
SQS_TEST_ENDPOINT=http://localhost:4566 cargo test -- --test-threads=1