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

WIP

Browse Source
This commit is contained in:
David Turanski
2020-10-19 19:02:39 -04:00
parent e768a4210a
commit 718fc0efc8
7 changed files with 219 additions and 397 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
stream-applications-integration-tests/pom.xml
View File

@@ -5,7 +5,7 @@
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.3.3.RELEASE</version>
<version>2.3.4.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>org.springframework.cloud.stream.apps</groupId>
@@ -146,14 +146,27 @@
<artifactId>spring-boot-starter-jdbc</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<!-- <plugin>-->
<!-- <groupId>org.apache.maven.plugins</groupId>-->
<!-- <artifactId>maven-surefire-plugin</artifactId>-->
<!-- <version>3.0.0-M4</version>-->
<!-- <configuration>-->
<!-- <includes>-->
<!-- <include>**/*Tests.java</include>-->
<!-- <include>**/*Test.java</include>-->
<!-- </includes>-->
<!-- </configuration>-->
<!-- <dependencies>-->
<!-- <dependency>-->
<!-- <groupId>org.junit.jupiter</groupId>-->
<!-- <artifactId>junit-jupiter-engine</artifactId>-->
<!-- <version>5.5.2</version>-->
<!-- </dependency>-->
<!-- </dependencies>-->
<!-- </plugin>-->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>

71
stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/TikTokBaselineTests.java
View File

@@ -1,71 +0,0 @@
/*
* 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.apps.integration.test;
import java.util.regex.Pattern;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;
import org.springframework.cloud.stream.app.test.integration.LogMatcher;
import org.springframework.cloud.stream.apps.integration.test.support.KafkaStreamIntegrationTestSupport;
import static org.springframework.cloud.stream.app.test.integration.FluentMap.fluentMap;
@Testcontainers
public class TikTokBaselineTests extends KafkaStreamIntegrationTestSupport {
private static Logger logger = LoggerFactory.getLogger(TikTokBaselineTests.class);
// "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}");
private final static LogMatcher logMatcher = new LogMatcher();
@Container
static GenericContainer timeSource = new GenericContainer(
DockerImageName.parse("springcloudstream/time-source-kafka:3.0.0-SNAPSHOT"))
.withNetwork(kafka.getNetwork())
.withEnv(fluentMap().withEntry("SPRING_CLOUD_STREAM_BINDINGS_OUTPUT_DESTINATION", "test-topic")
.withEntry("SPRING_CLOUD_STREAM_KAFKA_BINDER_BROKERS",
kafka.getNetworkAliases().get(0) + ":9092"))
.dependsOn(kafka);
// @Container
// static GenericContainer logSink = new GenericContainer(
// DockerImageName.parse("springcloudstream/log-sink-kafka:3.0.0-SNAPSHOT"))
// .withNetwork(kafka.getNetwork())
// .withEnv(fluentMap().withEntry("SPRING_CLOUD_STREAM_BINDINGS_INPUT_DESTINATION",
// "test-topic")
// .withEntry("SPRING_CLOUD_STREAM_KAFKA_BINDER_BROKERS",
// kafka.getNetworkAliases().get(0) + ":9092")
// .withEntry("SPRING_CLOUD_STREAM_BINDINGS_INPUT_GROUP", "TikTok"))
// .withLogConsumer(logMatcher)
// // .withLogConsumer(appLog("log-sink"))
// .dependsOn(kafka);
@Test
void tiktok() {
// await().atMost(Duration.ofMinutes(2)).until(verifyOutputMessage((Message<String> m) ->
// pattern.matcher(m.getPayload()).matches()));
}
}

51
stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/TikTokTests.java
View File

@@ -1,51 +0,0 @@
/*
* 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.apps.integration.test;
import java.time.Duration;
import java.util.regex.Pattern;
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.StreamApps;
import org.springframework.cloud.stream.apps.integration.test.support.KafkaStreamIntegrationTestSupport;
import static org.awaitility.Awaitility.await;
import static org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamApps.kafkaStreamApps;
public class TikTokTests extends KafkaStreamIntegrationTestSupport {
// "MM/dd/yy HH:mm:ss";
private final static Pattern DATE_PATTERN = Pattern.compile(".*\\d{2}/\\d{2}/\\d{2}\\s+\\d{2}:\\d{2}:\\d{2}");
private final static LogMatcher logMatcher = new LogMatcher();
//
// @Container
// static StreamApps streamApps = kafkaStreamApps("tikTok", kafka)
// .withSourceContainer(defaultKafkaContainerFor("time-source"))
// .withSinkContainer(defaultKafkaContainerFor("log-sink")
// .withLogConsumer(logMatcher))
// .build();
//
// @Test
// void tiktok() {
// await().atMost(Duration.ofMinutes(2)).until(logMatcher.verifies(log -> log.contains("Started LogSink")));
// await().atMost(Duration.ofSeconds(30))
// .until(logMatcher.verifies(log -> log.matchesRegex(DATE_PATTERN.pattern())));
// }
}

100
stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/processor/HttpRequestProcessorTests.java
View File

@@ -16,31 +16,12 @@
package org.springframework.cloud.stream.apps.integration.test.processor;
import java.net.InetAddress;
import java.time.Duration;
import okhttp3.mockwebserver.Dispatcher;
import okhttp3.mockwebserver.MockResponse;
import okhttp3.mockwebserver.MockWebServer;
import okhttp3.mockwebserver.RecordedRequest;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.testcontainers.junit.jupiter.Container;
import reactor.core.publisher.Mono;
import org.springframework.cloud.stream.app.test.integration.LogMatcher;
import org.springframework.cloud.stream.app.test.integration.StreamApps;
import org.springframework.cloud.stream.apps.integration.test.support.KafkaStreamIntegrationTestSupport;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
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.awaitility.Awaitility.await;
import static org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamApps.kafkaStreamApps;
public class HttpRequestProcessorTests extends KafkaStreamIntegrationTestSupport {
private static MockWebServer server = new MockWebServer();
@@ -52,45 +33,46 @@ public class HttpRequestProcessorTests extends KafkaStreamIntegrationTestSupport
private static int sourcePort = findAvailablePort();
// @Container
// private static final StreamApps streamApps = kafkaStreamApps(
// HttpRequestProcessorTests.class.getSimpleName(), kafka)
// .withSourceContainer(httpSource(sourcePort))
// .withProcessorContainer(defaultKafkaProcessorContainerFor("http-request-processor")
// .withEnv("HTTP_REQUEST_URL_EXPRESSION",
// "'http://" + localHostAddress() + ":" + serverPort + "'")
// .withEnv("HTTP_REQUEST_HTTP_METHOD_EXPRESSION", "'POST'"))
// .withSinkContainer(
// defaultKafkaContainerFor("log-sink").withLogConsumer(logMatcher))
// .build();
//
// @BeforeAll
// static void startServer() throws Exception {
// server.start(InetAddress.getLocalHost(), serverPort);
// }
//
// @Test
// void get() {
// server.setDispatcher(new Dispatcher() {
// @Override
// public MockResponse dispatch(RecordedRequest recordedRequest) {
// return new MockResponse().setHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
// .setBody("{\"response\":\"" + recordedRequest.getBody().readUtf8() + "\"}")
// .setResponseCode(HttpStatus.OK.value());
// }
// });
//
// await().atMost(Duration.ofSeconds(30))
// .until(logMatcher.verifies(log -> log.when(() -> {
// ClientResponse response = webClient
// .post()
// .uri("http://localhost:" + streamApps.sourceContainer().getMappedPort(sourcePort))
// .contentType(MediaType.TEXT_PLAIN)
// .body(Mono.just("ping"), String.class)
// .exchange()
// .block();
// assertThat(response.statusCode().is2xxSuccessful()).isTrue();
// }).matchesRegex(".*\\{\"response\":\"ping\"\\}")));
// }
// @Container
// private static final StreamApps streamApps = kafkaStreamApps(
// HttpRequestProcessorTests.class.getSimpleName(), kafka)
// .withSourceContainer(httpSource(sourcePort))
// .withProcessorContainer(defaultKafkaProcessorContainerFor("http-request-processor")
// .withEnv("HTTP_REQUEST_URL_EXPRESSION",
// "'http://" + localHostAddress() + ":" + serverPort + "'")
// .withEnv("HTTP_REQUEST_HTTP_METHOD_EXPRESSION", "'POST'"))
// .withSinkContainer(
// defaultKafkaContainerFor("log-sink").withLogConsumer(logMatcher))
// .build();
//
// @BeforeAll
// static void startServer() throws Exception {
// server.start(InetAddress.getLocalHost(), serverPort);
// }
//
// @Test
// void get() {
// server.setDispatcher(new Dispatcher() {
// @Override
// public MockResponse dispatch(RecordedRequest recordedRequest) {
// return new MockResponse().setHeader(HttpHeaders.CONTENT_TYPE,
// MediaType.APPLICATION_JSON_VALUE)
// .setBody("{\"response\":\"" + recordedRequest.getBody().readUtf8() + "\"}")
// .setResponseCode(HttpStatus.OK.value());
// }
// });
//
// await().atMost(Duration.ofSeconds(30))
// .until(logMatcher.verifies(log -> log.when(() -> {
// ClientResponse response = webClient
// .post()
// .uri("http://localhost:" + streamApps.sourceContainer().getMappedPort(sourcePort))
// .contentType(MediaType.TEXT_PLAIN)
// .body(Mono.just("ping"), String.class)
// .exchange()
// .block();
// assertThat(response.statusCode().is2xxSuccessful()).isTrue();
// }).matchesRegex(".*\\{\"response\":\"ping\"\\}")));
// }
}

134
stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/sink/JdbcSinkTests.java
View File

@@ -16,29 +16,10 @@
package org.springframework.cloud.stream.apps.integration.test.sink;
import java.time.Duration;
import com.zaxxer.hikari.HikariDataSource;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
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 reactor.core.publisher.Mono;
import org.springframework.cloud.stream.app.test.integration.StreamApps;
import org.springframework.cloud.stream.apps.integration.test.support.KafkaStreamIntegrationTestSupport;
import org.springframework.http.MediaType;
import org.springframework.jdbc.core.JdbcTemplate;
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.awaitility.Awaitility.await;
import static org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamApps.kafkaStreamApps;
public class JdbcSinkTests extends KafkaStreamIntegrationTestSupport {
private static int serverPort = findAvailablePort();
@@ -47,60 +28,63 @@ public class JdbcSinkTests extends KafkaStreamIntegrationTestSupport {
private static WebClient webClient = WebClient.builder().build();
// @Container
// private static MySQLContainer mySQL = new MySQLContainer<>(DockerImageName.parse("mysql:5.7"))
// .withUsername("test")
// .withPassword("secret")
// .withExposedPorts(3306)
// .withNetwork(kafka.getNetwork())
// .withClasspathResourceMapping("init.sql", "/init.sql", BindMode.READ_ONLY)
// .withCommand("--init-file", "/init.sql");
//
// @Container
// private static StreamApps streamApps = kafkaStreamApps(JdbcSinkTests.class.getSimpleName(), kafka)
// .withSourceContainer(defaultKafkaContainerFor("http-source")
// .withEnv("SERVER_PORT", String.valueOf(serverPort))
// .withExposedPorts(serverPort)
// .waitingFor(Wait.forListeningPort().withStartupTimeout(Duration.ofMinutes(2))))
// .withSinkContainer(defaultKafkaContainerFor("jdbc-sink")
// .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:mysql://" + mySQL.getNetworkAliases().get(0) + ":3306/test"))
// .build();
//
// @BeforeAll
// static void startStreamApps() {
// 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");
// }
//
// @Test
// void postData() {
// String json = "{\"name\":\"My Name\",\"address\":{ \"city\": \"Big City\", \"street\":\"Narrow Alley\"}}";
// ClientResponse response = webClient
// .post()
// .uri("http://localhost:" + streamApps.sourceContainer().getMappedPort(serverPort))
// .contentType(MediaType.APPLICATION_JSON)
// .body(Mono.just(json), String.class)
// .exchange()
// .block();
// assertThat(response.statusCode().is2xxSuccessful()).isTrue();
//
// await().atMost(Duration.ofSeconds(30))
// .untilAsserted(
// () -> assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) from People",
// Integer.class))
// .isOne());
// assertThat(jdbcTemplate.queryForObject("SELECT name from People",
// String.class)).isEqualTo("My Name");
// }
// @Container
// private static MySQLContainer mySQL = new
// MySQLContainer<>(DockerImageName.parse("mysql:5.7"))
// .withUsername("test")
// .withPassword("secret")
// .withExposedPorts(3306)
// .withNetwork(kafka.getNetwork())
// .withClasspathResourceMapping("init.sql", "/init.sql", BindMode.READ_ONLY)
// .withCommand("--init-file", "/init.sql");
//
// @Container
// private static StreamApps streamApps =
// kafkaStreamApps(JdbcSinkTests.class.getSimpleName(), kafka)
// .withSourceContainer(defaultKafkaContainerFor("http-source")
// .withEnv("SERVER_PORT", String.valueOf(serverPort))
// .withExposedPorts(serverPort)
// .waitingFor(Wait.forListeningPort().withStartupTimeout(Duration.ofMinutes(2))))
// .withSinkContainer(defaultKafkaContainerFor("jdbc-sink")
// .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:mysql://" + mySQL.getNetworkAliases().get(0) + ":3306/test"))
// .build();
//
// @BeforeAll
// static void startStreamApps() {
// 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");
// }
//
// @Test
// void postData() {
// String json = "{\"name\":\"My Name\",\"address\":{ \"city\": \"Big City\",
// \"street\":\"Narrow Alley\"}}";
// ClientResponse response = webClient
// .post()
// .uri("http://localhost:" + streamApps.sourceContainer().getMappedPort(serverPort))
// .contentType(MediaType.APPLICATION_JSON)
// .body(Mono.just(json), String.class)
// .exchange()
// .block();
// assertThat(response.statusCode().is2xxSuccessful()).isTrue();
//
// await().atMost(Duration.ofSeconds(30))
// .untilAsserted(
// () -> assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) from People",
// Integer.class))
// .isOne());
// assertThat(jdbcTemplate.queryForObject("SELECT name from People",
// String.class)).isEqualTo("My Name");
// }
}

116
stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/sink/MongoDBSinkTests.java
View File

@@ -16,73 +16,57 @@
package org.springframework.cloud.stream.apps.integration.test.sink;
import java.time.Duration;
import java.util.List;
import org.bson.Document;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.MongoDBContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.utility.DockerImageName;
import reactor.core.publisher.Mono;
import org.springframework.cloud.stream.app.test.integration.StreamApps;
import org.springframework.cloud.stream.apps.integration.test.support.KafkaStreamIntegrationTestSupport;
import org.springframework.data.mongodb.MongoDatabaseFactory;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.SimpleMongoClientDatabaseFactory;
import org.springframework.http.MediaType;
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.test.integration.kafka.KafkaStreamApps.kafkaStreamApps;
public class MongoDBSinkTests extends KafkaStreamIntegrationTestSupport {
//
// private static int serverPort = findAvailablePort();
//
// private static MongoTemplate mongoTemplate;
//
// private static WebClient webClient = WebClient.builder().build();
//
// @Container
// private static MongoDBContainer mongoDBContainer = new MongoDBContainer(DockerImageName.parse("mongo:4.0.10"))
// .withExposedPorts(27017)
// .withStartupTimeout(Duration.ofMinutes(2));
//
// private static String mongoConnectionString() {
// return String.format("mongodb://%s:%s/%s", localHostAddress(), mongoDBContainer.getMappedPort(27017), "test");
// }
//
// @Container
// private StreamApps streamApps = kafkaStreamApps(MongoDBSinkTests.class.getSimpleName(), kafka)
// .withSourceContainer(httpSource(serverPort))
// .withSinkContainer(defaultKafkaContainerFor("mongodb-sink")
// .withEnv("MONGO_DB_CONSUMER_COLLECTION", "test")
// .withEnv("SPRING_DATA_MONGODB_URL", mongoConnectionString()))
// .build();
//
// @BeforeAll
// static void buildMongoTemplate() {
// MongoDatabaseFactory mongoDatabaseFactory = new SimpleMongoClientDatabaseFactory(
// mongoConnectionString());
// mongoTemplate = new MongoTemplate(mongoDatabaseFactory);
// }
//
// @Test
// void postData() {
// String json = "{\"name\":\"My Name\",\"address\":{ \"city\": \"Big City\", \"street\": \"Narrow Alley\"}}";
// ClientResponse response = webClient
// .post()
// .uri("http://localhost:" + streamApps.sourceContainer().getMappedPort(serverPort))
// .contentType(MediaType.APPLICATION_JSON)
// .body(Mono.just(json), String.class)
// .exchange()
// .block(Duration.ofSeconds(30));
// assertThat(response.statusCode().is2xxSuccessful()).isTrue();
// List<Document> docs = mongoTemplate.findAll(Document.class, "test");
// assertThat(docs).allMatch(document -> document.get("name", String.class).equals("My Name"));
// }
//
// private static int serverPort = findAvailablePort();
//
// private static MongoTemplate mongoTemplate;
//
// private static WebClient webClient = WebClient.builder().build();
//
// @Container
// private static MongoDBContainer mongoDBContainer = new
// MongoDBContainer(DockerImageName.parse("mongo:4.0.10"))
// .withExposedPorts(27017)
// .withStartupTimeout(Duration.ofMinutes(2));
//
// private static String mongoConnectionString() {
// return String.format("mongodb://%s:%s/%s", localHostAddress(),
// mongoDBContainer.getMappedPort(27017), "test");
// }
//
// @Container
// private StreamApps streamApps = kafkaStreamApps(MongoDBSinkTests.class.getSimpleName(),
// kafka)
// .withSourceContainer(httpSource(serverPort))
// .withSinkContainer(defaultKafkaContainerFor("mongodb-sink")
// .withEnv("MONGO_DB_CONSUMER_COLLECTION", "test")
// .withEnv("SPRING_DATA_MONGODB_URL", mongoConnectionString()))
// .build();
//
// @BeforeAll
// static void buildMongoTemplate() {
// MongoDatabaseFactory mongoDatabaseFactory = new SimpleMongoClientDatabaseFactory(
// mongoConnectionString());
// mongoTemplate = new MongoTemplate(mongoDatabaseFactory);
// }
//
// @Test
// void postData() {
// String json = "{\"name\":\"My Name\",\"address\":{ \"city\": \"Big City\", \"street\":
// \"Narrow Alley\"}}";
// ClientResponse response = webClient
// .post()
// .uri("http://localhost:" + streamApps.sourceContainer().getMappedPort(serverPort))
// .contentType(MediaType.APPLICATION_JSON)
// .body(Mono.just(json), String.class)
// .exchange()
// .block(Duration.ofSeconds(30));
// assertThat(response.statusCode().is2xxSuccessful()).isTrue();
// List<Document> docs = mongoTemplate.findAll(Document.class, "test");
// assertThat(docs).allMatch(document -> document.get("name", String.class).equals("My
// Name"));
// }
}

119
stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/sink/TcpSinkTests.java
View File

@@ -16,77 +16,58 @@
package org.springframework.cloud.stream.apps.integration.test.sink;
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.BeforeAll;
import org.junit.jupiter.api.Test;
import org.testcontainers.junit.jupiter.Container;
import reactor.core.publisher.Mono;
import org.springframework.cloud.stream.app.test.integration.StreamApps;
import org.springframework.cloud.stream.apps.integration.test.support.KafkaStreamIntegrationTestSupport;
import org.springframework.http.MediaType;
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.awaitility.Awaitility.await;
import static org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamApps.kafkaStreamApps;
public class TcpSinkTests extends KafkaStreamIntegrationTestSupport {
// private static final int port = findAvailablePort();
//
// private static final int tcpPort = findAvailablePort();
//
// private static Socket socket;
//
// private static final AtomicBoolean socketReady = new AtomicBoolean();
//
// private static WebClient webClient = WebClient.builder().build();
//
// @Container
// private static StreamApps streamApps = kafkaStreamApps(TcpSinkTests.class.getSimpleName(), kafka)
// .withSourceContainer(httpSource(port))
// .withSinkContainer(defaultKafkaContainerFor("tcp-sink")
// .withEnv("TCP_CONSUMER_HOST", localHostAddress())
// .withEnv("TCP_PORT", String.valueOf(tcpPort))
// .withEnv("TCP_CONSUMER_ENCODER", "CRLF"))
// .build();
//
// @BeforeAll
// static void startTcpServer() {
// new Thread(() -> {
// try {
// socket = new ServerSocket(tcpPort, 50, InetAddress.getLocalHost()).accept();
// socketReady.set(true);
// }
// catch (IOException exception) {
// exception.printStackTrace();
// }
// }).start();
// }
//
// @Test
// void postData() throws IOException {
// String text = "Hello, world!";
// ClientResponse response = webClient
// .post()
// .uri("http://localhost:" + streamApps.sourceContainer().getMappedPort(port))
// .contentType(MediaType.TEXT_PLAIN)
// .body(Mono.just(text), String.class)
// .exchange()
// .block();
// assertThat(response.statusCode().is2xxSuccessful()).isTrue();
// await().atMost(Duration.ofSeconds(10)).untilTrue(socketReady);
// BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
// await().atMost(Duration.ofSeconds(10)).until(() -> reader.readLine().equals("Hello, world!"));
// }
// private static final int port = findAvailablePort();
//
// private static final int tcpPort = findAvailablePort();
//
// private static Socket socket;
//
// private static final AtomicBoolean socketReady = new AtomicBoolean();
//
// private static WebClient webClient = WebClient.builder().build();
//
// @Container
// private static StreamApps streamApps =
// kafkaStreamApps(TcpSinkTests.class.getSimpleName(), kafka)
// .withSourceContainer(httpSource(port))
// .withSinkContainer(defaultKafkaContainerFor("tcp-sink")
// .withEnv("TCP_CONSUMER_HOST", localHostAddress())
// .withEnv("TCP_PORT", String.valueOf(tcpPort))
// .withEnv("TCP_CONSUMER_ENCODER", "CRLF"))
// .build();
//
// @BeforeAll
// static void startTcpServer() {
// new Thread(() -> {
// try {
// socket = new ServerSocket(tcpPort, 50, InetAddress.getLocalHost()).accept();
// socketReady.set(true);
// }
// catch (IOException exception) {
// exception.printStackTrace();
// }
// }).start();
// }
//
// @Test
// void postData() throws IOException {
// String text = "Hello, world!";
// ClientResponse response = webClient
// .post()
// .uri("http://localhost:" + streamApps.sourceContainer().getMappedPort(port))
// .contentType(MediaType.TEXT_PLAIN)
// .body(Mono.just(text), String.class)
// .exchange()
// .block();
// assertThat(response.statusCode().is2xxSuccessful()).isTrue();
// await().atMost(Duration.ofSeconds(10)).untilTrue(socketReady);
// BufferedReader reader = new BufferedReader(new
// InputStreamReader(socket.getInputStream()));
// await().atMost(Duration.ofSeconds(10)).until(() -> reader.readLine().equals("Hello,
// world!"));
// }
}