SpringCloud/stream-applications
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactored to allow BaseContainer as a Type annotation

Browse Source
This commit is contained in:
David Turanski
2020-11-04 11:03:09 -05:00
parent d61ac4665e
commit 2aa59e0613
15 changed files with 298 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
applications/stream-applications-core/stream-applications-test-support/README.adoc
View File

@@ -110,8 +110,8 @@ public class KafkaTimeSourceTests {
private OutputMatcher outputMatcher;
@Container
static StreamAppContainer timeSource = KafkaConfig.prepackagedContainerFor("time-source", VERSION)
.withLogConsumer(logMatcher);
static StreamAppContainer source = new KafkaStreamAppContainer(StreamAppContainerTestUtils
.imageName(StreamAppContainerTestUtils.SPRINGCLOUDSTREAM_REPOSITOTRY, "time-source-kafka", VERSION));
@Test
void test() {
await().atMost(DEFAULT_DURATION).until(logMatcher.matches());
@@ -122,7 +122,6 @@ public class KafkaTimeSourceTests {
We start an ApplicationContext `KafkaStreamAppContainerTestConfiguration` which starts the `KafkaConfig.kafka` TestContainer in a static initializer.
The Time Source emits the time every second. In this case, it's hard to know what the expected output payload is, but it should at least match the date pattern.
This test also uses a `LogMatcher`, which is not strictly necessary, but used here to verify that the app logged the standard start up message - always a good sign.
Then we wait for a message on the output topic that matches the pattern.
@@ -159,27 +158,35 @@ This realizes the concepts of `source`, `processor`, and `sink` , and similar Sp
Here is a test for the canonical `TikTok` stream:
```java
public class RabbitMQTikTokTests extends RabbitMQStreamApplicationIntegrationTestSupport {
@RabbitMQStreamAppTest
public class RabbitTikTokTests {
private static LogMatcher logMatcher = LogMatcher.matchesRegex(".*\\d{2}/\\d{2}/\\d{2}\\s+\\d{2}:\\d{2}:\\d{2}")
.times(3);
@Container
private static final StreamApps streamApp = rabbitMQStreamApps(RabbitMQTikTokTests.class.getSimpleName(), rabbitmq)
.withSourceContainer(prepackagedRabbitMQContainerFor("time-source", VERSION))
.withSinkContainer(prepackagedRabbitMQContainerFor("log-sink", VERSION)
.withLogConsumer(logMatcher))
.build();
private static final StreamApps streamApp = kafkaStreamApps(KafkaTikTokTests.class.getSimpleName(),
KafkaConfig.kafka)
.withSourceContainer(
new RabbitMQStreamAppContainer(StreamAppContainerTestUtils.imageName(
"time-source-rabbit",
VERSION)))
.withSinkContainer(
new RabbitMQStreamAppContainer(StreamAppContainerTestUtils.imageName(
"log-sink-rabbit",
VERSION)).withLogConsumer(logMatcher)
.log())
.build();
@Test
void test() {
await().atMost(DEFAULT_DURATION).until(logMatcher.matches());
}
}
```
Here, the link:src/main/java/org/springframework/cloud/stream/app/test/integration/LogMatcher.java[LogMatcher] can be declared statically since it doesn't depend on Spring beans.
This is an extension of TestContainer's `LogConsumer` so it is created before the container starts. Here, we
verify the LogSink logs at least 3 messages that match the pattern.
Here, the link:src/main/java/org/springframework/cloud/stream/app/test/integration/LogMatcher.java[LogMatcher].
This is an extension of TestContainer's `LogConsumer`. Here, we verify the LogSink logs at least 3 messages that match the pattern.
link:src/main/java/org/springframework/cloud/stream/app/test/integration/AppLog.java[AppLog] is also another useful LogConsumer
to enable container logging.

2
applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamAppContainer.java
View File

@@ -94,7 +94,7 @@ public abstract class StreamAppContainer extends GenericContainer<StreamAppConta
/**
* Assign a destination name to the standard input.
* @param destination the destination name.
* @return
* @return this.
*/
public StreamAppContainer withInputDestination(String destination) {
Assert.hasText(destination, "'destination' is required.");

12
applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamAppContainerTestUtils.java
View File

@@ -31,12 +31,16 @@ import org.springframework.util.SocketUtils;
public abstract class StreamAppContainerTestUtils {
/**
* Default docker org.
* Default docker repository.
*/
public static String DOCKER_ORG = "springcloudstream";
public static final String SPRINGCLOUDSTREAM_REPOSITOTRY = "springcloudstream";
public static final String prePackagedStreamAppImageName(String appName, String binderName, String version) {
return DOCKER_ORG + "/" + appName + "-" + binderName + ":" + version;
public static final String imageName(String appName, String version) {
return imageName(SPRINGCLOUDSTREAM_REPOSITOTRY, appName, version);
}
public static final String imageName(String repository, String appName, String version) {
return repository + "/" + appName + ":" + version;
}
public static final String localHostAddress() {

4
applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamApps.java
View File

@@ -32,9 +32,9 @@ import org.springframework.util.CollectionUtils;
/**
* The base class used for testing end-to-end Stream applications.
* @see {@link org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamApps},
* {@link org.springframework.cloud.stream.app.test.integration.rabbitmq.RabbitMQStreamApps}.
* @author David Turanski
* @see org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamApps
* @see org.springframework.cloud.stream.app.test.integration.rabbitmq.RabbitMQStreamApps
*/
public abstract class StreamApps implements AutoCloseable, Startable {

33
applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/BaseContainer.java
View File

@@ -17,9 +17,13 @@
package org.springframework.cloud.stream.app.test.integration.junit.jupiter;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import java.util.function.Supplier;
import static org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils.SPRINGCLOUDSTREAM_REPOSITOTRY;
/**
* Marker for a
@@ -29,7 +33,34 @@ import java.lang.annotation.Target;
* must be public.
* @author David Turanski
*/
@Target(ElementType.FIELD)
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
public @interface BaseContainer {
/**
* @return the version or image tag.
*/
String version() default "";
/**
* @return A {@code Class<? extends Supplier<String>>} used to set the version
* dynamically.
*/
Class<? extends Supplier<String>> versionSupplier() default NullVersionSupplier.class;
/**
* @return the image name.
*/
String name() default "";
/**
* @return the type of Spring Cloud Stream binder used by the application.
*/
Binder binder();
/**
* @return the Docker repository name of the image.
*/
String repository() default SPRINGCLOUDSTREAM_REPOSITOTRY;
}

86
applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/BaseContainerExtension.java
View File

@@ -16,49 +16,91 @@
package org.springframework.cloud.stream.app.test.integration.junit.jupiter;
import java.util.concurrent.atomic.AtomicInteger;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.function.Supplier;
import org.junit.jupiter.api.extension.ConditionEvaluationResult;
import org.junit.jupiter.api.extension.ExecutionCondition;
import org.junit.jupiter.api.extension.ExtensionConfigurationException;
import org.junit.jupiter.api.extension.ExtensionContext;
import org.junit.platform.commons.util.AnnotationUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.app.test.integration.StreamAppContainer;
import org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils;
import org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamAppContainer;
import org.springframework.cloud.stream.app.test.integration.rabbitmq.RabbitMQStreamAppContainer;
import org.springframework.core.annotation.AnnotatedElementUtils;
/**
* A junit Jupiter extension used to discover {@link StreamAppContainer}s annotated with {code @BaseContainer}.
* A junit Jupiter extension used to discover {@link StreamAppContainer}s annotated with
* {@link BaseContainer}.
* @author David Turanski
*/
public class BaseContainerExtension implements ExecutionCondition {
private static StreamAppContainer baseContainer;
private static Logger logger = LoggerFactory.getLogger(BaseContainerExtension.class);
public static StreamAppContainer containerInstance() {
return baseContainer;
}
@Override
public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext extensionContext) {
AtomicInteger count = new AtomicInteger();
AnnotationUtils.findAnnotatedFields(extensionContext.getRequiredTestClass(), BaseContainer.class, field -> {
try {
StreamAppContainer base = (StreamAppContainer) field.get(null);
if (base == null) {
throw new ExtensionConfigurationException("@BaseContainer is not initialized");
}
count.getAndIncrement();
baseContainer = base;
}
catch (Exception e) {
throw new ExtensionConfigurationException(e.getMessage(), e);
}
return true;
});
if (count.get() != 1) {
throw new ExtensionConfigurationException(
"Expecting exactly one @BaseContainer instance, found " + count.get());
BaseContainer annotation = AnnotatedElementUtils.getMergedAnnotation(extensionContext.getRequiredTestClass(),
BaseContainer.class);
if (annotation == null) {
throw new ExtensionConfigurationException("@BaseContainer is required for this extension");
}
return ConditionEvaluationResult.enabled("@BaseContainer found");
String version = getVersion(annotation);
switch (annotation.binder()) {
case Kafka:
baseContainer = new KafkaStreamAppContainer(
StreamAppContainerTestUtils.imageName(annotation.repository(), annotation.name(), version));
break;
case RabbitMQ:
baseContainer = new RabbitMQStreamAppContainer(
StreamAppContainerTestUtils.imageName(annotation.repository(), annotation.name(), version));
break;
default:
throw new ExtensionConfigurationException(
"the binder type " + annotation.binder().name() + " is not supported");
}
logger.debug("StreamAppContainer created using base container image " + baseContainer.getDockerImageName());
return ConditionEvaluationResult
.enabled("StreamAppContainer created using base container image " + baseContainer.getDockerImageName());
}
private String getVersion(BaseContainer annotation) {
String version = null;
if (annotation.version().isEmpty() && annotation.versionSupplier().equals(NullVersionSupplier.class)) {
throw new ExtensionConfigurationException(
"either 'version' or 'versionSupplier' must be set in @BaseContainer");
}
if (!annotation.version().isEmpty() && !annotation.versionSupplier().equals(NullVersionSupplier.class)) {
throw new ExtensionConfigurationException(
"only one of 'version' or 'versionSupplier' must be set in @BaseContainer");
}
if (annotation.version().isEmpty()) {
try {
Constructor<? extends Supplier<String>> constructor = annotation.versionSupplier().getConstructor();
version = constructor.newInstance().get();
}
catch (NoSuchMethodException | InstantiationException | IllegalAccessException
| InvocationTargetException e) {
throw new ExtensionConfigurationException(
"No accessible default constructor found for " + annotation.versionSupplier().getName());
}
}
else {
version = annotation.version();
}
return version;
}
}

33
applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/Binder.java Normal file
View File

@@ -0,0 +1,33 @@
/*
* Copyright 2020-2020 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.cloud.stream.app.test.integration.junit.jupiter;
/**
* Supported Spring Cloud Stream binder types.
* @author David Turanski
*/
public enum Binder {
/**
* Kafka Binder.
*/
Kafka,
/**
* RabbitMQ Binder.
*/
RabbitMQ,
}

50
applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/KafkaBaseContainer.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* Copyright 2020-2020 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.cloud.stream.app.test.integration.junit.jupiter;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import java.util.function.Supplier;
import org.springframework.core.annotation.AliasFor;
import static org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils.SPRINGCLOUDSTREAM_REPOSITOTRY;
/**
* Annotation For configuring a {@link org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamAppContainer}.
* @author David Turanski
* @see org.springframework.cloud.stream.app.test.integration.junit.jupiter.BaseContainer
*/
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@BaseContainer(binder = Binder.Kafka)
public @interface KafkaBaseContainer {
@AliasFor(annotation = BaseContainer.class)
String version() default "";
@AliasFor(annotation = BaseContainer.class)
Class<? extends Supplier<String>> versionSupplier() default NullVersionSupplier.class;
@AliasFor(annotation = BaseContainer.class)
String name() default "";
@AliasFor(annotation = BaseContainer.class)
String repository() default SPRINGCLOUDSTREAM_REPOSITOTRY;
}

31
applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/NullVersionSupplier.java Normal file
View File

@@ -0,0 +1,31 @@
/*
* Copyright 2020-2020 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.cloud.stream.app.test.integration.junit.jupiter;
import java.util.function.Supplier;
/**
* A placeHolder for {@link org.springframework.cloud.stream.app.test.integration.junit.jupiter.BaseContainer}
* indicating no "versionSupplier" has been set.
* @author David Turanski
*/
class NullVersionSupplier implements Supplier<String> {
@Override
public String get() {
return null;
}
}

50
applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/RabbitMQBaseContainer.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* Copyright 2020-2020 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.cloud.stream.app.test.integration.junit.jupiter;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import java.util.function.Supplier;
import org.springframework.core.annotation.AliasFor;
import static org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils.SPRINGCLOUDSTREAM_REPOSITOTRY;
/**
* Annotation For configuring a
* {@link org.springframework.cloud.stream.app.test.integration.rabbitmq.RabbitMQStreamAppContainer}.
* @author David Turanski
* @see org.springframework.cloud.stream.app.test.integration.junit.jupiter.BaseContainer
*/
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@BaseContainer(binder = Binder.RabbitMQ)
public @interface RabbitMQBaseContainer {
@AliasFor(annotation = BaseContainer.class)
String version() default "";
@AliasFor(annotation = BaseContainer.class)
Class<? extends Supplier<String>> versionSupplier() default NullVersionSupplier.class;
@AliasFor(annotation = BaseContainer.class)
String name() default "";
@AliasFor(annotation = BaseContainer.class)
String repository() default SPRINGCLOUDSTREAM_REPOSITOTRY;
}

12
applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaConfig.java
View File

@@ -20,24 +20,14 @@ import org.testcontainers.containers.KafkaContainer;
import org.testcontainers.containers.Network;
import org.testcontainers.utility.DockerImageName;
import org.springframework.cloud.stream.app.test.integration.StreamAppContainer;
import org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils;
/**
* Initializes and starts a Kafka TestContainer and provides associated utilities for configuring a {@link KafkaStreamAppContainer}.
* Initializes and starts a {@link KafkaContainer}.
* @author David Turanski
*/
public abstract class KafkaConfig {
final static String BINDER = "kafka";
final static Network network = Network.SHARED;
public static StreamAppContainer prepackagedContainerFor(String appName, String version) {
return new KafkaStreamAppContainer(
StreamAppContainerTestUtils.prePackagedStreamAppImageName(appName, BINDER, version),
kafka);
}
/**
* The KafkaContainer.
*/

7
applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppContainer.java
View File

@@ -16,8 +16,6 @@
package org.springframework.cloud.stream.app.test.integration.kafka;
import org.testcontainers.containers.GenericContainer;
import org.springframework.cloud.stream.app.test.integration.StreamAppContainer;
/**
@@ -30,10 +28,9 @@ public class KafkaStreamAppContainer extends StreamAppContainer {
/**
* @param imageName the image name.
* @param kafka a running kafka TestContainer instance.
*/
public KafkaStreamAppContainer(String imageName, GenericContainer kafka) {
super(imageName, kafka);
public KafkaStreamAppContainer(String imageName) {
super(imageName, KafkaConfig.kafka);
}
@Override

16
applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQConfig.java
View File

@@ -20,13 +20,8 @@ import org.testcontainers.containers.Network;
import org.testcontainers.containers.RabbitMQContainer;
import org.testcontainers.utility.DockerImageName;
import org.springframework.cloud.stream.app.test.integration.StreamAppContainer;
import org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils;
/**
* Initializes and starts a RabbitMQ TestContainer and provides associated utilities for
* configuring a
* {@link org.springframework.cloud.stream.app.test.integration.rabbitmq.RabbitMQStreamAppContainer}.
* Initializes and starts a {@link RabbitMQContainer}.
* @author David Turanski
*/
public abstract class RabbitMQConfig {
@@ -35,8 +30,6 @@ public abstract class RabbitMQConfig {
*/
public static RabbitMQContainer rabbitmq;
final static String BINDER = "rabbit";
final static Network network = Network.SHARED;
static {
@@ -45,11 +38,4 @@ public abstract class RabbitMQConfig {
.withExposedPorts(5672, 15672);
rabbitmq.start();
}
public static StreamAppContainer prepackagedContainerFor(String appName, String version) {
return new RabbitMQStreamAppContainer(
StreamAppContainerTestUtils.prePackagedStreamAppImageName(appName, BINDER, version),
rabbitmq);
}
}

8
applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainer.java
View File

@@ -16,8 +16,6 @@
package org.springframework.cloud.stream.app.test.integration.rabbitmq;
import org.testcontainers.containers.GenericContainer;
import org.springframework.cloud.stream.app.test.integration.StreamAppContainer;
/**
@@ -30,10 +28,10 @@ public class RabbitMQStreamAppContainer extends StreamAppContainer {
/**
* @param imageName the image name.
* @param rabbitmq a running rabbitMQ TestContainer instance.
*/
public RabbitMQStreamAppContainer(String imageName, GenericContainer rabbitmq) {
super(imageName, rabbitmq);
public RabbitMQStreamAppContainer(String imageName) {
super(imageName, RabbitMQConfig.rabbitmq);
}
@Override

2
applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainerTestConfiguration.java
View File

@@ -292,7 +292,7 @@ public abstract class RabbitMQStreamAppContainerTestConfiguration {
@Override
public <P> void send(String topic, P payload) {
await().atMost(Duration.ofSeconds(30)).pollDelay(Duration.ofSeconds(1)).pollInterval(Duration.ofSeconds(1))
await().atMost(Duration.ofSeconds(30)).pollInterval(Duration.ofSeconds(1))
.until(topicExistsAndIsBound(topic));
rabbitTemplate.convertAndSend(topic, routingKey, payload);
}