对正在重试的流进行单元测试

Question

我编写了一个调用 foobarMethod 的流，可能会触发 SocketTimeoutException 的单声道。我希望这个方法重试2次。手动地，此代码可以工作，但在 junit 测试中我无法让它重试。这是因为单元测试的环境问题还是我的测试没有错误？非常感谢任何帮助。谢谢。

public void foobarServiceMethod() {
   foobarClient.foobarMethod(foobarParameter)
            .onErrorMap(ex -> {
                log.error("Code has failed");
                if (ex instanceof SocketTimeoutException) {
                    return new SocketTimeoutException("");
                } else {
                    return ex;
                }
            })
            .retryWhen(
                    Retry.fixedDelay(3, Duration.ofSeconds(1))
                            .filter(SocketTimeoutException.class::isInstance))
            .subscribe(foobarResponse -> handleFoobarResoponse(foobarResponse));

@Test
void foobarRest(CapturedOutput capturedOutput) throws SocketTimeoutException, InterruptedException {
    foobarParameter = "test";
    when(foobarClient.foobarMethod(any()))
            .thenReturn(Mono.error(new SocketTimeoutException("First timeout")))
            .thenReturn(Mono.error(new SocketTimeoutException("First timeout")));

    foobarMethod(foobarParameter);

    assertEquals(2, capturedOutputSplit.contains("Code has failed");
}

Answer 1

要在反应式上运行测试，您必须使用

reactor-test

依赖项，而不是使用

subscribe

单独执行反应式代码，您必须使用

StepVerifier.create(<your publisher>)

调用它。更多详情请参阅这里

<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-test</artifactId>
    <scope>test</scope>
    <version>3.7.1</version>
</dependency>

public Mono<Foo> foobarServiceMethod() {//return Mono/Flux dependig of result of foobarClient.foobarMethod
   return foobarClient.foobarMethod(foobarParameter)
            .onErrorMap(ex -> {
                log.error("Code has failed");
                if (ex instanceof SocketTimeoutException) {
                    return new SocketTimeoutException("");
                } else {
                    return ex;
                }
            })
            .retryWhen(
                    Retry.fixedDelay(3, Duration.ofSeconds(1))
                            .filter(SocketTimeoutException.class::isInstance));

@Test
void foobarRest(CapturedOutput capturedOutput) throws SocketTimeoutException, InterruptedException {
    foobarParameter = "test";
    when(foobarClient.foobarMethod(any()))
            .thenReturn(Mono.error(new SocketTimeoutException("First timeout")))
            .thenReturn(Mono.error(new SocketTimeoutException("First timeout")));

    StepVerifier.create(foobarMethod(foobarParameter))
    .expectErrorSatisfies(..)// error verify logic
    //.expectNextMatches(...) OR check on success
     .expectComplete()
    .verify();
}