From 8fbdcf3306d2eab9da8b4d78246b32bb3b1d14ca Mon Sep 17 00:00:00 2001 From: David Turanski Date: Thu, 17 Sep 2020 16:22:12 -0400 Subject: [PATCH] Implement S3SourceTests --- .../test/source/S3SourceTests.java | 83 ++++++++++++++++--- .../integration/test/support/LogMatcher.java | 1 + .../test/support/TemplateProcessor.java | 3 + .../test/resources/source/s3-source-tests.yml | 7 +- 4 files changed, 82 insertions(+), 12 deletions(-) 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 7874b97..9f422d3 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 @@ -29,6 +29,7 @@ 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.Test; import org.testcontainers.containers.DockerComposeContainer; @@ -38,6 +39,7 @@ 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.TemplateProcessor; import static org.awaitility.Awaitility.await; import static org.springframework.cloud.stream.apps.integration.test.support.AbstractStreamApplicationTests.AppLog.appLog; @@ -76,19 +78,20 @@ public class S3SourceTests extends AbstractStreamApplicationTests { } - @Container - private final DockerComposeContainer environment = new DockerComposeContainer( - templateProcessor("source/s3-source-tests.yml", - fluentMap().withEntry("s3.local.dir", resourceAsFile("minio")) - .withEntry("s3.endpoint.url", - "http://minio:" + minio.getMappedPort(9000)) - .withEntry("minioHost", localHostAddress())).processTemplate()) - .withLogConsumer("log-sink", logMatcher) - .withLogConsumer("s3-source", logMatcher) - .withLogConsumer("log-sink", appLog("logSink")); + private DockerComposeContainer environment; @Test - void test() { + void testLines() { + startContainer( + templateProcessor("source/s3-source-tests.yml", + fluentMap().withEntry("s3.local.dir", resourceAsFile("minio")) + .withEntry("s3.endpoint.url", + "http://minio:" + minio.getMappedPort(9000)) + .withEntry("functionDefinition", "s3Supplier") + .withEntry("consumerMode", "lines") + .withEntry("listOnly", false) + .withEntry("minioHost", localHostAddress()))); + 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"); @@ -96,4 +99,62 @@ public class S3SourceTests extends AbstractStreamApplicationTests { }).contains("Bart Simpson"))); } + + @Test + void testTaskLaunchRequest() { + + startContainer(templateProcessor("source/s3-source-tests.yml", + fluentMap().withEntry("s3.local.dir", resourceAsFile("minio")) + .withEntry("s3.endpoint.url", + "http://minio:" + minio.getMappedPort(9000)) + .withEntry("functionDefinition", "s3Supplier|taskLaunchRequestFunction") + .withEntry("consumerMode", "ref") + .withEntry("listOnly", false) + .withEntry("minioHost", localHostAddress()))); + environment.start(); + 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"))); + }).endsWith( + "\\{\"args\":\\[\"filename=/tmp/s3-supplier/test\"\\],\"deploymentProps\":\\{\\},\"name\":\"myTask\"\\}"))); + } + + @Test + void testListOnly() { + + startContainer(templateProcessor("source/s3-source-tests.yml", + fluentMap().withEntry("s3.local.dir", resourceAsFile("minio")) + .withEntry("s3.endpoint.url", + "http://minio:" + minio.getMappedPort(9000)) + .withEntry("functionDefinition", "s3Supplier") + .withEntry("consumerMode", "ref") + .withEntry("listOnly", true) + .withEntry("minioHost", localHostAddress()))); + environment.start(); + 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("\"bucketName\":\"bucket\",\"key\":\"test\""))); + } + + + private void startContainer(TemplateProcessor templateProcessor) { + environment = new DockerComposeContainer( + templateProcessor.processTemplate()) + .withLogConsumer("log-sink", logMatcher) + .withLogConsumer("s3-source", logMatcher) + .withLogConsumer("log-sink", appLog("log-sink")); + environment.start(); + } + + @AfterEach + void stop() { + if (s3Client.doesBucketExistV2("bucket")) { + s3Client.deleteObject("bucket", "test"); + s3Client.deleteBucket("bucket"); + } + environment.stop(); + } } 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 8e40edd..11841e8 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 @@ -59,6 +59,7 @@ public class LogMatcher implements Consumer { if (pattern.matcher(s.trim()).matches()) { logger.debug(" MATCHED " + s.trim()); matched.set(true); + listeners.remove(this); } } diff --git a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/support/TemplateProcessor.java b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/support/TemplateProcessor.java index 085d06b..719b9e2 100644 --- a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/support/TemplateProcessor.java +++ b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/support/TemplateProcessor.java @@ -55,6 +55,9 @@ public final class TemplateProcessor { Objects.requireNonNull(template.getInputStream()))) { Template resourceTemplate = Mustache.compiler().escapeHTML(false).compile(resourcesTemplateReader); Path temporaryFile = outputDirectory.resolve(Paths.get(template.getFilename())); + if (Files.exists(temporaryFile)) { + Files.delete(temporaryFile); + } Files.createFile(temporaryFile); Files.write(temporaryFile, resourceTemplate.execute(addGlobalProperties(templateProperties)).getBytes()).toFile(); diff --git a/stream-applications-integration-tests/src/test/resources/source/s3-source-tests.yml b/stream-applications-integration-tests/src/test/resources/source/s3-source-tests.yml index bd1aef5..9af7697 100644 --- a/stream-applications-integration-tests/src/test/resources/source/s3-source-tests.yml +++ b/stream-applications-integration-tests/src/test/resources/source/s3-source-tests.yml @@ -3,16 +3,21 @@ services: s3-source: image: springcloudstream/s3-source-kafka:{{stream.apps.version}} environment: - - FILE_CONSUMER_MODE=lines + - SPRING_CLOUD_FUNCTION_DEFINITION={{functionDefinition}} + - FILE_CONSUMER_MODE={{consumerMode}} - S3_COMMON_ENDPOINT_URL={{s3.endpoint.url}} - S3_COMMON_PATH_STYLE_ACCESS=true - S3_SUPPLIER_REMOTE_DIR=bucket + - S3_SUPPLIER_LIST_ONLY={{listOnly}} - CLOUD_AWS_STACK_AUTO=false - CLOUD_AWS_CREDENTIALS_ACCESS_KEY=minio - CLOUD_AWS_CREDENTIALS_SECRET_KEY=minio123 - CLOUD_AWS_REGION_STATIC=us-east-1 - SPRING_CLOUD_STREAM_BINDINGS_OUTPUT_DESTINATION=log - SPRING_CLOUD_STREAM_KAFKA_BINDER_BROKERS={{kafkaBootStrapServers}} +# For Task Launch Request + - TASK_LAUNCH_REQUEST_ARG_EXPRESSIONS=filename=payload + - TASK_LAUNCH_REQUEST_TASK_NAME=myTask extra_hosts: - minio:{{minioHost}} log-sink: