diff --git a/applications/pom.xml b/applications/pom.xml index 0cd8a89e..96e1c2aa 100644 --- a/applications/pom.xml +++ b/applications/pom.xml @@ -13,6 +13,7 @@ source sink processor + stream-applications-integration-tests diff --git a/applications/source/file-source/pom.xml b/applications/source/file-source/pom.xml index 20a3284f..3469bc17 100644 --- a/applications/source/file-source/pom.xml +++ b/applications/source/file-source/pom.xml @@ -85,6 +85,14 @@ Spring Milestone Release https://repo.spring.io/milestone + + + false + + spring-release + Spring Release + https://repo.spring.io/release + @@ -103,5 +111,14 @@ Spring Milestones https://repo.spring.io/milestone + + + false + + spring-releases + Spring Releases + https://repo.spring.io/release + + diff --git a/applications/source/geode-source/pom.xml b/applications/source/geode-source/pom.xml index b0259789..73170980 100644 --- a/applications/source/geode-source/pom.xml +++ b/applications/source/geode-source/pom.xml @@ -40,8 +40,6 @@ ${spring-data-geode-test.version} test - - @@ -115,5 +113,13 @@ Spring Milestones https://repo.spring.io/milestone + + + false + + spring-releases + Spring Releases + https://repo.spring.io/release + diff --git a/applications/stream-applications-integration-tests/README.adoc b/applications/stream-applications-integration-tests/README.adoc new file mode 100644 index 00000000..361577ff --- /dev/null +++ b/applications/stream-applications-integration-tests/README.adoc @@ -0,0 +1,42 @@ += Stream Applications Integration Tests + +This contains integration tests for pre-packaged stream-applications for Docker using https://www.testcontainers.org/[TestContainers]. +These are end-to-end integration tests running apps and required resources, using docker-compose. +The goal is to have an end-to-end integration test for each pre-packaged application. +We don't aim to test all different configuration options, as this is the responsibility of the stream application and function components. +One of the major benefits is to verify the built Docker images run correctly, especially when we introduce global changes, +such as upgrading the base JDK image, the maven plugins, or other pervasive changes. + +== Test Strategy + +See https://github.com/spring-cloud/stream-applications/tree/master/applications/stream-applications-core/common/stream-applications-test-support[] for a full description. + + +The tests use following patterns: + +== Source +To test a source, we may require some application specific setup or event to trigger the source. +For example, the jdbc source needs some data in the database to which it is listening. +Then use an `OutputMatcher` to verify the output. + +== Sink +To test a sink, we need to publish a message to its input. Simply use the provided TestTopicSender. +Then we need to verify the result by checking the sink's external resource. + +== Processor +To test a processor we publish a message and use an `OutputMatcher` to verify the output. + +== Configuration +See link:src/test/java/org/springframework/cloud/stream/apps/integration/test/common/Configuration.java[Configuration] for configuration +options. These tests use Spring but not boot currently. +The most important setting is the image versions to test. +To override it, set the System property, e.g., + +./mvnw clean test -Dspring.cloud.stream.applications.version=3.0.0-M3 + + + + + + + diff --git a/applications/stream-applications-integration-tests/pom.xml b/applications/stream-applications-integration-tests/pom.xml new file mode 100644 index 00000000..3a4bb4c2 --- /dev/null +++ b/applications/stream-applications-integration-tests/pom.xml @@ -0,0 +1,187 @@ + + + + 4.0.0 + + org.springframework.cloud.stream.app + stream-applications-core + 3.1.0-SNAPSHOT + + + + stream-applications-integration-tests + stream-applications-integration-tests + Integration Tests for stream applications + + + 2.6.2 + 2.27.1 + 1.11.415 + 8.0.16 + + + + + org.springframework.boot + spring-boot-starter-webflux + test + + + org.springframework.boot + spring-boot-starter-test + test + + + org.testcontainers + testcontainers + test + + + + org.testcontainers + junit-jupiter + test + + + + org.testcontainers + kafka + test + + + org.testcontainers + rabbitmq + + + org.testcontainers + mongodb + test + + + org.testcontainers + mysql + test + + + mysql + mysql-connector-java + ${mysql-connector-java.version} + test + + + org.springframework.cloud.stream.app + stream-applications-test-support + ${stream-apps-core.version} + test + + + org.springframework.cloud.fn + function-test-support + test + + + org.springframework.data + spring-data-geode + + + org.apache.logging.log4j + log4j + + + test + + + com.amazonaws + aws-java-sdk-s3 + ${aws.version} + test + + + com.squareup.okhttp3 + mockwebserver + test + + + org.springframework.boot + spring-boot-starter-data-mongodb + test + + + org.mariadb.jdbc + mariadb-java-client + ${mariadb-client.version} + test + + + org.springframework.boot + spring-boot-starter-jdbc + test + + + + + + + org.testcontainers + testcontainers-bom + ${test-containers.version} + pom + import + + + + + + + spring-snapshots + Spring Snapshots + https://repo.spring.io/snapshot + + true + + + + spring-milestone-release + Spring Milestone Release + https://repo.spring.io/milestone + + + spring-milestones + Spring Milestones + https://repo.spring.io/milestone + + + spring-releases + Spring Releases + https://repo.spring.io/release + + false + + + + + + spring-snapshots + Spring Snapshots + https://repo.spring.io/snapshot + + true + + + + spring-milestones + Spring Milestones + https://repo.spring.io/milestone + + + spring-releases + Spring Releases + https://repo.spring.io/release + + false + + + + + diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/common/Configuration.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/common/Configuration.java new file mode 100644 index 00000000..d5d1a14d --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/common/Configuration.java @@ -0,0 +1,48 @@ +/* + * 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.integration.test.common; + +import java.time.Duration; +import java.util.function.Supplier; + +public abstract class Configuration { + + /** + * Version. + */ + public static String VERSION; + + /** + * Duration. + */ + public static final Duration DEFAULT_DURATION = Duration.ofMinutes(1); + + private static final String SPRING_CLOUD_STREAM_APPLICATIONS_VERSION = "spring.cloud.stream.applications.version"; + + static { + VERSION = System.getProperty(SPRING_CLOUD_STREAM_APPLICATIONS_VERSION, "latest"); + } + + public static class VersionSupplier implements Supplier { + + @Override + public String get() { + return VERSION; + } + } + +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/processor/httprequest/HttpRequestProcessorTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/processor/httprequest/HttpRequestProcessorTests.java new file mode 100644 index 00000000..26ddaf00 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/processor/httprequest/HttpRequestProcessorTests.java @@ -0,0 +1,95 @@ +/* + * 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.integration.test.processor.httprequest; + +import java.io.IOException; +import java.net.InetAddress; + +import okhttp3.mockwebserver.Dispatcher; +import okhttp3.mockwebserver.MockResponse; +import okhttp3.mockwebserver.MockWebServer; +import okhttp3.mockwebserver.RecordedRequest; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cloud.stream.app.test.integration.OutputMatcher; +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.TestTopicSender; +import org.springframework.http.HttpStatus; + +import static org.awaitility.Awaitility.await; +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.DEFAULT_DURATION; +import static org.springframework.cloud.stream.app.test.integration.AppLog.appLog; + +@Tag("integration") +abstract class HttpRequestProcessorTests { + + private static MockWebServer server; + + private static int serverPort; + + @Autowired + private TestTopicSender testTopicSender; + + @Autowired + private OutputMatcher outputMatcher; + + private static StreamAppContainer processor; + + protected static StreamAppContainer configureProcessor(StreamAppContainer baseContainer) { + serverPort = StreamAppContainerTestUtils.findAvailablePort(); + processor = baseContainer.withLogConsumer(appLog("http-request-processor")) + .withEnv("HTTP_REQUEST_URL_EXPRESSION", + "'http://" + StreamAppContainerTestUtils.localHostAddress() + ":" + serverPort + "'") + .withEnv("HTTP_REQUEST_HTTP_METHOD_EXPRESSION", "'POST'"); + return processor; + } + + @BeforeAll + static void startServer() throws Exception { + server = new MockWebServer(); + server.start(InetAddress.getLocalHost(), serverPort); + } + + @Test + void get() { + server.setDispatcher(new Dispatcher() { + @Override + public MockResponse dispatch(RecordedRequest recordedRequest) { + return new MockResponse() + .setBody("{\"response\":\"" + recordedRequest.getBody().readUtf8() + "\"}") + .setResponseCode(HttpStatus.OK.value()); + } + }); + testTopicSender.send(processor.getInputDestination(), "ping"); + await().atMost(DEFAULT_DURATION) + .until(outputMatcher.messageMatches(message -> message.getPayload().equals("{\"response\":\"ping\"}"))); + // See https://github.com/spring-cloud/spring-cloud-stream/issues/2190 .This condition is no longer true. + // && message.getHeaders().get(MessageHeaders.CONTENT_TYPE) + // .equals(MediaType.APPLICATION_JSON_VALUE))); + } + + @AfterAll + static void cleanUp() throws IOException { + server.shutdown(); + } + +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/processor/httprequest/KafkaHttpRequestProcessorTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/processor/httprequest/KafkaHttpRequestProcessorTests.java new file mode 100644 index 00000000..e670f871 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/processor/httprequest/KafkaHttpRequestProcessorTests.java @@ -0,0 +1,35 @@ +/* + * 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.integration.test.processor.httprequest; + +import org.testcontainers.junit.jupiter.Container; + +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.junit.jupiter.KafkaStreamAppTest; +import org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamAppContainer; + +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.VERSION; + +@KafkaStreamAppTest +class KafkaHttpRequestProcessorTests extends HttpRequestProcessorTests { + @Container + private static StreamAppContainer container = configureProcessor( + new KafkaStreamAppContainer(StreamAppContainerTestUtils.imageName( + "http-request-processor-kafka", VERSION))); + +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/processor/httprequest/RabbitMQHttpRequestProcessorTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/processor/httprequest/RabbitMQHttpRequestProcessorTests.java new file mode 100644 index 00000000..72e8352e --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/processor/httprequest/RabbitMQHttpRequestProcessorTests.java @@ -0,0 +1,35 @@ +/* + * 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.integration.test.processor.httprequest; + +import org.testcontainers.junit.jupiter.Container; + +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.junit.jupiter.RabbitMQStreamAppTest; +import org.springframework.cloud.stream.app.test.integration.rabbitmq.RabbitMQStreamAppContainer; + +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.VERSION; + +@RabbitMQStreamAppTest +class RabbitMQHttpRequestProcessorTests extends HttpRequestProcessorTests { + + @Container + private static StreamAppContainer container = configureProcessor( + new RabbitMQStreamAppContainer(StreamAppContainerTestUtils.imageName( + "http-request-processor-rabbit", VERSION))); +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/jdbc/JdbcSinkTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/jdbc/JdbcSinkTests.java new file mode 100644 index 00000000..9d9711ea --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/jdbc/JdbcSinkTests.java @@ -0,0 +1,115 @@ +/* + * 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.integration.test.sink.jdbc; + +import com.zaxxer.hikari.HikariDataSource; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.testcontainers.containers.BindMode; +import org.testcontainers.containers.MySQLContainer; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.utility.DockerImageName; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cloud.stream.app.test.integration.StreamAppContainer; +import org.springframework.cloud.stream.app.test.integration.TestTopicSender; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.BaseContainerExtension; +import org.springframework.cloud.stream.app.test.integration.kafka.KafkaConfig; +import org.springframework.jdbc.core.JdbcTemplate; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.DEFAULT_DURATION; +import static org.springframework.cloud.stream.app.test.integration.AppLog.appLog; + +@Tag("integration") +@ExtendWith(BaseContainerExtension.class) +public abstract class JdbcSinkTests { + + private static JdbcTemplate jdbcTemplate; + + private static StreamAppContainer sink; + + @Autowired + private TestTopicSender testTopicSender; + + @Container + private static MySQLContainer mySQL = new MySQLContainer<>(DockerImageName.parse("mysql:5.7")) + .withUsername("test") + .withPassword("secret") + .withExposedPorts(3306) + .withNetwork(KafkaConfig.kafka.getNetwork()) + .withNetworkAliases("mysql-for-sink") + .withClasspathResourceMapping("init.sql", "/init.sql", BindMode.READ_ONLY) + .withLogConsumer(appLog("mysql-for-sink")) + .withCommand("--init-file", "/init.sql"); + + @BeforeAll + static void init() { + sink = BaseContainerExtension.containerInstance() + .dependsOn(mySQL) + .withEnv("JDBC_CONSUMER_COLUMNS", "name,city:address.city,street:address.street") + .withEnv("JDBC_CONSUMER_TABLE_NAME", "People") + .withEnv("SPRING_DATASOURCE_USERNAME", "test") + .withEnv("SPRING_DATASOURCE_PASSWORD", "secret") + .withEnv("SPRING_DATASOURCE_DRIVER_CLASS_NAME", "org.mariadb.jdbc.Driver") + .withEnv("SPRING_DATASOURCE_URL", + "jdbc:mariadb://mysql-for-sink:3306/test") + .waitingFor(Wait.forLogMessage(".*Started JdbcSink.*", 1)); + startSink(); + } + + static void startSink() { + + HikariDataSource dataSource = new HikariDataSource(); + dataSource.setDriverClassName("org.mariadb.jdbc.Driver"); + dataSource.setUsername(mySQL.getUsername()); + dataSource.setPassword(mySQL.getPassword()); + dataSource.setJdbcUrl("jdbc:mysql://localhost:" + mySQL.getMappedPort(3306) + "/test"); + jdbcTemplate = new JdbcTemplate(dataSource); + jdbcTemplate.execute("DELETE FROM People"); + await().atMost(DEFAULT_DURATION) + .until(() -> jdbcTemplate.queryForObject("SELECT COUNT(*) from People", Integer.class) + .intValue() == 0); + sink.start(); + } + + @Test + void test() { + + String json = "{\"name\":\"My Name\",\"address\":{ \"city\": \"Big City\",\"street\":\"Narrow Alley\"}}"; + testTopicSender.send(sink.getInputDestination(), json); + + await().atMost(DEFAULT_DURATION) + .untilAsserted( + () -> assertThat( + jdbcTemplate.queryForObject("SELECT COUNT(*) from People", Integer.class).intValue()) + .isOne()); + assertThat(jdbcTemplate.queryForObject("SELECT name from People", + String.class)).isEqualTo("My Name"); + } + + @AfterAll + static void cleanUp() { + sink.stop(); + } + +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/jdbc/KafkaJdbcSinkTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/jdbc/KafkaJdbcSinkTests.java new file mode 100644 index 00000000..35bc4d3d --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/jdbc/KafkaJdbcSinkTests.java @@ -0,0 +1,26 @@ +/* + * 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.integration.test.sink.jdbc; + +import org.springframework.cloud.stream.app.integration.test.common.Configuration; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaBaseContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaStreamAppTest; + +@KafkaStreamAppTest +@KafkaBaseContainer(name = "jdbc-sink-kafka", versionSupplier = Configuration.VersionSupplier.class) +public class KafkaJdbcSinkTests extends JdbcSinkTests { +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/jdbc/RabbitMQJdbcSinkTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/jdbc/RabbitMQJdbcSinkTests.java new file mode 100644 index 00000000..b12b5582 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/jdbc/RabbitMQJdbcSinkTests.java @@ -0,0 +1,26 @@ +/* + * 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.integration.test.sink.jdbc; + +import org.springframework.cloud.stream.app.integration.test.common.Configuration; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.RabbitMQBaseContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.RabbitMQStreamAppTest; + +@RabbitMQStreamAppTest +@RabbitMQBaseContainer(name = "jdbc-sink-rabbit", versionSupplier = Configuration.VersionSupplier.class) +public class RabbitMQJdbcSinkTests extends JdbcSinkTests { +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/mongodb/KafkaMongoDBSinkTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/mongodb/KafkaMongoDBSinkTests.java new file mode 100644 index 00000000..032eb9a4 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/mongodb/KafkaMongoDBSinkTests.java @@ -0,0 +1,26 @@ +/* + * 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.integration.test.sink.mongodb; + +import org.springframework.cloud.stream.app.integration.test.common.Configuration; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaBaseContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaStreamAppTest; + +@KafkaStreamAppTest +@KafkaBaseContainer(name = "mongodb-sink-kafka", versionSupplier = Configuration.VersionSupplier.class) +public class KafkaMongoDBSinkTests extends MongoDBSinkTests { +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/mongodb/MongoDBSinkTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/mongodb/MongoDBSinkTests.java new file mode 100644 index 00000000..323fa246 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/mongodb/MongoDBSinkTests.java @@ -0,0 +1,99 @@ +/* + * 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.integration.test.sink.mongodb; + +import java.time.Duration; +import java.util.List; + +import org.bson.Document; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.testcontainers.containers.MongoDBContainer; +import org.testcontainers.containers.wait.strategy.Wait; + +import org.springframework.beans.factory.annotation.Autowired; +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.TestTopicSender; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.BaseContainerExtension; +import org.springframework.data.mongodb.MongoDatabaseFactory; +import org.springframework.data.mongodb.core.MongoTemplate; +import org.springframework.data.mongodb.core.SimpleMongoClientDatabaseFactory; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.DEFAULT_DURATION; +@Tag("integration") +@ExtendWith(BaseContainerExtension.class) +abstract class MongoDBSinkTests { + + private static MongoTemplate mongoTemplate; + + @Autowired + private TestTopicSender testTopicSender; + + private static final MongoDBContainer mongoDBContainer = new MongoDBContainer("mongo:4.0.10") + .withExposedPorts(27017) + .withStartupTimeout(Duration.ofMinutes(2)); + + private static String mongoConnectionString() { + return String.format("mongodb://%s:%s/%s", StreamAppContainerTestUtils.localHostAddress(), + mongoDBContainer.getMappedPort(27017), "test"); + } + + private static StreamAppContainer sink; + + @BeforeAll + protected static void configureSink() { + mongoDBContainer.start(); + sink = BaseContainerExtension.containerInstance() + .withEnv("MONGODB_CONSUMER_COLLECTION", "test") + .withEnv("SPRING_DATA_MONGODB_URL", mongoConnectionString()) + .waitingFor(Wait.forLogMessage(".*Started MongodbSink.*", 1)); + + sink.start(); + buildMongoTemplate(); + } + + static void buildMongoTemplate() { + mongoDBContainer.start(); + MongoDatabaseFactory mongoDatabaseFactory = new SimpleMongoClientDatabaseFactory( + mongoConnectionString()); + mongoTemplate = new MongoTemplate(mongoDatabaseFactory); + } + + @Test + void postData() { + String json = "{\"name\":\"My Name\",\"address\":{ \"city\": \"Big City\", \"street\":\"Narrow Alley\"}}"; + testTopicSender.send(sink.getInputDestination(), json); + + await().atMost(DEFAULT_DURATION).untilAsserted(() -> { + List docs = mongoTemplate.findAll(Document.class, "test"); + assertThat(docs).allMatch(document -> document.get("name", String.class).equals("My Name")); + }); + } + + @AfterAll + static void cleanUp() { + mongoDBContainer.close(); + sink.stop(); + } + +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/mongodb/RabbitMQMongoDBSinkTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/mongodb/RabbitMQMongoDBSinkTests.java new file mode 100644 index 00000000..6b6cca94 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/mongodb/RabbitMQMongoDBSinkTests.java @@ -0,0 +1,26 @@ +/* + * 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.integration.test.sink.mongodb; + +import org.springframework.cloud.stream.app.integration.test.common.Configuration; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.RabbitMQBaseContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.RabbitMQStreamAppTest; + +@RabbitMQStreamAppTest +@RabbitMQBaseContainer(name = "mongodb-sink-rabbit", versionSupplier = Configuration.VersionSupplier.class) +public class RabbitMQMongoDBSinkTests extends MongoDBSinkTests { +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/tcp/KafkaTcpSinkTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/tcp/KafkaTcpSinkTests.java new file mode 100644 index 00000000..cd500d8a --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/tcp/KafkaTcpSinkTests.java @@ -0,0 +1,26 @@ +/* + * 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.integration.test.sink.tcp; + +import org.springframework.cloud.stream.app.integration.test.common.Configuration; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaBaseContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaStreamAppTest; + +@KafkaStreamAppTest +@KafkaBaseContainer(name = "tcp-sink-kafka", versionSupplier = Configuration.VersionSupplier.class) +public class KafkaTcpSinkTests extends TcpSinkTests { +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/tcp/RabbitMQTcpSinkTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/tcp/RabbitMQTcpSinkTests.java new file mode 100644 index 00000000..f977e3ab --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/tcp/RabbitMQTcpSinkTests.java @@ -0,0 +1,26 @@ +/* + * 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.integration.test.sink.tcp; + +import org.springframework.cloud.stream.app.integration.test.common.Configuration; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.RabbitMQBaseContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.RabbitMQStreamAppTest; + +@RabbitMQStreamAppTest +@RabbitMQBaseContainer(name = "tcp-sink-rabbit", versionSupplier = Configuration.VersionSupplier.class) +public class RabbitMQTcpSinkTests extends TcpSinkTests { +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/tcp/TcpSinkTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/tcp/TcpSinkTests.java new file mode 100644 index 00000000..0af448ec --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/sink/tcp/TcpSinkTests.java @@ -0,0 +1,99 @@ +/* + * 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.integration.test.sink.tcp; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.time.Duration; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.testcontainers.containers.wait.strategy.Wait; + +import org.springframework.beans.factory.annotation.Autowired; +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.TestTopicSender; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.BaseContainerExtension; + +import static org.awaitility.Awaitility.await; +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.DEFAULT_DURATION; +@Tag("integration") +@ExtendWith(BaseContainerExtension.class) +abstract class TcpSinkTests { + + private static int tcpPort; + + private static Socket socket; + + private static final AtomicBoolean socketReady = new AtomicBoolean(); + + private static StreamAppContainer sink; + + @Autowired + private TestTopicSender testTopicSender; + + @BeforeAll + static void configureSink() { + tcpPort = StreamAppContainerTestUtils.findAvailablePort(); + startTcpServer(); + sink = BaseContainerExtension.containerInstance() + .withEnv("TCP_CONSUMER_HOST", StreamAppContainerTestUtils.localHostAddress()) + .withEnv("TCP_PORT", String.valueOf(tcpPort)) + .withEnv("TCP_CONSUMER_ENCODER", "CRLF") + .waitingFor(Wait.forLogMessage(".*Started TcpSink.*", 1)); + sink.start(); + } + + static void startTcpServer() { + socketReady.set(false); + new Thread(() -> { + try { + socket = new ServerSocket(tcpPort, 50, InetAddress.getLocalHost()).accept(); + socketReady.set(true); + } + catch (IOException e) { + throw new RuntimeException("failed to bind to port " + tcpPort + ": " + e.getMessage(), e); + } + }).start(); + } + + @Test + void postData() throws IOException { + // Sink will not connect until it receives a message. + String text = "Hello, world!"; + testTopicSender.send(sink.getInputDestination(), text); + + await().atMost(DEFAULT_DURATION).untilTrue(socketReady); + BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream())); + await().atMost(Duration.ofSeconds(10)).until(() -> reader.readLine().equals(text)); + } + + @AfterAll + static void cleanUp() throws IOException { + sink.stop(); + socket.close(); + } +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/geode/GeodeSourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/geode/GeodeSourceTests.java new file mode 100644 index 00000000..8bdf60be --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/geode/GeodeSourceTests.java @@ -0,0 +1,119 @@ +/* + * 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.integration.test.source.geode; + +import java.time.Duration; +import java.util.UUID; +import java.util.function.Consumer; + +import com.github.dockerjava.api.command.CreateContainerCmd; +import com.github.dockerjava.api.model.ExposedPort; +import com.github.dockerjava.api.model.HostConfig; +import com.github.dockerjava.api.model.PortBinding; +import com.github.dockerjava.api.model.Ports; +import org.apache.geode.cache.Region; +import org.apache.geode.cache.client.ClientCache; +import org.apache.geode.cache.client.ClientCacheFactory; +import org.apache.geode.cache.client.ClientRegionShortcut; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.images.builder.ImageFromDockerfile; +import org.testcontainers.junit.jupiter.Container; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cloud.fn.test.support.geode.GeodeContainer; +import org.springframework.cloud.stream.app.test.integration.OutputMatcher; +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.junit.jupiter.BaseContainerExtension; + +import static org.awaitility.Awaitility.await; +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.DEFAULT_DURATION; +@Tag("integration") +@ExtendWith(BaseContainerExtension.class) +abstract class GeodeSourceTests { + private static int locatorPort = StreamAppContainerTestUtils.findAvailablePort(); + + private static int cacheServerPort = StreamAppContainerTestUtils.findAvailablePort(); + + private static Region clientRegion; + + private static ClientCache clientCache; + + protected static StreamAppContainer source; + + @Autowired + private OutputMatcher outputMatcher; + + @Container + private static final GeodeContainer geode = (GeodeContainer) new GeodeContainer(new ImageFromDockerfile() + .withFileFromClasspath("Dockerfile", "geode-test/Dockerfile") + .withBuildArg("CACHE_SERVER_PORT", String.valueOf(cacheServerPort)) + .withBuildArg("LOCATOR_PORT", String.valueOf(locatorPort)), + locatorPort, cacheServerPort) + .withCreateContainerCmdModifier( + (Consumer) createContainerCmd -> createContainerCmd + .withHostName("geode").withHostConfig(new HostConfig().withPortBindings( + new PortBinding(Ports.Binding.bindPort(cacheServerPort), + new ExposedPort(cacheServerPort)), + new PortBinding(Ports.Binding.bindPort(locatorPort), + new ExposedPort(locatorPort))))) + .withCommand("tail", "-f", "/dev/null") + .withStartupTimeout(DEFAULT_DURATION.multipliedBy(2)); + + @BeforeAll + static void initializeGeodeCacheThenStartSource() { + // Not using locator is faster. + System.out.println(geode.execGfsh( + "start server --name=Server1 " + "--hostname-for-clients=geode" + " --server-port=" + + cacheServerPort + " --J=-Dgemfire.jmx-manager=true --J=-Dgemfire.jmx-manager-start=true") + .getStdout()); + System.out.println(geode.execGfsh("connect --jmx-manager=localhost[1099]", + "create region --name=myRegion --type=REPLICATE").getStdout()); + clientCache = new ClientCacheFactory().addPoolServer("localhost", cacheServerPort) + .create(); + clientCache.readyForEvents(); + clientRegion = clientCache + .createClientRegionFactory(ClientRegionShortcut.PROXY) + .create("myRegion"); + + source = BaseContainerExtension.containerInstance().withEnv("GEODE_POOL_CONNECT_TYPE", "server") + .withEnv("GEODE_REGION_REGION_NAME", "myRegion") + .withEnv("GEODE_POOL_HOST_ADDRESSES", + StreamAppContainerTestUtils.localHostAddress() + ":" + cacheServerPort) + .waitingFor(Wait.forLogMessage(".*Started GeodeSource.*", 1)); + source.start(); + } + + @Test + void test() throws InterruptedException { + String random = UUID.randomUUID().toString(); + clientRegion.put(random, random); + await().atMost(Duration.ofSeconds(30)) + .until(outputMatcher.payloadMatches((String s) -> s.contains(random))); + } + + @AfterAll + static void cleanup() { + source.stop(); + clientCache.close(); + } +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/geode/KafkaGeodeSourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/geode/KafkaGeodeSourceTests.java new file mode 100644 index 00000000..75a2c3e5 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/geode/KafkaGeodeSourceTests.java @@ -0,0 +1,26 @@ +/* + * 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.integration.test.source.geode; + +import org.springframework.cloud.stream.app.integration.test.common.Configuration; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaBaseContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaStreamAppTest; + +@KafkaStreamAppTest +@KafkaBaseContainer(name = "geode-source-kafka", versionSupplier = Configuration.VersionSupplier.class) +public class KafkaGeodeSourceTests extends GeodeSourceTests { +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/geode/RabbitMQGeodeSourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/geode/RabbitMQGeodeSourceTests.java new file mode 100644 index 00000000..ea465837 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/geode/RabbitMQGeodeSourceTests.java @@ -0,0 +1,26 @@ +/* + * 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.integration.test.source.geode; + +import org.springframework.cloud.stream.app.integration.test.common.Configuration; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.RabbitMQBaseContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.RabbitMQStreamAppTest; + +@RabbitMQStreamAppTest +@RabbitMQBaseContainer(name = "geode-source-rabbit", versionSupplier = Configuration.VersionSupplier.class) +public class RabbitMQGeodeSourceTests extends GeodeSourceTests { +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/http/HttpSourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/http/HttpSourceTests.java new file mode 100644 index 00000000..199a8c1d --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/http/HttpSourceTests.java @@ -0,0 +1,116 @@ +/* + * 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.integration.test.source.http; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.testcontainers.containers.wait.strategy.Wait; +import reactor.core.publisher.Mono; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cloud.stream.app.test.integration.AppLog; +import org.springframework.cloud.stream.app.test.integration.OutputMatcher; +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.junit.jupiter.BaseContainerExtension; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.web.reactive.function.client.WebClient; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.DEFAULT_DURATION; +@Tag("integration") +@ExtendWith(BaseContainerExtension.class) +public abstract class HttpSourceTests { + + private static int serverPort = StreamAppContainerTestUtils.findAvailablePort(); + + private static WebClient webClient = WebClient.builder().build(); + + private static StreamAppContainer source; + + @BeforeAll + static void configureSource() { + source = BaseContainerExtension.containerInstance() + .withEnv("SERVER_PORT", String.valueOf(serverPort)) + .withExposedPorts(serverPort) + .waitingFor(Wait.forListeningPort().withStartupTimeout(DEFAULT_DURATION)); + source.withLogConsumer(AppLog.appLog("http")).start(); + } + + @Autowired + private OutputMatcher outputMatcher; + + @AfterEach + void reset() { + outputMatcher.clearMessageMatchers(); + } + + @Test + void plaintext() throws InterruptedException { + CountDownLatch countDownLatch = new CountDownLatch(1); + AtomicReference httpStatus = new AtomicReference<>(); + webClient + .post() + .uri("http://localhost:" + source.getMappedPort(serverPort)) + .contentType(MediaType.valueOf("application/x-www-form-url-encoded")) + .body(Mono.just("Hello".getBytes()), byte[].class) + .exchange() + .subscribe(r -> { + httpStatus.set(r.statusCode()); + countDownLatch.countDown(); + }); + countDownLatch.await(30, TimeUnit.SECONDS); + assertThat(httpStatus.get().is2xxSuccessful()).isTrue(); + await().atMost(DEFAULT_DURATION) + .until(outputMatcher.payloadMatches(s -> s.equals("Hello"))); + } + + @Test + void json() throws InterruptedException { + + CountDownLatch countDownLatch = new CountDownLatch(1); + webClient + .post() + .uri("http://localhost:" + source.getMappedPort(serverPort)) + .contentType(MediaType.APPLICATION_JSON) + .body(Mono.just("{\"Hello\":\"world\"}"), String.class) + .exchange() + .subscribe(r -> { + countDownLatch.countDown(); + assertThat(r.statusCode().is2xxSuccessful()).isTrue(); + }); + countDownLatch.await(30, TimeUnit.SECONDS); + await().atMost(DEFAULT_DURATION) + .until(outputMatcher.payloadMatches(s -> s.equals("{\"Hello\":\"world\"}"))); + } + + @AfterAll + static void cleanUp() { + source.stop(); + } + +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/http/KafkaHttpSourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/http/KafkaHttpSourceTests.java new file mode 100644 index 00000000..d7ca9137 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/http/KafkaHttpSourceTests.java @@ -0,0 +1,26 @@ +/* + * 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.integration.test.source.http; + +import org.springframework.cloud.stream.app.integration.test.common.Configuration; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaBaseContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaStreamAppTest; + +@KafkaStreamAppTest +@KafkaBaseContainer(name = "http-source-kafka", versionSupplier = Configuration.VersionSupplier.class) +public class KafkaHttpSourceTests extends HttpSourceTests { +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/http/RabbitMQHttpSourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/http/RabbitMQHttpSourceTests.java new file mode 100644 index 00000000..1e4b4735 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/http/RabbitMQHttpSourceTests.java @@ -0,0 +1,26 @@ +/* + * 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.integration.test.source.http; + +import org.springframework.cloud.stream.app.integration.test.common.Configuration; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.RabbitMQBaseContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.RabbitMQStreamAppTest; + +@RabbitMQStreamAppTest +@RabbitMQBaseContainer(name = "http-source-rabbit", versionSupplier = Configuration.VersionSupplier.class) +public class RabbitMQHttpSourceTests extends HttpSourceTests { +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/jdbc/JdbcSourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/jdbc/JdbcSourceTests.java new file mode 100644 index 00000000..365c7e0d --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/jdbc/JdbcSourceTests.java @@ -0,0 +1,85 @@ +/* + * 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.integration.test.source.jdbc; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.testcontainers.containers.BindMode; +import org.testcontainers.containers.MySQLContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.utility.DockerImageName; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cloud.stream.app.test.integration.OutputMatcher; +import org.springframework.cloud.stream.app.test.integration.StreamAppContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.BaseContainerExtension; +import org.springframework.cloud.stream.app.test.integration.kafka.KafkaConfig; + +import static org.awaitility.Awaitility.await; +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.DEFAULT_DURATION; +import static org.springframework.cloud.stream.app.test.integration.AppLog.appLog; +@Tag("integration") +@ExtendWith(BaseContainerExtension.class) +abstract class JdbcSourceTests { + + @Autowired + private OutputMatcher outputMatcher; + + private static StreamAppContainer source; + + @Container + private static MySQLContainer mySQL = new MySQLContainer<>(DockerImageName.parse("mysql:5.7")) + .withUsername("test") + .withPassword("secret") + .withExposedPorts(3306) + .withNetwork(KafkaConfig.kafka.getNetwork()) + .withNetworkAliases("mysql-for-source") + .withLogConsumer(appLog("mysql-for-source")) + .withClasspathResourceMapping("init.sql", "/init.sql", BindMode.READ_ONLY) + .withCommand("--init-file", "/init.sql"); + + @BeforeAll + protected static void configureSource() { + source = BaseContainerExtension.containerInstance() + .withEnv("JDBC_SUPPLIER_QUERY", "SELECT * FROM People WHERE deleted='N'") + .withEnv("JDBC_SUPPLIER_UPDATE", "UPDATE People SET deleted='Y' WHERE id=:id") + .withEnv("SPRING_DATASOURCE_USERNAME", "test") + .withEnv("SPRING_DATASOURCE_PASSWORD", "secret") + .withEnv("SPRING_DATASOURCE_DRIVER_CLASS_NAME", "org.mariadb.jdbc.Driver") + .withEnv("SPRING_DATASOURCE_URL", "jdbc:mariadb://mysql-for-source:3306/test"); + startSource(); + } + + private static void startSource() { + await().atMost(DEFAULT_DURATION).until(() -> mySQL.isRunning()); + source.start(); + } + + @Test + void test() { + await().atMost(DEFAULT_DURATION) + .until(outputMatcher.payloadMatches((String s) -> s.contains("Bart Simpson"))); + } + + @AfterAll + private static void cleanUp() { + source.stop(); + } +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/jdbc/KafkaJdbcSourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/jdbc/KafkaJdbcSourceTests.java new file mode 100644 index 00000000..2fbe1c8c --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/jdbc/KafkaJdbcSourceTests.java @@ -0,0 +1,26 @@ +/* + * 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.integration.test.source.jdbc; + +import org.springframework.cloud.stream.app.integration.test.common.Configuration; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaBaseContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaStreamAppTest; + +@KafkaStreamAppTest +@KafkaBaseContainer(name = "jdbc-source-kafka", versionSupplier = Configuration.VersionSupplier.class) +public class KafkaJdbcSourceTests extends JdbcSourceTests { +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/jdbc/RabbitMQJdbcSourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/jdbc/RabbitMQJdbcSourceTests.java new file mode 100644 index 00000000..30eede73 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/jdbc/RabbitMQJdbcSourceTests.java @@ -0,0 +1,26 @@ +/* + * 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.integration.test.source.jdbc; + +import org.springframework.cloud.stream.app.integration.test.common.Configuration; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.RabbitMQBaseContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.RabbitMQStreamAppTest; + +@RabbitMQStreamAppTest +@RabbitMQBaseContainer(name = "jdbc-source-rabbit", versionSupplier = Configuration.VersionSupplier.class) +public class RabbitMQJdbcSourceTests extends JdbcSourceTests { +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/s3/KafkaS3SourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/s3/KafkaS3SourceTests.java new file mode 100644 index 00000000..5558c138 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/s3/KafkaS3SourceTests.java @@ -0,0 +1,26 @@ +/* + * 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.integration.test.source.s3; + +import org.springframework.cloud.stream.app.integration.test.common.Configuration; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaBaseContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaStreamAppTest; + +@KafkaStreamAppTest +@KafkaBaseContainer(name = "s3-source-kafka", versionSupplier = Configuration.VersionSupplier.class) +public class KafkaS3SourceTests extends S3SourceTests { +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/s3/RabbitMQS3SourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/s3/RabbitMQS3SourceTests.java new file mode 100644 index 00000000..cba6bda1 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/s3/RabbitMQS3SourceTests.java @@ -0,0 +1,26 @@ +/* + * 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.integration.test.source.s3; + +import org.springframework.cloud.stream.app.integration.test.common.Configuration; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.RabbitMQBaseContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.RabbitMQStreamAppTest; + +@RabbitMQStreamAppTest +@RabbitMQBaseContainer(name = "s3-source-rabbit", versionSupplier = Configuration.VersionSupplier.class) +public class RabbitMQS3SourceTests extends S3SourceTests { +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/s3/S3SourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/s3/S3SourceTests.java new file mode 100644 index 00000000..5b7cb1bc --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/s3/S3SourceTests.java @@ -0,0 +1,163 @@ +/* + * 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.integration.test.source.s3; + +import java.util.Map; +import java.util.function.Consumer; + +import com.amazonaws.ClientConfiguration; +import com.amazonaws.auth.AWSCredentials; +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.s3.AmazonS3; +import com.amazonaws.services.s3.AmazonS3ClientBuilder; +import com.amazonaws.services.s3.model.PutObjectRequest; +import com.github.dockerjava.api.command.CreateContainerCmd; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.utility.DockerImageName; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cloud.stream.app.test.integration.OutputMatcher; +import org.springframework.cloud.stream.app.test.integration.StreamAppContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.BaseContainerExtension; + +import static org.awaitility.Awaitility.await; +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.DEFAULT_DURATION; +import static org.springframework.cloud.stream.app.test.integration.AppLog.appLog; +import static org.springframework.cloud.stream.app.test.integration.FluentMap.fluentMap; +import static org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils.localHostAddress; +import static org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils.resourceAsFile; +@Tag("integration") +@ExtendWith(BaseContainerExtension.class) +abstract class S3SourceTests { + + private static AmazonS3 s3Client; + + private StreamAppContainer source; + + @Autowired + private OutputMatcher outputMatcher; + + @Container + private static final GenericContainer minio = new GenericContainer( + DockerImageName.parse("minio/minio:RELEASE.2020-10-18T21-54-12Z")) + .withExposedPorts(9000) + .withEnv("MINIO_ACCESS_KEY", "minio") + .withEnv("MINIO_SECRET_KEY", "minio123") + .waitingFor(Wait.forHttp("/minio/health/live")) + .withCreateContainerCmdModifier( + (Consumer) createContainerCmd -> createContainerCmd + .withHostName("minio")) + .withLogConsumer(appLog("minio")) + .withCommand("minio", "server", "/data"); + + @BeforeAll + static void initS3() { + AWSCredentials credentials = new BasicAWSCredentials("minio", "minio123"); + ClientConfiguration clientConfiguration = new ClientConfiguration(); + s3Client = AmazonS3ClientBuilder + .standard() + .withEndpointConfiguration( + new AwsClientBuilder.EndpointConfiguration("http://localhost:" + minio.getMappedPort(9000), + Regions.US_EAST_1.name())) + .withPathStyleAccessEnabled(true) + .withClientConfiguration(clientConfiguration) + .withCredentials(new AWSStaticCredentialsProvider(credentials)) + .build(); + + } + + @BeforeEach + void configureSource() { + source = BaseContainerExtension.containerInstance() + .withEnv("S3_COMMON_ENDPOINT_URL", "http://" + localHostAddress() + ":" + minio.getMappedPort(9000)) + .withEnv("S3_COMMON_PATH_STYLE_ACCESS", "true") + .withEnv("CLOUD_AWS_STACK_AUTO", "false") + .withEnv("CLOUD_AWS_CREDENTIALS_ACCESS_KEY", "minio") + .withEnv("CLOUD_AWS_CREDENTIALS_SECRET_KEY", "minio123") + .withEnv("LOGGING_LEVEL_ORG_SPRINGFRAMEWORK_INTEGRATION", "DEBUG") + .withEnv("CLOUD_AWS_REGION_STATIC", "us-east-1").log(); + s3Client.createBucket("bucket"); + } + + @Test + void testLines() { + startContainer( + fluentMap().withEntry("FILE_CONSUMER_MODE", "lines")); + s3Client.putObject(new PutObjectRequest("bucket", "test", + resourceAsFile("minio/data"))); + + await().atMost(DEFAULT_DURATION).until(outputMatcher.payloadMatches((String s) -> s.contains("Bart Simpson"))); + + } + + @Test + void testTaskLaunchRequest() { + + startContainer(fluentMap() + .withEntry("SPRING_CLOUD_FUNCTION_DEFINITION", "s3Supplier|taskLaunchRequestFunction") + .withEntry("TASK_LAUNCH_REQUEST_ARG_EXPRESSIONS", "filename=payload") + .withEntry("TASK_LAUNCH_REQUEST_TASK_NAME", "myTask") + .withEntry("FILE_CONSUMER_MODE", "ref")); + + s3Client.putObject(new PutObjectRequest("bucket", "test", + resourceAsFile("minio/data"))); + await().atMost(DEFAULT_DURATION).until(outputMatcher.payloadMatches(s -> s.equals( + "{\"args\":[\"filename=/tmp/s3-supplier/test\"],\"deploymentProps\":{},\"name\":\"myTask\"}"))); + } + + @Test + void testListOnly() { + startContainer( + fluentMap() + .withEntry("FILE_CONSUMER_MODE", "ref") + .withEntry("S3_SUPPLIER_LIST_ONLY", "true")); + + s3Client.putObject(new PutObjectRequest("bucket", "test", + resourceAsFile("minio/data"))); + await().atMost(DEFAULT_DURATION) + .until(outputMatcher + .payloadMatches((String s) -> s.contains("\"bucketName\":\"bucket\",\"key\":\"test\""))); + } + + private void startContainer(Map environment) { + source.withEnv(environment); + source.waitingFor(Wait.forLogMessage(".*Started S3Source.*", 1)).start(); + environment.keySet().forEach(k -> source.getEnvMap().remove(k)); + } + + @AfterEach + void stop() { + if (s3Client.doesBucketExistV2("bucket")) { + s3Client.deleteObject("bucket", "test"); + s3Client.deleteBucket("bucket"); + } + source.stop(); + outputMatcher.clearMessageMatchers(); + } + +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/sftp/KafkaSftpSourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/sftp/KafkaSftpSourceTests.java new file mode 100644 index 00000000..a011a969 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/sftp/KafkaSftpSourceTests.java @@ -0,0 +1,26 @@ +/* + * 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.integration.test.source.sftp; + +import org.springframework.cloud.stream.app.integration.test.common.Configuration; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaBaseContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaStreamAppTest; + +@KafkaStreamAppTest +@KafkaBaseContainer(name = "sftp-source-kafka", versionSupplier = Configuration.VersionSupplier.class) +public class KafkaSftpSourceTests extends SftpSourceTests { +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/sftp/RabbitMQSftpSourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/sftp/RabbitMQSftpSourceTests.java new file mode 100644 index 00000000..3d71aaf1 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/sftp/RabbitMQSftpSourceTests.java @@ -0,0 +1,26 @@ +/* + * 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.integration.test.source.sftp; + +import org.springframework.cloud.stream.app.integration.test.common.Configuration; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.RabbitMQBaseContainer; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.RabbitMQStreamAppTest; + +@RabbitMQStreamAppTest +@RabbitMQBaseContainer(name = "sftp-source-rabbit", versionSupplier = Configuration.VersionSupplier.class) +public class RabbitMQSftpSourceTests extends SftpSourceTests { +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/sftp/SftpSourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/sftp/SftpSourceTests.java new file mode 100644 index 00000000..7e7790b3 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/sftp/SftpSourceTests.java @@ -0,0 +1,88 @@ +/* + * 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.integration.test.source.sftp; + +import java.util.Collections; +import java.util.Map; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.testcontainers.containers.BindMode; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.utility.DockerImageName; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cloud.stream.app.test.integration.OutputMatcher; +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.junit.jupiter.BaseContainerExtension; + +import static org.awaitility.Awaitility.await; +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.DEFAULT_DURATION; +@Tag("integration") +@ExtendWith(BaseContainerExtension.class) +abstract class SftpSourceTests { + + private StreamAppContainer source; + + @Container + private static final GenericContainer sftp = new GenericContainer(DockerImageName.parse("atmoz/sftp")) + .withExposedPorts(22) + .withCommand("user:pass:::remote") + .withClasspathResourceMapping("sftp", "/home/user/remote", BindMode.READ_ONLY) + .withStartupTimeout(DEFAULT_DURATION); + + @BeforeEach + void configureSource() { + await().atMost(DEFAULT_DURATION).until(() -> sftp.isRunning()); + source = BaseContainerExtension.containerInstance() + .withEnv("SFTP_SUPPLIER_FACTORY_ALLOW_UNKNOWN_KEYS", "true") + .withEnv("SFTP_SUPPLIER_REMOTE_DIR", "/remote") + .withEnv("SFTP_SUPPLIER_FACTORY_USERNAME", "user") + .withEnv("SFTP_SUPPLIER_FACTORY_PASSWORD", "pass") + .withEnv("SFTP_SUPPLIER_FACTORY_PORT", String.valueOf(sftp.getMappedPort(22))) + .withEnv("SFTP_SUPPLIER_FACTORY_HOST", StreamAppContainerTestUtils.localHostAddress()); + } + + @Autowired + private OutputMatcher outputMatcher; + + // TODO: This fixture supports additional tests with different modes, etc. + @Test + void test() { + startContainer(Collections.singletonMap("FILE_CONSUMER_MODE", "ref")); + + await().atMost(DEFAULT_DURATION) + .until(outputMatcher.payloadMatches((String s) -> s.equals("\"/tmp/sftp-supplier/data.txt\""))); + } + + private void startContainer(Map environment) { + source.withEnv(environment); + source.start(); + environment.keySet().forEach(k -> source.getEnvMap().remove(k)); + } + + @AfterEach + private void cleanUp() { + source.stop(); + } + +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/time/KafkaTimeSourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/time/KafkaTimeSourceTests.java new file mode 100644 index 00000000..8a37e58f --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/time/KafkaTimeSourceTests.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.integration.test.source.time; + +import org.junit.jupiter.api.Test; +import org.testcontainers.junit.jupiter.Container; + +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.junit.jupiter.KafkaStreamAppTest; +import org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamAppContainer; +import org.springframework.http.HttpStatus; +import org.springframework.web.reactive.function.client.ClientResponse; +import org.springframework.web.reactive.function.client.WebClient; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.VERSION; + +@KafkaStreamAppTest +class KafkaTimeSourceTests extends TimeSourceTests { + + @Container + static StreamAppContainer source = new KafkaStreamAppContainer(StreamAppContainerTestUtils + .imageName(StreamAppContainerTestUtils.SPRINGCLOUDSTREAM_REPOSITOTRY, "time-source-kafka", VERSION)) + .withCommand("--server.port", "8080") + .withEnv("SPRING_CLOUD_STREAMAPP_SECURITY_ENABLED", "false") + .withExposedPorts(8080); + + @Test + void testActuator() { + WebClient webClient = WebClient.create(); + ClientResponse response = webClient.get() + .uri("http://localhost:" + source.getMappedPort(8080) + "/actuator/health").exchange().block(); + assertThat(response.statusCode()).isEqualTo(HttpStatus.OK); + response = webClient.get().uri("http://localhost:" + source.getMappedPort(8080) + "/actuator/info").exchange() + .block(); + assertThat(response.statusCode()).isEqualTo(HttpStatus.OK); + + } + +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/time/RabbitMQTimeSourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/time/RabbitMQTimeSourceTests.java new file mode 100644 index 00000000..e56a1d72 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/time/RabbitMQTimeSourceTests.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.integration.test.source.time; + +import org.testcontainers.junit.jupiter.Container; + +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.junit.jupiter.RabbitMQStreamAppTest; +import org.springframework.cloud.stream.app.test.integration.rabbitmq.RabbitMQStreamAppContainer; + +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.VERSION; + +@RabbitMQStreamAppTest + +class RabbitMQTimeSourceTests extends TimeSourceTests { + + @Container + static StreamAppContainer source = new RabbitMQStreamAppContainer(StreamAppContainerTestUtils + .imageName(StreamAppContainerTestUtils.SPRINGCLOUDSTREAM_REPOSITOTRY, "time-source-rabbit", VERSION)); + +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/time/TimeSourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/time/TimeSourceTests.java new file mode 100644 index 00000000..c103173f --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/time/TimeSourceTests.java @@ -0,0 +1,49 @@ +/* + * 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.integration.test.source.time; + +import java.util.regex.Pattern; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cloud.stream.app.test.integration.OutputMatcher; + +import static org.awaitility.Awaitility.await; +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.DEFAULT_DURATION; +@Tag("integration") +abstract class TimeSourceTests { + + // "MM/dd/yy HH:mm:ss"; + private final static Pattern pattern = Pattern.compile(".*\\d{2}/\\d{2}/\\d{2}\\s+\\d{2}:\\d{2}:\\d{2}"); + + @Autowired + private OutputMatcher outputMatcher; + + @Test + void test() { + await().atMost(DEFAULT_DURATION) + .until(outputMatcher.payloadMatches((String s) -> pattern.matcher(s).matches())); + } + + @AfterEach + void cleanUp() { + outputMatcher.clearMessageMatchers(); + } +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/stream/jdbclog/KafkaJdbcLogStreamTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/stream/jdbclog/KafkaJdbcLogStreamTests.java new file mode 100644 index 00000000..b0723b5b --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/stream/jdbclog/KafkaJdbcLogStreamTests.java @@ -0,0 +1,80 @@ +/* + * 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.integration.test.stream.jdbclog; + +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.BindMode; +import org.testcontainers.containers.MySQLContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.utility.DockerImageName; + +import org.springframework.cloud.stream.app.test.integration.LogMatcher; +import org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils; +import org.springframework.cloud.stream.app.test.integration.StreamApps; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaStreamAppTest; +import org.springframework.cloud.stream.app.test.integration.kafka.KafkaConfig; +import org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamAppContainer; + +import static org.awaitility.Awaitility.await; +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.DEFAULT_DURATION; +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.VERSION; +import static org.springframework.cloud.stream.app.test.integration.AppLog.appLog; +import static org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamApps.kafkaStreamApps; + +@Tag("integration") +@KafkaStreamAppTest +public class KafkaJdbcLogStreamTests { + + private static LogMatcher logMatcher = LogMatcher.contains("Bart Simpson"); + + @Container + private static MySQLContainer mySQL = new MySQLContainer<>(DockerImageName.parse("mysql:5.7")) + .withUsername("test") + .withPassword("secret") + .withExposedPorts(3306) + .withNetwork(KafkaConfig.kafka.getNetwork()) + .withNetworkAliases("mysql-for-stream") + .withLogConsumer(appLog("mySQL")) + .withClasspathResourceMapping("init.sql", "/init.sql", BindMode.READ_ONLY) + .withCommand("--init-file", "/init.sql"); + + @Container + private static final StreamApps streamApp = kafkaStreamApps(KafkaJdbcLogStreamTests.class.getSimpleName(), + KafkaConfig.kafka) + .withSourceContainer( + new KafkaStreamAppContainer(StreamAppContainerTestUtils.imageName( + "jdbc-source-kafka", + VERSION)) + .withEnv("JDBC_SUPPLIER_QUERY", "SELECT * FROM People WHERE deleted='N'") + .withEnv("JDBC_SUPPLIER_UPDATE", + "UPDATE People SET deleted='Y' WHERE id=:id") + .withEnv("SPRING_DATASOURCE_PASSWORD", "secret") + .withEnv("SPRING_DATASOURCE_USERNAME", "test") + .withEnv("SPRING_DATASOURCE_DRIVER_CLASS_NAME", "org.mariadb.jdbc.Driver") + .withEnv("SPRING_DATASOURCE_URL", + "jdbc:mariadb://mysql-for-stream:3306/test")) + .withSinkContainer( + new KafkaStreamAppContainer(StreamAppContainerTestUtils.imageName( + "log-sink-kafka", VERSION)).withLogConsumer(logMatcher)) + .build(); + + @Test + void test() { + await().atMost(DEFAULT_DURATION).until(logMatcher.matches()); + } +} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/stream/tiktok/KafkaTikTokTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/stream/tiktok/KafkaTikTokTests.java new file mode 100644 index 00000000..31dcc0f2 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/stream/tiktok/KafkaTikTokTests.java @@ -0,0 +1,60 @@ +/* + * 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.integration.test.stream.tiktok; + +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.testcontainers.junit.jupiter.Container; + +import org.springframework.cloud.stream.app.test.integration.LogMatcher; +import org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils; +import org.springframework.cloud.stream.app.test.integration.StreamApps; +import org.springframework.cloud.stream.app.test.integration.junit.jupiter.KafkaStreamAppTest; +import org.springframework.cloud.stream.app.test.integration.kafka.KafkaConfig; +import org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamAppContainer; + +import static org.awaitility.Awaitility.await; +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.DEFAULT_DURATION; +import static org.springframework.cloud.stream.app.integration.test.common.Configuration.VERSION; +import static org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamApps.kafkaStreamApps; + +@Tag("integration") +@KafkaStreamAppTest +public class KafkaTikTokTests { + + 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 = kafkaStreamApps(KafkaTikTokTests.class.getSimpleName(), + KafkaConfig.kafka) + .withSourceContainer( + new KafkaStreamAppContainer(StreamAppContainerTestUtils.imageName( + "time-source-kafka", + VERSION))) + .withSinkContainer( + new KafkaStreamAppContainer(StreamAppContainerTestUtils.imageName( + "log-sink-kafka", + VERSION)).withLogConsumer(logMatcher) + .log()) + .build(); + + @Test + void test() { + await().atMost(DEFAULT_DURATION).until(logMatcher.matches()); + } +} diff --git a/applications/stream-applications-integration-tests/src/test/resources/geode-test/Dockerfile b/applications/stream-applications-integration-tests/src/test/resources/geode-test/Dockerfile new file mode 100644 index 00000000..340ca31a --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/resources/geode-test/Dockerfile @@ -0,0 +1,6 @@ +FROM apachegeode/geode:1.13.2 + +ARG CACHE_SERVER_PORT=40404 +ARG LOCATOR_PORT=10334 + +EXPOSE ${LOCATOR_PORT} ${CACHE_SERVER_PORT} \ No newline at end of file diff --git a/applications/stream-applications-integration-tests/src/test/resources/init.sql b/applications/stream-applications-integration-tests/src/test/resources/init.sql new file mode 100644 index 00000000..c21c3ad2 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/resources/init.sql @@ -0,0 +1,10 @@ +CREATE DATABASE IF NOT EXISTS test; +USE test; +CREATE TABLE IF NOT EXISTS People ( + id INT NOT NULL AUTO_INCREMENT, + name VARCHAR(255) NOT NULL, + street VARCHAR(255) NOT NULL, + city VARCHAR(255) NOT NULL, + deleted CHAR(1) DEFAULT 'N', + PRIMARY KEY (id)); +INSERT INTO People (name, street, city, deleted) VALUES ("Bart Simpson", "Main Street", "Springfield","N"); \ No newline at end of file diff --git a/applications/stream-applications-integration-tests/src/test/resources/logback-test.xml b/applications/stream-applications-integration-tests/src/test/resources/logback-test.xml new file mode 100644 index 00000000..5f482feb --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/resources/logback-test.xml @@ -0,0 +1,17 @@ + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n + + + + + + + + + + + + + \ No newline at end of file diff --git a/applications/stream-applications-integration-tests/src/test/resources/minio/data b/applications/stream-applications-integration-tests/src/test/resources/minio/data new file mode 100644 index 00000000..01c51bc2 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/resources/minio/data @@ -0,0 +1 @@ +Bart Simpson diff --git a/applications/stream-applications-integration-tests/src/test/resources/sftp/data.txt b/applications/stream-applications-integration-tests/src/test/resources/sftp/data.txt new file mode 100644 index 00000000..01c51bc2 --- /dev/null +++ b/applications/stream-applications-integration-tests/src/test/resources/sftp/data.txt @@ -0,0 +1 @@ +Bart Simpson diff --git a/setup-build.sh b/setup-build.sh index 55fb1f59..284311d6 100755 --- a/setup-build.sh +++ b/setup-build.sh @@ -1,6 +1,6 @@ #!/usr/bin/env bash -./mvnw clean install -f stream-applications-build -./mvnw clean install -f functions -N -./mvnw clean install -f functions/function-dependencies -N -./mvnw clean install -f applications/stream-applications-core -N +./mvnw clean install -f stream-applications-build -U +./mvnw clean install -f functions -N -U +./mvnw clean install -f functions/function-dependencies -N -U +./mvnw clean install -f applications/stream-applications-core -N -U diff --git a/stream-applications-build/pom.xml b/stream-applications-build/pom.xml index 8f7b2881..18e87b53 100644 --- a/stream-applications-build/pom.xml +++ b/stream-applications-build/pom.xml @@ -338,6 +338,10 @@ org.apache.maven.plugins maven-surefire-plugin + + !integration + integration +