测试文件或管道服务

测试用 TestApp 运行服务自己的应用，运行在进程内传输或者真实流文件上。

被测服务负责结账。订单到达 orders，处理器在 receipts 下以一张收据作为回复。没有金额的订单被丢弃。超出反欺诈检查限额的订单会被暂缓：处理器用 retry_after 要求一分钟后再投递它。经过三次投递后，订单进入死信键 orders.review。应用由 service(file) 构建，所以测试可以选择文件，而 main 使用生产环境的文件。被测服务就是示例文件自己的服务，examples/guide/testing.rs。测试需要在 [dev-dependencies] 中启用 ruststream-sea-file 的 testing feature。

订单及其收据

宏
手写

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

async fn an_order_gets_a_receipt_in(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<FileBroker>()
        .message(&Order {
            id: 1,
            amount_cents: 2_500,
        })
        .publish()
        .await?;

    tb.broker::<FileBroker>()
        .subscriber("orders")
        .assert_called_once()
        .settled(HandlerOutcome::ack());
    tb.broker::<FileBroker>()
        .published::<Receipt>("receipts")
        .assert_called_once()
        .with(&Receipt {
            order: 1,
            amount_cents: 2_500,
        });
    tb.shutdown().await?;
    Ok(())
}

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

async fn an_order_gets_a_receipt_in(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<FileBroker>()
        .message(&Order {
            id: 1,
            amount_cents: 2_500,
        })
        .publish()
        .await?;

    tb.broker::<FileBroker>()
        .subscriber("orders")
        .assert_called_once()
        .settled(HandlerOutcome::ack());
    tb.broker::<FileBroker>()
        .published::<Receipt>("receipts")
        .assert_called_once()
        .with(&Receipt {
            order: 1,
            amount_cents: 2_500,
        });
    tb.shutdown().await?;
    Ok(())
}

TestApp::start 把 FileBroker 连接到内存中的一份日志。路径从不被打开，所以多个测试可以在一个进程中并行运行。测试按生产类型访问 Broker：tb.broker::<FileBroker>()。 publish 在处理器及其回复都完成结算后才返回。所以接下来的几行断言的是已经完成的状态：订阅者如何结算，以及什么到达了 receipts。测试主体把已启动的应用作为参数接收，所以下面的实时测试也运行它。

没有金额的订单

宏
手写

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

    tb.broker::<FileBroker>()
        .message(&Order {
            id: 2,
            amount_cents: 0,
        })
        .publish()
        .await?;

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

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

    tb.broker::<FileBroker>()
        .message(&Order {
            id: 2,
            amount_cents: 0,
        })
        .publish()
        .await?;

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

settled 检查处理器返回的结果。 assert_not_called 检查没有追加任何收据。

不必等待的延迟

宏
手写

src/main.rs
#[tokio::test(start_paused = true)]
async fn an_order_held_three_times_goes_to_review() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let large = Order {
        id: 3,
        amount_cents: 250_000,
    };

    tb.broker::<FileBroker>().message(&large).publish().await?;
    tb.broker::<FileBroker>()
        .subscriber("orders")
        .assert_called_once()
        .settled(HandlerOutcome::retry_after(HOLD));

    // The delay ends in a copy of the order, appended under `orders`.
    tb.advance(HOLD).await?;
    tb.broker::<FileBroker>()
        .published::<Order>("orders")
        .assert_called(2)
        .with(&large)
        .with_header(RETRY_COUNT_HEADER, "1");

    // The third delivery is the last one: it is written under `orders.review` as it arrived.
    tb.advance(HOLD).await?;
    assert_eq!(
        tb.broker::<FileBroker>().subscriber("orders").outcomes(),
        [Outcome::Nack, Outcome::Nack, Outcome::Nack],
    );
    tb.broker::<FileBroker>()
        .published::<Order>("orders.review")
        .assert_called_once()
        .with(&large);
    tb.shutdown().await?;
    Ok(())
}

src/main.rs
#[tokio::test(start_paused = true)]
async fn an_order_held_three_times_goes_to_review() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let large = Order {
        id: 3,
        amount_cents: 250_000,
    };

    tb.broker::<FileBroker>().message(&large).publish().await?;
    tb.broker::<FileBroker>()
        .subscriber("orders")
        .assert_called_once()
        .settled(HandlerOutcome::retry_after(HOLD));

    // The delay ends in a copy of the order, appended under `orders`.
    tb.advance(HOLD).await?;
    tb.broker::<FileBroker>()
        .published::<Order>("orders")
        .assert_called(2)
        .with(&large)
        .with_header(RETRY_COUNT_HEADER, "1");

    // The third delivery is the last one: it is written under `orders.review` as it arrived.
    tb.advance(HOLD).await?;
    assert_eq!(
        tb.broker::<FileBroker>().subscriber("orders").outcomes(),
        [Outcome::Nack, Outcome::Nack, Outcome::Nack],
    );
    tb.broker::<FileBroker>()
        .published::<Order>("orders.review")
        .assert_called_once()
        .with(&large);
    tb.shutdown().await?;
    Ok(())
}

测试在暂停的时钟上启动，advance 让时钟前进。一分钟的 HOLD 不花费任何时间。延迟结束时，retry_after 把订单的副本追加到 orders 下。副本在消息头中带着重试次数，published 把它和测试自己的订单一起显示在 orders 下。第三次投递是挂载点允许的最后一次，所以这个订单按到达时的样子进入 orders.review。每次被暂缓的投递都记录为 Outcome::Nack。

进程内传输模拟了什么

进程内流文件是内存中的一份日志，每个 Broker 值各有一份新的日志，由它的克隆共享。它保留文件所保留的行为：

每个流键的序号从一开始，实时订阅从末端开始。
重放读取日志中已有的内容，读到末尾即完成。
位置和定位按日志解析，拒绝的情形与文件相同。
end_with_eos 在关闭时结束每个订阅。

服务在启动时读取的日志，在 TestApp::start 之前记录在其 Broker 的一个克隆上，恢复与定位的测试就是这样做的。

进程内管道由测试输入的标准输入和服务写出的标准输出组成。 tb.broker::<StdioBroker>().message(..) 是以其键到达的一行，published::<T>(key) 读取这一级写出的一行。只有在 loopback 下，一级才会读到自己的输出，与真实管道上一样。两种传输都拒绝真实传输所拒绝的内容：客户端不接受的流键、管道上的空行，以及不是 1024 倍数的信标间隔。

.ss 文件的字节、文件在多次运行之间保留的内容，以及真实管道上的行格式，只在真实传输上才能看到。

在流文件上运行同一个测试

宏
手写

src/main.rs
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_order_gets_a_receipt_on_a_stream_file() -> Result<(), Box<dyn Error>> {
    // A file of this run's own, so nothing an earlier run appended is read.
    let path = env::temp_dir().join(format!("checkout-{}.ss", process::id()));
    let file = FileBroker::new(path.to_string_lossy());
    let tb = TestApp::start_live(service(file)).await?;
    an_order_gets_a_receipt_in(tb).await?;
    fs::remove_file(&path)?;
    Ok(())
}

src/main.rs
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_order_gets_a_receipt_on_a_stream_file() -> Result<(), Box<dyn Error>> {
    // A file of this run's own, so nothing an earlier run appended is read.
    let path = env::temp_dir().join(format!("manual-checkout-{}.ss", process::id()));
    let file = FileBroker::new(path.to_string_lossy());
    let tb = TestApp::start_live(service(file)).await?;
    an_order_gets_a_receipt_in(tb).await?;
    fs::remove_file(&path)?;
    Ok(())
}

TestApp::start_live 在临时目录中的真实流文件上运行同一个应用。传输就是本地文件系统，所以实时测试不需要服务器，与其他测试一起在 cargo test 中运行。文件以进程命名，所以之前运行追加的内容不会被读到，测试结束时会删除该文件。断言与进程内的完全相同。管道级的实时测试把它作为子进程启动，标准输入和标准输出都接到管道上，shell 管道展示了这一点。

订单及其收据​

没有金额的订单​

不必等待的延迟​

进程内传输模拟了什么​

在流文件上运行同一个测试​

订单及其收据

没有金额的订单

不必等待的延迟

进程内传输模拟了什么

在流文件上运行同一个测试