跳到主要内容
英文版本为准

本页译自英文，由模型翻译。若与英文原文存在差异，以英文原文为准。

测试 Redis 服务

测试用 TestApp 运行服务自己的应用，连接的是进程内 Redis 或者服务器。

被测服务负责银行卡扣款。 付款到达流 card-payments，处理器在 card-receipts 上以一张收据作为回复。 没有金额的付款被丢弃。 每个账户每天最多花费 1,000.00：处理器通过 Broker 的连接池把金额加到键 spent:<account> 上，超出限额的付款用 retry_after 暂缓一分钟。 经过三次投递后，付款进入死信流 card-payments.review。 退款到达可靠列表 card-refunds，并把金额退回。 应用由 service(url) 构建，所以测试可以选择服务器，而 main 使用生产环境的服务器。 处理器通过 fred 的命令 trait 调用连接池，所以服务把 fred 列入依赖。

测试需要 ruststream-fred 的 testing feature 和一个 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
}

TestApp::start 把 RedisBroker 连接到在测试进程内模拟的 Redis。 测试从不连接这个 URL。 测试按生产类型访问 Broker：tb.broker::<RedisBroker>()。 测试消息按读取该名字的订阅所期望的方式写入：列表用 LPUSH，频道或模式用 PUBLISH，流以及没有订阅读取的名字用 XADD。 publish 在处理器及其回复都完成结算后才返回。 所以接下来的几行断言的是已经完成的状态：订阅者如何结算，以及什么到达了 card-receipts。 测试主体把已启动的应用作为参数接收，所以下面的 live 测试也运行它。

没有金额的付款​

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(())
}

处理器的 INCRBY 和 DECRBY 在进程内 Redis 上执行，本 crate 的每条命令都是如此。 第二笔付款让账户超出限额，所以它被暂缓。 退款是一个列表条目，它的处理器把额度退回。

测试在暂停的时钟上启动，advance 让时钟前进。 一分钟的 HOLD 不花费任何时间。 进程内 Redis 也读取测试的时钟，所以在时钟前进之前，条目 id、空闲时间和延迟集合的分值都保持不变。 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(())
}

超出每日限额的付款在每次投递时都被暂缓。 每次暂缓都把一份副本发布回流，重试次数加一，published 读回写入某个名字的每条内容，这些副本也包括在内。 第三次投递是挂载点允许的最后一次，所以这笔付款按到达时的样子进入 card-payments.review。

进程内 Redis 模拟了什么​

进程内 Redis 接收本 crate 发送的命令，并像 Redis 8.4 那样应答，所以认领订阅可以挂载。 它从 Broker 读取每项设置：URL、连接池大小、默认组和集群拓扑。 它维护：

  • 键的类型，所以写入错误类型会返回 WRONGTYPE 错误；
  • 消费者组及其游标、待处理条目和投递次数，XAUTOCLAIM、XREADGROUP .. CLAIM 和 seek；
  • 列表，包括可靠列表的处理中列表和它的恢复集合；
  • 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
}

TestApp::start_live 把同一个应用连接到正在运行的 Redis。 这时，发布会在真实时钟上等待，直到每个订阅都处理完它，最多十秒。 测试用 live::url（示例自带的辅助函数）从 REDIS_TEST_URL 读取地址，未设置时跳过。 服务器保留着之前运行留下的数据，所以测试从一个专属账户中扣款。 断言与进程内的完全相同。

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