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

Updated SFTP Integration test with file.consumer.mode=lines

Browse Source
This commit is contained in:
Corneil du Plessis
2023-10-20 17:49:05 +02:00
parent ae97e3e31d
commit dff63b54da
1 changed files with 30 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/sftp/SftpSourceTests.java
View File

@@ -24,12 +24,16 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.BindMode;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.Network;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.utility.DockerImageName;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.app.test.integration.AppLog;
import org.springframework.cloud.stream.app.test.integration.OutputMatcher;
import org.springframework.cloud.stream.app.test.integration.StreamAppContainer;
import org.springframework.cloud.stream.app.test.integration.junit.jupiter.BaseContainerExtension;
@@ -37,42 +41,59 @@ import org.springframework.cloud.stream.app.test.integration.junit.jupiter.BaseC
import static org.awaitility.Awaitility.await;
import static org.springframework.cloud.stream.app.integration.test.common.Configuration.DEFAULT_DURATION;
@SuppressWarnings({"rawtypes", "unchecked"})
@Tag("integration")
@ExtendWith(BaseContainerExtension.class)
abstract class SftpSourceTests {
private static final Logger logger = LoggerFactory.getLogger(SftpSourceTests.class);
private StreamAppContainer source;
@Container
private static final GenericContainer sftp = new GenericContainer(DockerImageName.parse("atmoz/sftp"))
.withExposedPorts(22)
.withNetwork(Network.SHARED)
.withNetworkAliases("sftp-host")
.withCommand("user:pass:::remote")
.withClasspathResourceMapping("sftp", "/home/user/remote", BindMode.READ_ONLY)
.withStartupTimeout(DEFAULT_DURATION);
.withStartupTimeout(DEFAULT_DURATION)
.withLogConsumer(AppLog.appLog("sftp-host"));
@BeforeEach
void configureSource() {
await().atMost(DEFAULT_DURATION).until(sftp::isRunning);
source = BaseContainerExtension.containerInstance()
.withNetwork(Network.SHARED)
.withEnv("SFTP_SUPPLIER_FACTORY_ALLOW_UNKNOWN_KEYS", "true")
.withEnv("SFTP_SUPPLIER_REMOTE_DIR", "/remote")
.withEnv("SFTP_SUPPLIER_FACTORY_USERNAME", "user")
.withEnv("SFTP_SUPPLIER_FACTORY_PASSWORD", "pass")
.withEnv("SFTP_SUPPLIER_FACTORY_PORT", String.valueOf(sftp.getMappedPort(22)))
.withEnv("SFTP_SUPPLIER_FACTORY_HOST", "sftp-host");
.withEnv("SFTP_SUPPLIER_FACTORY_PORT", "22")
.withEnv("SFTP_SUPPLIER_FACTORY_HOST", "sftp-host")
.withEnv("LOGGING_LEVEL_ORG_SPRINGFRAMEWORK_INTEGRATION", "DEBUG")
.withEnv("LOGGING_LEVEL_ORG_SPRINGFRAMEWORK_CLOUD", "DEBUG")
.log();
}
@Autowired
private OutputMatcher outputMatcher;
// TODO: This fixture supports additional tests with different modes, etc.
@Test
void test() {
void testRef() {
startContainer(Collections.singletonMap("FILE_CONSUMER_MODE", "ref"));
await().atMost(DEFAULT_DURATION)
.until(outputMatcher.payloadMatches((String s) -> s.equals("\"/tmp/sftp-supplier/data.txt\"")));
await().atMost(DEFAULT_DURATION).until(outputMatcher.payloadMatches((String s) -> {
logger.info("payload:{}", s);
return s.equals("\"/tmp/sftp-supplier/data.txt\"");
}));
}
@Test
void testLines() {
startContainer(Collections.singletonMap("FILE_CONSUMER_MODE", "lines"));
await().atMost(DEFAULT_DURATION).until(outputMatcher.payloadMatches((String s) -> {
logger.info("payload:{}", s);
return s.contains("Bart Simpson");
}));
}
private void startContainer(Map<String, String> environment) {
@@ -82,7 +103,7 @@ abstract class SftpSourceTests {
}
@AfterEach
private void cleanUp() {
public void cleanUp() {
source.stop();
}