diff --git a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/TickTockTests.java b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/TickTockTests.java index a8f851e..bdb5992 100644 --- a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/TickTockTests.java +++ b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/TickTockTests.java @@ -20,11 +20,12 @@ import java.time.Duration; import java.util.regex.Pattern; import org.junit.jupiter.api.Test; -import org.springframework.cloud.stream.apps.integration.test.support.AbstractStreamApplicationTests; -import org.springframework.cloud.stream.apps.integration.test.support.LogMatcher; import org.testcontainers.containers.DockerComposeContainer; import org.testcontainers.junit.jupiter.Container; +import org.springframework.cloud.stream.apps.integration.test.support.AbstractStreamApplicationTests; +import org.springframework.cloud.stream.apps.integration.test.support.LogMatcher; + import static org.awaitility.Awaitility.await; import static org.springframework.cloud.stream.apps.integration.test.support.AbstractStreamApplicationTests.AppLog.appLog; @@ -42,7 +43,7 @@ public class TickTockTests extends AbstractStreamApplicationTests { @Test void ticktock() { - await().atMost(Duration.ofMinutes(2)).untilTrue(logMatcher.contains("Started LogSink").matches()); - await().atMost(Duration.ofSeconds(30)).untilTrue(logMatcher.withRegex(pattern.pattern()).matches()); + await().atMost(Duration.ofMinutes(2)).until(logMatcher.verifies(log -> log.contains("Started LogSink"))); + await().atMost(Duration.ofSeconds(30)).until(logMatcher.verifies(log -> log.matchesRegex(pattern.pattern()))); } } diff --git a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/processor/HttpRequestProcessorTests.java b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/processor/HttpRequestProcessorTests.java index 0faad65..780e93e 100644 --- a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/processor/HttpRequestProcessorTests.java +++ b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/processor/HttpRequestProcessorTests.java @@ -78,16 +78,17 @@ public class HttpRequestProcessorTests extends AbstractStreamApplicationTests { .setResponseCode(HttpStatus.OK.value()); } }); - ClientResponse response = webClient() - .post() - .uri("http://localhost:" + sourcePort) - .contentType(MediaType.TEXT_PLAIN) - .body(Mono.just("ping"), String.class) - .exchange() - .block(); - assertThat(response.statusCode().is2xxSuccessful()).isTrue(); await().atMost(Duration.ofSeconds(30)) - .untilTrue(logMatcher.withRegex(".*\\{\"response\":\"ping\"\\}").matches()); + .until(logMatcher.verifies(log -> log.when(() -> { + ClientResponse response = webClient() + .post() + .uri("http://localhost:" + sourcePort) + .contentType(MediaType.TEXT_PLAIN) + .body(Mono.just("ping"), String.class) + .exchange() + .block(); + assertThat(response.statusCode().is2xxSuccessful()).isTrue(); + }).matchesRegex(".*\\{\"response\":\"ping\"\\}"))); } } diff --git a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/GeodeSourceTests.java b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/GeodeSourceTests.java index a0868c4..b580ddd 100644 --- a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/GeodeSourceTests.java +++ b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/GeodeSourceTests.java @@ -45,20 +45,20 @@ import static org.springframework.cloud.stream.apps.integration.test.support.Flu public class GeodeSourceTests extends AbstractStreamApplicationTests { - private static LogMatcher logMatcher = new LogMatcher(); + private static final LogMatcher logMatcher = new LogMatcher(); - private static LogMatcher geodeLogMatcher = new LogMatcher(); + private static final LogMatcher geodeLogMatcher = new LogMatcher(); - private static int locatorPort = findAvailablePort(); + private static final int locatorPort = findAvailablePort(); - private static int cacheServerPort = findAvailablePort(); + private static final int cacheServerPort = findAvailablePort(); private static Region clientRegion; private static ClientCache clientCache; @Container - private static GeodeContainer geode = (GeodeContainer) new GeodeContainer(new ImageFromDockerfile() + private static final GeodeContainer geode = (GeodeContainer) new GeodeContainer(new ImageFromDockerfile() .withFileFromClasspath("Dockerfile", "geode/Dockerfile") .withBuildArg("CACHE_SERVER_PORT", String.valueOf(cacheServerPort)) .withBuildArg("LOCATOR_PORT", String.valueOf(locatorPort)), @@ -91,7 +91,7 @@ public class GeodeSourceTests extends AbstractStreamApplicationTests { } @Container - private DockerComposeContainer environment = new DockerComposeContainer( + private final DockerComposeContainer environment = new DockerComposeContainer( templateProcessor("source/geode-source-tests.yml", fluentMap() .withEntry("geode.host-addresses", "geode:" + cacheServerPort) .withEntry("geodeHost", localHostAddress()) @@ -103,12 +103,16 @@ public class GeodeSourceTests extends AbstractStreamApplicationTests { @Test void test() { - LogMatcher.LogListener logListener = logMatcher.contains("world"); + // logMatcher.when(() -> clientRegion.put("hello", "world")).contains("world")); + // LogMatcher.LogListener logListener = logMatcher.contains("world").when(() -> + // clientRegion.put("hello", "world")); await().atMost(Duration.ofMinutes(2)) - .untilTrue(geodeLogMatcher.contains("Started GeodeSource").matches()); - clientRegion.put("hello", "world"); + .until(geodeLogMatcher.verifies(log -> log.contains("Started GeodeSource"))); + // clientRegion.put("hello", "world"); await().atMost(Duration.ofSeconds(30)) - .untilTrue(logListener.matches()); + .until(logMatcher.verifies(log -> log + .when(() -> clientRegion.put("hello", "world")) + .contains("world"))); } @AfterAll diff --git a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/HttpSourceTests.java b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/HttpSourceTests.java index f590e69..992ad54 100644 --- a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/HttpSourceTests.java +++ b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/HttpSourceTests.java @@ -50,30 +50,31 @@ public class HttpSourceTests extends AbstractStreamApplicationTests { @Test void plaintext() { - ClientResponse response = webClient() - .post() - .uri("http://localhost:" + port) - .contentType(MediaType.TEXT_PLAIN) - .body(Mono.just("Hello"), String.class) - .exchange() - .block(); - assertThat(response.statusCode().is2xxSuccessful()).isTrue(); - await().atMost(Duration.ofSeconds(30)) - .untilTrue(logMatcher.endsWith("Hello").matches()); + .until(logMatcher.verifies(log -> log.when(() -> { + ClientResponse response = webClient() + .post() + .uri("http://localhost:" + port) + .contentType(MediaType.TEXT_PLAIN) + .body(Mono.just("Hello"), String.class) + .exchange() + .block(); + assertThat(response.statusCode().is2xxSuccessful()).isTrue(); + }).endsWith("Hello"))); } @Test void json() { - ClientResponse response = webClient() - .post() - .uri("http://localhost:" + port) - .contentType(MediaType.APPLICATION_JSON) - .body(Mono.just("{\"Hello\":\"world\"}"), String.class) - .exchange() - .block(); - assertThat(response.statusCode().is2xxSuccessful()).isTrue(); await().atMost(Duration.ofSeconds(30)) - .untilTrue(logMatcher.withRegex(".*\\{\"Hello\":\"world\"\\}").matches()); + .until(logMatcher.verifies(log -> log.when(() -> { + ClientResponse response = webClient() + .post() + .uri("http://localhost:" + port) + .contentType(MediaType.APPLICATION_JSON) + .body(Mono.just("{\"Hello\":\"world\"}"), String.class) + .exchange() + .block(); + assertThat(response.statusCode().is2xxSuccessful()).isTrue(); + }).matchesRegex(".*\\{\"Hello\":\"world\"\\}"))); } } diff --git a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/JdbcSourceTests.java b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/JdbcSourceTests.java index f3552a7..71ef218 100644 --- a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/JdbcSourceTests.java +++ b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/JdbcSourceTests.java @@ -43,6 +43,6 @@ public class JdbcSourceTests extends AbstractStreamApplicationTests { @Test void test() { - await().atMost(Duration.ofSeconds(30)).untilTrue(logMatcher.contains("Bart Simpson").matches()); + await().atMost(Duration.ofSeconds(30)).until(logMatcher.verifies(log -> log.contains("Bart Simpson"))); } } diff --git a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/S3SourceTests.java b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/S3SourceTests.java index 3b931ac..7874b97 100644 --- a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/S3SourceTests.java +++ b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/S3SourceTests.java @@ -38,7 +38,6 @@ import org.testcontainers.junit.jupiter.Container; import org.springframework.cloud.stream.apps.integration.test.support.AbstractStreamApplicationTests; import org.springframework.cloud.stream.apps.integration.test.support.LogMatcher; -import org.springframework.cloud.stream.apps.integration.test.support.LogMatcher.LogListener; import static org.awaitility.Awaitility.await; import static org.springframework.cloud.stream.apps.integration.test.support.AbstractStreamApplicationTests.AppLog.appLog; @@ -90,11 +89,11 @@ public class S3SourceTests extends AbstractStreamApplicationTests { @Test void test() { + await().atMost(Duration.ofMinutes(2)).until(logMatcher.verifies(log -> log.contains("Started S3Source"))); + await().atMost(Duration.ofSeconds(30)).until(logMatcher.verifies(log -> log.when(() -> { + s3Client.createBucket("bucket"); + s3Client.putObject(new PutObjectRequest("bucket", "test", resourceAsFile("minio/data"))); + }).contains("Bart Simpson"))); - await().atMost(Duration.ofMinutes(2)).untilTrue(logMatcher.contains("Started S3Source").matches()); - LogListener logListener = logMatcher.contains("Bart Simpson"); - s3Client.createBucket("bucket"); - s3Client.putObject(new PutObjectRequest("bucket", "test", resourceAsFile("minio/data"))); - await().atMost(Duration.ofSeconds(30)).untilTrue(logListener.matches()); } } diff --git a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/support/LogMatcher.java b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/support/LogMatcher.java index fdd7ab9..8e40edd 100644 --- a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/support/LogMatcher.java +++ b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/support/LogMatcher.java @@ -19,6 +19,7 @@ package org.springframework.cloud.stream.apps.integration.test.support; import java.util.LinkedList; import java.util.List; import java.util.Optional; +import java.util.concurrent.Callable; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import java.util.regex.Pattern; @@ -32,12 +33,12 @@ public class LogMatcher implements Consumer { private List> listeners = new LinkedList<>(); - public LogListener contains(String string) { - return withRegex(".*" + string + ".*"); - } - - public LogListener endsWith(String string) { - return withRegex(".*" + string); + public Callable verifies(Consumer consumer) { + LogListener logListener = new LogListener(); + consumer.accept(logListener); + logListener.runnable.ifPresent(runnable -> runnable.run()); + listeners.add(logListener); + return () -> logListener.matches().get(); } @Override @@ -45,25 +46,12 @@ public class LogMatcher implements Consumer { listeners.forEach(m -> m.accept(outputFrame.getUtf8String())); } - public LogListener withRegex(String regex) { - LogListener logListener = new LogListener(regex); - if (logListener.runnable.isPresent()) { - logListener.runnable.get().run(); - } - listeners.add(logListener); - return logListener; - } - public class LogListener implements Consumer { private AtomicBoolean matched = new AtomicBoolean(); private Optional runnable = Optional.empty(); - private final Pattern pattern; - - LogListener(String regex) { - pattern = Pattern.compile(regex); - } + private Pattern pattern; @Override public void accept(String s) { @@ -74,6 +62,19 @@ public class LogMatcher implements Consumer { } } + public LogListener contains(String string) { + return matchesRegex(".*" + string + ".*"); + } + + public LogListener endsWith(String string) { + return matchesRegex(".*" + string); + } + + public LogListener matchesRegex(String regex) { + this.pattern = Pattern.compile(regex); + return this; + } + public LogListener when(Runnable runnable) { this.runnable = Optional.of(runnable); return this;