如何测试订阅Uni的代码？解决测试竞态条件难题

一、不修改业务代码：同步等待Uni完成 #

如果暂时不想改动业务逻辑，可以通过Mock控制Uni的执行节奏，结合同步工具类或Mockito捕获回调来避免竞态条件：

方法1：用CountDownLatch等待Uni完成

@Test
public void shouldLogSuccessAfterReprocessing() throws InterruptedException {
    CountDownLatch completionLatch = new CountDownLatch(1);
    
    // 构造一个完成时触发latch的mock Uni
    Uni<Void> mockUni = Uni.createFrom().voidItem()
        .invoke(completionLatch::countDown);

    // 让service返回这个mock Uni
    when(service.reprocessAll()).thenReturn(mockUni);

    // 执行被测方法
    service.execute();

    // 等待Uni完成，超时时间按需设置
    if (!completionLatch.await(3, TimeUnit.SECONDS)) {
        fail("Uni未在预期时间内执行完成");
    }

    // 此时再验证日志调用
    Mockito.verify(log, times(1)).info("Reprocessing ran successfully.");
}

方法2：捕获订阅回调手动触发

直接捕获subscribe().with()中的成功/错误回调，手动触发后验证日志：

@Test
public void shouldLogSuccessAfterReprocessing() {
    // 构造空Uni用于捕获回调
    Uni<Void> mockUni = Uni.createFrom().voidItem();
    when(service.reprocessAll()).thenReturn(mockUni);

    // 执行被测方法
    service.execute();

    // 捕获订阅时传入的成功回调
    ArgumentCaptor<Consumer<Void>> successCaptor = ArgumentCaptor.forClass(Consumer.class);
    verify(mockUni).subscribe().with(successCaptor.capture(), any(Consumer.class));

    // 手动触发成功回调
    successCaptor.getValue().accept(null);

    // 验证日志输出
    Mockito.verify(log, times(1)).info("Reprocessing ran successfully.");
}

二、重构业务代码：提升可测性（推荐） #

从响应式编程的最佳实践出发，让业务代码本身更易测试，推荐以下两种重构方式：

方式1：让execute方法返回Uni

把订阅逻辑交给调用方，方法仅负责构建Uni的处理流程：

public Uni<Void> execute() {
    return service.reprocessAll()
        .onItem().invoke(() -> log.info("Reprocessing ran successfully."))
        .onFailure().invoke(error -> log.severe("Reprocessing failed: " + error.getMessage()));
}

测试时直接用你熟悉的UniAssertSubscriber即可，完全避免竞态：

@Test
public void shouldLogSuccessAfterReprocessing() {
    when(service.reprocessAll()).thenReturn(Uni.createFrom().voidItem());

    service.execute()
        .subscribe().withSubscriber(UniAssertSubscriber.create())
        .assertCompleted();

    Mockito.verify(log, times(1)).info("Reprocessing ran successfully.");
}

方式2：抽离回调逻辑作为参数

将日志处理等回调逻辑从execute中抽离，作为方法参数传入：

public void execute(Consumer<Void> onSuccess, Consumer<Throwable> onError) {
    service.reprocessAll().subscribe().with(onSuccess, onError);
}

业务代码调用时传入日志逻辑：

// 业务侧调用
execute(
    success -> log.info("Reprocessing ran successfully."),
    error -> log.severe("Reprocessing failed: " + error.getMessage())
);

测试时传入自定义回调验证逻辑，同时等待Uni完成：

@Test
public void shouldLogSuccessAfterReprocessing() {
    AtomicBoolean successTriggered = new AtomicBoolean(false);
    when(service.reprocessAll()).thenReturn(Uni.createFrom().voidItem());

    service.execute(
        success -> successTriggered.set(true),
        error -> fail("错误回调不应被触发")
    );

    // 等待Uni执行完成
    service.reprocessAll()
        .subscribe().withSubscriber(UniAssertSubscriber.create())
        .assertCompleted();

    assertTrue(successTriggered.get());
    // 若需验证日志，可单独测试业务侧的回调传入逻辑
}

内容的提问来源于stack exchange，提问作者Allan Juan