如何使用接收器和 @StreamListener 编写测试 Junit 消息传递测试

Question

我有一个带有 kafka 绑定的 spring-cloud-stream 应用程序。我想创建一个测试 Junit。

\n

我的类定义如下：-

\n

@EnableBinding(Sink.class)\n@Slf4j\npublic class Messaging {\n\n    @Autowired\n    private RestTemplate restTemplate;\n\n    @Value("${messaging}")\n    private String url;\n\n    @Value("${messaging.prefix}")\n    private String messaging;\n    \n    @StreamListener(Sink.INPUT) \n    public void handle(Message<String> request) {\n        log.info("Topic name ==> %s :", request.getPayload());\n        try {   \n            String jsonString = request.getPayload().replace("\\\\", "").replace("\\"{", "{").replace("}\\"", "}");\n            JsonObject jsonObject = (JsonObject)jsonParser.parse(request.getPayload());\n            String urlRequest =url.concat(jsonObject.get("targetClass").getAsString()).concat(messaging);\n            HttpEntity<Object> entity = new HttpEntity<Object>(jsonString, getHeaderMap(request.getHeaders()));\n            ResponseEntity<String> response = restTemplate.postForEntity(urlRequest, entity, String.class);\n\n        \n        } catch (ValidationException validationException) {\n            log.error("Error de validaci\xc3\xb3n: {}", validationException.getMessage());\n        } catch (Exception e) {\n            log.error("Error ", e.getMessage());\n        }\n    }\n

\n

我的频道定义如下：-

\n

public interface Sink {\n    String INPUT = "input";\n\n    @Input(Sink.INPUT)\n    SubscribableChannel input();\n}\n

\n

这是我的应用程序属性，用于将您的通道绑定到同一队列：

\n

spring:\n  cloud: \n    bus: \n      destination:CONFIG\n      enabled: true\n    stream: \n      bindings:\n        input:\n          group: input_messaging          \n          contentType: application/json\n          destination: CONFIG_Test1,CONFIG_Test2\n

\n

这是我创建的测试，但它给出了这个错误：\n原因：org.springframework.context.ApplicationContextException：由于缺少 ServletWebServerFactory bean，无法启动 ServletWebServerApplicationContext。

\n

@RunWith(SpringRunner.class)\n@SpringBootTest\n@TestPropertySource("classpath:test.properties")\n@EnableConfigurationProperties\n@ContextConfiguration(classes = MessagingTestConfig.class)\npublic class MessagingListenerTest {\n\n    @Before\n    public void setup() {\n\n    }\n    \n    @Test\n    @SuppressWarnings("unchecked")\n    public void testReturn() throws Exception {\n            \n        ConfigurableApplicationContext context = SpringApplication.run(MessagingTest.class,\n                "--server.port=0",\n                "--spring.jmx.enabled=false",\n                "--spring.cloud.stream.bindings.input.contentType=text/plain",\n                "--spring.cloud.stream.bindings.output.contentType=text/plain");\n        \n        MessageCollector collector = context.getBean(MessageCollector.class);\n        Processor processor = context.getBean(Processor.class);\n        Sink inputChannel = context.getBean(Sink.class);\n        \n        Message<String> request = MessageBuilder.withPayload("headers").setHeader("topicName", "topic-1").build();\n        \n        inputChannel.input()\n                .send(request);\n        Message<String> message = (Message<String>) collector\n                .forChannel(processor.output()).poll(1, TimeUnit.SECONDS);\n        \n        assertThat(message).isNotNull();\n        assertThat(message.getPayload()).contains("topicName");\n        context.close();\n    }\n    \n    @EnableBinding(Sink.class)\n    @EnableAutoConfiguration\n    public class TestProcessor  {\n        \n        @StreamListener(Sink.INPUT)\n        public Message<String> hundle(Message<String> messageHundle) {\n            return  messageHundle;\n        }\n    }\n\n    }\n

\n

我想知道是否有方法来测试我的课程，如何进行，感谢您的帮助

\n

这是使用最新版本的 Spring Cloud Stream 进行的更改

\n

package com.common.messaging;\n\nimport static org.assertj.core.api.Assertions.assertThat;\nimport static org.mockito.ArgumentMatchers.any;\nimport static org.mockito.ArgumentMatchers.eq;\nimport static org.mockito.Mockito.mock;\nimport static org.mockito.Mockito.when;\n\nimport java.nio.charset.StandardCharsets;\n\nimport org.junit.Before;\nimport org.junit.Test;\nimport org.junit.runner.RunWith;\nimport org.mockito.Mock;\nimport org.mockito.Mockito;\nimport org.springframework.beans.factory.annotation.Autowired;\nimport org.springframework.boot.test.context.SpringBootTest;\nimport org.springframework.cloud.stream.binder.test.InputDestination;\nimport org.springframework.cloud.stream.binder.test.OutputDestination;\nimport org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;\nimport org.springframework.context.annotation.Configuration;\nimport org.springframework.context.annotation.Import;\nimport org.springframework.http.HttpStatus;\nimport org.springframework.http.ResponseEntity;\nimport org.springframework.integration.config.EnableIntegration;\nimport org.springframework.messaging.Message;\nimport org.springframework.messaging.support.GenericMessage;\nimport org.springframework.messaging.support.MessageBuilder;\nimport org.springframework.test.annotation.DirtiesContext;\nimport org.springframework.test.context.ActiveProfiles;\nimport org.springframework.test.context.ContextConfiguration;\nimport org.springframework.test.context.TestPropertySource;\nimport org.springframework.test.context.junit4.SpringRunner;\nimport org.springframework.test.util.ReflectionTestUtils;\nimport org.springframework.web.client.RestTemplate;\n\nimport lombok.extern.slf4j.Slf4j;\n\n@EnableIntegration\n@Configuration\n@TestPropertySource(locations="classpath:/msc-test.properties")\n@Slf4j\n@RunWith(SpringRunner.class)\n@ActiveProfiles("test")\n@ContextConfiguration(classes = MessagingListenerTestConfig.class)\n@Import(TestChannelBinderConfiguration.class)\n@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)\n@DirtiesContext\npublic class MessagingListenerTest {\n\n    @Autowired\n    private MessagingListener listener;\n\n    @Autowired\n    private InputDestination inputDestination;\n\n    @Autowired\n    private OutputDestination outputDestination;\n\n    @Mock\n    private RestTemplate restTemplate;\n\n    private static final String URL = "http://localhost:8080/";\n\n    @Before\n    public void setup() {\n\n        restTemplate = mock(RestTemplate.class);\n        ReflectionTestUtils.setField(listener, "restTemplate", restTemplate);\n\n        ResponseEntity<String> mockResponse = new ResponseEntity<>("{}", HttpStatus.ACCEPTED);\n        when(restTemplate.postForEntity(any(), any(), eq(String.class))).thenReturn(mockResponse);\n\n    }\n\n    @Test\n    public void testHundleMessage() {\n\n        String expectedUrl = URL;\n        Message<String> request = MessageBuilder.withPayload("headers").setHeader("topicName", "topic-1").build();\n        log.info("request Test :", request.getPayload());\n         \n        //inputDestination.send(new GenericMessage<byte[]>(request.getPayload().getBytes(StandardCharsets.UTF_8)));\n        listener.handle(request);\n\n        //Verificar la recepci\xc3\xb3n de los mensajes\n        assertThat(outputDestination.receive()).isNotNull();\n        assertThat(outputDestination.receive().getPayload().toString()).contains("topicName");\n\n        //Verificar la url del restTemplate\n        Mockito.verify(restTemplate, Mockito.times(1)).postForEntity(expectedUrl, any(), eq(String.class));\n\n    }\n}\n

\n

但鉴于此错误，我不知道为什么

\n

java.lang.IndexOutOfBoundsException: Index: 0, Size: 0\nat java.util.ArrayList.rangeCheck(ArrayList.java:653)\nat java.util.ArrayList.get(ArrayList.java:429)\nat org.springframework.cloud.stream.binder.test.OutputDestination.receive(OutputDestination.java:59)\nat org.springframework.cloud.stream.binder.test.OutputDestination.receive(OutputDestination.java:73)\nat com.common.messaging.MessagingListenerTest.testHundleMessage(MessagingListenerTest.java:87)\nat sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nat sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\nat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\nat java.lang.reflect.Method.invoke(Method.java:497)\nat org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)\nat org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)\nat org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)\nat org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)\nat org.springframework.test.context.junit4.statements.RunBeforeTestExecutionCallbacks.evaluate(RunBeforeTestExecutionCallbacks.java:74)\nat org.springframework.test.context.junit4.statements.RunAfterTestExecutionCallbacks.evaluate(RunAfterTestExecutionCallbacks.java:84)\nat org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)\n

\n

Answer 1

我仍然遇到同样的问题

\n

@EnableIntegration\n@Configuration\n@TestPropertySource(locations="classpath:/msc-test.properties")\n@Slf4j\n@RunWith(SpringRunner.class)\n@ActiveProfiles("test")\n@ContextConfiguration(classes = MessagingListenerTestConfig.class)\n@Import(TestChannelBinderConfiguration.class)\n@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)\n@DirtiesContext\npublic class MessagingListenerTest {\n\n    @Autowired\n    private MessagingListener listener;\n\n    @Autowired\n    private InputDestination inputDestination;\n\n    @Autowired\n    private OutputDestination outputDestination;\n\n    @Mock\n    private RestTemplate restTemplate;\n\n    private static final String EXPECTED_URL = "http://localhost:11000/test/v2/verification/messaging";\n\n    @Before\n    public void setup() {\n\n        restTemplate = mock(RestTemplate.class);\n        ReflectionTestUtils.setField(listener, "restTemplate", restTemplate);\n\n        ResponseEntity<String> mockResponse = new ResponseEntity<>("{}", HttpStatus.ACCEPTED);\n        when(restTemplate.postForEntity(any(), any(), eq(String.class))).thenReturn(mockResponse);\n\n    }\n\n    @Test\n    public void testHundleMessage() {\n\n        \n        JSONObject obj1 = new JSONObject()\n                .put("id", 1)\n                .put("targetClass", "/test/v2/verification");\n        \n        \n        Message<String> request = MessageBuilder.withPayload(obj1.toString()).build();\n        log.info("request Test : "+ request.getPayload());\n         \n        \n        inputDestination.send(new GenericMessage<byte[]>(request.getPayload().getBytes(StandardCharsets.UTF_8)));\n        listener.handle(request);\n\n        //Verificar la url del restTemplate\n        Mockito.verify(restTemplate, Mockito.times(1)).postForEntity(eq(EXPECTED_URL), any(), eq(String.class));\n        \n        //Verificar la recepci\xc3\xb3n de los mensajes\n        assertThat(outputDestination.receive()).isNotNull();\n        assertThat(outputDestination.receive().getPayload().toString()).contains("topicName");\n\n\n    }\n\n}\n

\n

就在这一行

\n

inputDestination.send(new GenericMessage<byte[]>(request.getPayload().getBytes(StandardCharsets.UTF_8)));\n

\n

这是错误 Junit

\n

java.lang.IndexOutOfBoundsException: Index: 0, Size: 0\n    at java.util.ArrayList.rangeCheck(ArrayList.java:653)\n    at java.util.ArrayList.get(ArrayList.java:429)\n    at org.springframework.cloud.stream.binder.test.AbstractDestination.getChannel(AbstractDestination.java:34)\n    at org.springframework.cloud.stream.binder.test.InputDestination.send(InputDestination.java:37)\n    at com.common.messaging.MessagingListenerTest.testHundleMessage(MessagingListenerTest.java:93)\n

\n

和控制台错误

\n

2020-07-14 11:29:16.850  INFO 25240 --- [           main] c.b.a.m.c.m.MessagingListenerTest        : The following profiles are active: test\n2020-07-14 11:29:18.171  INFO 25240 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.\n2020-07-14 11:29:18.192  INFO 25240 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'taskScheduler' has been explicitly defined. Therefore, a default ThreadPoolTaskScheduler will be created.\n2020-07-14 11:29:18.212  INFO 25240 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'integrationHeaderChannelRegistry' has been explicitly defined. Therefore, a default DefaultHeaderChannelRegistry will be created.\n2020-07-14 11:29:18.392  INFO 25240 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationChannelResolver' of type [org.springframework.integration.support.channel.BeanFactoryChannelResolver] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)\n2020-07-14 11:29:18.429  INFO 25240 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationDisposableAutoCreatedBeans' of type [org.springframework.integration.config.annotation.Disposables] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)\n2020-07-14 11:29:20.113  INFO 25240 --- [           main] o.s.s.c.ThreadPoolTaskScheduler          : Initializing ExecutorService 'taskScheduler'\n2020-07-14 11:29:20.356  INFO 25240 --- [           main] o.s.i.e.EventDrivenConsumer              : Adding {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel\n2020-07-14 11:29:20.358  INFO 25240 --- [           main] o.s.i.c.PublishSubscribeChannel          : Channel 'application.errorChannel' has 1 subscriber(s).\n2020-07-14 11:29:20.361  INFO 25240 --- [           main] o.s.i.e.EventDrivenConsumer              : started bean '_org.springframework.integration.errorLogger'\n2020-07-14 11:29:20.382  INFO 25240 --- [           main] c.b.a.m.c.m.MessagingListenerTest        : Started MessagingListenerTest in 4.629 seconds (JVM running for 9.331)\n2020-07-14 11:29:23.255  INFO 25240 --- [           main] c.b.a.m.c.m.MessagingListenerTest        : request Test : {"targetClass":"/test/v2/verification","id":1}\n2020-07-14 11:29:28.207  INFO 25240 --- [           main] o.s.i.e.EventDrivenConsumer              : Removing {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel\n2020-07-14 11:29:28.207  INFO 25240 --- [           main] o.s.i.c.PublishSubscribeChannel          : Channel 'application.errorChannel' has 0 subscriber(s).\n2020-07-14 11:29:28.207  INFO 25240 --- [           main] o.s.i.e.EventDrivenConsumer              : stopped bean '_org.springframework.integration.errorLogger'\n2020-07-14 11:29:28.208  INFO 25240 --- [           main] o.s.s.c.ThreadPoolTaskScheduler          : Shutting down ExecutorService 'taskScheduler'\nPicked up JAVA_TOOL_OPTIONS: -agentpath:"C:\\windows\\system32\\Aternity\\Java\\JavaHookLoader.dll"="C:\\ProgramData\\Aternity\\hooks"\n

\n

可能是什么问题呢

\n