使用StepVerifier对Flux.take(持续时间)进行单元测试

Question

我正在使用弹簧反应堆核心 3.0.6，我有一个返回 Flux 的方法：

public Flux<Foo> createFlux(){
return Flux.<List<Foo>,String>generate(/* generator omitted for clarity's sake */ )
.take(Duration.ofSeconds(10)
.flatMap(Flux::fromIterable);
}

生成器函数调用分页的 REST API 来获取结果，如果 API 继续返回数据，我希望 Flux 只运行 10 秒。

它工作正常，但我想创建一些单元测试，并且在创建测试以验证 Flux 最多只运行 10 秒时遇到麻烦。

我嘲笑了其余服务，以便它始终返回数据并编写了以下内容：

StepVerifier.withVirtualTime(() -> createFlux())
.thenAwait(Duration.ofSeconds(10))
.verifyComplete();

但它失败了：

java.lang.AssertionError: expectation "expectComplete" failed (expected: onComplete(); actual: onNext([my toString() Foo bean]))

我想我应该以某种方式使用生成的项目，但我找不到正确的 StepVerifier 方法来执行此操作。

编辑

我尝试跳过每个项目thenConsumeWhile：

StepVerifier.withVirtualTime(() -> createFlux())
.thenAwait(Duration.ofSeconds(10))
.thenConsumeWhile(t -> true)
.verifyComplete();

但现在测试只是无限期运行，永远不会结束。

Answer 1

生成器实际上可能非常重要...StepVerifier受到无限序列的限制，在使用虚拟时间时更是如此。问题是生成器和thenAwait都在主线程中运行，因此生成器是无限的，阻止步进验证器前进时间，这反过来又阻止序列超时。

由于您想测试take的持续时间，我认为虚拟时间是不对的(您正在测试时间的模拟)。我会使 createFlux 方法可以使用拍摄持续时间进行参数化，并执行一个StepVerifier.create()，持续时间要短得多。

如果你真的想使用某种形式的虚拟时间，我发现让它工作的最低要求是

1. 通过在测试开始时实例化Scheduler来隔离非虚拟线程上的生成器循环，然后在 StepVerifier 的Supplier中使用subscribeOn(scheduler)。
2. 通过先调用.expectNextCount(1)，确保在尝试提前时间之前订阅所有内容并且数据开始流动。

喜欢这个：

public Flux<Integer> createFlux() {
return Flux.<List<Integer>>generate(sink -> {
sink.next(Arrays.asList(1, 2, 3));
})
.take(Duration.ofSeconds(10))
.flatMap(Flux::fromIterable);
}
@Test
public void so44657525() throws InterruptedException {
Scheduler scheduler = Schedulers.newSingle("test");
AtomicInteger adder = new AtomicInteger();
StepVerifier.withVirtualTime(() -> createFlux()
.subscribeOn(scheduler)
.doOnNext(v -> adder.incrementAndGet())
)
.expectNextCount(1)
.thenAwait(Duration.ofSeconds(10))
.thenConsumeWhile(t -> true)
.verifyComplete();
System.out.println("Total number of values in generated lists: " + adder.get());
}

将expectNextCount(1)修改为expectNextCount(100_000)，我运行了一次打印Total number of values in generated lists: 102405并花费了 40 毫秒。

Answer 2

如果您错过了参考指南，也许可以使您走上正确的道路？

除了最常见的expectNext，您必须对序列中的每个项目重复，如果您知道元素的数量，您可以使用expectNextCount或thenConsumeWhile基于谓词跳过元素，