FTP: run tests against (S)FTP server in Docker

.gitignore

@@ -10,6 +10,7 @@ target/
 .project
 .settings
 .tmpBin/
+tmp/
 *.sublime-project
 /bin/
 ext-lib-src/

.travis.yml

@@ -57,6 +57,7 @@ jobs:
       - DIR=file
     - env:
       - DIR=ftp
+      - PRE_CMD="docker-compose up -d ftp sftp"
     - env:
       - DIR=geode
       - PRE_CMD="docker-compose up -d geode"

docker-compose.yml

@@ -66,6 +66,22 @@ services:
     image: deangiberson/aws-dynamodb-local
     ports:
       - "8001:8000"
+  ftp:
+    image: stilliard/pure-ftpd:latest
+    ports:
+      - "21000:21"
+      - "30000-30009:30000-30009"
+    volumes:
+      - ./ftp/tmp/home:/home/username/
+      - ./ftp/src/test/resources/ftpd/:/etc/ssl/private/
+    environment:
+      PUBLICHOST: "localhost"
+      FTP_USER_NAME: username
+      FTP_USER_PASS: userpass
+      FTP_USER_HOME: /home/username
+      FTP_USER_UID: 2000
+      FTP_USER_GID: 2000
+      ADDED_FLAGS: "--tls=1"
   geode:
     container_name: geode
     image: apachegeode/geode:1.8.0
@@ -199,3 +215,14 @@ services:
     environment:
       - "ORIENTDB_ROOT_PASSWORD=root"
     command: /orientdb/bin/server.sh -Dmemory.chunk.size=268435456
+  sftp:
+    image: atmoz/sftp
+    volumes:
+      - ./ftp/tmp/home:/home/username/upload
+      - ./ftp/src/test/resources/id_rsa.pub:/home/username/.ssh/keys/id_rsa.pub:ro
+      - ./ftp/src/test/resources/sftpd/ssh_host_ed25519_key:/tmp/ssh_host_ed25519_key
+      - ./ftp/src/test/resources/sftpd/ssh_host_rsa_key:/tmp/ssh_host_rsa_key
+      - ./ftp/src/test/resources/sftpd/init.sh:/etc/sftp.d/init.sh
+    ports:
+      - "2222:22"
+    command: username:userpass:2000:2000

ftp/src/main/scala/akka/stream/alpakka/ftp/impl/CommonFtpOperations.scala

@@ -7,6 +7,7 @@ package akka.stream.alpakka.ftp.impl
 import java.io.{IOException, InputStream, OutputStream}
 import java.nio.file.Paths
 import java.nio.file.attribute.PosixFilePermission
+import java.util.TimeZone
 
 import akka.annotation.InternalApi
 import akka.stream.alpakka.ftp.FtpFile
@@ -28,6 +29,8 @@ private[ftp] trait CommonFtpOperations {
       .listFiles(path)
       .collect {
         case file: FTPFile if file.getName != "." && file.getName != ".." =>
+          val calendar = file.getTimestamp
+          calendar.setTimeZone(TimeZone.getTimeZone("UTC"))
           FtpFile(
             file.getName,
             if (java.io.File.separatorChar == '\\')
@@ -36,7 +39,7 @@ private[ftp] trait CommonFtpOperations {
               Paths.get(s"$path/${file.getName}").normalize.toString,
             file.isDirectory,
             file.getSize,
-            file.getTimestamp.getTimeInMillis,
+            calendar.getTimeInMillis,
             getPosixFilePermissions(file)
           )
       }
@@ -75,4 +78,7 @@ private[ftp] trait CommonFtpOperations {
 
   def remove(path: String, handler: Handler): Unit =
     handler.deleteFile(path)
+
+  def completePendingCommand(handler: Handler): Boolean =
+    handler.completePendingCommand()
 }

ftp/src/main/scala/akka/stream/alpakka/ftp/impl/FtpIOGraphStage.scala

@@ -87,6 +87,12 @@ private[ftp] trait FtpIOSourceStage[FtpClient, S <: RemoteFileSettings]
           isOpt.foreach { os =>
             try {
               os.close()
+              ftpLike match {
+                case cfo: CommonFtpOperations =>
+                  if (!cfo.completePendingCommand(handler.get.asInstanceOf[cfo.Handler]))
+                    throw new IOException("File transfer failed.")
+                case _ =>
+              }
             } catch {
               case e: IOException =>
                 matFailure(e)
@@ -184,6 +190,12 @@ private[ftp] trait FtpIOSinkStage[FtpClient, S <: RemoteFileSettings]
           osOpt.foreach { os =>
             try {
               os.close()
+              ftpLike match {
+                case cfo: CommonFtpOperations =>
+                  if (!cfo.completePendingCommand(handler.get.asInstanceOf[cfo.Handler]))
+                    throw new IOException("File transfer failed.")
+                case _ =>
+              }
             } catch {
               case e: IOException =>
                 matFailure(e)

ftp/src/test/java/akka/stream/alpakka/ftp/BaseFtpSupport.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright (C) 2016-2019 Lightbend Inc. <http://www.lightbend.com>
+ */
+
+package akka.stream.alpakka.ftp;
+
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+public class BaseFtpSupport extends BaseSupportImpl {
+
+  private final Path ROOT_DIR = Paths.get("tmp/home");
+  public final String HOSTNAME = "localhost";
+  public final int PORT = 21000;
+  public final FtpCredentials CREDENTIALS = FtpCredentials.create("username", "userpass");
+
+  @Override
+  public Path getRootDir() {
+    return ROOT_DIR;
+  }
+}

ftp/src/test/java/akka/stream/alpakka/ftp/BaseSftpSupport.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright (C) 2016-2019 Lightbend Inc. <http://www.lightbend.com>
+ */
+
+package akka.stream.alpakka.ftp;
+
+import java.io.File;
+import java.nio.charset.Charset;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+public class BaseSftpSupport extends BaseSupportImpl {
+
+  private final Path ROOT_DIR = Paths.get("tmp/home");
+  final String HOSTNAME = "localhost";
+  final int PORT = 2222;
+  final FtpCredentials CREDENTIALS = FtpCredentials.create("username", "userpass");
+  // Issue: the root folder of the sftp server is not writable so tests must happen inside a
+  // sub-folder
+  final String ROOT_PATH = "upload/";
+
+  public static final byte[] CLIENT_PRIVATE_KEY_PASSPHRASE =
+      "secret".getBytes(Charset.forName("UTF-8"));
+
+  private File clientPrivateKeyFile;
+  private File knownHostsFile;
+
+  BaseSftpSupport() {
+    clientPrivateKeyFile = new File(getClass().getResource("/id_rsa").getPath());
+    knownHostsFile = new File(getClass().getResource("/known_hosts").getPath());
+  }
+
+  public File getClientPrivateKeyFile() {
+    return clientPrivateKeyFile;
+  }
+
+  public File getKnownHostsFile() {
+    return knownHostsFile;
+  }
+
+  @Override
+  public Path getRootDir() {
+    return ROOT_DIR;
+  }
+}

ftp/src/test/java/akka/stream/alpakka/ftp/BaseSupport.java

@@ -0,0 +1,22 @@
+/*
+ * Copyright (C) 2016-2019 Lightbend Inc. <http://www.lightbend.com>
+ */
+
+package akka.stream.alpakka.ftp;
+
+interface BaseSupport {
+
+  void cleanFiles();
+
+  void generateFiles(int numFiles, int pageSize, String basePath);
+
+  void putFileOnFtp(String filePath);
+
+  void putFileOnFtpWithContents(String filePath, byte[] fileContents);
+
+  byte[] getFtpFileContents(String filePath);
+
+  boolean fileExists(String filePath);
+
+  String getDefaultContent();
+}

ftp/src/test/java/akka/stream/alpakka/ftp/BaseSupportImpl.java

@@ -0,0 +1,130 @@
+/*
+ * Copyright (C) 2016-2019 Lightbend Inc. <http://www.lightbend.com>
+ */
+
+package akka.stream.alpakka.ftp;
+
+import akka.actor.ActorSystem;
+import akka.stream.ActorMaterializer;
+import akka.stream.Materializer;
+import org.junit.After;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.*;
+import java.nio.file.attribute.BasicFileAttributes;
+
+public abstract class BaseSupportImpl implements BaseSupport, AkkaSupport {
+
+  private ActorSystem system = ActorSystem.create("alpakka-ftp");
+  private Materializer materializer = ActorMaterializer.create(system);
+
+  private String loremIpsum =
+      "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Praesent auctor imperdiet "
+          + "velit, eu dapibus nisl dapibus vitae. Sed quam lacus, fringilla posuere ligula at, "
+          + "aliquet laoreet nulla. Aliquam id fermentum justo. Aliquam et massa consequat, "
+          + "pellentesque dolor nec, gravida libero. Phasellus elit eros, finibus eget "
+          + "sollicitudin ac, consectetur sed ante. Etiam ornare lacus blandit nisi gravida "
+          + "accumsan. Sed in lorem arcu. Vivamus et eleifend ligula. Maecenas ut commodo ante. "
+          + "Suspendisse sit amet placerat arcu, porttitor sagittis velit. Quisque gravida mi a "
+          + "porttitor ornare. Cras lorem nisl, sollicitudin vitae odio at, vehicula maximus "
+          + "mauris. Sed ac purus ac turpis pellentesque cursus ac eget est. Pellentesque "
+          + "habitant morbi tristique senectus et netus et malesuada fames ac turpis egestas.";
+
+  abstract Path getRootDir();
+
+  @Override
+  @After
+  public void cleanFiles() {
+    try {
+      Files.walkFileTree(
+          getRootDir(),
+          new SimpleFileVisitor<Path>() {
+            @Override
+            public FileVisitResult postVisitDirectory(Path dir, IOException exc)
+                throws IOException {
+              if (!dir.equals(getRootDir())) Files.delete(dir);
+              return FileVisitResult.CONTINUE;
+            }
+
+            @Override
+            public FileVisitResult visitFile(Path file, BasicFileAttributes attrs)
+                throws IOException {
+              Files.deleteIfExists(file);
+              return FileVisitResult.CONTINUE;
+            }
+          });
+    } catch (Throwable t) {
+      t.printStackTrace();
+    }
+  }
+
+  @Override
+  public void generateFiles(int numFiles, int pageSize, String basePath) {
+    String path = basePath.endsWith("/") ? basePath.substring(0, basePath.length() - 1) : basePath;
+    int i = 1;
+    while (i <= numFiles) {
+      int j = i / pageSize;
+      String subDir = (j > 0) ? "dir_" + j + "/" : "";
+      putFileOnFtp(path + "/" + subDir + "sample_" + i);
+      i++;
+    }
+  }
+
+  @Override
+  public void putFileOnFtp(String filePath) {
+    putFileOnFtpWithContents(filePath, getDefaultContent().getBytes());
+  }
+
+  @Override
+  public void putFileOnFtpWithContents(String filePath, byte[] fileContents) {
+    String relativePath = filePath.startsWith("/") ? filePath.substring(1) : filePath;
+    try {
+      File parent = getRootDir().resolve(relativePath).getParent().toFile();
+      if (!parent.exists()) {
+        parent.mkdirs();
+      }
+      Path path = getRootDir().resolve(relativePath);
+      path.toFile().createNewFile();
+      Files.write(path, fileContents, StandardOpenOption.SYNC);
+    } catch (Throwable t) {
+      throw new RuntimeException(t);
+    }
+  }
+
+  @Override
+  public byte[] getFtpFileContents(String filePath) {
+    String relativePath = filePath.startsWith("/") ? filePath.substring(1) : filePath;
+    try {
+      Path path = getRootDir().resolve(relativePath);
+      return Files.readAllBytes(path);
+    } catch (Throwable t) {
+      throw new RuntimeException(t);
+    }
+  }
+
+  @Override
+  public boolean fileExists(String filePath) {
+    String relativePath = filePath.startsWith("/") ? filePath.substring(1) : filePath;
+    try {
+      return getRootDir().resolve(relativePath).toFile().exists();
+    } catch (Throwable t) {
+      throw new RuntimeException(t);
+    }
+  }
+
+  @Override
+  public String getDefaultContent() {
+    return loremIpsum;
+  }
+
+  @Override
+  public ActorSystem getSystem() {
+    return system;
+  }
+
+  @Override
+  public Materializer getMaterializer() {
+    return materializer;
+  }
+}