如何在spring云流中模拟KStream进行单元测试

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.MessageListener;

public class KafkaConsumer implements MessageListener<String, String> {

    private String result;

    @Override
    public void onMessage(ConsumerRecord<String, String> record) {
        result = record.value();
    }

    public String getResult() {
        return result;
    }
}

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.context.annotation.Import;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.test.context.ActiveProfiles;

@SpringBootTest
@ActiveProfiles("test")
@EnableBinding(Processor.class)
@Import(KafkaConsumer.class)
public class KafkaStreamUnitTest {

    @Autowired
    private Processor processor;

    @Autowired
    private KafkaConsumer kafkaConsumer;

    @Test
    public void testKStream() {
        MessageChannel inputChannel = processor.input();
        SubscribableChannel outputChannel = processor.output();

        inputChannel.send(MessageBuilder.withPayload("test").build());

        //等待KStream处理完成
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        String result = kafkaConsumer.getResult();
        //执行断言或其他操作
    }

    @StreamListener(Processor.INPUT)
    @Output(Processor.OUTPUT)
    public void processKStream(String input) {
        // 对KStream进行处理
        // 将结果发送到Processor的输出通道
    }

}