Implement S3SourceTests

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();
+	}
 }

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<OutputFrame> {
 			if (pattern.matcher(s.trim()).matches()) {
 				logger.debug(" MATCHED " + s.trim());
 				matched.set(true);
+				listeners.remove(this);
 			}
 		}
 

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();

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: