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

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

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

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

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

Платёж с ключом​

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

async fn a_receipt_names_the_account_of_the_key(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<KinesisBroker>()
        .message(&Payment {
            id: 1,
            amount_cents: 2_500,
        })
        .with_headers(keyed("acct-7"))
        .publish()
        .await?;

    tb.broker::<KinesisBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::ack());
    tb.broker::<KinesisBroker>()
        .published::<Receipt>("card-receipts")
        .assert_called_once()
        .with(&Receipt {
            payment: 1,
            account: "acct-7".to_owned(),
        });
    tb.shutdown().await?;
    Ok(())
}

TestApp::start подключает KinesisBroker к Kinesis внутри процесса. Тест не обращается ни к одной конечной точке AWS, и учётные данные ему не нужны. Тест обращается к брокеру по его рабочему типу: tb.broker::<KinesisBroker>(). Ключ сегмента записи передаётся заголовком partition-key (PARTITION_KEY_HEADER), поэтому тест задаёт его через with_headers. Брокер превращает этот заголовок в ключ сегмента записи, а доставка возвращает его в том же заголовке. publish возвращает управление, когда обработчик и его ответ завершены. Поэтому следующие строки проверяют уже готовое состояние: как подписчик завершил доставку и что дошло до card-receipts. Тело теста принимает запущенное приложение аргументом, поэтому его же выполняет и тест на стенде ниже.

Платёж без суммы​

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

    tb.broker::<KinesisBroker>()
        .message(&Payment {
            id: 2,
            amount_cents: 0,
        })
        .with_headers(keyed("acct-7"))
        .publish()
        .await?;

    tb.broker::<KinesisBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::drop());
    tb.broker::<KinesisBroker>()
        .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(KinesisBroker::new())).await?;
    let held = Payment {
        id: 3,
        amount_cents: 250_000,
    };

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

    // The delay is a copy of the record, published back to the stream after `HOLD`.
    tb.advance(HOLD).await?;
    tb.broker::<KinesisBroker>()
        .published::<Payment>("card-payments")
        .assert_called(2)
        .with(&held)
        .with_header(RETRY_COUNT_HEADER, "1")
        .with_header(PARTITION_KEY_HEADER, "acct-7");

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

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

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

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

  • Брокер читает поток одной подпиской: она читает каждую его запись и открывается с конца. Вторая подписка на тот же поток отклоняется, как её отклоняет брокер сервиса.
  • start_at и дескриптор перемотки обработчика читают журнал потока заново с позиции, которую они называют (Повторное чтение и перемотка).
  • На retry без предела Kinesis внутри процесса отвечает так же, как Kinesis: не поддерживается, и запись не возвращается.
  • Отложенный повтор возвращается копией со счётчиком попыток, а запись, исчерпавшая попытки, уходит в поток мёртвых писем.
  • Запись, которую отвергает сервис Kinesis, отвергается и здесь: имя потока длиннее 128 символов или вне [a-zA-Z0-9_.-], пустой ключ сегмента или ключ длиннее 256 символов, запись больше 10 МиБ, наибольшего размера, который можно настроить для потока. Тест, который публикует такую запись, получает ошибку сервиса.
  • Издатель или дескриптор перемотки, использованный после shutdown, возвращает ошибку KinesisError::NotConnected.

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

Тот же тест на стенде​

src/main.rs
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_keyed_payment_gets_a_receipt_on_a_stack() -> Result<(), Box<dyn Error>> {
    let Some(endpoint) = support::endpoint() else {
        return Ok(());
    };
    // The service reads `card-payments` and creates it; the receipts are only published.
    support::ensure_stream(&support::config(&endpoint).await, "card-receipts", 1).await;
    let broker = KinesisBroker::new()
        .endpoint(endpoint)
        .region("us-east-1")
        .test_credentials();
    let tb = TestApp::start_live(service(broker)).await?;
    a_receipt_names_the_account_of_the_key(tb).await
}

TestApp::start_live подключает то же приложение к работающему Kinesis, здесь к LocalStack. Публикация ждёт завершения реакции до десяти секунд. start_live_within задаёт более долгий срок. Тест читает адрес из KINESIS_TEST_ENDPOINT через support::endpoint, вспомогательную функцию примеров, и пропускается, если переменная не задана. Сервис создаёт card-payments, когда открывается его подписка. Тест создаёт card-receipts, в который только пишут. Подписка открывается с конца потока, поэтому записи, которые оставил в потоке прошлый прогон, не читаются. Проверки те же, что и внутри процесса, без изменений.

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