diff --git a/applications/stream-applications-core/common/stream-applications-test-support/pom.xml b/applications/stream-applications-core/common/stream-applications-test-support/pom.xml index dfd8e29d..e051154a 100644 --- a/applications/stream-applications-core/common/stream-applications-test-support/pom.xml +++ b/applications/stream-applications-core/common/stream-applications-test-support/pom.xml @@ -25,6 +25,33 @@ spring-boot-starter-data-redis true + + org.testcontainers + testcontainers + ${test-containers.version} + true + + + + org.testcontainers + junit-jupiter + ${test-containers.version} + true + + + + org.testcontainers + kafka + ${test-containers.version} + true + + + + org.awaitility + awaitility + true + + diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/AppLog.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/AppLog.java new file mode 100644 index 00000000..f892e15e --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/AppLog.java @@ -0,0 +1,38 @@ +/* + * 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; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.output.Slf4jLogConsumer; + +public final class AppLog extends Slf4jLogConsumer { + private static final Map appLogs = new ConcurrentHashMap<>(); + + public static AppLog appLog(String appName) { + if (!appLogs.containsKey(appName)) { + appLogs.put(appName, new AppLog(appName)); + } + return appLogs.get(appName); + } + + private AppLog(String appName) { + super(LoggerFactory.getLogger(appName)); + } +} diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/FluentMap.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/FluentMap.java new file mode 100644 index 00000000..6d56885d --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/FluentMap.java @@ -0,0 +1,36 @@ +/* + * 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; + +import java.util.LinkedHashMap; + +/** + * Fluent API wrapper for {@link java.util.LinkedHashMap}. + * @param key type. + * @param value type. + * @author David Turanski + */ +public class FluentMap extends LinkedHashMap { + public static FluentMap fluentMap() { + return new FluentMap<>(); + } + + public FluentMap withEntry(K key, V value) { + put(key, value); + return this; + } +} diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/LogMatcher.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/LogMatcher.java new file mode 100644 index 00000000..633fd0e9 --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/LogMatcher.java @@ -0,0 +1,92 @@ +/* + * 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; + +import java.util.LinkedList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.Callable; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Consumer; +import java.util.regex.Pattern; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.output.OutputFrame; + +/** + * Utility for matching test container log contents. + * @author David Turanski + */ +public class LogMatcher implements Consumer { + private static Logger logger = LoggerFactory.getLogger(LogMatcher.class); + + private List> listeners = new LinkedList<>(); + + public Callable verifies(Consumer consumer) { + LogListener logListener = new LogListener(); + consumer.accept(logListener); + logListener.runnable.ifPresent(runnable -> runnable.run()); + listeners.add(logListener); + return () -> logListener.matches().get(); + } + + @Override + public void accept(OutputFrame outputFrame) { + listeners.forEach(m -> m.accept(outputFrame.getUtf8String())); + } + + public class LogListener implements Consumer { + private AtomicBoolean matched = new AtomicBoolean(); + + private Optional runnable = Optional.empty(); + + private Pattern pattern; + + @Override + public void accept(String s) { + logger.trace(this + "matching " + s.trim() + " using pattern " + pattern.pattern()); + if (pattern.matcher(s.trim()).matches()) { + logger.debug(" MATCHED " + s.trim()); + matched.set(true); + listeners.remove(this); + } + } + + public LogListener contains(String string) { + return matchesRegex(".*" + string + ".*"); + } + + public LogListener endsWith(String string) { + return matchesRegex(".*" + string); + } + + public LogListener matchesRegex(String regex) { + this.pattern = Pattern.compile(regex); + return this; + } + + public LogListener when(Runnable runnable) { + this.runnable = Optional.of(runnable); + return this; + } + + public AtomicBoolean matches() { + return matched; + } + } +} diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamApps.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamApps.java new file mode 100644 index 00000000..e9953f39 --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamApps.java @@ -0,0 +1,184 @@ +/* + * 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; + +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.function.Consumer; +import java.util.stream.IntStream; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.lifecycle.Startable; + +import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; + +import static org.springframework.cloud.stream.app.test.integration.AppLog.appLog; + +public abstract class StreamApps implements AutoCloseable, Startable { + + protected Logger logger = LoggerFactory.getLogger(this.getClass()); + + private final GenericContainer sourceContainer; + + private final GenericContainer sinkContainer; + + private List processorContainers; + + protected StreamApps(GenericContainer sourceContainer, List processorContainers, + GenericContainer sinkContainer) { + this.sourceContainer = sourceContainer; + this.sinkContainer = sinkContainer; + this.processorContainers = processorContainers; + } + + public GenericContainer sourceContainer() { + return sourceContainer; + } + + public GenericContainer sinkContainer() { + return sinkContainer; + } + + public List processorContainers() { + return processorContainers; + } + + public void start() { + if (logger.isDebugEnabled()) { + logger.debug("Starting apps..."); + + logger.debug("Source container environment:"); + sourceContainer().getEnv().forEach((Consumer) env -> logger.debug(env)); + sourceContainer().withLogConsumer(appLog(sourceContainer().getImage().get())); + + if (!CollectionUtils.isEmpty(processorContainers)) { + logger.debug("\nProcessor containers environment:"); + processorContainers().forEach(container -> { + logger.debug("Processor container environment:"); + container.getEnv().forEach((Consumer) env -> logger.debug(env)); + container.withLogConsumer(appLog(container.getImage().get())); + + }); + } + + logger.debug("\nSink container environment:"); + sinkContainer().getEnv().forEach((Consumer) env -> logger.debug(env)); + sinkContainer().withLogConsumer(appLog(sinkContainer().getImage().get())); + } + + sinkContainer.start(); + processorContainers.forEach(GenericContainer::start); + sourceContainer.start(); + } + + public void stop() { + sinkContainer.stop(); + processorContainers.forEach(GenericContainer::stop); + sourceContainer.stop(); + } + + public static abstract class Builder { + private final String streamName; + + private GenericContainer source; + + private GenericContainer sink; + + private List processors = new LinkedList<>(); + + protected final GenericContainer messageBrokerContainer; + + protected Builder(String streamName, GenericContainer messageBrokerContainer) { + Assert.hasText(streamName, "Stream name is required"); + Assert.notNull(messageBrokerContainer, "A Message broker container is required."); + Assert.isTrue(messageBrokerContainer.isRunning(), "Message broker container must be started first."); + this.messageBrokerContainer = messageBrokerContainer; + this.streamName = streamName; + } + + public Builder withSourceContainer(GenericContainer sourceContainer) { + this.source = sourceContainer; + return this; + } + + public Builder withSinkContainer(GenericContainer sinkContainer) { + this.sink = sinkContainer; + return this; + } + + public Builder withProcessorContainer(GenericContainer processorContainer) { + this.processors.add(processorContainer); + return this; + } + + public StreamApps build() { + + Assert.notNull(source, "A Source container is required."); + Assert.notNull(sink, "A Sink container is required."); + + return streamAppsInstance(setupSourceContainer(), setupProcessorContainers(), setupSinkContainer()); + } + + protected abstract StreamApps streamAppsInstance(GenericContainer sourceContainer, + List processorContainers, GenericContainer sinkContainer); + + private GenericContainer setupSourceContainer() { + return source.withNetwork(messageBrokerContainer.getNetwork()) + .withEnv("SPRING_CLOUD_STREAM_BINDINGS_OUTPUT_DESTINATION", sourceOutputDestination()) + .withEnv(binderProperties()) + .dependsOn(messageBrokerContainer); + } + + private GenericContainer setupSinkContainer() { + return sink + .withNetwork(messageBrokerContainer.getNetwork()) + .withEnv("SPRING_CLOUD_STREAM_BINDINGS_INPUT_DESTINATION", sinkInputDestination()) + .withEnv("SPRING_CLOUD_STREAM_BINDINGS_INPUT_GROUP", streamName) + .withEnv(binderProperties()) + .dependsOn(messageBrokerContainer); + } + + private List setupProcessorContainers() { + IntStream.range(0, processors.size()) + .forEach(i -> processors.get(i).withNetwork(messageBrokerContainer.getNetwork()) + .withEnv("SPRING_CLOUD_STREAM_BINDINGS_INPUT_DESTINATION", + i == 0 ? sourceOutputDestination() : "processor_ " + i) + .withEnv("SPRING_CLOUD_STREAM_BINDINGS_OUTPUT_DESTINATION", + i == (processors.size() - 1) ? sinkInputDestination() + : "processor_" + (i + 1)) + .withEnv("SPRING_CLOUD_STREAM_BINDINGS_INPUT_GROUP", streamName) + .withEnv(binderProperties()) + .dependsOn(messageBrokerContainer)); + return processors; + } + + private String sourceOutputDestination() { + return CollectionUtils.isEmpty(processors) ? streamName : "processor_0"; + } + + private String sinkInputDestination() { + return (CollectionUtils.isEmpty(processors) || processors.size() <= 1) ? streamName + : "processor_" + (processors.size() - 1); + } + + protected abstract Map binderProperties(); + } +} diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamIApplicationIntegrationTestSupport.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamIApplicationIntegrationTestSupport.java new file mode 100644 index 00000000..5d3b2063 --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamIApplicationIntegrationTestSupport.java @@ -0,0 +1,58 @@ +/* + * 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; + +import java.io.File; +import java.io.IOException; +import java.net.InetAddress; +import java.net.UnknownHostException; + +import org.testcontainers.junit.jupiter.Testcontainers; + +import org.springframework.core.io.ClassPathResource; +import org.springframework.util.SocketUtils; + +/** + * Support utility for stream application integration testing . + * @author David Turanski + */ +@Testcontainers +public abstract class StreamIApplicationIntegrationTestSupport { + + protected static String localHostAddress() { + try { + return InetAddress.getLocalHost().getHostAddress(); + } + catch (UnknownHostException e) { + throw new IllegalStateException(e.getMessage(), e); + } + } + + protected static File resourceAsFile(String path) { + try { + return new ClassPathResource(path).getFile(); + } + catch (IOException e) { + throw new IllegalStateException("Unable to access resource " + path); + } + } + + protected static final int findAvailablePort() { + return SocketUtils.findAvailableTcpPort(10000, 20000); + } + +} diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/AbstractKafkaStreamApplicationIntegrationTests.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/AbstractKafkaStreamApplicationIntegrationTests.java new file mode 100644 index 00000000..cfe456e5 --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/AbstractKafkaStreamApplicationIntegrationTests.java @@ -0,0 +1,42 @@ +/* + * 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.kafka; + +import org.testcontainers.containers.KafkaContainer; +import org.testcontainers.containers.Network; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; + +import org.springframework.cloud.stream.app.test.integration.StreamIApplicationIntegrationTestSupport; + +/** + * Base class for stream application integration testing with Test Containers and Kafka + * binder + */ +@Testcontainers +public abstract class AbstractKafkaStreamApplicationIntegrationTests extends StreamIApplicationIntegrationTestSupport { + + final static Network network = Network.SHARED; + + protected final static KafkaContainer kafka = new KafkaContainer( + DockerImageName.parse("confluentinc/cp-kafka:5.5.1")) + .withNetwork(network); + + static { + kafka.start(); + } +} diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApps.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApps.java new file mode 100644 index 00000000..3e44ea1e --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApps.java @@ -0,0 +1,55 @@ +/* + * 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.kafka; + +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import org.testcontainers.containers.GenericContainer; + +import org.springframework.cloud.stream.app.test.integration.StreamApps; + +public class KafkaStreamApps extends StreamApps { + + protected KafkaStreamApps(GenericContainer sourceContainer, List processorContainers, + GenericContainer sinkContainer) { + super(sourceContainer, processorContainers, sinkContainer); + } + + public static Builder kafkaStreamApps(String streamName, GenericContainer messageBrokerContainer) { + return new KafkaBuilder(streamName, messageBrokerContainer); + } + + public static final class KafkaBuilder extends Builder { + + protected KafkaBuilder(String streamName, GenericContainer messageBrokerContainer) { + super(streamName, messageBrokerContainer); + } + + protected Map binderProperties() { + return Collections.singletonMap("SPRING_CLOUD_STREAM_KAFKA_BINDER_BROKERS", + messageBrokerContainer.getNetworkAliases().get(0) + ":9092"); + } + + @Override + protected StreamApps streamAppsInstance(GenericContainer sourceContainer, + List processorContainers, GenericContainer sinkContainer) { + return new KafkaStreamApps(sourceContainer, processorContainers, sinkContainer); + } + } +}