如何单独测试Quarkus SmallRye Reactive Messaging的process方法？

核心思路：隔离目标方法 #

不用手动移除注解，Quarkus Reactive Messaging提供了专门的测试工具，通过替换通道为内存实现或禁用非目标通道，就能单独测试process方法。

具体实现方案 #

1. 用内存连接器隔离通道（推荐集成测试） #

借助Quarkus的InMemoryConnector，把process方法的输入/输出通道替换成内存通道，同时禁用produce和consume的通道：

首先确保测试依赖已添加（根据你用的消息中间件调整，比如AMQP/Kafka）：

<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-reactive-messaging-amqp-test</artifactId>
    <scope>test</scope>
</dependency>

编写测试类：

import io.quarkus.test.junit.QuarkusTest;
import io.smallrye.reactive.messaging.MutinyEmitter;
import io.smallrye.reactive.messaging.test.common.InMemoryConnector;
import io.smallrye.reactive.messaging.test.common.InMemorySink;
import jakarta.inject.Inject;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;

@QuarkusTest
public class MyReactiveMessagingProcessTest {

    @Inject
    InMemoryConnector connector;

    @Test
    void testProcessLogic() {
        // 配置：绑定process方法的输入/输出通道到内存连接器
        connector.source("classA-channel", InMemoryConnector.sourceConfig()); // 对应process的@Incoming通道名
        connector.sink("classB-channel", InMemoryConnector.sinkConfig()); // 对应process的@Outgoing通道名

        // 禁用produce和consume的通道，避免自动触发
        System.setProperty("mp.messaging.outgoing.produce-channel.enabled", "false");
        System.setProperty("mp.messaging.incoming.consume-channel.enabled", "false");

        // 发送测试用的ClassA对象到process的输入通道
        MutinyEmitter<ClassA> emitter = connector.source("classA-channel").mutinyEmitter();
        emitter.send(new ClassA("test-content"));

        // 从输出通道获取结果并验证
        InMemorySink<ClassB> sink = connector.sink("classB-channel");
        ClassB processedResult = sink.received().get(0).getPayload();
        assertEquals("processed-test-content", processedResult.getContent()); // 替换为你的预期值
    }
}

2. 禁用特定通道启动 #

如果只想阻止produce/consume自动运行，直接在测试类上通过配置禁用对应通道：

@QuarkusTest
@ConfigProperty(name = "mp.messaging.outgoing.produce-channel.enabled", value = "false")
@ConfigProperty(name = "mp.messaging.incoming.consume-channel.enabled", value = "false")
public class MyReactiveMessagingProcessTest {

    @Inject
    @Channel("classA-channel") // process的@Incoming通道
    MutinyEmitter<ClassA> inputEmitter;

    @Inject
    @Channel("classB-channel") // process的@Outgoing通道
    Multi<ClassB> outputStream;

    @Test
    void testProcess() {
        // 发送测试数据
        inputEmitter.send(new ClassA("test"));

        // 订阅输出流并验证结果
        ClassB result = outputStream.first().await().indefinitely();
        assertEquals("expected-result", result.getContent());
    }
}

3. 纯单元测试（不启动Quarkus容器） #

如果不需要容器上下文，直接实例化类调用process方法：

import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;

public class MyReactiveMessagingUnitTest {

    @Test
    void testProcessMethod() {
        MyReactiveMessaging service = new MyReactiveMessaging();
        // 若有依赖，用Mockito等工具mock
        ClassA input = new ClassA("test-data");
        ClassB result = service.process(input); // 如果是Reactive类型，用.await()处理
        assertEquals("processed-data", result.getValue());
    }
}

针对你的入门项目的自定义测试建议 #

你的项目基于Quarkus Reactive Messaging官方模板，直接套用第一种方案即可：

1. 找到process方法对应的@Incoming和@Outgoing通道名称（比如模板里的"prices-in"/"prices-out"）
2. 用InMemoryConnector绑定这些通道，发送测试数据并验证输出
3. 通过配置禁用produce和consume对应的通道，避免自动触发消息

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