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

Migrate to JUnit 5; apply Local Stack Docker

Browse Source
This commit is contained in:
Artem Bilan
2019-10-18 15:25:41 -04:00
parent db77753a84
commit 2298560e3f
28 changed files with 465 additions and 675 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
build.gradle
View File

@@ -27,6 +27,7 @@ ext {
awaitilityVersion = '4.0.1'
dynamodbLockClientVersion = '1.1.0'
jacksonVersion = '2.10.0'
junitVersion = '5.5.2'
servletApiVersion = '4.0.1'
localstackVersion = '0.1.22'
log4jVersion = '2.12.1'
@@ -77,6 +78,7 @@ dependencyManagement {
mavenBom "org.springframework.cloud:spring-cloud-aws-dependencies:$springCloudAwsVersion"
mavenBom "org.springframework.integration:spring-integration-bom:$springIntegrationVersion"
mavenBom "com.fasterxml.jackson:jackson-bom:$jacksonVersion"
mavenBom "org.junit:junit-bom:$junitVersion"
}
}
@@ -107,16 +109,22 @@ dependencies {
compile("javax.servlet:javax.servlet-api:$servletApiVersion", provided)
testCompile 'org.springframework.integration:spring-integration-test'
testCompile ('org.springframework.integration:spring-integration-test') {
exclude group: 'junit'
}
testCompile "org.assertj:assertj-core:$assertjVersion"
testCompile "cloud.localstack:localstack-utils:$localstackVersion"
testCompile ("org.awaitility:awaitility:$awaitilityVersion") {
exclude group: 'org.hamcrest'
}
compile 'org.junit.jupiter:junit-jupiter-api'
testRuntime "org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion"
testRuntime "org.apache.logging.log4j:log4j-jcl:$log4jVersion"
testRuntime 'org.junit.jupiter:junit-jupiter-engine'
testRuntime 'org.junit.platform:junit-platform-launcher'
}
eclipse.project.natures += 'org.springframework.ide.eclipse.core.springnature'

100
src/test/java/org/springframework/integration/aws/DynamoDbLocalRunning.java
View File

@@ -1,100 +0,0 @@
/*
* Copyright 2017-2019 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws;
import static org.junit.Assume.assumeNoException;
import static org.junit.Assume.assumeTrue;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.rules.TestWatcher;
import org.junit.runner.Description;
import org.junit.runners.model.Statement;
import com.amazonaws.ClientConfiguration;
import com.amazonaws.SdkClientException;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsyncClientBuilder;
/**
* The {@link TestWatcher} implementation for local Amazon DynamoDB service. See
* https://github.com/mhart/dynalite.
*
* @author Artem Bilan
* @since 1.1
*/
public final class DynamoDbLocalRunning extends TestWatcher {
public static final int DEFAULT_PORT = 4569;
private static Log logger = LogFactory.getLog(DynamoDbLocalRunning.class);
// Static so that we only test once on failure: speeds up test suite
private static Map<Integer, Boolean> dynamoDbOnline = new HashMap<>();
private final int port;
private AmazonDynamoDBAsync amazonDynamoDB;
private DynamoDbLocalRunning(int port) {
this.port = port;
dynamoDbOnline.put(port, true);
}
public AmazonDynamoDBAsync getDynamoDB() {
return this.amazonDynamoDB;
}
@Override
public Statement apply(Statement base, Description description) {
assumeTrue(dynamoDbOnline.get(this.port));
String url = "http://localhost:" + this.port;
this.amazonDynamoDB = AmazonDynamoDBAsyncClientBuilder.standard()
.withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("", "")))
.withClientConfiguration(new ClientConfiguration().withMaxErrorRetry(0).withConnectionTimeout(1000))
.withEndpointConfiguration(
new AwsClientBuilder.EndpointConfiguration(url, Regions.DEFAULT_REGION.getName()))
.build();
try {
this.amazonDynamoDB.listTables();
}
catch (SdkClientException e) {
logger.warn("Tests not running because no DynamoDb on " + url, e);
assumeNoException(e);
}
return super.apply(base, description);
}
public static DynamoDbLocalRunning isRunning() {
return isRunning(DEFAULT_PORT);
}
public static DynamoDbLocalRunning isRunning(int port) {
return new DynamoDbLocalRunning(port);
}
}

69
src/test/java/org/springframework/integration/aws/ExtendedDockerTestUtils.java Normal file
View File

@@ -0,0 +1,69 @@
/*
* Copyright 2019 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws;
import static cloud.localstack.TestUtils.DEFAULT_REGION;
import java.util.function.Supplier;
import cloud.localstack.TestUtils;
import cloud.localstack.docker.LocalstackDocker;
import com.amazonaws.ClientConfiguration;
import com.amazonaws.client.builder.AwsAsyncClientBuilder;
import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsyncClientBuilder;
import com.amazonaws.services.kinesis.AmazonKinesisAsync;
import com.amazonaws.services.kinesis.AmazonKinesisAsyncClientBuilder;
/**
* @author Artem Bilan
*
* @since 2.3
*/
public final class ExtendedDockerTestUtils {
public static AmazonKinesisAsync getClientKinesisAsync() {
AmazonKinesisAsyncClientBuilder amazonKinesisAsyncClientBuilder =
AmazonKinesisAsyncClientBuilder.standard()
.withEndpointConfiguration(
createEndpointConfiguration(LocalstackDocker.INSTANCE::getEndpointKinesis));
return applyConfigurationAndBuild(amazonKinesisAsyncClientBuilder);
}
public static AmazonDynamoDBAsync getClientDynamoDbAsync() {
AmazonDynamoDBAsyncClientBuilder dynamoDBAsyncClientBuilder =
AmazonDynamoDBAsyncClientBuilder.standard()
.withEndpointConfiguration(
createEndpointConfiguration(LocalstackDocker.INSTANCE::getEndpointDynamoDB));
return applyConfigurationAndBuild(dynamoDBAsyncClientBuilder);
}
private static AwsClientBuilder.EndpointConfiguration createEndpointConfiguration(Supplier<String> supplier) {
return new AwsClientBuilder.EndpointConfiguration(supplier.get(), DEFAULT_REGION);
}
private static <T, C extends AwsAsyncClientBuilder<C, T>> T applyConfigurationAndBuild(C builder) {
return builder.withCredentials(TestUtils.getCredentialsProvider())
.withClientConfiguration(new ClientConfiguration().withMaxErrorRetry(0).withConnectionTimeout(1000))
.build();
}
private ExtendedDockerTestUtils() {
}
}

117
src/test/java/org/springframework/integration/aws/KinesisLocalRunning.java
View File

@@ -1,117 +0,0 @@
/*
* Copyright 2017-2019 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws;
import static org.junit.Assume.assumeNoException;
import static org.junit.Assume.assumeTrue;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.rules.TestWatcher;
import org.junit.runner.Description;
import org.junit.runners.model.Statement;
import com.amazonaws.ClientConfiguration;
import com.amazonaws.SDKGlobalConfiguration;
import com.amazonaws.SdkClientException;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.kinesis.AmazonKinesisAsync;
import com.amazonaws.services.kinesis.AmazonKinesisAsyncClientBuilder;
/**
* The {@link TestWatcher} implementation for local Amazon Kinesis service. See
* https://github.com/mhart/kinesalite.
*
* @author Artem Bilan
* @since 1.1
*/
public final class KinesisLocalRunning extends TestWatcher {
public static final int DEFAULT_PORT = 4568;
private static Log logger = LogFactory.getLog(KinesisLocalRunning.class);
// Static so that we only test once on failure: speeds up test suite
private static Map<Integer, Boolean> kinesisOnline = new HashMap<>();
private final int port;
private AmazonKinesisAsync amazonKinesis;
private KinesisLocalRunning(int port) {
this.port = port;
kinesisOnline.put(port, true);
}
public AmazonKinesisAsync getKinesis() {
return this.amazonKinesis;
}
@Override
public Statement apply(Statement base, Description description) {
assumeTrue(kinesisOnline.get(this.port));
String url = "http://localhost:" + this.port;
// See https://github.com/mhart/kinesalite#cbor-protocol-issues-with-the-java-sdk
System.setProperty(SDKGlobalConfiguration.AWS_CBOR_DISABLE_SYSTEM_PROPERTY, "true");
this.amazonKinesis = AmazonKinesisAsyncClientBuilder.standard()
.withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("", "")))
.withClientConfiguration(new ClientConfiguration().withMaxErrorRetry(0).withConnectionTimeout(1000))
.withEndpointConfiguration(
new AwsClientBuilder.EndpointConfiguration(url, Regions.DEFAULT_REGION.getName()))
.build();
try {
this.amazonKinesis.listStreams();
}
catch (SdkClientException e) {
logger.warn("Tests not running because no Kinesis on " + url, e);
assumeNoException(e);
}
return new Statement() {
@Override
public void evaluate() throws Throwable {
try {
base.evaluate();
}
finally {
System.clearProperty(SDKGlobalConfiguration.AWS_CBOR_DISABLE_SYSTEM_PROPERTY);
}
}
};
}
public static KinesisLocalRunning isRunning() {
return isRunning(DEFAULT_PORT);
}
public static KinesisLocalRunning isRunning(int port) {
return new KinesisLocalRunning(port);
}
}

40
src/test/java/org/springframework/integration/aws/MyCloudAppTests.java
View File

@@ -1,40 +0,0 @@
/*
* Copyright 2019 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.Test;
import org.junit.runner.RunWith;
import cloud.localstack.DockerTestUtils;
import cloud.localstack.docker.LocalstackDockerTestRunner;
import cloud.localstack.docker.annotation.LocalstackDockerProperties;
import com.amazonaws.services.s3.AmazonS3;
@RunWith(LocalstackDockerTestRunner.class)
@LocalstackDockerProperties(randomizePorts = true, services = { "sqs", "kinesis", "s3", "sns", "dynamodb", "cloudwatch" })
public class MyCloudAppTests {
@Test
public void testLocalS3API() {
AmazonS3 s3 = DockerTestUtils.getClientS3();
assertThat(s3).isNotNull();
}
}

35
src/test/java/org/springframework/integration/aws/config/xml/S3InboundChannelAdapterParserTests.java
View File

@@ -27,8 +27,7 @@ import java.util.Set;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.expression.Expression;
@@ -45,17 +44,15 @@ import org.springframework.integration.file.remote.synchronizer.AbstractInboundF
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.util.ReflectionUtils;
/**
* @author Artem Bilan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@SpringJUnitConfig
@DirtiesContext
public class S3InboundChannelAdapterParserTests {
class S3InboundChannelAdapterParserTests {
@Autowired
private SourcePollingChannelAdapter s3Inbound;
@@ -74,7 +71,7 @@ public class S3InboundChannelAdapterParserTests {
@Test
@SuppressWarnings("unchecked")
public void testS3InboundChannelAdapterComplete() throws Exception {
void testS3InboundChannelAdapterComplete() throws Exception {
assertThat(TestUtils.getPropertyValue(this.s3Inbound, "autoStartup", Boolean.class)).isFalse();
PriorityBlockingQueue<?> blockingQueue = TestUtils.getPropertyValue(this.s3Inbound,
"source.fileSource.toBeReceived", PriorityBlockingQueue.class);
@@ -91,7 +88,7 @@ public class S3InboundChannelAdapterParserTests {
S3InboundFileSynchronizer.class);
assertThat(
TestUtils.getPropertyValue(fisync, "remoteDirectoryExpression", Expression.class).getExpressionString())
.isEqualTo("'foo/bar'");
.isEqualTo("'foo/bar'");
assertThat(TestUtils.getPropertyValue(fisync, "localFilenameGeneratorExpression")).isNotNull();
assertThat(TestUtils.getPropertyValue(fisync, "preserveTimestamp", Boolean.class)).isTrue();
assertThat(TestUtils.getPropertyValue(fisync, "temporaryFileSuffix", String.class)).isEqualTo(".foo");
@@ -112,18 +109,14 @@ public class S3InboundChannelAdapterParserTests {
.isSameAs(this.s3SessionFactory);
assertThat(TestUtils.getPropertyValue(inbound, "fileSource.scanner.filter.fileFilters", Collection.class)
.contains(this.acceptAllFilter)).isTrue();
final AtomicReference<Method> genMethod = new AtomicReference<Method>();
ReflectionUtils.doWithMethods(AbstractInboundFileSynchronizer.class, new ReflectionUtils.MethodCallback() {
@Override
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
if ("generateLocalFileName".equals(method.getName())) {
method.setAccessible(true);
genMethod.set(method);
}
}
});
final AtomicReference<Method> genMethod = new AtomicReference<>();
ReflectionUtils.doWithMethods(AbstractInboundFileSynchronizer.class,
method -> {
if ("generateLocalFileName".equals(method.getName())) {
method.setAccessible(true);
genMethod.set(method);
}
});
assertThat(genMethod.get().invoke(fisync, "foo")).isEqualTo("FOO.afoo");
}

17
src/test/java/org/springframework/integration/aws/config/xml/S3MessageHandlerParserTests.java
View File

@@ -18,8 +18,7 @@ package org.springframework.integration.aws.config.xml;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -34,8 +33,7 @@ import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.transfer.TransferManager;
@@ -44,10 +42,9 @@ import com.amazonaws.services.s3.transfer.internal.S3ProgressListener;
/**
* @author Artem Bilan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@SpringJUnitConfig
@DirtiesContext
public class S3MessageHandlerParserTests {
class S3MessageHandlerParserTests {
@Autowired
private AmazonS3 amazonS3;
@@ -88,7 +85,7 @@ public class S3MessageHandlerParserTests {
private BeanFactory beanFactory;
@Test
public void testS3OutboundChannelAdapterParser() {
void testS3OutboundChannelAdapterParser() {
assertThat(TestUtils.getPropertyValue(this.s3OutboundChannelAdapterHandler, "transferManager.s3"))
.isSameAs(this.amazonS3);
assertThat(TestUtils.getPropertyValue(this.s3OutboundChannelAdapterHandler, "bucketExpression.literalValue"))
@@ -97,7 +94,7 @@ public class S3MessageHandlerParserTests {
"destinationBucketExpression.expression")).isEqualTo("'bar'");
assertThat(
TestUtils.getPropertyValue(this.s3OutboundChannelAdapterHandler, "destinationKeyExpression.expression"))
.isEqualTo("'baz'");
.isEqualTo("'baz'");
assertThat(TestUtils.getPropertyValue(this.s3OutboundChannelAdapterHandler, "keyExpression.expression"))
.isEqualTo("payload.name");
assertThat(TestUtils.getPropertyValue(this.s3OutboundChannelAdapterHandler, "objectAclExpression.expression"))
@@ -125,7 +122,7 @@ public class S3MessageHandlerParserTests {
}
@Test
public void testS3OutboundGatewayParser() {
void testS3OutboundGatewayParser() {
assertThat(TestUtils.getPropertyValue(this.s3OutboundGatewayHandler, "transferManager"))
.isSameAs(this.transferManager);
assertThat(TestUtils.getPropertyValue(this.s3OutboundGatewayHandler, "bucketExpression.expression"))

34
src/test/java/org/springframework/integration/aws/config/xml/S3StreamingInboundChannelAdapterParserTests.java
View File

@@ -22,8 +22,7 @@ import java.lang.reflect.Method;
import java.util.Comparator;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.expression.Expression;
@@ -35,17 +34,16 @@ import org.springframework.integration.file.remote.synchronizer.AbstractInboundF
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.util.ReflectionUtils;
/**
* @author Christian Tzolov
* @author Artem Bilan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@SpringJUnitConfig
@DirtiesContext
public class S3StreamingInboundChannelAdapterParserTests {
class S3StreamingInboundChannelAdapterParserTests {
@Autowired
private SourcePollingChannelAdapter s3Inbound;
@@ -63,7 +61,7 @@ public class S3StreamingInboundChannelAdapterParserTests {
private SessionFactory<?> s3SessionFactory;
@Test
public void testS3StreamingInboundChannelAdapterComplete() throws Exception {
void testS3StreamingInboundChannelAdapterComplete() {
assertThat(TestUtils.getPropertyValue(this.s3Inbound, "autoStartup", Boolean.class)).isFalse();
assertThat(this.s3Inbound.getComponentName()).isEqualTo("s3Inbound");
@@ -75,7 +73,7 @@ public class S3StreamingInboundChannelAdapterParserTests {
assertThat(
TestUtils.getPropertyValue(source, "remoteDirectoryExpression", Expression.class).getExpressionString())
.isEqualTo("foo/bar");
.isEqualTo("foo/bar");
assertThat(TestUtils.getPropertyValue(source, "comparator")).isSameAs(this.comparator);
String remoteFileSeparator = (String) TestUtils.getPropertyValue(source, "remoteFileSeparator");
@@ -89,17 +87,13 @@ public class S3StreamingInboundChannelAdapterParserTests {
.isSameAs(this.s3SessionFactory);
final AtomicReference<Method> genMethod = new AtomicReference<Method>();
ReflectionUtils.doWithMethods(AbstractInboundFileSynchronizer.class, new ReflectionUtils.MethodCallback() {
@Override
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
if ("generateLocalFileName".equals(method.getName())) {
method.setAccessible(true);
genMethod.set(method);
}
}
});
ReflectionUtils.doWithMethods(AbstractInboundFileSynchronizer.class,
method -> {
if ("generateLocalFileName".equals(method.getName())) {
method.setAccessible(true);
genMethod.set(method);
}
});
}
}

13
src/test/java/org/springframework/integration/aws/config/xml/SnsInboundChannelAdapterParserTests.java
View File

@@ -18,8 +18,7 @@ package org.springframework.integration.aws.config.xml;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
@@ -28,18 +27,16 @@ import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import com.amazonaws.services.sns.AmazonSNS;
/**
* @author Artem Bilan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@SpringJUnitConfig
@DirtiesContext
public class SnsInboundChannelAdapterParserTests {
class SnsInboundChannelAdapterParserTests {
@Autowired
private AmazonSNS amazonSns;
@@ -55,7 +52,7 @@ public class SnsInboundChannelAdapterParserTests {
private SnsInboundChannelAdapter snsInboundChannelAdapter;
@Test
public void testSnsInboundChannelAdapterParser() {
void testSnsInboundChannelAdapterParser() {
assertThat(TestUtils.getPropertyValue(this.snsInboundChannelAdapter, "notificationStatusResolver.amazonSns"))
.isSameAs(this.amazonSns);
assertThat(TestUtils.getPropertyValue(this.snsInboundChannelAdapter, "handleNotificationStatus", Boolean.class))

16
src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests.java
View File

@@ -18,8 +18,7 @@ package org.springframework.integration.aws.config.xml;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
@@ -30,8 +29,7 @@ import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import com.amazonaws.handlers.AsyncHandler;
import com.amazonaws.services.sns.AmazonSNSAsync;
@@ -39,10 +37,9 @@ import com.amazonaws.services.sns.AmazonSNSAsync;
/**
* @author Artem Bilan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@SpringJUnitConfig
@DirtiesContext
public class SnsOutboundChannelAdapterParserTests {
class SnsOutboundChannelAdapterParserTests {
@Autowired
private AmazonSNSAsync amazonSns;
@@ -76,9 +73,7 @@ public class SnsOutboundChannelAdapterParserTests {
private MessageChannel successChannel;
@Test
public void testSnsOutboundChannelAdapterDefaultParser() {
Object handler = TestUtils.getPropertyValue(this.defaultAdapter, "handler");
void testSnsOutboundChannelAdapterDefaultParser() {
assertThat(TestUtils.getPropertyValue(this.defaultAdapter, "inputChannel")).isSameAs(this.notificationChannel);
assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "amazonSns")).isSameAs(this.amazonSns);
@@ -106,7 +101,6 @@ public class SnsOutboundChannelAdapterParserTests {
assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "sendTimeoutExpression.literalValue"))
.isEqualTo("202");
}
}

13
src/test/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParserTests.java
View File

@@ -20,8 +20,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.BDDMockito.willThrow;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
@@ -37,8 +36,7 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.core.DestinationResolutionException;
import org.springframework.messaging.core.DestinationResolver;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import com.amazonaws.services.sqs.AmazonSQS;
@@ -46,8 +44,7 @@ import com.amazonaws.services.sqs.AmazonSQS;
* @author Artem Bilan
* @author Patrick Fitzsimons
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@SpringJUnitConfig
@DirtiesContext
public class SqsMessageDrivenChannelAdapterParserTests {
@@ -73,14 +70,14 @@ public class SqsMessageDrivenChannelAdapterParserTests {
private SqsMessageDrivenChannelAdapter sqsMessageDrivenChannelAdapter;
@Bean
public DestinationResolver<?> destinationResolver() {
DestinationResolver<?> destinationResolver() {
DestinationResolver<?> destinationResolver = Mockito.mock(DestinationResolver.class);
willThrow(DestinationResolutionException.class).given(destinationResolver).resolveDestination(anyString());
return destinationResolver;
}
@Test
public void testSqsMessageDrivenChannelAdapterParser() {
void testSqsMessageDrivenChannelAdapterParser() {
SimpleMessageListenerContainer listenerContainer = TestUtils.getPropertyValue(
this.sqsMessageDrivenChannelAdapter, "listenerContainer", SimpleMessageListenerContainer.class);
assertThat(TestUtils.getPropertyValue(listenerContainer, "amazonSqs")).isSameAs(this.amazonSqs);

15
src/test/java/org/springframework/integration/aws/config/xml/SqsMessageHandlerParserTests.java
View File

@@ -18,8 +18,7 @@ package org.springframework.integration.aws.config.xml;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
@@ -31,8 +30,7 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import com.amazonaws.handlers.AsyncHandler;
import com.amazonaws.services.sqs.AmazonSQS;
@@ -40,10 +38,9 @@ import com.amazonaws.services.sqs.AmazonSQS;
/**
* @author Artem Bilan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@SpringJUnitConfig
@DirtiesContext
public class SqsMessageHandlerParserTests {
class SqsMessageHandlerParserTests {
@Autowired
private AmazonSQS amazonSqs;
@@ -77,7 +74,7 @@ public class SqsMessageHandlerParserTests {
private MessageHandler sqsOutboundChannelAdapterHandler;
@Test
public void testSqsMessageHandlerParser() {
void testSqsMessageHandlerParser() {
assertThat(TestUtils.getPropertyValue(this.sqsOutboundChannelAdapterHandler, "amazonSqs"))
.isSameAs(this.amazonSqs);
assertThat(TestUtils.getPropertyValue(this.sqsOutboundChannelAdapterHandler,
@@ -120,7 +117,7 @@ public class SqsMessageHandlerParserTests {
assertThat(
TestUtils.getPropertyValue(this.sqsOutboundChannelAdapterHandler, "sendTimeoutExpression.literalValue"))
.isEqualTo("202");
.isEqualTo("202");
}
}

42
src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests.java
View File

@@ -16,7 +16,9 @@
package org.springframework.integration.aws.config.xml;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -25,26 +27,38 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
* @author Rahul Pilani
* @author Artem Bilan
*/
public class SqsOutboundChannelAdapterParserTests {
class SqsOutboundChannelAdapterParserTests {
@Test(expected = BeanDefinitionStoreException.class)
public void test_sqs_resource_resolver_defined_with_queue_messaging_template() {
new ClassPathXmlApplicationContext("SqsOutboundChannelAdapterParserTests-context-bad.xml", getClass()).close();
@Test
void test_sqs_resource_resolver_defined_with_queue_messaging_template() {
assertThatExceptionOfType(BeanDefinitionStoreException.class)
.isThrownBy(() ->
new ClassPathXmlApplicationContext("SqsOutboundChannelAdapterParserTests-context-bad.xml",
getClass()));
}
@Test(expected = BeanDefinitionStoreException.class)
public void test_sqs_defined_with_queue_messaging_template() {
new ClassPathXmlApplicationContext("SqsOutboundChannelAdapterParserTests-context-bad2.xml", getClass()).close();
@Test
void test_sqs_defined_with_queue_messaging_template() {
assertThatExceptionOfType(BeanDefinitionStoreException.class)
.isThrownBy(() ->
new ClassPathXmlApplicationContext("SqsOutboundChannelAdapterParserTests-context-bad2.xml",
getClass()));
}
@Test(expected = BeanDefinitionStoreException.class)
public void test_resource_resolver_defined_with_queue_messaging_template() {
new ClassPathXmlApplicationContext("SqsOutboundChannelAdapterParserTests-context-bad3.xml", getClass()).close();
@Test
void test_resource_resolver_defined_with_queue_messaging_template() {
assertThatExceptionOfType(BeanDefinitionStoreException.class)
.isThrownBy(() ->
new ClassPathXmlApplicationContext("SqsOutboundChannelAdapterParserTests-context-bad3.xml",
getClass()));
}
@Test(expected = BeanDefinitionStoreException.class)
public void test_neither_sqs_nor_queue_messaging_template_defined() {
new ClassPathXmlApplicationContext("SqsOutboundChannelAdapterParserTests-context-bad4.xml", getClass()).close();
@Test
void test_neither_sqs_nor_queue_messaging_template_defined() {
assertThatExceptionOfType(BeanDefinitionStoreException.class)
.isThrownBy(() ->
new ClassPathXmlApplicationContext("SqsOutboundChannelAdapterParserTests-context-bad4.xml",
getClass()));
}
}

21
src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java
View File

@@ -29,9 +29,8 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
@@ -55,7 +54,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.PollableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import com.amazonaws.services.kinesis.AmazonKinesis;
import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
@@ -75,7 +74,7 @@ import com.amazonaws.services.kinesis.model.StreamStatus;
* @author Artem Bilan
* @since 1.1
*/
@RunWith(SpringRunner.class)
@SpringJUnitConfig
@DirtiesContext
public class KinesisMessageDrivenChannelAdapterTests {
@@ -98,14 +97,14 @@ public class KinesisMessageDrivenChannelAdapterTests {
@Autowired
private AmazonKinesis amazonKinesisForResharding;
@Before
public void setup() {
@BeforeEach
void setup() {
this.kinesisChannel.purge(null);
}
@Test
@SuppressWarnings({ "unchecked", "rawtypes" })
public void testKinesisMessageDrivenChannelAdapter() {
void testKinesisMessageDrivenChannelAdapter() {
this.kinesisMessageDrivenChannelAdapter.start();
final Set<KinesisShardOffset> shardOffsets = TestUtils.getPropertyValue(this.kinesisMessageDrivenChannelAdapter,
"shardOffsets", Set.class);
@@ -195,7 +194,7 @@ public class KinesisMessageDrivenChannelAdapterTests {
@Test
@SuppressWarnings("rawtypes")
public void testResharding() throws InterruptedException {
void testResharding() throws InterruptedException {
this.reshardingChannelAdapter.start();
assertThat(this.kinesisChannel.receive(10000)).isNotNull();
@@ -269,7 +268,7 @@ public class KinesisMessageDrivenChannelAdapterTests {
given(amazonKinesis.getShardIterator(
KinesisShardOffset.afterSequenceNumber(STREAM1, "1", "1").toShardIteratorRequest()))
.willReturn(new GetShardIteratorResult().withShardIterator(shard1Iterator4));
.willReturn(new GetShardIteratorResult().withShardIterator(shard1Iterator4));
given(amazonKinesis.getRecords(new GetRecordsRequest().withShardIterator(shard1Iterator4).withLimit(25)))
.willReturn(new GetRecordsResult().withNextShardIterator(shard1Iterator3)
@@ -329,7 +328,7 @@ public class KinesisMessageDrivenChannelAdapterTests {
given(amazonKinesis.getShardIterator(
KinesisShardOffset.latest(STREAM_FOR_RESHARDING, "closedShard").toShardIteratorRequest()))
.willReturn(new GetShardIteratorResult().withShardIterator(shard1Iterator1));
.willReturn(new GetShardIteratorResult().withShardIterator(shard1Iterator1));
given(amazonKinesis.getRecords(new GetRecordsRequest().withShardIterator(shard1Iterator1).withLimit(25)))
.willReturn(new GetRecordsResult().withNextShardIterator(null)

41
src/test/java/org/springframework/integration/aws/inbound/S3InboundChannelAdapterTests.java
View File

@@ -25,15 +25,14 @@ import java.io.File;
import java.io.FileInputStream;
import java.io.FileReader;
import java.io.IOException;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
@@ -52,8 +51,7 @@ import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.util.FileCopyUtils;
import com.amazonaws.services.s3.AmazonS3;
@@ -65,10 +63,9 @@ import com.amazonaws.services.s3.model.S3ObjectSummary;
/**
* @author Artem Bilan
* @autor Jim Krygowski
* @author Jim Krygowski
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@SpringJUnitConfig
@DirtiesContext
public class S3InboundChannelAdapterTests {
@@ -76,8 +73,8 @@ public class S3InboundChannelAdapterTests {
private static final String S3_BUCKET = "S3_BUCKET";
@ClassRule
public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder();
@TempDir
static Path TEMPORARY_FOLDER;
private static List<S3Object> S3_OBJECTS;
@@ -86,15 +83,19 @@ public class S3InboundChannelAdapterTests {
@Autowired
private PollableChannel s3FilesChannel;
@BeforeClass
public static void setup() throws IOException {
File remoteFolder = TEMPORARY_FOLDER.newFolder("remote");
@BeforeAll
static void setup() throws IOException {
File remoteFolder = new File(TEMPORARY_FOLDER.toFile(), "remote");
remoteFolder.mkdir();
File aFile = new File(remoteFolder, "a.test");
aFile.createNewFile();
FileCopyUtils.copy("Hello".getBytes(), aFile);
File bFile = new File(remoteFolder, "b.test");
bFile.createNewFile();
FileCopyUtils.copy("Bye".getBytes(), bFile);
File otherFile = new File(remoteFolder, "otherFile");
otherFile.createNewFile();
FileCopyUtils.copy("Other".getBytes(), otherFile);
S3_OBJECTS = new ArrayList<>();
@@ -103,15 +104,17 @@ public class S3InboundChannelAdapterTests {
S3Object s3Object = new S3Object();
s3Object.setBucketName(S3_BUCKET);
s3Object.setKey("subdir/" + file.getName());
s3Object.setObjectContent(new FileInputStream(file));
if (!"otherFile".equals(file.getName())) {
s3Object.setObjectContent(new FileInputStream(file));
}
S3_OBJECTS.add(s3Object);
}
LOCAL_FOLDER = TEMPORARY_FOLDER.newFolder("local");
LOCAL_FOLDER = TEMPORARY_FOLDER.resolve("local").toFile();
}
@Test
public void testS3InboundChannelAdapter() throws IOException {
void testS3InboundChannelAdapter() throws IOException {
Message<?> message = this.s3FilesChannel.receive(10000);
assertThat(message).isNotNull();
assertThat(message.getPayload()).isInstanceOf(File.class);
@@ -199,7 +202,7 @@ public class S3InboundChannelAdapterTests {
s3InboundFileSynchronizer());
messageSource.setAutoCreateLocalDirectory(true);
messageSource.setLocalDirectory(LOCAL_FOLDER);
messageSource.setLocalFilter(new AcceptOnceFileListFilter<File>());
messageSource.setLocalFilter(new AcceptOnceFileListFilter<>());
return messageSource;
}

35
src/test/java/org/springframework/integration/aws/inbound/S3StreamingChannelAdapterTests.java
View File

@@ -26,17 +26,16 @@ import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.Charset;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import org.apache.commons.io.IOUtils;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
@@ -54,8 +53,7 @@ import org.springframework.integration.metadata.SimpleMetadataStore;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.util.FileCopyUtils;
import com.amazonaws.services.s3.AmazonS3;
@@ -68,15 +66,15 @@ import com.amazonaws.services.s3.model.S3ObjectSummary;
/**
* @author Christian Tzolov
* @author Artem Bilan
*
* @since 1.1
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@SpringJUnitConfig
@DirtiesContext
public class S3StreamingChannelAdapterTests {
@ClassRule
public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder();
@TempDir
static Path TEMPORARY_FOLDER;
private static final String S3_BUCKET = "S3_BUCKET";
@@ -85,13 +83,15 @@ public class S3StreamingChannelAdapterTests {
@Autowired
private PollableChannel s3FilesChannel;
@BeforeClass
public static void setup() throws IOException {
File remoteFolder = TEMPORARY_FOLDER.newFolder("remote");
@BeforeAll
static void setup() throws IOException {
File remoteFolder = new File(TEMPORARY_FOLDER.toFile(), "remote");
remoteFolder.mkdir();
File aFile = new File(remoteFolder, "a.test");
aFile.createNewFile();
FileCopyUtils.copy("Hello".getBytes(), aFile);
File bFile = new File(remoteFolder, "b.test");
bFile.createNewFile();
FileCopyUtils.copy("Bye".getBytes(), bFile);
S3_OBJECTS = new ArrayList<>();
@@ -107,7 +107,7 @@ public class S3StreamingChannelAdapterTests {
}
@Test
public void testS3InboundStreamingChannelAdapter() throws IOException {
void testS3InboundStreamingChannelAdapter() throws IOException {
Message<?> message = this.s3FilesChannel.receive(10000);
assertThat(message).isNotNull();
assertThat(message.getPayload()).isInstanceOf(InputStream.class);
@@ -116,6 +116,7 @@ public class S3StreamingChannelAdapterTests {
InputStream inputStreamA = (InputStream) message.getPayload();
assertThat(inputStreamA).isNotNull();
assertThat(IOUtils.toString(inputStreamA, Charset.defaultCharset())).isEqualTo("Hello");
inputStreamA.close();
message = this.s3FilesChannel.receive(10000);
assertThat(message).isNotNull();
@@ -127,6 +128,8 @@ public class S3StreamingChannelAdapterTests {
assertThat(IOUtils.toString(inputStreamB, Charset.defaultCharset())).isEqualTo("Bye");
assertThat(this.s3FilesChannel.receive(10)).isNull();
inputStreamB.close();
}
@Configuration

23
src/test/java/org/springframework/integration/aws/inbound/SnsInboundChannelAdapterTests.java
View File

@@ -23,9 +23,8 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.
import java.util.Map;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.BDDMockito;
import org.springframework.beans.factory.annotation.Autowired;
@@ -41,9 +40,7 @@ import org.springframework.integration.config.EnableIntegration;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.web.WebAppConfiguration;
import org.springframework.test.context.junit.jupiter.web.SpringJUnitWebConfig;
import org.springframework.test.web.servlet.MockMvc;
import org.springframework.test.web.servlet.setup.MockMvcBuilders;
import org.springframework.util.StreamUtils;
@@ -56,9 +53,7 @@ import com.amazonaws.services.sns.AmazonSNS;
* @author Artem Bilan
* @author Kamil Przerwa
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@WebAppConfiguration
@SpringJUnitWebConfig
@DirtiesContext
public class SnsInboundChannelAdapterTests {
@@ -82,13 +77,13 @@ public class SnsInboundChannelAdapterTests {
private MockMvc mockMvc;
@Before
public void setUp() throws Exception {
@BeforeEach
void setUp() {
this.mockMvc = MockMvcBuilders.webAppContextSetup(this.context).build();
}
@Test
public void testSubscriptionConfirmation() throws Exception {
void testSubscriptionConfirmation() throws Exception {
this.mockMvc
.perform(post("/mySampleTopic").header("x-amz-sns-message-type", "SubscriptionConfirmation")
.contentType(MediaType.APPLICATION_JSON)
@@ -111,7 +106,7 @@ public class SnsInboundChannelAdapterTests {
@Test
@SuppressWarnings("unchecked")
public void testNotification() throws Exception {
void testNotification() throws Exception {
this.mockMvc
.perform(post("/mySampleTopic").header("x-amz-sns-message-type", "Notification")
.contentType(MediaType.TEXT_PLAIN)
@@ -127,7 +122,7 @@ public class SnsInboundChannelAdapterTests {
}
@Test
public void testUnsubscribe() throws Exception {
void testUnsubscribe() throws Exception {
this.mockMvc
.perform(post("/mySampleTopic").header("x-amz-sns-message-type", "UnsubscribeConfirmation")
.contentType(MediaType.TEXT_PLAIN)

25
src/test/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapterTests.java
View File

@@ -22,8 +22,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.mock;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
@@ -39,8 +38,7 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import com.amazonaws.services.sqs.AmazonSQSAsync;
import com.amazonaws.services.sqs.model.GetQueueAttributesRequest;
@@ -54,8 +52,7 @@ import com.amazonaws.services.sqs.model.ReceiveMessageResult;
/**
* @author Artem Bilan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@SpringJUnitConfig
@DirtiesContext
public class SqsMessageDrivenChannelAdapterTests {
@@ -72,10 +69,10 @@ public class SqsMessageDrivenChannelAdapterTests {
private PollableChannel controlBusOutput;
@Test
public void testSqsMessageDrivenChannelAdapter() {
void testSqsMessageDrivenChannelAdapter() {
assertThat(
TestUtils.getPropertyValue(this.sqsMessageDrivenChannelAdapter, "listenerContainer.queueStopTimeout"))
.isEqualTo(10000L);
.isEqualTo(10000L);
org.springframework.messaging.Message<?> receive = this.inputChannel.receive(1000);
assertThat(receive).isNotNull();
assertThat((String) receive.getPayload()).isIn("messageContent", "messageContent2");
@@ -107,8 +104,8 @@ public class SqsMessageDrivenChannelAdapterTests {
assertThatThrownBy(
() -> this.controlBusInput.send(new GenericMessage<>("@sqsMessageDrivenChannelAdapter.start('foo')")))
.hasCauseExactlyInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Queue with name 'foo' does not exist");
.hasCauseExactlyInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Queue with name 'foo' does not exist");
assertThat(this.sqsMessageDrivenChannelAdapter.getQueues()).isEqualTo(new String[] { "testQueue" });
}
@@ -126,10 +123,10 @@ public class SqsMessageDrivenChannelAdapterTests {
given(sqs.receiveMessage(
new ReceiveMessageRequest("http://testQueue.amazonaws.com").withAttributeNames("All")
.withMessageAttributeNames("All").withMaxNumberOfMessages(10).withWaitTimeSeconds(20)))
.willReturn(new ReceiveMessageResult().withMessages(
new Message().withBody("messageContent"),
new Message().withBody("messageContent2")))
.willReturn(new ReceiveMessageResult());
.willReturn(new ReceiveMessageResult().withMessages(
new Message().withBody("messageContent"),
new Message().withBody("messageContent2")))
.willReturn(new ReceiveMessageResult());
given(sqs.getQueueAttributes(any(GetQueueAttributesRequest.class)))
.willReturn(new GetQueueAttributesResult());

48
src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java
View File

@@ -23,11 +23,12 @@ import java.util.Date;
import java.util.HashSet;
import java.util.Set;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
@@ -35,7 +36,7 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.aws.KinesisLocalRunning;
import org.springframework.integration.aws.ExtendedDockerTestUtils;
import org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter;
import org.springframework.integration.aws.inbound.kinesis.KinesisMessageHeaderErrorMessageStrategy;
import org.springframework.integration.aws.outbound.KinesisMessageHandler;
@@ -55,21 +56,27 @@ import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import cloud.localstack.docker.LocalstackDockerExtension;
import cloud.localstack.docker.annotation.LocalstackDockerProperties;
import com.amazonaws.services.kinesis.AmazonKinesisAsync;
/**
* @author Artem Bilan
* @since 1.1
*/
@RunWith(SpringRunner.class)
@DisabledOnOs(OS.WINDOWS)
@SpringJUnitConfig
@ExtendWith(LocalstackDockerExtension.class)
@LocalstackDockerProperties(services = "kinesis")
@DirtiesContext
public class KinesisIntegrationTests {
@ClassRule
public static final KinesisLocalRunning KINESIS_LOCAL_RUNNING = KinesisLocalRunning.isRunning();
private static final String TEST_STREAM = "TestStream";
private static AmazonKinesisAsync AMAZON_KINESIS_ASYNC;
@Autowired
private MessageChannel kinesisSendChannel;
@@ -79,18 +86,19 @@ public class KinesisIntegrationTests {
@Autowired
private PollableChannel errorChannel;
@BeforeClass
public static void setup() {
KINESIS_LOCAL_RUNNING.getKinesis().createStream(TEST_STREAM, 1);
@BeforeAll
static void setup() {
AMAZON_KINESIS_ASYNC = ExtendedDockerTestUtils.getClientKinesisAsync();
AMAZON_KINESIS_ASYNC.createStream(TEST_STREAM, 1);
}
@AfterClass
public static void tearDown() {
KINESIS_LOCAL_RUNNING.getKinesis().deleteStream(TEST_STREAM);
@AfterAll
static void tearDown() {
AMAZON_KINESIS_ASYNC.deleteStream(TEST_STREAM);
}
@Test
public void testKinesisInboundOutbound() {
void testKinesisInboundOutbound() {
this.kinesisSendChannel
.send(MessageBuilder.withPayload("foo").setHeader(AwsHeaders.STREAM, TEST_STREAM).build());
@@ -138,7 +146,7 @@ public class KinesisIntegrationTests {
@Bean
@ServiceActivator(inputChannel = "kinesisSendChannel")
public MessageHandler kinesisMessageHandler() {
KinesisMessageHandler kinesisMessageHandler = new KinesisMessageHandler(KINESIS_LOCAL_RUNNING.getKinesis());
KinesisMessageHandler kinesisMessageHandler = new KinesisMessageHandler(AMAZON_KINESIS_ASYNC);
kinesisMessageHandler.setPartitionKey("1");
kinesisMessageHandler.setEmbeddedHeadersMapper(new EmbeddedJsonHeadersMessageMapper("foo"));
return kinesisMessageHandler;
@@ -156,7 +164,7 @@ public class KinesisIntegrationTests {
private KinesisMessageDrivenChannelAdapter kinesisMessageDrivenChannelAdapter() {
KinesisMessageDrivenChannelAdapter adapter = new KinesisMessageDrivenChannelAdapter(
KINESIS_LOCAL_RUNNING.getKinesis(), TEST_STREAM);
AMAZON_KINESIS_ASYNC, TEST_STREAM);
adapter.setOutputChannel(kinesisReceiveChannel());
adapter.setErrorChannel(errorChannel());
adapter.setErrorMessageStrategy(new KinesisMessageHeaderErrorMessageStrategy());

51
src/test/java/org/springframework/integration/aws/leader/DynamoDbLockRegistryLeaderInitiatorTests.java
View File

@@ -24,12 +24,14 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.integration.aws.DynamoDbLocalRunning;
import org.springframework.integration.aws.ExtendedDockerTestUtils;
import org.springframework.integration.aws.lock.DynamoDbLockRegistry;
import org.springframework.integration.leader.Context;
import org.springframework.integration.leader.DefaultCandidate;
@@ -37,6 +39,8 @@ import org.springframework.integration.leader.event.LeaderEventPublisher;
import org.springframework.integration.support.leader.LockRegistryLeaderInitiator;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import cloud.localstack.docker.LocalstackDockerExtension;
import cloud.localstack.docker.annotation.LocalstackDockerProperties;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync;
import com.amazonaws.services.dynamodbv2.model.DescribeTableRequest;
import com.amazonaws.waiters.FixedDelayStrategy;
@@ -47,46 +51,47 @@ import com.amazonaws.waiters.WaiterParameters;
/**
* @author Artem Bilan
*
* @since 2.0
*/
public class DynamoDbLockRegistryLeaderInitiatorTests {
@DisabledOnOs(OS.WINDOWS)
@ExtendWith(LocalstackDockerExtension.class)
@LocalstackDockerProperties(services = "dynamodb")
class DynamoDbLockRegistryLeaderInitiatorTests {
@ClassRule
public static final DynamoDbLocalRunning DYNAMO_DB_RUNNING = DynamoDbLocalRunning.isRunning();
private static AmazonDynamoDBAsync DYNAMO_DB;
private static AmazonDynamoDBAsync dynamoDB;
@BeforeClass
public static void init() {
dynamoDB = DYNAMO_DB_RUNNING.getDynamoDB();
@BeforeAll
static void init() {
DYNAMO_DB = ExtendedDockerTestUtils.getClientDynamoDbAsync();
try {
dynamoDB.deleteTableAsync(DynamoDbLockRegistry.DEFAULT_TABLE_NAME);
DYNAMO_DB.deleteTableAsync(DynamoDbLockRegistry.DEFAULT_TABLE_NAME);
Waiter<DescribeTableRequest> waiter = dynamoDB.waiters().tableNotExists();
Waiter<DescribeTableRequest> waiter = DYNAMO_DB.waiters().tableNotExists();
waiter.run(new WaiterParameters<>(new DescribeTableRequest(DynamoDbLockRegistry.DEFAULT_TABLE_NAME))
.withPollingStrategy(
new PollingStrategy(new MaxAttemptsRetryStrategy(25), new FixedDelayStrategy(1))));
}
catch (Exception e) {
// Ignore
}
}
@AfterClass
public static void destroy() {
dynamoDB.deleteTable(DynamoDbLockRegistry.DEFAULT_TABLE_NAME);
@AfterAll
static void destroy() {
DYNAMO_DB.deleteTable(DynamoDbLockRegistry.DEFAULT_TABLE_NAME);
}
@Test
public void testDistributedLeaderElection() throws Exception {
void testDistributedLeaderElection() throws Exception {
CountDownLatch granted = new CountDownLatch(1);
CountingPublisher countingPublisher = new CountingPublisher(granted);
List<DynamoDbLockRegistry> registries = new ArrayList<>();
List<LockRegistryLeaderInitiator> initiators = new ArrayList<>();
for (int i = 0; i < 2; i++) {
DynamoDbLockRegistry lockRepository = new DynamoDbLockRegistry(dynamoDB);
DynamoDbLockRegistry lockRepository = new DynamoDbLockRegistry(DYNAMO_DB);
lockRepository.afterPropertiesSet();
registries.add(lockRepository);
@@ -173,11 +178,11 @@ public class DynamoDbLockRegistryLeaderInitiatorTests {
}
@Test
public void testLostConnection() throws Exception {
void testLostConnection() throws Exception {
CountDownLatch granted = new CountDownLatch(1);
CountingPublisher countingPublisher = new CountingPublisher(granted);
DynamoDbLockRegistry lockRepository = new DynamoDbLockRegistry(dynamoDB);
DynamoDbLockRegistry lockRepository = new DynamoDbLockRegistry(DYNAMO_DB);
lockRepository.afterPropertiesSet();
LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator(lockRepository);

72
src/test/java/org/springframework/integration/aws/lock/DynamoDbLockRegistryTests.java
View File

@@ -25,22 +25,25 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.aws.DynamoDbLocalRunning;
import org.springframework.integration.aws.ExtendedDockerTestUtils;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import cloud.localstack.docker.LocalstackDockerExtension;
import cloud.localstack.docker.annotation.LocalstackDockerProperties;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync;
import com.amazonaws.services.dynamodbv2.model.DescribeTableRequest;
import com.amazonaws.waiters.FixedDelayStrategy;
@@ -51,46 +54,49 @@ import com.amazonaws.waiters.WaiterParameters;
/**
* @author Artem Bilan
*
* @since 2.0
*/
@RunWith(SpringRunner.class)
@DisabledOnOs(OS.WINDOWS)
@SpringJUnitConfig
@ExtendWith(LocalstackDockerExtension.class)
@LocalstackDockerProperties(services = "dynamodb")
@DirtiesContext
public class DynamoDbLockRegistryTests {
@ClassRule
public static final DynamoDbLocalRunning DYNAMO_DB_RUNNING = DynamoDbLocalRunning.isRunning();
private final AsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
private static AmazonDynamoDBAsync DYNAMO_DB;
@Autowired
private DynamoDbLockRegistry dynamoDbLockRegistry;
@BeforeClass
public static void setup() {
AmazonDynamoDBAsync dynamoDB = DYNAMO_DB_RUNNING.getDynamoDB();
@BeforeAll
static void setup() {
DYNAMO_DB = ExtendedDockerTestUtils.getClientDynamoDbAsync();
try {
dynamoDB.deleteTableAsync(DynamoDbLockRegistry.DEFAULT_TABLE_NAME);
DYNAMO_DB.deleteTableAsync(DynamoDbLockRegistry.DEFAULT_TABLE_NAME);
Waiter<DescribeTableRequest> waiter = dynamoDB.waiters().tableNotExists();
Waiter<DescribeTableRequest> waiter = DYNAMO_DB.waiters().tableNotExists();
waiter.run(new WaiterParameters<>(new DescribeTableRequest(DynamoDbLockRegistry.DEFAULT_TABLE_NAME))
.withPollingStrategy(
new PollingStrategy(new MaxAttemptsRetryStrategy(25), new FixedDelayStrategy(1))));
}
catch (Exception e) {
// Ignore
}
}
@Before
public void clear() {
@BeforeEach
void clear() {
this.dynamoDbLockRegistry.expireUnusedOlderThan(0);
}
@Test
@SuppressWarnings("unchecked")
public void testLock() {
void testLock() {
for (int i = 0; i < 10; i++) {
Lock lock = this.dynamoDbLockRegistry.obtain("foo");
lock.lock();
@@ -105,7 +111,7 @@ public class DynamoDbLockRegistryTests {
@Test
@SuppressWarnings("unchecked")
public void testLockInterruptibly() throws Exception {
void testLockInterruptibly() throws Exception {
for (int i = 0; i < 10; i++) {
Lock lock = this.dynamoDbLockRegistry.obtain("foo");
lock.lockInterruptibly();
@@ -119,7 +125,7 @@ public class DynamoDbLockRegistryTests {
}
@Test
public void testReentrantLock() {
void testReentrantLock() {
for (int i = 0; i < 10; i++) {
Lock lock1 = this.dynamoDbLockRegistry.obtain("foo");
lock1.lock();
@@ -136,7 +142,7 @@ public class DynamoDbLockRegistryTests {
}
@Test
public void testReentrantLockInterruptibly() throws Exception {
void testReentrantLockInterruptibly() throws Exception {
for (int i = 0; i < 10; i++) {
Lock lock1 = this.dynamoDbLockRegistry.obtain("foo");
lock1.lockInterruptibly();
@@ -153,7 +159,7 @@ public class DynamoDbLockRegistryTests {
}
@Test
public void testTwoLocks() throws Exception {
void testTwoLocks() throws Exception {
for (int i = 0; i < 10; i++) {
Lock lock1 = this.dynamoDbLockRegistry.obtain("foo");
lock1.lockInterruptibly();
@@ -170,13 +176,13 @@ public class DynamoDbLockRegistryTests {
}
@Test
public void testTwoThreadsSecondFailsToGetLock() throws Exception {
void testTwoThreadsSecondFailsToGetLock() throws Exception {
final Lock lock1 = this.dynamoDbLockRegistry.obtain("foo");
lock1.lockInterruptibly();
final AtomicBoolean locked = new AtomicBoolean();
final CountDownLatch latch = new CountDownLatch(1);
Future<Object> result = this.taskExecutor.submit(() -> {
DynamoDbLockRegistry registry2 = new DynamoDbLockRegistry(DYNAMO_DB_RUNNING.getDynamoDB());
DynamoDbLockRegistry registry2 = new DynamoDbLockRegistry(DYNAMO_DB);
registry2.setHeartbeatPeriod(1);
registry2.setRefreshPeriod(10);
registry2.setLeaseDuration(2);
@@ -204,7 +210,7 @@ public class DynamoDbLockRegistryTests {
}
@Test
public void testTwoThreads() throws Exception {
void testTwoThreads() throws Exception {
final Lock lock1 = this.dynamoDbLockRegistry.obtain("foo");
final AtomicBoolean locked = new AtomicBoolean();
final CountDownLatch latch1 = new CountDownLatch(1);
@@ -212,7 +218,7 @@ public class DynamoDbLockRegistryTests {
final CountDownLatch latch3 = new CountDownLatch(1);
lock1.lockInterruptibly();
this.taskExecutor.submit(() -> {
DynamoDbLockRegistry registry2 = new DynamoDbLockRegistry(DYNAMO_DB_RUNNING.getDynamoDB());
DynamoDbLockRegistry registry2 = new DynamoDbLockRegistry(DYNAMO_DB);
registry2.setHeartbeatPeriod(1);
registry2.setRefreshPeriod(10);
registry2.setLeaseDuration(2);
@@ -246,14 +252,14 @@ public class DynamoDbLockRegistryTests {
}
@Test
public void testTwoThreadsDifferentRegistries() throws Exception {
final DynamoDbLockRegistry registry1 = new DynamoDbLockRegistry(DYNAMO_DB_RUNNING.getDynamoDB());
void testTwoThreadsDifferentRegistries() throws Exception {
final DynamoDbLockRegistry registry1 = new DynamoDbLockRegistry(DYNAMO_DB);
registry1.setHeartbeatPeriod(1);
registry1.setRefreshPeriod(10);
registry1.setLeaseDuration(2);
registry1.afterPropertiesSet();
final DynamoDbLockRegistry registry2 = new DynamoDbLockRegistry(DYNAMO_DB_RUNNING.getDynamoDB());
final DynamoDbLockRegistry registry2 = new DynamoDbLockRegistry(DYNAMO_DB);
registry2.setHeartbeatPeriod(1);
registry2.setRefreshPeriod(10);
registry2.setLeaseDuration(2);
@@ -295,7 +301,7 @@ public class DynamoDbLockRegistryTests {
}
@Test
public void testTwoThreadsWrongOneUnlocks() throws Exception {
void testTwoThreadsWrongOneUnlocks() throws Exception {
final Lock lock = this.dynamoDbLockRegistry.obtain("foo");
lock.lockInterruptibly();
final AtomicBoolean locked = new AtomicBoolean();
@@ -325,7 +331,7 @@ public class DynamoDbLockRegistryTests {
@Bean
public DynamoDbLockRegistry dynamoDbLockRegistry() {
DynamoDbLockRegistry dynamoDbLockRegistry = new DynamoDbLockRegistry(DYNAMO_DB_RUNNING.getDynamoDB());
DynamoDbLockRegistry dynamoDbLockRegistry = new DynamoDbLockRegistry(DYNAMO_DB);
dynamoDbLockRegistry.setHeartbeatPeriod(1);
dynamoDbLockRegistry.setRefreshPeriod(10);
dynamoDbLockRegistry.setLeaseDuration(2);

53
src/test/java/org/springframework/integration/aws/metadata/DynamoDbMetadataStoreTests.java
View File

@@ -21,14 +21,18 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.util.Collections;
import java.util.concurrent.CountDownLatch;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.integration.aws.DynamoDbLocalRunning;
import org.springframework.integration.aws.ExtendedDockerTestUtils;
import org.springframework.integration.test.util.TestUtils;
import cloud.localstack.docker.LocalstackDockerExtension;
import cloud.localstack.docker.annotation.LocalstackDockerProperties;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync;
import com.amazonaws.services.dynamodbv2.model.AttributeValue;
import com.amazonaws.services.dynamodbv2.model.DescribeTableRequest;
@@ -40,13 +44,15 @@ import com.amazonaws.waiters.WaiterParameters;
/**
* @author Artem Bilan
* @since 1.1
*
* @since 1.1
*/
public class DynamoDbMetadataStoreTests {
@DisabledOnOs(OS.WINDOWS)
@ExtendWith(LocalstackDockerExtension.class)
@LocalstackDockerProperties(services = "dynamodb")
class DynamoDbMetadataStoreTests {
@ClassRule
public static final DynamoDbLocalRunning DYNAMO_DB_RUNNING = DynamoDbLocalRunning.isRunning();
private static AmazonDynamoDBAsync DYNAMO_DB;
private static final String TEST_TABLE = "testMetadataStore";
@@ -56,39 +62,38 @@ public class DynamoDbMetadataStoreTests {
private final String file1Id = "12345";
@BeforeClass
public static void setup() {
AmazonDynamoDBAsync dynamoDB = DYNAMO_DB_RUNNING.getDynamoDB();
@BeforeAll
static void setup() {
DYNAMO_DB = ExtendedDockerTestUtils.getClientDynamoDbAsync();
try {
dynamoDB.deleteTableAsync(TEST_TABLE);
DYNAMO_DB.deleteTableAsync(TEST_TABLE);
Waiter<DescribeTableRequest> waiter = dynamoDB.waiters().tableNotExists();
Waiter<DescribeTableRequest> waiter = DYNAMO_DB.waiters().tableNotExists();
waiter.run(new WaiterParameters<>(new DescribeTableRequest(TEST_TABLE)).withPollingStrategy(
new PollingStrategy(new MaxAttemptsRetryStrategy(25), new FixedDelayStrategy(1))));
}
catch (Exception e) {
// Ignore
}
store = new DynamoDbMetadataStore(dynamoDB, TEST_TABLE);
store = new DynamoDbMetadataStore(DYNAMO_DB, TEST_TABLE);
store.setTimeToLive(10); // Dynalite doesn't support TTL
store.afterPropertiesSet();
}
@Before
public void clear() throws InterruptedException {
@BeforeEach
void clear() throws InterruptedException {
CountDownLatch createTableLatch = TestUtils.getPropertyValue(store, "createTableLatch", CountDownLatch.class);
createTableLatch.await();
DYNAMO_DB_RUNNING.getDynamoDB().deleteItem(TEST_TABLE,
Collections.singletonMap("KEY", new AttributeValue().withS(this.file1)));
DYNAMO_DB.deleteItem(TEST_TABLE, Collections.singletonMap("KEY", new AttributeValue().withS(this.file1)));
}
@Test
public void testGetFromStore() {
void testGetFromStore() {
String fileID = store.get(this.file1);
assertThat(fileID).isNull();
@@ -100,7 +105,7 @@ public class DynamoDbMetadataStoreTests {
}
@Test
public void testPutIfAbsent() {
void testPutIfAbsent() {
String fileID = store.get(this.file1);
assertThat(fileID).describedAs("Get First time, Value must not exist").isNull();
@@ -115,7 +120,7 @@ public class DynamoDbMetadataStoreTests {
}
@Test
public void testRemove() {
void testRemove() {
String fileID = store.remove(this.file1);
assertThat(fileID).isNull();
@@ -131,7 +136,7 @@ public class DynamoDbMetadataStoreTests {
}
@Test
public void testReplace() {
void testReplace() {
boolean removedValue = store.replace(this.file1, this.file1Id, "4567");
assertThat(removedValue).isFalse();

47
src/test/java/org/springframework/integration/aws/outbound/KinesisMessageHandlerTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.integration.aws.outbound;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.assertj.core.api.Assertions.entry;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
@@ -27,8 +28,7 @@ import static org.mockito.Mockito.verify;
import java.nio.ByteBuffer;
import java.util.concurrent.Future;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.annotation.Autowired;
@@ -48,7 +48,7 @@ import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import com.amazonaws.handlers.AsyncHandler;
import com.amazonaws.services.kinesis.AmazonKinesisAsync;
@@ -59,9 +59,10 @@ import com.amazonaws.services.kinesis.model.PutRecordsRequestEntry;
/**
* @author Artem Bilan
*
* @since 1.1
*/
@RunWith(SpringRunner.class)
@SpringJUnitConfig
@DirtiesContext
public class KinesisMessageHandlerTests {
@@ -79,31 +80,25 @@ public class KinesisMessageHandlerTests {
@Test
@SuppressWarnings("unchecked")
public void testKinesisMessageHandler() throws Exception {
Message<?> message = MessageBuilder.withPayload("message").build();
try {
this.kinesisSendChannel.send(message);
}
catch (Exception e) {
assertThat(e).isInstanceOf(MessageHandlingException.class);
assertThat(e.getCause()).isInstanceOf(IllegalStateException.class);
assertThat(e.getMessage()).contains("'stream' must not be null for sending a Kinesis record");
}
void testKinesisMessageHandler() {
final Message<?> message = MessageBuilder.withPayload("message").build();
assertThatExceptionOfType(MessageHandlingException.class)
.isThrownBy(() -> this.kinesisSendChannel.send(message))
.withCauseInstanceOf(IllegalStateException.class)
.withMessageContaining("'stream' must not be null for sending a Kinesis record");
this.kinesisMessageHandler.setStream("foo");
try {
this.kinesisSendChannel.send(message);
}
catch (Exception e) {
assertThat(e).isInstanceOf(MessageHandlingException.class);
assertThat(e.getCause()).isInstanceOf(IllegalStateException.class);
assertThat(e.getMessage()).contains("'partitionKey' must not be null for sending a Kinesis record");
}
message = MessageBuilder.fromMessage(message).setHeader(AwsHeaders.PARTITION_KEY, "fooKey")
assertThatExceptionOfType(MessageHandlingException.class)
.isThrownBy(() -> this.kinesisSendChannel.send(message))
.withCauseInstanceOf(IllegalStateException.class)
.withMessageContaining("'partitionKey' must not be null for sending a Kinesis record");
Message<?> message2 = MessageBuilder.fromMessage(message).setHeader(AwsHeaders.PARTITION_KEY, "fooKey")
.setHeader(AwsHeaders.SEQUENCE_NUMBER, "10").setHeader("foo", "bar").build();
this.kinesisSendChannel.send(message);
this.kinesisSendChannel.send(message2);
ArgumentCaptor<PutRecordRequest> putRecordRequestArgumentCaptor = ArgumentCaptor
.forClass(PutRecordRequest.class);
@@ -133,10 +128,10 @@ public class KinesisMessageHandlerTests {
verify(this.asyncHandler).onError(eq(testingException));
message = new GenericMessage<>(new PutRecordsRequest().withStreamName("myStream").withRecords(
message2 = new GenericMessage<>(new PutRecordsRequest().withStreamName("myStream").withRecords(
new PutRecordsRequestEntry().withData(ByteBuffer.wrap("test".getBytes())).withPartitionKey("testKey")));
this.kinesisSendChannel.send(message);
this.kinesisSendChannel.send(message2);
ArgumentCaptor<PutRecordsRequest> putRecordsRequestArgumentCaptor = ArgumentCaptor
.forClass(PutRecordsRequest.class);

54
src/test/java/org/springframework/integration/aws/outbound/KinesisProducingMessageHandlerTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.integration.aws.outbound;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.BDDMockito.given;
import static org.mockito.Mockito.mock;
@@ -24,8 +25,7 @@ import static org.mockito.Mockito.mock;
import java.nio.ByteBuffer;
import java.util.concurrent.Future;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
@@ -46,7 +46,7 @@ import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import com.amazonaws.handlers.AsyncHandler;
import com.amazonaws.services.kinesis.AmazonKinesisAsync;
@@ -58,9 +58,11 @@ import com.amazonaws.services.kinesis.model.PutRecordsResult;
/**
* @author Jacob Severson
* @author Artem Bilan
*
* @since 1.1
*/
@RunWith(SpringRunner.class)
@SpringJUnitConfig
@DirtiesContext
public class KinesisProducingMessageHandlerTests {
@@ -79,40 +81,34 @@ public class KinesisProducingMessageHandlerTests {
@Test
@SuppressWarnings("unchecked")
public void testKinesisMessageHandler() {
Message<?> message = MessageBuilder.withPayload("message").build();
try {
this.kinesisSendChannel.send(message);
}
catch (Exception e) {
assertThat(e).isInstanceOf(MessageHandlingException.class);
assertThat(e.getCause()).isInstanceOf(IllegalStateException.class);
assertThat(e.getMessage()).contains("'stream' must not be null for sending a Kinesis record");
}
final Message<?> message = MessageBuilder.withPayload("message").build();
assertThatExceptionOfType(MessageHandlingException.class)
.isThrownBy(() -> this.kinesisSendChannel.send(message))
.withCauseInstanceOf(IllegalStateException.class)
.withMessageContaining("'stream' must not be null for sending a Kinesis record");
this.kinesisMessageHandler.setStream("foo");
try {
this.kinesisSendChannel.send(message);
}
catch (Exception e) {
assertThat(e).isInstanceOf(MessageHandlingException.class);
assertThat(e.getCause()).isInstanceOf(IllegalStateException.class);
assertThat(e.getMessage()).contains("'partitionKey' must not be null for sending a Kinesis record");
}
message = MessageBuilder.fromMessage(message).setHeader(AwsHeaders.PARTITION_KEY, "fooKey")
assertThatExceptionOfType(MessageHandlingException.class)
.isThrownBy(() -> this.kinesisSendChannel.send(message))
.withCauseInstanceOf(IllegalStateException.class)
.withMessageContaining("'partitionKey' must not be null for sending a Kinesis record");
Message<?> message2 = MessageBuilder.fromMessage(message).setHeader(AwsHeaders.PARTITION_KEY, "fooKey")
.setHeader(AwsHeaders.SEQUENCE_NUMBER, "10").build();
this.kinesisSendChannel.send(message);
this.kinesisSendChannel.send(message2);
Message<?> success = this.successChannel.receive(10000);
assertThat(success.getHeaders().get(AwsHeaders.PARTITION_KEY)).isEqualTo("fooKey");
assertThat(success.getHeaders().get(AwsHeaders.SEQUENCE_NUMBER)).isEqualTo("10");
assertThat(success.getPayload()).isEqualTo("message");
message = MessageBuilder.fromMessage(message).setHeader(AwsHeaders.PARTITION_KEY, "fooKey")
message2 = MessageBuilder.fromMessage(message).setHeader(AwsHeaders.PARTITION_KEY, "fooKey")
.setHeader(AwsHeaders.SEQUENCE_NUMBER, "10").build();
this.kinesisSendChannel.send(message);
this.kinesisSendChannel.send(message2);
Message<?> failed = this.errorChannel.receive(10000);
AwsRequestFailureException putRecordFailure = (AwsRequestFailureException) failed.getPayload();
@@ -124,19 +120,19 @@ public class KinesisProducingMessageHandlerTests {
assertThat(((PutRecordRequest) putRecordFailure.getRequest()).getData())
.isEqualTo(ByteBuffer.wrap("message".getBytes()));
message = new GenericMessage<>(new PutRecordsRequest().withStreamName("myStream").withRecords(
message2 = new GenericMessage<>(new PutRecordsRequest().withStreamName("myStream").withRecords(
new PutRecordsRequestEntry().withData(ByteBuffer.wrap("test".getBytes())).withPartitionKey("testKey")));
this.kinesisSendChannel.send(message);
this.kinesisSendChannel.send(message2);
success = this.successChannel.receive(10000);
assertThat(((PutRecordsRequest) success.getPayload()).getRecords()).containsExactlyInAnyOrder(
new PutRecordsRequestEntry().withData(ByteBuffer.wrap("test".getBytes())).withPartitionKey("testKey"));
message = new GenericMessage<>(new PutRecordsRequest().withStreamName("myStream").withRecords(
message2 = new GenericMessage<>(new PutRecordsRequest().withStreamName("myStream").withRecords(
new PutRecordsRequestEntry().withData(ByteBuffer.wrap("test".getBytes())).withPartitionKey("testKey")));
this.kinesisSendChannel.send(message);
this.kinesisSendChannel.send(message2);
failed = this.errorChannel.receive(10000);
AwsRequestFailureException putRecordsFailure = (AwsRequestFailureException) failed.getPayload();

60
src/test/java/org/springframework/integration/aws/outbound/S3MessageHandlerTests.java
View File

@@ -17,8 +17,8 @@
package org.springframework.integration.aws.outbound;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Fail.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willAnswer;
@@ -33,8 +33,10 @@ import java.io.FileInputStream;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedList;
@@ -44,10 +46,8 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.http.client.methods.HttpRequestBase;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.annotation.Autowired;
@@ -70,8 +70,7 @@ import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.util.FileCopyUtils;
import com.amazonaws.event.ProgressEvent;
@@ -108,8 +107,7 @@ import com.amazonaws.util.StringUtils;
* @author John Logan
* @author Jim Krygowski
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@SpringJUnitConfig
@DirtiesContext
public class S3MessageHandlerTests {
@@ -122,8 +120,8 @@ public class S3MessageHandlerTests {
private static final String S3_FILE_KEY_FOO = "subdir/foo";
@Rule
public TemporaryFolder temporaryFolder = new TemporaryFolder();
@TempDir
static Path temporaryFolder;
@Autowired
private AmazonS3 amazonS3;
@@ -148,8 +146,9 @@ public class S3MessageHandlerTests {
private S3MessageHandler s3MessageHandler;
@Test
public void testUploadFile() throws IOException, InterruptedException {
File file = this.temporaryFolder.newFile("foo.mp3");
void testUploadFile() throws IOException, InterruptedException {
File file = new File(temporaryFolder.toFile(), "foo.mp3");
file.createNewFile();
Message<?> message = MessageBuilder.withPayload(file)
.setHeader("s3Command", S3MessageHandler.Command.UPLOAD.name()).build();
@@ -189,7 +188,7 @@ public class S3MessageHandlerTests {
}
@Test
public void testUploadInputStream() throws IOException {
void testUploadInputStream() throws IOException {
Expression actualKeyExpression = TestUtils.getPropertyValue(this.s3MessageHandler, "keyExpression",
Expression.class);
@@ -225,25 +224,23 @@ public class S3MessageHandlerTests {
}
@Test
public void testUploadInputStreamNoMarkSupported() throws IOException, InterruptedException {
File file = this.temporaryFolder.newFile("foo.mp3");
void testUploadInputStreamNoMarkSupported() throws IOException {
File file = new File(temporaryFolder.toFile(), "foo.mp3");
file.createNewFile();
FileInputStream fileInputStream = new FileInputStream(file);
Message<?> message = MessageBuilder.withPayload(fileInputStream)
.setHeader("s3Command", S3MessageHandler.Command.UPLOAD.name()).setHeader("key", "myStream").build();
try {
this.s3SendChannel.send(message);
fail("Expected send() failure with FileInputStream, got success.");
}
catch (Exception e) {
assertThat(e).isInstanceOf(MessageHandlingException.class);
assertThat(e.getCause()).isInstanceOf(IllegalStateException.class);
}
assertThatExceptionOfType(MessageHandlingException.class)
.isThrownBy(() -> this.s3SendChannel.send(message))
.withCauseInstanceOf(IllegalStateException.class);
fileInputStream.close();
}
@Test
public void testUploadByteArray() throws IOException {
byte[] payload = "b".getBytes("UTF-8");
void testUploadByteArray() {
byte[] payload = "b".getBytes(StandardCharsets.UTF_8);
Message<?> message = MessageBuilder.withPayload(payload)
.setHeader("s3Command", S3MessageHandler.Command.UPLOAD.name()).setHeader("key", "myStream").build();
@@ -267,8 +264,9 @@ public class S3MessageHandlerTests {
}
@Test
public void testDownloadDirectory() throws IOException {
File directoryForDownload = this.temporaryFolder.newFolder("myFolder");
void testDownloadDirectory() throws IOException {
File directoryForDownload = new File(temporaryFolder.toFile(), "myFolder");
directoryForDownload.mkdir();
Message<?> message = MessageBuilder.withPayload(directoryForDownload)
.setHeader("s3Command", S3MessageHandler.Command.DOWNLOAD).build();
@@ -288,7 +286,7 @@ public class S3MessageHandlerTests {
assertThat(fileArray.length).isEqualTo(2);
List<File> files = Arrays.asList(fileArray);
Collections.sort(files, (o1, o2) -> o1.getName().compareTo(o2.getName()));
files.sort(Comparator.comparing(File::getName));
File file1 = files.get(0);
assertThat(file1.getName()).isEqualTo(S3_FILE_KEY_BAR.split("/", 2)[1]);
@@ -300,7 +298,7 @@ public class S3MessageHandlerTests {
}
@Test
public void testCopy() throws InterruptedException {
void testCopy() throws InterruptedException {
Map<String, String> payload = new HashMap<>();
payload.put("key", "mySource");
payload.put("destination", "theirBucket");

42
src/test/java/org/springframework/integration/aws/outbound/SnsMessageBuilderTests.java
View File

@@ -17,47 +17,33 @@
package org.springframework.integration.aws.outbound;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.integration.aws.support.SnsBodyBuilder;
/**
* @author Artem Bilan
*/
public class SnsMessageBuilderTests {
class SnsMessageBuilderTests {
@Test
public void testSnsMessageBuilder() {
try {
SnsBodyBuilder.withDefault("");
fail("IllegalArgumentException expected");
}
catch (Exception e) {
assertThat(e).isInstanceOf(IllegalArgumentException.class);
assertThat(e.getMessage()).contains("defaultMessage must not be empty.");
}
void testSnsMessageBuilder() {
assertThatIllegalArgumentException()
.isThrownBy(() -> SnsBodyBuilder.withDefault(""))
.withMessageContaining("defaultMessage must not be empty.");
String message = SnsBodyBuilder.withDefault("foo").build();
assertThat(message).isEqualTo("{\"default\":\"foo\"}");
try {
SnsBodyBuilder.withDefault("foo").forProtocols("{\"foo\" : \"bar\"}").build();
fail("IllegalArgumentException expected");
}
catch (Exception e) {
assertThat(e).isInstanceOf(IllegalArgumentException.class);
assertThat(e.getMessage()).contains("protocols must not be empty.");
}
try {
SnsBodyBuilder.withDefault("foo").forProtocols("{\"foo\" : \"bar\"}", "").build();
fail("IllegalArgumentException expected");
}
catch (Exception e) {
assertThat(e).isInstanceOf(IllegalArgumentException.class);
assertThat(e.getMessage()).contains("protocols must not contain empty elements.");
}
assertThatIllegalArgumentException()
.isThrownBy(() -> SnsBodyBuilder.withDefault("foo").forProtocols("{\"foo\" : \"bar\"}").build())
.withMessageContaining("protocols must not be empty.");
assertThatIllegalArgumentException()
.isThrownBy(() -> SnsBodyBuilder.withDefault("foo").forProtocols("{\"foo\" : \"bar\"}", "").build())
.withMessageContaining("protocols must not contain empty elements.");
message = SnsBodyBuilder.withDefault("foo").forProtocols("{\"foo\" : \"bar\"}", "sms").build();

11
src/test/java/org/springframework/integration/aws/outbound/SnsMessageHandlerTests.java
View File

@@ -24,8 +24,7 @@ import static org.mockito.Mockito.verify;
import java.util.Map;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.annotation.Autowired;
@@ -46,8 +45,7 @@ import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.PollableChannel;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import com.amazonaws.handlers.AsyncHandler;
import com.amazonaws.services.sns.AmazonSNSAsync;
@@ -58,8 +56,7 @@ import com.amazonaws.services.sns.model.PublishResult;
/**
* @author Artem Bilan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@SpringJUnitConfig
@DirtiesContext
public class SnsMessageHandlerTests {
@@ -76,7 +73,7 @@ public class SnsMessageHandlerTests {
@Test
@SuppressWarnings("unchecked")
public void testSnsMessageHandler() {
void testSnsMessageHandler() {
SnsBodyBuilder payload = SnsBodyBuilder.withDefault("foo").forProtocols("{\"foo\" : \"bar\"}", "sms");
Message<?> message = MessageBuilder.withPayload(payload).setHeader("topic", "topic")

36
src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.integration.aws.outbound;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.Mockito.mock;
@@ -25,8 +26,7 @@ import static org.mockito.Mockito.verify;
import java.util.Map;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.annotation.Autowired;
@@ -45,8 +45,7 @@ import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import com.amazonaws.handlers.AsyncHandler;
import com.amazonaws.services.sqs.AmazonSQSAsync;
@@ -64,8 +63,7 @@ import com.amazonaws.services.sqs.model.SendMessageRequest;
* @author Rahul Pilani
* @author Seth Kelly
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@SpringJUnitConfig
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
public class SqsMessageHandlerTests {
@@ -86,15 +84,12 @@ public class SqsMessageHandlerTests {
@Test
@SuppressWarnings("unchecked")
public void testSqsMessageHandler() {
Message<String> message = MessageBuilder.withPayload("message").build();
try {
this.sqsSendChannel.send(message);
}
catch (Exception e) {
assertThat(e).isInstanceOf(MessageHandlingException.class);
assertThat(e.getCause()).isInstanceOf(IllegalStateException.class);
}
void testSqsMessageHandler() {
final Message<String> message = MessageBuilder.withPayload("message").build();
assertThatExceptionOfType(MessageHandlingException.class)
.isThrownBy(() -> this.sqsSendChannel.send(message))
.withCauseInstanceOf(IllegalStateException.class);
this.sqsMessageHandler.setQueue("foo");
this.sqsSendChannel.send(message);
@@ -103,8 +98,8 @@ public class SqsMessageHandlerTests {
verify(this.amazonSqs).sendMessageAsync(sendMessageRequestArgumentCaptor.capture(), any(AsyncHandler.class));
assertThat(sendMessageRequestArgumentCaptor.getValue().getQueueUrl()).isEqualTo("https://queue-url.com/foo");
message = MessageBuilder.withPayload("message").setHeader(AwsHeaders.QUEUE, "bar").build();
this.sqsSendChannel.send(message);
Message<String> message2 = MessageBuilder.withPayload("message").setHeader(AwsHeaders.QUEUE, "bar").build();
this.sqsSendChannel.send(message2);
verify(this.amazonSqs, times(2)).sendMessageAsync(sendMessageRequestArgumentCaptor.capture(),
any(AsyncHandler.class));
@@ -113,8 +108,8 @@ public class SqsMessageHandlerTests {
SpelExpressionParser spelExpressionParser = new SpelExpressionParser();
Expression expression = spelExpressionParser.parseExpression("headers.foo");
this.sqsMessageHandler.setQueueExpression(expression);
message = MessageBuilder.withPayload("message").setHeader("foo", "baz").build();
this.sqsSendChannel.send(message);
message2 = MessageBuilder.withPayload("message").setHeader("foo", "baz").build();
this.sqsSendChannel.send(message2);
verify(this.amazonSqs, times(3)).sendMessageAsync(sendMessageRequestArgumentCaptor.capture(),
any(AsyncHandler.class));
@@ -128,12 +123,11 @@ public class SqsMessageHandlerTests {
assertThat(messageAttributes).doesNotContainKey(MessageHeaders.TIMESTAMP);
assertThat(messageAttributes).containsKey("foo");
assertThat(messageAttributes.get("foo").getStringValue()).isEqualTo("baz");
}
@Test
@SuppressWarnings("unchecked")
public void testSqsMessageHandlerWithAutoQueueCreate() {
void testSqsMessageHandlerWithAutoQueueCreate() {
Message<String> message = MessageBuilder.withPayload("message").build();
this.sqsMessageHandlerWithAutoQueueCreate.setQueue("foo");