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

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

在 SQL 上测试服务

测试用 TestApp 运行服务自己的应用，连接的是真实的数据库：内存中的 SQLite，或者服务器上测试专用的数据库。

队列的行为存在于它的 SQL 中：领取、结算，以及服务自己的 Publish。因此测试框架没有数据库的模型。它的进程内模式在测试交给服务的数据库上执行服务的语句，并记下测试框架所等待的账目。

被测服务生成缩略图。小图片立即处理。缩放程序只在第二次尝试时才接受大图片，所以大图片的第一次投递用 retry_after 推迟一分钟。应用由 service(url) 构建，所以测试可以选择数据库，而 main 使用生产环境的数据库。测试需要在 [dev-dependencies] 中启用 ruststream-sqlx 的 testing feature。

小图片​

src/main.rs
/// A database in memory that every connection naming it shares, alive while one of them is
/// open: the test holds that one beside the pool it reads the table through. The service
/// creates the table when it starts.
async fn database(
    name: &str,
) -> Result<(String, SqlitePool, SqliteConnection), Box<dyn Error>> {
    let url = format!("sqlite:file:{name}?mode=memory&cache=shared");
    let options: SqliteConnectOptions = url.parse()?;
    let keeper = SqliteConnection::connect_with(&options).await?;
    let pool = SqlitePoolOptions::new()
        .min_connections(1)
        .idle_timeout(None)
        .connect_with(options)
        .await?;
    Ok((url, pool, keeper))
}

/// The scenario, on whichever database the app runs.
async fn a_small_image_is_made_at_once(broker: BrokerHandle<'_>) -> Result<(), Box<dyn Error>> {
    let image = Image {
        path: "cat.png".to_owned(),
        large: false,
    };
    broker.message(&image).publish().await?;

    broker
        .subscriber("thumbnails")
        .assert_called_once()
        .with(&image)
        .settled(HandlerOutcome::ack());
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn in_process() -> Result<(), Box<dyn Error>> {
    let (url, pool, _keeper) = database("guide_testing").await?;
    let tb = TestApp::start(service(&url)).await?;
    a_small_image_is_made_at_once(tb.broker::<SqlxBroker<Sqlite>>()).await?;

    let left: i64 = sqlx::query_scalar("SELECT count(*) FROM thumbnail_jobs")
        .fetch_one(&pool)
        .await?;
    assert_eq!(left, 0, "the acknowledgement deleted the row");
    Ok(())
}

database 在内存中创建一个 SQLite 数据库，所有指名它的连接共享这个数据库。只要其中一个连接仍然打开，它就存在，所以测试在连接池之外再持有一个连接，并通过这个连接池读取表。表由服务在启动时自己创建，与生产环境一样。TestApp::start 在进程内把生产环境的应用连接到这个数据库，测试按生产类型访问 Broker：tb.broker::<SqlxBroker<Sqlite>>()。

图片像在生产环境中一样，经由路由和服务自己的 Publish 进入队列。publish 在它唤醒的处理器完成结算后才返回，所以它之后的几行读到的是已经完成的状态：thumbnail 如何结算该行，以及测试自己的查询在表中找到什么。服务在应用启动之前用自己的 SQL 写入的行（初始数据或迁移），在它的订阅打开时就被计入，所以 settle 像等待已发布的消息一样等待它。测试运行期间这样写入的行，只有在领取取走它时才被计入，所以 settle 不等待它；这样的测试在真实时钟上运行，并让轮询间隔过去，任务入队一页就是这样做的。场景把 Broker 的句柄作为参数接收，所以下面的 Postgres 测试也运行它。

没有载荷列的表把行本身交给它的处理器（行）。测试这样的处理器时，用 with_value 和 received_values 检查行，代替 with 和 received。

延迟，而无需等待​

src/main.rs
#[tokio::test]
async fn a_large_image_is_made_a_minute_later() -> Result<(), Box<dyn Error>> {
    let (url, _pool, _keeper) = database("guide_testing_clock").await?;
    // The queue reads "now" from this clock, and the database calls run off it. A paused
    // clock takes the single-threaded runtime `#[tokio::test]` starts.
    tokio::time::pause();
    let tb = TestApp::start(service(&url)).await?;
    tb.broker::<SqlxBroker<Sqlite>>()
        .message(&Image {
            path: "panorama.png".to_owned(),
            large: true,
        })
        .publish()
        .await?;
    tb.advance(Duration::from_secs(61)).await?;

    // Put off for a minute, then made.
    let outcomes = tb
        .broker::<SqlxBroker<Sqlite>>()
        .subscriber("thumbnails")
        .assert_called(2)
        .outcomes();
    assert_eq!(outcomes, [Outcome::Nack, Outcome::Ack]);
    Ok(())
}

在进程内，队列从 tokio 时钟读取“现在”，并在 Broker 连接时以系统时钟为基准。测试暂停这个时钟，再用 advance 推动它：大图片的行在一分钟后到期，第二次投递生成缩略图。这一分钟不花任何时间。数据库调用脱离暂停的时钟运行，所以数据库应答期间时钟不会跳动。时钟越过租约时，租约也以同样的方式到期。使用数据库时钟的表读取数据库的时间，测试时钟推动不了它，服务自己的时钟也照常生效；这些情形的测试在真实时钟上运行。

在 Postgres 上运行同一个测试​

src/main.rs
// Skips without `POSTGRES_TEST_URL`; fails the skip under `RUSTSTREAM_REQUIRE_LIVE`.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn on_postgres() -> Result<(), Box<dyn Error>> {
    // A database of the test's own on the server, dropped at the end.
    let Some(db) = live::postgres::database().await else {
        return Ok(());
    };
    let tb = TestApp::start_live(service_on_postgres(&db.url())).await?;
    a_small_image_is_made_at_once(tb.broker::<SqlxBroker<Postgres>>()).await?;

    tb.shutdown().await?;
    db.finish().await;
    Ok(())
}

在生产环境中运行于 Postgres 的服务，也要在 Postgres 上测试：SQLite 执行的是另一套 SQL，没有行锁、没有通知，也没有可选的隔离级别。TestApp::start_live 以普通方式连接应用，并在真实时钟上等待。测试在服务器上取得一个专用的数据库，服务在其中创建自己的表，因此不会读到先前运行留下的任何东西；live::postgres::database 是本仓库测试中的辅助函数，#[sqlx::test] 为测试提供同样的东西，即一个已应用迁移的数据库。测试从 POSTGRES_TEST_URL 读取服务器地址，没有这个变量时跳过，所以在没有服务器的机器上 cargo test 也能通过；在必须有服务器的地方，RUSTSTREAM_REQUIRE_LIVE=1 把这种跳过变成失败。场景就是进程内的那一个，原样不变。

cargo test
POSTGRES_TEST_URL=postgres://user:password@localhost:5432/postgres cargo test

测试中的 outbox​

测试构建默认关闭 outbox，除非 RUSTSTREAM_SQLX_OUTBOX=on 启用它，所以服务自己的测试在不跟踪的情况下运行处理器。测试跟踪本身的测试在设置了开关时运行，没有开关时跳过（开关）。