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

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

Тестируемый сервис списывает платежи по картам. Платёж приходит в поток card-payments, а обработчик отвечает квитанцией в card-receipts. Платёж без суммы отбрасывается. Каждый счёт может тратить 1 000,00 в день: обработчик прибавляет сумму к ключу spent:<account> через пул соединений брокера и откладывает платёж сверх лимита на минуту через retry_after. После трёх доставок платёж уходит в card-payments.review, поток dead-letter. Возврат приходит в надёжный список card-refunds и возвращает сумму. Приложение строит service(url), поэтому тест выбирает сервер, а main использует промышленный. Обработчик вызывает пул через трейты команд fred, поэтому fred входит в зависимости сервиса.

Тестам нужны фича testing крейта ruststream-fred и рантайм Tokio:

Cargo.toml

[dev-dependencies]
ruststream-fred = { version = "0.7", features = ["testing"] }
tokio = { version = "1", features = ["macros", "rt", "rt-multi-thread"] }

Платёж и его квитанция

Макросы

src/main.rs

/// The body both modes run; only the start call differs.
async fn a_payment_is_answered_with_a_receipt(
    tb: TestApp<()>,
    account: &str,
) -> Result<(), Box<dyn Error>> {
    tb.broker::<RedisBroker>()
        .message(&payment(1, account, 2500))
        .publish()
        .await?;

    tb.broker::<RedisBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::ack());
    tb.broker::<RedisBroker>()
        .published::<Receipt>("card-receipts")
        .assert_called_once()
        .with(&Receipt { payment_id: 1 });
    tb.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    a_payment_is_answered_with_a_receipt(TestApp::start(app()).await?, "acct-1").await
}

Вручную

src/main.rs

/// The body both modes run; only the start call differs.
async fn a_payment_is_answered_with_a_receipt(
    tb: TestApp<()>,
    account: &str,
) -> Result<(), Box<dyn Error>> {
    tb.broker::<RedisBroker>()
        .message(&payment(1, account, 2500))
        .publish()
        .await?;

    tb.broker::<RedisBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::ack());
    tb.broker::<RedisBroker>()
        .published::<Receipt>("card-receipts")
        .assert_called_once()
        .with(&Receipt { payment_id: 1 });
    tb.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    a_payment_is_answered_with_a_receipt(TestApp::start(app()).await?, "acct-1").await
}

TestApp::start подключает RedisBroker к Redis, смоделированному внутри тестового процесса. К URL никто не подключается. Тест обращается к брокеру по его промышленному типу, tb.broker::<RedisBroker>(). Сообщение теста записывается в своё имя так, как этого ждёт подписка, которая читает это имя: LPUSH для списка, PUBLISH для канала или шаблона, XADD для потока и для имени, которое не читает ни одна подписка. 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(app()).await?;

    tb.broker::<RedisBroker>()
        .message(&payment(2, "acct-2", 0))
        .publish()
        .await?;

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

Вручную

src/main.rs

#[tokio::test]
async fn a_payment_without_an_amount_is_dropped() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;

    tb.broker::<RedisBroker>()
        .message(&payment(2, "acct-2", 0))
        .publish()
        .await?;

    tb.broker::<RedisBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::drop());
    tb.broker::<RedisBroker>()
        .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_held_payment_passes_once_a_refund_frees_room() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let redis = tb.broker::<RedisBroker>();

    redis
        .message(&payment(3, "acct-3", 80_000))
        .publish()
        .await?;
    redis
        .message(&payment(4, "acct-3", 30_000))
        .publish()
        .await?;
    redis
        .subscriber("card-payments")
        .assert_called(2)
        .settled(HandlerOutcome::retry_after(HOLD));

    // The refund is a list entry; the handler gives the room back through the pool.
    redis
        .message(&Refund {
            account: "acct-3".to_owned(),
            amount_cents: 50_000,
        })
        .publish()
        .await?;
    tb.advance(HOLD).await?;

    assert_eq!(
        redis.subscriber("card-payments").outcomes(),
        [Outcome::Ack, Outcome::Nack, Outcome::Ack],
    );
    redis
        .published::<Receipt>("card-receipts")
        .assert_called(2)
        .with(&Receipt { payment_id: 4 });
    tb.shutdown().await?;
    Ok(())
}

Вручную

src/main.rs

#[tokio::test(start_paused = true)]
async fn a_held_payment_passes_once_a_refund_frees_room() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let redis = tb.broker::<RedisBroker>();

    redis
        .message(&payment(3, "acct-3", 80_000))
        .publish()
        .await?;
    redis
        .message(&payment(4, "acct-3", 30_000))
        .publish()
        .await?;
    redis
        .subscriber("card-payments")
        .assert_called(2)
        .settled(HandlerOutcome::retry_after(HOLD));

    // The refund is a list entry; the handler gives the room back through the pool.
    redis
        .message(&Refund {
            account: "acct-3".to_owned(),
            amount_cents: 50_000,
        })
        .publish()
        .await?;
    tb.advance(HOLD).await?;

    assert_eq!(
        redis.subscriber("card-payments").outcomes(),
        [Outcome::Ack, Outcome::Nack, Outcome::Ack],
    );
    redis
        .published::<Receipt>("card-receipts")
        .assert_called(2)
        .with(&Receipt { payment_id: 4 });
    tb.shutdown().await?;
    Ok(())
}

INCRBY и DECRBY обработчика выполняются на Redis внутри процесса, как и любая команда крейта. Второй платёж превышает лимит счёта, поэтому его откладывают. Возврат - это запись списка, и его обработчик освобождает лимит.

Тест начинается с остановленными часами, а advance сдвигает их вперёд. Минута HOLD проходит мгновенно. Redis внутри процесса тоже читает часы теста, поэтому идентификаторы записей, время простоя и оценки во множестве задержек не меняются, пока часы стоят. outcomes перечисляет, как завершилась каждая доставка, по порядку. Отложенная доставка записана как Outcome::Nack.

Предел попыток

Макросы

src/main.rs

#[tokio::test(start_paused = true)]
async fn a_payment_held_three_times_goes_to_review() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;

    tb.broker::<RedisBroker>()
        .message(&payment(5, "acct-5", 150_000))
        .publish()
        .await?;
    tb.advance(HOLD).await?;
    tb.advance(HOLD).await?;

    tb.broker::<RedisBroker>()
        .subscriber("card-payments")
        .assert_called(3);
    tb.broker::<RedisBroker>()
        .published::<Payment>("card-payments.review")
        .assert_called_once()
        .with(&payment(5, "acct-5", 150_000))
        .with_header(RETRY_COUNT_HEADER, "3");
    tb.shutdown().await?;
    Ok(())
}

Вручную

src/main.rs

#[tokio::test(start_paused = true)]
async fn a_payment_held_three_times_goes_to_review() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;

    tb.broker::<RedisBroker>()
        .message(&payment(5, "acct-5", 150_000))
        .publish()
        .await?;
    tb.advance(HOLD).await?;
    tb.advance(HOLD).await?;

    tb.broker::<RedisBroker>()
        .subscriber("card-payments")
        .assert_called(3);
    tb.broker::<RedisBroker>()
        .published::<Payment>("card-payments.review")
        .assert_called_once()
        .with(&payment(5, "acct-5", 150_000))
        .with_header(RETRY_COUNT_HEADER, "3");
    tb.shutdown().await?;
    Ok(())
}

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

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

Redis внутри процесса принимает команды, которые отправляет крейт, и отвечает на них так же, как Redis 8.4, поэтому подписка с перехватом монтируется. Все настройки он читает из брокера: URL, размер пула, группу по умолчанию и топологию кластера. Он хранит:

• типы ключей, поэтому запись неверного типа возвращает ошибку WRONGTYPE;
• группы потребителей с курсором, неподтверждёнными записями и счётчиками доставок, XAUTOCLAIM, XREADGROUP .. CLAIM и перемотку;
• списки вместе со списком обработки и множеством для восстановления надёжного списка;
• каналы Pub/Sub, шаблоны и шардированные каналы, которые ничего не хранят для отсутствующего подписчика;
• упорядоченные множества, срок жизни ключей и MULTI / EXEC;
• в топологии кластера отказ выполнять команду или транзакцию, которая охватывает несколько хеш-слотов.

Собственные команды обработчика, через пул или в конвейерном раунде, тоже выполняются на нём. Скрипты он не выполняет: EVAL, EVALSHA, FCALL и незнакомая ему команда возвращают ошибку, которая их называет. Рукопожатие не выполняется, поэтому аутентификация и TLS не проверяются, как и переключение при отказе и перераспределение слотов. Для них нужен сервер.

Тот же тест с сервером

Макросы

src/main.rs

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_a_server() -> Result<(), Box<dyn Error>> {
    let Some(url) = live::url("REDIS_TEST_URL") else {
        return Ok(());
    };
    // A server keeps what earlier runs left, so each run spends from an account of its own.
    let run = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos();
    let account = format!("acct-{run}");
    a_payment_is_answered_with_a_receipt(TestApp::start_live(service(&url)).await?, &account)
        .await
}

Вручную

src/main.rs

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_a_server() -> Result<(), Box<dyn Error>> {
    let Some(url) = live::url("REDIS_TEST_URL") else {
        return Ok(());
    };
    // A server keeps what earlier runs left, so each run spends from an account of its own.
    let run = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos();
    let account = format!("acct-{run}");
    a_payment_is_answered_with_a_receipt(TestApp::start_live(service(&url)).await?, &account)
        .await
}

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

cargo test
docker run --name redis -p 6379:6379 redis:8.4
REDIS_TEST_URL=redis://localhost:6379 cargo test -- --test-threads=1