From adabb50693c53f2bcb2190e83d59f188f88e5453 Mon Sep 17 00:00:00 2001 From: David Turanski Date: Fri, 18 Sep 2020 15:46:29 -0400 Subject: [PATCH] Implement SftpSourceTests --- .../test/source/SftpSourceTests.java | 74 +++++++++++++++++++ .../src/test/resources/sftp/data.txt | 1 + .../resources/source/sftp-source-tests.yml | 27 +++++++ 3 files changed, 102 insertions(+) create mode 100644 stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/SftpSourceTests.java create mode 100755 stream-applications-integration-tests/src/test/resources/sftp/data.txt create mode 100644 stream-applications-integration-tests/src/test/resources/source/sftp-source-tests.yml diff --git a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/SftpSourceTests.java b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/SftpSourceTests.java new file mode 100644 index 0000000..0109976 --- /dev/null +++ b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/source/SftpSourceTests.java @@ -0,0 +1,74 @@ +/* + * 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.source; + +import java.time.Duration; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.BindMode; +import org.testcontainers.containers.DockerComposeContainer; +import org.testcontainers.containers.GenericContainer; +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; +import static org.springframework.cloud.stream.apps.integration.test.support.FluentMap.fluentMap; + +public class SftpSourceTests extends AbstractStreamApplicationTests { + + private static LogMatcher logMatcher = new LogMatcher(); + + @Container + private static final GenericContainer sftp = (GenericContainer) new GenericContainer("atmoz/sftp") + .withExposedPorts(22) + .withCommand("user:pass:::remote") + .withClasspathResourceMapping("sftp", "/home/user/remote", BindMode.READ_ONLY) + .withStartupTimeout(Duration.ofMinutes(1)); + + private DockerComposeContainer environment; + + @Test + void test() { + startContainer(templateProcessor("source/sftp-source-tests.yml", + fluentMap().withEntry("sftpPort", sftp.getMappedPort(22)) + .withEntry("functionDefinition", "sftpSupplier") + .withEntry("consumerMode", "ref") + .withEntry("listOnly", false) + .withEntry("sftpHost", localHostAddress()))); + await().atMost(Duration.ofSeconds(30)) + .until(logMatcher.verifies(logListener -> logListener.endsWith("\"/tmp/sftp-supplier/data.txt\""))); + } + + private void startContainer(TemplateProcessor templateProcessor) { + environment = new DockerComposeContainer( + templateProcessor.processTemplate()) + .withLogConsumer("log-sink", logMatcher) + .withLogConsumer("sftp-source", logMatcher) + .withLogConsumer("log-sink", appLog("log-sink")); + environment.start(); + } + + @AfterEach + void stop() { + environment.stop(); + } +} diff --git a/stream-applications-integration-tests/src/test/resources/sftp/data.txt b/stream-applications-integration-tests/src/test/resources/sftp/data.txt new file mode 100755 index 0000000..01c51bc --- /dev/null +++ b/stream-applications-integration-tests/src/test/resources/sftp/data.txt @@ -0,0 +1 @@ +Bart Simpson diff --git a/stream-applications-integration-tests/src/test/resources/source/sftp-source-tests.yml b/stream-applications-integration-tests/src/test/resources/source/sftp-source-tests.yml new file mode 100644 index 0000000..ad9a795 --- /dev/null +++ b/stream-applications-integration-tests/src/test/resources/source/sftp-source-tests.yml @@ -0,0 +1,27 @@ +version: {{docker.compose.version}} +services: + sftp-source: + image: springcloudstream/sftp-source-kafka:{{stream.apps.version}} + environment: + - SPRING_CLOUD_FUNCTION_DEFINITION={{functionDefinition}} + - FILE_CONSUMER_MODE={{consumerMode}} + - SFTP_SUPPLIER_FACTORY_ALLOW_UNKNOWN_KEYS=true + - SFTP_SUPPLIER_REMOTE_DIR=/remote + - SFTP_SUPPLIER_FACTORY_USERNAME=user + - SFTP_SUPPLIER_FACTORY_PASSWORD=pass + - SFTP_SUPPLIER_FACTORY_HOST=sftp + - SFTP_SUPPLIER_FACTORY_PORT={{sftpPort}} + - SFTP_SUPPLIER_LIST_ONLY={{listOnly}} + - 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: + - sftp:{{sftpHost}} + log-sink: + image: springcloudstream/log-sink-kafka:{{stream.apps.version}} + environment: + - SPRING_CLOUD_STREAM_KAFKA_BINDER_BROKERS={{kafkaBootStrapServers}} + - SPRING_CLOUD_STREAM_BINDINGS_INPUT_DESTINATION=log + - SPRING_CLOUD_STREAM_BINDINGS_INPUT_GROUP=sftp-source-tests \ No newline at end of file