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

Various tests fixes

Browse Source 
* Use a tmp file in the `FileTests.testFileReadingFlow()`
for writing content and then rename it to the target file.
It looks like we may already have a file just with a short content
and it is picked up by the watch service for processing in the flow
* Rework `AggregatorWithCustomReleaseStrategyTests` to JUnit 5 and remove
extra looping logic since it just does not add any extra coverage just
performance overhead via crating and destroying the same application ctx
* Rework `ChannelAdapterParserTests` to `@SpringJUnitConfig` managed test
This commit is contained in:
Artem Bilan
2021-02-25 11:29:28 -05:00
parent 51b6ba8de8
commit c023816c7e
5 changed files with 111 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/aggregator-with-custom-release-strategy.xml → spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/AggregatorWithCustomReleaseStrategyTests-context.xml
View File

@@ -5,6 +5,8 @@
xsi:schemaLocation="http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd">
<bean id="executor" class="java.util.concurrent.Executors" factory-method="newCachedThreadPool"
destroy-method="shutdown"/>
<int:splitter input-channel="in" output-channel="aggregationChannelFromSplitter"/>
@@ -20,4 +22,5 @@
<int:channel id="resultChannel">
<int:queue/>
</int:channel>
</beans>

86
spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/AggregatorWithCustomReleaseStrategyTests.java
View File

@@ -20,20 +20,18 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.junit.AfterClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.support.LongRunningIntegrationTest;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Oleg Zhurakousky
@@ -41,51 +39,43 @@ import org.springframework.messaging.MessageChannel;
* @author Artem Bilan
*
*/
@SpringJUnitConfig
@DirtiesContext
public class AggregatorWithCustomReleaseStrategyTests {
@ClassRule
public static LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
private static ExecutorService executor = Executors.newCachedThreadPool();
@Autowired
private ExecutorService executor;
@AfterClass
public static void tearDown() {
executor.shutdownNow();
}
@Autowired
@Qualifier("aggregationChannelCustomCorrelation")
private MessageChannel inputChannel;
@Autowired
private QueueChannel resultChannel;
@Autowired
@Qualifier("in")
private MessageChannel inChannel;
@Test
public void testAggregatorsUnderStressWithConcurrency() throws Exception {
// this is to be sure after INT-2502
for (int i = 0; i < 10; i++) {
this.validateSequenceSizeHasNoAffectCustomCorrelator();
}
for (int i = 0; i < 10; i++) {
this.validateSequenceSizeHasNoAffectWithSplitter();
}
}
public void validateSequenceSizeHasNoAffectCustomCorrelator() throws Exception {
AbstractApplicationContext context =
new ClassPathXmlApplicationContext("aggregator-with-custom-release-strategy.xml", this.getClass());
final MessageChannel inputChannel = context.getBean("aggregationChannelCustomCorrelation", MessageChannel.class);
QueueChannel resultChannel = context.getBean("resultChannel", QueueChannel.class);
final CountDownLatch latch = new CountDownLatch(1800);
CountDownLatch latch = new CountDownLatch(1800);
for (int i = 0; i < 600; i++) {
final int counter = i;
executor.execute(() -> {
inputChannel.send(MessageBuilder.withPayload("foo").
this.executor.execute(() -> {
this.inputChannel.send(MessageBuilder.withPayload("foo").
setHeader("correlation", "foo" + counter).build());
latch.countDown();
});
executor.execute(() -> {
inputChannel.send(MessageBuilder.withPayload("bar").
this.executor.execute(() -> {
this.inputChannel.send(MessageBuilder.withPayload("bar").
setHeader("correlation", "foo" + counter).build());
latch.countDown();
});
executor.execute(() -> {
inputChannel.send(MessageBuilder.withPayload("baz").
this.executor.execute(() -> {
this.inputChannel.send(MessageBuilder.withPayload("baz").
setHeader("correlation", "foo" + counter).build());
latch.countDown();
});
@@ -94,35 +84,30 @@ public class AggregatorWithCustomReleaseStrategyTests {
assertThat(latch.await(120, TimeUnit.SECONDS)).as("Sends failed to complete: " + latch.getCount() + " remain")
.isTrue();
Message<?> message = resultChannel.receive(1000);
Message<?> message = this.resultChannel.receive(1000);
int counter = 0;
while (message != null) {
counter++;
message = resultChannel.receive(1000);
message = this.resultChannel.receive(1000);
}
assertThat(counter).isEqualTo(600);
context.close();
}
@Test
public void validateSequenceSizeHasNoAffectWithSplitter() throws Exception {
AbstractApplicationContext context =
new ClassPathXmlApplicationContext("aggregator-with-custom-release-strategy.xml", this.getClass());
final MessageChannel inputChannel = context.getBean("in", MessageChannel.class);
QueueChannel resultChannel = context.getBean("resultChannel", QueueChannel.class);
final CountDownLatch latch = new CountDownLatch(1800);
for (int i = 0; i < 600; i++) {
executor.execute(() -> {
inputChannel.send(MessageBuilder.withPayload(new Integer[]{1, 2, 3, 4, 5, 6, 7, 8}).build());
this.executor.execute(() -> {
this.inChannel.send(MessageBuilder.withPayload(new Integer[]{ 1, 2, 3, 4, 5, 6, 7, 8 }).build());
latch.countDown();
});
executor.execute(() -> {
inputChannel.send(MessageBuilder.withPayload(new Integer[]{9, 10, 11, 12, 13, 14, 15, 16}).build());
this.inChannel.send(MessageBuilder.withPayload(new Integer[]{ 9, 10, 11, 12, 13, 14, 15, 16 }).build());
latch.countDown();
});
executor.execute(() -> {
inputChannel.send(MessageBuilder.withPayload(new Integer[]{17, 18, 19, 20, 21, 22, 23, 24}).build());
this.inChannel.send(MessageBuilder.withPayload(new Integer[]{ 17, 18, 19, 20, 21, 22, 23, 24 }).build());
latch.countDown();
});
}
@@ -130,13 +115,12 @@ public class AggregatorWithCustomReleaseStrategyTests {
assertThat(latch.await(60, TimeUnit.SECONDS)).as("Sends failed to complete: " + latch.getCount() + " remain")
.isTrue();
Message<?> message = resultChannel.receive(1000);
Message<?> message = this.resultChannel.receive(1000);
int counter = 0;
while (message != null && ++counter < 7200) {
message = resultChannel.receive(1000);
message = this.resultChannel.receive(1000);
}
assertThat(counter).isEqualTo(7200);
context.close();
}
}

4
spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2021 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -60,6 +60,7 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.util.StringUtils;
@@ -72,6 +73,7 @@ import org.springframework.util.StringUtils;
* @author Gary Russell
*/
@SpringJUnitConfig
@DirtiesContext
public class ChainParserTests {
@Autowired

172
spring-integration-core/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,12 +17,11 @@
package org.springframework.integration.config;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -39,98 +38,76 @@ import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.core.DestinationResolutionException;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Artem Bilan
*/
@SpringJUnitConfig
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
public class ChannelAdapterParserTests {
@Autowired
private AbstractApplicationContext applicationContext;
private AbstractApplicationContext applicationContextInner;
@Before
public void setUp() {
this.applicationContext = new ClassPathXmlApplicationContext(
"ChannelAdapterParserTests-context.xml", this.getClass());
this.applicationContextInner = new ClassPathXmlApplicationContext(
"ChannelAdapterParserTests-inner-context.xml", this.getClass());
}
@After
public void tearDown() {
this.applicationContext.close();
this.applicationContextInner.close();
}
@Autowired
private TestBean testBean;
@Test
public void methodInvokingSourceStoppedByApplicationContext() {
String beanName = "methodInvokingSource";
PollableChannel channel = (PollableChannel) this.applicationContext.getBean("queueChannel");
TestBean testBean = (TestBean) this.applicationContext.getBean("testBean");
testBean.store("source test");
Object adapter = this.applicationContext.getBean(beanName);
assertThat(adapter).isNotNull();
assertThat(adapter instanceof SourcePollingChannelAdapter).isTrue();
PollableChannel channel = this.applicationContext.getBean("queueChannel", PollableChannel.class);
Object adapter = this.applicationContext.getBean("methodInvokingSource");
assertThat(adapter).isInstanceOf(SourcePollingChannelAdapter.class);
assertThat(((SourcePollingChannelAdapter) adapter).getPhase()).isEqualTo(-1);
this.applicationContext.start();
Message<?> message = channel.receive(10000);
assertThat(message).isNotNull();
assertThat(testBean.getMessage()).isEqualTo("source test");
this.applicationContext.stop();
message = channel.receive(100);
message = channel.receive(0);
assertThat(message).isNull();
}
@Test
public void methodInvokingSourceStoppedByApplicationContextInner() {
String beanName = "methodInvokingSource";
PollableChannel channel = (PollableChannel) this.applicationContextInner.getBean("queueChannel");
// TestBean testBean = (TestBean) this.applicationContextInner.getBean("testBean");
// testBean.store("source test");
Object adapter = this.applicationContextInner.getBean(beanName);
assertThat(adapter).isNotNull();
assertThat(adapter instanceof SourcePollingChannelAdapter).isTrue();
this.applicationContextInner.start();
AbstractApplicationContext applicationContextInner =
new ClassPathXmlApplicationContext("ChannelAdapterParserTests-inner-context.xml", this.getClass());
PollableChannel channel = applicationContextInner.getBean("queueChannel", PollableChannel.class);
Object adapter = applicationContextInner.getBean("methodInvokingSource");
assertThat(adapter).isInstanceOf(SourcePollingChannelAdapter.class);
applicationContextInner.start();
Message<?> message = channel.receive(10000);
assertThat(message).isNotNull();
//assertEquals("source test", testBean.getMessage());
this.applicationContextInner.stop();
message = channel.receive(100);
applicationContextInner.stop();
message = channel.receive(0);
assertThat(message).isNull();
applicationContextInner.close();
}
@Test
public void targetOnly() {
String beanName = "outboundWithImplicitChannel";
Object channel = this.applicationContext.getBean(beanName);
assertThat(channel instanceof DirectChannel).isTrue();
assertThat(channel).isInstanceOf(DirectChannel.class);
BeanFactoryChannelResolver channelResolver = new BeanFactoryChannelResolver(this.applicationContext);
assertThat(channelResolver.resolveDestination(beanName)).isNotNull();
Object adapter = this.applicationContext.getBean(beanName + ".adapter");
assertThat(adapter).isNotNull();
assertThat(adapter instanceof EventDrivenConsumer).isTrue();
assertThat(adapter).isInstanceOf(EventDrivenConsumer.class);
assertThat(((EventDrivenConsumer) adapter).isAutoStartup()).isFalse();
assertThat(((EventDrivenConsumer) adapter).getPhase()).isEqualTo(-1);
TestConsumer consumer = (TestConsumer) this.applicationContext.getBean("consumer");
assertThat(consumer.getLastMessage()).isNull();
Message<?> message = new GenericMessage<String>("test");
try {
((MessageChannel) channel).send(message);
fail("MessageDispatchingException is expected.");
}
catch (Exception e) {
assertThat(e).isInstanceOf(MessageDeliveryException.class);
assertThat(e.getCause()).isInstanceOf(MessageDispatchingException.class);
}
Message<?> message = new GenericMessage<>("test");
assertThatExceptionOfType(MessageDeliveryException.class)
.isThrownBy(() -> ((MessageChannel) channel).send(message))
.withCauseInstanceOf(MessageDispatchingException.class);
((EventDrivenConsumer) adapter).start();
((MessageChannel) channel).send(message);
assertThat(consumer.getLastMessage()).isNotNull();
assertThat(consumer.getLastMessage()).isEqualTo(message);
}
@@ -138,24 +115,19 @@ public class ChannelAdapterParserTests {
public void methodInvokingConsumer() {
String beanName = "methodInvokingConsumer";
Object channel = this.applicationContext.getBean(beanName);
assertThat(channel instanceof DirectChannel).isTrue();
assertThat(channel).isInstanceOf(DirectChannel.class);
BeanFactoryChannelResolver channelResolver = new BeanFactoryChannelResolver(this.applicationContext);
assertThat(channelResolver.resolveDestination(beanName)).isNotNull();
Object adapter = this.applicationContext.getBean(beanName + ".adapter");
assertThat(adapter).isNotNull();
assertThat(adapter instanceof EventDrivenConsumer).isTrue();
TestBean testBean = (TestBean) this.applicationContext.getBean("testBean");
assertThat(adapter).isInstanceOf(EventDrivenConsumer.class);
assertThat(testBean.getMessage()).isNull();
Message<?> message = new GenericMessage<String>("consumer test");
Message<?> message = new GenericMessage<>("consumer test");
assertThat(((MessageChannel) channel).send(message)).isTrue();
assertThat(testBean.getMessage()).isNotNull();
assertThat(testBean.getMessage()).isEqualTo("consumer test");
}
@Test
/**
* @since 2.1
*/
public void expressionConsumer() {
String beanName = "expressionConsumer";
Object channel = this.applicationContext.getBean(beanName);
@@ -163,11 +135,9 @@ public class ChannelAdapterParserTests {
BeanFactoryChannelResolver channelResolver = new BeanFactoryChannelResolver(this.applicationContext);
assertThat(channelResolver.resolveDestination(beanName)).isNotNull();
Object adapter = this.applicationContext.getBean(beanName + ".adapter");
assertThat(adapter).isNotNull();
assertThat(adapter instanceof EventDrivenConsumer).isTrue();
TestBean testBean = (TestBean) this.applicationContext.getBean("testBean");
assertThat(adapter).isInstanceOf(EventDrivenConsumer.class);
assertThat(testBean.getMessage()).isNull();
Message<?> message = new GenericMessage<String>("consumer test expression");
Message<?> message = new GenericMessage<>("consumer test expression");
assertThat(((MessageChannel) channel).send(message)).isTrue();
assertThat(testBean.getMessage()).isNotNull();
assertThat(testBean.getMessage()).isEqualTo("consumer test expression");
@@ -177,11 +147,9 @@ public class ChannelAdapterParserTests {
public void methodInvokingSource() {
String beanName = "methodInvokingSource";
PollableChannel channel = (PollableChannel) this.applicationContext.getBean("queueChannel");
TestBean testBean = (TestBean) this.applicationContext.getBean("testBean");
testBean.store("source test");
Object adapter = this.applicationContext.getBean(beanName);
assertThat(adapter).isNotNull();
assertThat(adapter instanceof SourcePollingChannelAdapter).isTrue();
assertThat(adapter).isInstanceOf(SourcePollingChannelAdapter.class);
((SourcePollingChannelAdapter) adapter).start();
Message<?> message = channel.receive(10000);
assertThat(message).isNotNull();
@@ -191,13 +159,10 @@ public class ChannelAdapterParserTests {
@Test
public void methodInvokingSourceWithHeaders() {
String beanName = "methodInvokingSourceWithHeaders";
PollableChannel channel = (PollableChannel) this.applicationContext.getBean("queueChannelForHeadersTest");
TestBean testBean = (TestBean) this.applicationContext.getBean("testBean");
PollableChannel channel = this.applicationContext.getBean("queueChannelForHeadersTest", PollableChannel.class);
testBean.store("source test");
Object adapter = this.applicationContext.getBean(beanName);
assertThat(adapter).isNotNull();
assertThat(adapter instanceof SourcePollingChannelAdapter).isTrue();
Object adapter = this.applicationContext.getBean("methodInvokingSourceWithHeaders");
assertThat(adapter).isInstanceOf(SourcePollingChannelAdapter.class);
((SourcePollingChannelAdapter) adapter).start();
Message<?> message = channel.receive(10000);
((SourcePollingChannelAdapter) adapter).stop();
@@ -210,44 +175,35 @@ public class ChannelAdapterParserTests {
@Test
public void methodInvokingSourceNotStarted() {
String beanName = "methodInvokingSource";
PollableChannel channel = (PollableChannel) this.applicationContext.getBean("queueChannel");
TestBean testBean = (TestBean) this.applicationContext.getBean("testBean");
PollableChannel channel = this.applicationContext.getBean("queueChannel", PollableChannel.class);
testBean.store("source test");
Object adapter = this.applicationContext.getBean(beanName);
assertThat(adapter).isNotNull();
assertThat(adapter instanceof SourcePollingChannelAdapter).isTrue();
Message<?> message = channel.receive(100);
Object adapter = this.applicationContext.getBean("methodInvokingSource");
assertThat(adapter).isInstanceOf(SourcePollingChannelAdapter.class);
Message<?> message = channel.receive(0);
assertThat(message).isNull();
}
@Test
public void methodInvokingSourceStopped() {
String beanName = "methodInvokingSource";
PollableChannel channel = (PollableChannel) this.applicationContext.getBean("queueChannel");
TestBean testBean = (TestBean) this.applicationContext.getBean("testBean");
PollableChannel channel = this.applicationContext.getBean("queueChannel", PollableChannel.class);
testBean.store("source test");
Object adapter = this.applicationContext.getBean(beanName);
assertThat(adapter).isNotNull();
assertThat(adapter instanceof SourcePollingChannelAdapter).isTrue();
Object adapter = this.applicationContext.getBean("methodInvokingSource");
assertThat(adapter).isInstanceOf(SourcePollingChannelAdapter.class);
((SourcePollingChannelAdapter) adapter).start();
Message<?> message = channel.receive(10000);
assertThat(message).isNotNull();
assertThat(testBean.getMessage()).isEqualTo("source test");
((SourcePollingChannelAdapter) adapter).stop();
message = channel.receive(100);
message = channel.receive(0);
assertThat(message).isNull();
}
@Test
public void methodInvokingSourceStartedByApplicationContext() {
String beanName = "methodInvokingSource";
PollableChannel channel = (PollableChannel) this.applicationContext.getBean("queueChannel");
TestBean testBean = (TestBean) this.applicationContext.getBean("testBean");
PollableChannel channel = this.applicationContext.getBean("queueChannel", PollableChannel.class);
testBean.store("source test");
Object adapter = this.applicationContext.getBean(beanName);
assertThat(adapter).isNotNull();
assertThat(adapter instanceof SourcePollingChannelAdapter).isTrue();
Object adapter = this.applicationContext.getBean("methodInvokingSource");
assertThat(adapter).isInstanceOf(SourcePollingChannelAdapter.class);
this.applicationContext.start();
Message<?> message = channel.receive(1000);
assertThat(message).isNotNull();
@@ -255,27 +211,29 @@ public class ChannelAdapterParserTests {
this.applicationContext.stop();
}
@Test(expected = DestinationResolutionException.class)
@Test
public void methodInvokingSourceAdapterIsNotChannel() {
BeanFactoryChannelResolver channelResolver = new BeanFactoryChannelResolver(this.applicationContext);
channelResolver.resolveDestination("methodInvokingSource");
assertThatExceptionOfType(DestinationResolutionException.class)
.isThrownBy(() -> channelResolver.resolveDestination("methodInvokingSource"));
}
@Test
public void methodInvokingSourceWithSendTimeout() throws Exception {
String beanName = "methodInvokingSourceWithTimeout";
public void methodInvokingSourceWithSendTimeout() {
SourcePollingChannelAdapter adapter =
this.applicationContext.getBean(beanName, SourcePollingChannelAdapter.class);
this.applicationContext.getBean("methodInvokingSourceWithTimeout", SourcePollingChannelAdapter.class);
assertThat(adapter).isNotNull();
long sendTimeout = TestUtils.getPropertyValue(adapter, "messagingTemplate.sendTimeout", Long.class);
assertThat(sendTimeout).isEqualTo(999);
}
@Test(expected = BeanDefinitionParsingException.class)
public void innerBeanAndExpressionFail() throws Exception {
new ClassPathXmlApplicationContext("InboundChannelAdapterInnerBeanWithExpression-fail-context.xml",
this.getClass()).close();
@Test
public void innerBeanAndExpressionFail() {
assertThatExceptionOfType(BeanDefinitionParsingException.class)
.isThrownBy(() ->
new ClassPathXmlApplicationContext(
"InboundChannelAdapterInnerBeanWithExpression-fail-context.xml",
this.getClass()));
}
@Test
@@ -310,10 +268,8 @@ public class ChannelAdapterParserTests {
public static class SampleBean {
private final String message = "hello";
String getMessage() {
return message;
public String getMessage() {
return "hello";
}
}

10
spring-integration-file/src/test/java/org/springframework/integration/file/dsl/FileTests.java
View File

@@ -201,10 +201,12 @@ public class FileTests {
if (even) {
evens.add(i);
}
FileOutputStream file = new FileOutputStream(new File(tmpDir, i + extension));
file.write(("" + i).getBytes());
file.flush();
file.close();
File tmpFile = new File(tmpDir, i + extension + ".tmp");
FileOutputStream stream = new FileOutputStream(tmpFile);
stream.write(("" + i).getBytes());
stream.flush();
stream.close();
tmpFile.renameTo(new File(tmpDir, i + extension));
}
Message<?> message = fileReadingResultChannel.receive(60000);