Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix AMQP & MQTT tests for LogAccessor

Browse Source
This commit is contained in:
Artem Bilan
2020-10-06 14:41:32 -04:00
parent c7ff99a4e8
commit fe0e54c46d
2 changed files with 22 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java
View File

@@ -19,7 +19,6 @@ package org.springframework.integration.amqp.outbound;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.startsWith;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.BDDMockito.willReturn;
@@ -29,9 +28,10 @@ import static org.mockito.Mockito.spy;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import org.apache.commons.logging.Log;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentMatchers;
import org.springframework.amqp.core.AmqpReplyTimeoutException;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
@@ -47,6 +47,7 @@ import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.utils.test.TestUtils;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.log.LogAccessor;
import org.springframework.integration.amqp.support.NackedAmqpMessageException;
import org.springframework.integration.amqp.support.ReturnedAmqpMessageException;
import org.springframework.integration.channel.DirectChannel;
@@ -104,14 +105,16 @@ class AsyncAmqpGatewayTests {
receiver.start();
AsyncAmqpOutboundGateway gateway = new AsyncAmqpOutboundGateway(asyncTemplate);
Log logger = spy(TestUtils.getPropertyValue(gateway, "logger", Log.class));
LogAccessor logger = spy(TestUtils.getPropertyValue(gateway, "logger", LogAccessor.class));
given(logger.isDebugEnabled()).willReturn(true);
final CountDownLatch replyTimeoutLatch = new CountDownLatch(1);
willAnswer(invocation -> {
invocation.callRealMethod();
replyTimeoutLatch.countDown();
return null;
}).given(logger).debug(startsWith("Reply not required and async timeout for"));
}).given(logger)
.debug(ArgumentMatchers.<Supplier<String>>argThat(logMessage ->
logMessage.get().startsWith("Reply not required and async timeout for")));
new DirectFieldAccessor(gateway).setPropertyValue("logger", logger);
QueueChannel outputChannel = new QueueChannel();
outputChannel.setBeanName("output");

26
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java
View File

@@ -18,8 +18,6 @@ package org.springframework.integration.mqtt;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.contains;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
@@ -28,15 +26,17 @@ import static org.mockito.Mockito.verify;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import org.apache.commons.logging.Log;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentMatchers;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.log.LogAccessor;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.test.util.TestUtils;
@@ -47,6 +47,8 @@ import org.springframework.test.context.junit4.SpringRunner;
/**
* @author Gary Russell
* @author Artem Bilan
*
* @since 4.0
*
*/
@@ -72,27 +74,29 @@ public class DownstreamExceptionTests {
@Test
public void testNoErrorChannel() throws Exception {
service.n = 0;
Log logger = spy(TestUtils.getPropertyValue(noErrorChannel, "logger", Log.class));
LogAccessor logger = spy(TestUtils.getPropertyValue(noErrorChannel, "logger", LogAccessor.class));
final CountDownLatch latch = new CountDownLatch(1);
doAnswer(invocation -> {
if (((String) invocation.getArgument(0)).contains("Unhandled")) {
if (((Supplier<String>) invocation.getArgument(1)).get().contains("Unhandled")) {
latch.countDown();
}
return null;
}).when(logger).error(anyString(), any(Throwable.class));
}).when(logger).error(any(Throwable.class), any(Supplier.class));
new DirectFieldAccessor(noErrorChannel).setPropertyValue("logger", logger);
MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out");
adapter.setDefaultTopic("mqtt-fooEx1");
adapter.setBeanFactory(mock(BeanFactory.class));
adapter.afterPropertiesSet();
adapter.start();
adapter.handleMessage(new GenericMessage<String>("foo"));
adapter.handleMessage(new GenericMessage<>("foo"));
service.barrier.await(10, TimeUnit.SECONDS);
service.barrier.reset();
adapter.handleMessage(new GenericMessage<String>("foo"));
adapter.handleMessage(new GenericMessage<>("foo"));
service.barrier.await(10, TimeUnit.SECONDS);
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
verify(logger).error(contains("Unhandled exception for"), any(Throwable.class));
verify(logger).error(any(Throwable.class),
ArgumentMatchers.<Supplier<String>>argThat(logMessage ->
logMessage.get().startsWith("Unhandled exception for")));
service.barrier.reset();
adapter.stop();
}
@@ -106,10 +110,10 @@ public class DownstreamExceptionTests {
adapter.setBeanFactory(mock(BeanFactory.class));
adapter.afterPropertiesSet();
adapter.start();
adapter.handleMessage(new GenericMessage<String>("foo"));
adapter.handleMessage(new GenericMessage<>("foo"));
service.barrier.await(10, TimeUnit.SECONDS);
service.barrier.reset();
adapter.handleMessage(new GenericMessage<String>("foo"));
adapter.handleMessage(new GenericMessage<>("foo"));
service.barrier.await(10, TimeUnit.SECONDS);
assertThat(errors.receive(10000)).isNotNull();
service.barrier.reset();