Fix mutual authentication

Last version from demo.

Change-Id: Ib41581bf6f9eb92a03edf8434261d3674b6e3e39
Issue-ID: DCAEGEN2-860
Signed-off-by: elinuxhenrik <henrik.b.andersson@est.tech>
Signed-off-by: Vijay Venkatesh Kumar <vv770d@att.com>

27 files changed, 1620 insertions, 32 deletions

diff --git a/datafile-dmaap-client/pom.xml b/datafile-dmaap-client/pom.xml
index 9f60f8da..21839ab1 100644
--- a/datafile-dmaap-client/pom.xml
+++ b/datafile-dmaap-client/pom.xml
@@ -19,12 +19,13 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0"
   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
   xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
   <parent>
     <groupId>org.onap.dcaegen2.collectors</groupId>
     <artifactId>datafile</artifactId>
     <version>1.0.0-SNAPSHOT</version>
   </parent>
-  <modelVersion>4.0.0</modelVersion>
 
   <groupId>org.onap.dcaegen2.collectors.datafile</groupId>
   <artifactId>datafile-dmaap-client</artifactId>
@@ -39,6 +40,11 @@
 
     <!-- DEVELOPMENT DEPENDENCIES -->
     <dependency>
+      <groupId>org.onap.dcaegen2.collectors.datafile</groupId>
+      <artifactId>datafile-commons</artifactId>
+      <version>${project.parent.version}</version>
+    </dependency>
+    <dependency>
       <groupId>org.immutables</groupId>
       <artifactId>value</artifactId>
     </dependency>
@@ -59,11 +65,6 @@
       <artifactId>httpclient</artifactId>
     </dependency>
     <dependency>
-      <groupId>org.onap.dcaegen2.collectors.datafile</groupId>
-      <artifactId>datafile-commons</artifactId>
-      <version>1.0.0-SNAPSHOT</version>
-    </dependency>
-    <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-reactor-netty</artifactId>
     </dependency>
@@ -72,6 +73,14 @@
       <artifactId>commons-io</artifactId>
     </dependency>
     <dependency>
+      <groupId>com.jcraft</groupId>
+      <artifactId>jsch</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>commons-net</groupId>
+      <artifactId>commons-net</artifactId>
+    </dependency>
+    <dependency>
       <groupId>org.junit.jupiter</groupId>
       <artifactId>junit-jupiter-api</artifactId>
       <scope>test</scope>
@@ -114,6 +123,16 @@
       <scope>test</scope>
     </dependency>
     <dependency>
+      <groupId>com.github.stefanbirkner</groupId>
+      <artifactId>fake-sftp-server-rule</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-test</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
       <groupId>com.google.code.findbugs</groupId>
       <artifactId>jsr305</artifactId>
       <version>2.0.1</version>
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FTPSClientWrapper.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FTPSClientWrapper.java
new file mode 100644
index 00000000..77048042
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FTPSClientWrapper.java
@@ -0,0 +1,92 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ *      http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ftp;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.TrustManager;
+
+import org.apache.commons.net.ftp.FTPSClient;
+
+public class FTPSClientWrapper implements IFTPSClient {
+    private FTPSClient ftpsClient = new FTPSClient();
+
+    @Override
+    public void setNeedClientAuth(boolean isNeedClientAuth) {
+        ftpsClient.setNeedClientAuth(isNeedClientAuth);
+    }
+
+    @Override
+    public void setKeyManager(KeyManager keyManager) {
+        ftpsClient.setKeyManager(keyManager);
+    }
+
+    @Override
+    public void setTrustManager(TrustManager trustManager) {
+        ftpsClient.setTrustManager(trustManager);
+    }
+
+    @Override
+    public void connect(String hostName, int port) throws IOException {
+        ftpsClient.connect(hostName, port);
+    }
+
+    @Override
+    public boolean login(String username, String password) throws IOException {
+        return ftpsClient.login(username, password);
+    }
+
+    @Override
+    public boolean logout() throws IOException {
+        return ftpsClient.logout();
+    }
+
+    @Override
+    public int getReplyCode() {
+        return ftpsClient.getReplyCode();
+    }
+
+    @Override
+    public void disconnect() throws IOException {
+        ftpsClient.disconnect();
+    }
+
+    @Override
+    public void enterLocalPassiveMode() {
+        ftpsClient.enterLocalPassiveMode();
+    }
+
+    @Override
+    public void execPBSZ(int psbz) throws IOException {
+        ftpsClient.execPBSZ(psbz);
+    }
+
+    @Override
+    public void execPROT(String prot) throws IOException {
+        ftpsClient.execPROT(prot);
+    }
+
+    @Override
+    public boolean retrieveFile(String remote, OutputStream local) throws IOException {
+        return ftpsClient.retrieveFile(remote, local);
+    }
+
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FileServerData.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FileServerData.java
new file mode 100644
index 00000000..d4eca4d7
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FileServerData.java
@@ -0,0 +1,31 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ * http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ftp;
+
+import org.immutables.value.Value;
+
+/**
+ * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
+ *
+ */
+@Value.Immutable
+public interface FileServerData {
+    public String serverAddress();
+    public String userId();
+    public String password();
+    public int port();
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClient.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClient.java
new file mode 100644
index 00000000..719013ea
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClient.java
@@ -0,0 +1,279 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ * http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ftp;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.security.GeneralSecurityException;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+
+import org.apache.commons.net.ftp.FTPReply;
+import org.onap.dcaegen2.collectors.datafile.io.FileSystemResourceWrapper;
+import org.onap.dcaegen2.collectors.datafile.io.FileWrapper;
+import org.onap.dcaegen2.collectors.datafile.io.IFile;
+import org.onap.dcaegen2.collectors.datafile.io.IFileSystemResource;
+import org.onap.dcaegen2.collectors.datafile.io.IOutputStream;
+import org.onap.dcaegen2.collectors.datafile.io.OutputStreamWrapper;
+import org.onap.dcaegen2.collectors.datafile.ssl.IKeyManagerUtils;
+import org.onap.dcaegen2.collectors.datafile.ssl.IKeyStore;
+import org.onap.dcaegen2.collectors.datafile.ssl.ITrustManagerFactory;
+import org.onap.dcaegen2.collectors.datafile.ssl.KeyManagerUtilsWrapper;
+import org.onap.dcaegen2.collectors.datafile.ssl.KeyStoreWrapper;
+import org.onap.dcaegen2.collectors.datafile.ssl.TrustManagerFactoryWrapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+/**
+ * Gets file from xNF with FTPS protocol.
+ *
+ * @author <a href="mailto:martin.c.yan@est.tech">Martin Yan</a>
+ */
+@Component
+public class FtpsClient {
+    private static final Logger logger = LoggerFactory.getLogger(FtpsClient.class);
+
+    private String keyCertPath;
+    private String keyCertPassword;
+    private String trustedCAPath;
+    private String trustedCAPassword;
+
+    private IFTPSClient realFtpsClient;
+    private IKeyManagerUtils kmu;
+    private IKeyStore keyStore;
+    private ITrustManagerFactory trustManagerFactory;
+    private IFile localFile;
+    private IFileSystemResource fileResource;
+    private IOutputStream os;
+
+    public boolean collectFile(FileServerData fileServerData, String remoteFile, String localFile) {
+        logger.trace("collectFile called with fileServerData: {}, remoteFile: {}, localFile: {}", fileServerData,
+                remoteFile, localFile);
+        boolean result = true;
+        IFTPSClient ftps = getFtpsClient();
+
+        ftps.setNeedClientAuth(true);
+
+        if (setUpKeyManager(ftps) && setUpTrustedCA(ftps) && setUpConnection(fileServerData, ftps)) {
+            result = getFileFromxNF(remoteFile, localFile, ftps, fileServerData);
+
+            closeDownConnection(ftps);
+        } else {
+            result = false;
+        }
+        logger.trace("collectFile left with result: {}", result);
+        return result;
+    }
+
+    private boolean setUpKeyManager(IFTPSClient ftps) {
+        boolean result = true;
+        try {
+            IKeyManagerUtils keyManagerUtils = getKeyManagerUtils();
+            keyManagerUtils.setCredentials(keyCertPath, keyCertPassword);
+            ftps.setKeyManager(keyManagerUtils.getClientKeyManager());
+        } catch (GeneralSecurityException | IOException e) {
+            logger.error("Unable to use own key store {}", keyCertPath, e);
+            result = false;
+        }
+        return result;
+    }
+
+    private boolean setUpTrustedCA(IFTPSClient ftps) {
+        boolean result = true;
+        try {
+            IFileSystemResource fileSystemResource = getFileSystemResource();
+            fileSystemResource.setPath(trustedCAPath);
+            InputStream fis = fileSystemResource.getInputStream();
+            IKeyStore ks = getKeyStore();
+            ks.load(fis, trustedCAPassword.toCharArray());
+            fis.close();
+            ITrustManagerFactory tmf = getTrustManagerFactory();
+            tmf.init(ks.getKeyStore());
+            ftps.setTrustManager(tmf.getTrustManagers()[0]);
+
+        } catch (Exception e) {
+            logger.error("Unable to trust xNF's CA, {}", trustedCAPath, e);
+            result = false;
+        }
+        return result;
+    }
+
+    private boolean setUpConnection(FileServerData fileServerData, IFTPSClient ftps) {
+        boolean result = true;
+        try {
+            ftps.connect(fileServerData.serverAddress(), fileServerData.port());
+
+            boolean loginSuccesful = ftps.login(fileServerData.userId(), fileServerData.password());
+            if (!loginSuccesful) {
+                ftps.logout();
+                logger.error("Unable to log in to xNF. {}", fileServerData);
+                result = false;
+            }
+
+            if (loginSuccesful && FTPReply.isPositiveCompletion(ftps.getReplyCode())) {
+                ftps.enterLocalPassiveMode();
+                // Set protection buffer size
+                ftps.execPBSZ(0);
+                // Set data channel protection to private
+                ftps.execPROT("P");
+            } else {
+                ftps.disconnect();
+                logger.error("Unable to connect to xNF. {}", fileServerData);
+                result = false;
+            }
+        } catch (Exception ex) {
+            logger.error("Unable to connect to xNF. Data: {}", fileServerData, ex);
+            result = false;
+        }
+        logger.trace("setUpConnection return value: {}", result);
+        return result;
+    }
+
+    private boolean getFileFromxNF(String remoteFile, String localFilePath, IFTPSClient ftps,
+            FileServerData fileServerData) {
+        logger.trace("starting to getFile");
+        boolean result = true;
+        try {
+            IFile outfile = getFile();
+            outfile.setPath(localFilePath);
+            outfile.createNewFile();
+
+            IOutputStream outputStream = getOutputStream();
+            OutputStream output = outputStream.getOutputStream(outfile.getFile());
+
+            ftps.retrieveFile(remoteFile, output);
+
+            output.close();
+            logger.debug("File {} Download Successfull from xNF", localFilePath);
+        } catch (IOException ex) {
+            logger.error("Unable to collect file from xNF. Data: {}", fileServerData, ex);
+            result = false;
+        }
+        return result;
+    }
+
+    private void closeDownConnection(IFTPSClient ftps) {
+        logger.trace("starting to closeDownConnection");
+        try {
+            if (ftps != null) {
+                ftps.logout();
+                ftps.disconnect();
+            }
+        } catch (Exception e) {
+            // Do nothing, file has been collected.
+        }
+    }
+
+    public void setKeyCertPath(String keyCertPath) {
+        this.keyCertPath = keyCertPath;
+    }
+
+    public void setKeyCertPassword(String keyCertPassword) {
+        this.keyCertPassword = keyCertPassword;
+    }
+
+    public void setTrustedCAPath(String trustedCAPath) {
+        this.trustedCAPath = trustedCAPath;
+    }
+
+    public void setTrustedCAPassword(String trustedCAPassword) {
+        this.trustedCAPassword = trustedCAPassword;
+    }
+
+    private ITrustManagerFactory getTrustManagerFactory() throws NoSuchAlgorithmException {
+        if (trustManagerFactory == null) {
+            trustManagerFactory = new TrustManagerFactoryWrapper();
+        }
+        return trustManagerFactory;
+    }
+
+    private IFTPSClient getFtpsClient() {
+        if (realFtpsClient == null) {
+            realFtpsClient = new FTPSClientWrapper();
+        }
+        return realFtpsClient;
+    }
+
+    private IKeyManagerUtils getKeyManagerUtils() {
+        if (kmu == null) {
+            kmu = new KeyManagerUtilsWrapper();
+        }
+
+        return kmu;
+    }
+
+    private IKeyStore getKeyStore() throws KeyStoreException {
+        if (keyStore == null) {
+            keyStore = new KeyStoreWrapper();
+        }
+
+        return keyStore;
+    }
+
+    private IFile getFile() {
+        if (localFile == null) {
+            localFile = new FileWrapper();
+        }
+
+        return localFile;
+    }
+
+    private IOutputStream getOutputStream() {
+        if (os == null) {
+            os = new OutputStreamWrapper();
+        }
+
+        return os;
+    }
+
+    private IFileSystemResource getFileSystemResource() {
+        if (fileResource == null) {
+            fileResource = new FileSystemResourceWrapper();
+        }
+        return fileResource;
+    }
+
+    protected void setFtpsClient(IFTPSClient ftpsClient) {
+        this.realFtpsClient = ftpsClient;
+    }
+
+    protected void setKeyManagerUtils(IKeyManagerUtils keyManagerUtils) {
+        this.kmu = keyManagerUtils;
+    }
+
+    protected void setKeyStore(IKeyStore keyStore) {
+        this.keyStore = keyStore;
+    }
+
+    protected void setTrustManagerFactory(ITrustManagerFactory tmf) {
+        trustManagerFactory = tmf;
+    }
+
+    protected void setFile(IFile file) {
+        localFile = file;
+    }
+
+    protected void setOutputStream(IOutputStream outputStream) {
+        os = outputStream;
+    }
+
+    protected void setFileSystemResource(IFileSystemResource fileSystemResource) {
+        fileResource = fileSystemResource;
+    }
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/IFTPSClient.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/IFTPSClient.java
new file mode 100644
index 00000000..2f370b98
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/IFTPSClient.java
@@ -0,0 +1,51 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ *      http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ftp;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.TrustManager;
+
+public interface IFTPSClient {
+    public void setNeedClientAuth(boolean isNeedClientAuth);
+
+    public void setKeyManager(KeyManager keyManager);
+
+    public void setTrustManager(TrustManager trustManager);
+
+    public void connect(String hostname, int port) throws IOException;
+
+    public boolean login(String username, String password) throws IOException;
+
+    public boolean logout() throws IOException;
+
+    public int getReplyCode();
+
+    public void disconnect() throws IOException;
+
+    public void enterLocalPassiveMode();
+
+    public void execPBSZ(int newParam) throws IOException;
+
+    public void execPROT(String prot) throws IOException;
+
+    public boolean retrieveFile(String remote, OutputStream local) throws IOException;
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClient.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClient.java
new file mode 100644
index 00000000..5bd95b16
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClient.java
@@ -0,0 +1,94 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ * http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ftp;
+
+import com.jcraft.jsch.Channel;
+import com.jcraft.jsch.ChannelSftp;
+import com.jcraft.jsch.JSch;
+import com.jcraft.jsch.JSchException;
+import com.jcraft.jsch.Session;
+import com.jcraft.jsch.SftpException;
+
+import org.apache.commons.io.FilenameUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+/**
+ * Gets file from xNF with SFTP protocol.
+ *
+ * @author <a href="mailto:martin.c.yan@est.tech">Martin Yan</a>
+ *
+ */
+@Component
+public class SftpClient {
+    private static final Logger logger = LoggerFactory.getLogger(SftpClient.class);
+
+    public boolean collectFile(FileServerData fileServerData, String remoteFile, String localFile) {
+        boolean result = true;
+        Session session = setUpSession(fileServerData);
+
+        if (session != null) {
+            ChannelSftp sftpChannel = getChannel(session, fileServerData);
+            if (sftpChannel != null) {
+                try {
+                    sftpChannel.get(remoteFile, localFile);
+                    logger.debug("File {} Download Successfull from xNF", FilenameUtils.getName(localFile));
+                } catch (SftpException e) {
+                    logger.error("Unable to get file from xNF. Data: {}", fileServerData, e);
+                    result = false;
+                }
+
+                sftpChannel.exit();
+            } else {
+                result = false;
+            }
+            session.disconnect();
+        } else {
+            result = false;
+        }
+        return result;
+    }
+
+    private Session setUpSession(FileServerData fileServerData) {
+        JSch jsch = new JSch();
+
+        Session session = null;
+        try {
+            session = jsch.getSession(fileServerData.userId(), fileServerData.serverAddress(), fileServerData.port());
+            session.setConfig("StrictHostKeyChecking", "no");
+            session.setPassword(fileServerData.password());
+            session.connect();
+        } catch (JSchException e) {
+            logger.error("Unable to set up SFTP connection to xNF. Data: {}", fileServerData, e);
+        }
+        return session;
+    }
+
+    private ChannelSftp getChannel(Session session, FileServerData fileServerData) {
+        ChannelSftp sftpChannel = null;
+        try {
+            Channel channel;
+            channel = session.openChannel("sftp");
+            channel.connect();
+            sftpChannel = (ChannelSftp) channel;
+        } catch (JSchException e) {
+            logger.error("Unable to get sftp channel to xNF. Data: {}", fileServerData, e);
+        }
+        return sftpChannel;
+    }
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/FileSystemResourceWrapper.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/FileSystemResourceWrapper.java
new file mode 100644
index 00000000..fa794e90
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/FileSystemResourceWrapper.java
@@ -0,0 +1,40 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ *      http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.io;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.springframework.core.io.FileSystemResource;
+
+/**
+ * @author
+ *
+ */
+public class FileSystemResourceWrapper implements IFileSystemResource {
+    private FileSystemResource realResource;
+
+    public void setPath(String path) {
+        realResource = new FileSystemResource(path);
+    }
+    @Override
+    public InputStream getInputStream() throws IOException {
+        return realResource.getInputStream();
+    }
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/FileWrapper.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/FileWrapper.java
new file mode 100644
index 00000000..f8c02f09
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/FileWrapper.java
@@ -0,0 +1,44 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ *      http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.io;
+
+import java.io.File;
+import java.io.IOException;
+
+public class FileWrapper implements IFile {
+    private File file;
+
+    @Override
+    public void setPath(String path) {
+        file = new File(path);
+    }
+
+    @Override
+    public boolean createNewFile() throws IOException {
+        if (file == null) {
+            throw new IOException("Path to file not set.");
+        }
+        return file.createNewFile();
+    }
+
+    @Override
+    public File getFile() {
+        return file;
+    }
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/IFile.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/IFile.java
new file mode 100644
index 00000000..47d868a0
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/IFile.java
@@ -0,0 +1,30 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ *      http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.io;
+
+import java.io.File;
+import java.io.IOException;
+
+public interface IFile {
+    public void setPath(String path);
+
+    public boolean createNewFile() throws IOException;
+
+    public File getFile();
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/IFileSystemResource.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/IFileSystemResource.java
new file mode 100644
index 00000000..99fe13ea
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/IFileSystemResource.java
@@ -0,0 +1,31 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ * http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.io;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * @author
+ *
+ */
+public interface IFileSystemResource {
+
+    public void setPath(String filePath);
+
+    public InputStream getInputStream() throws IOException;
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/IOutputStream.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/IOutputStream.java
new file mode 100644
index 00000000..cb9d8572
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/IOutputStream.java
@@ -0,0 +1,27 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ *      http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.io;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.OutputStream;
+
+public interface IOutputStream {
+    public OutputStream getOutputStream(File file) throws FileNotFoundException;
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/OutputStreamWrapper.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/OutputStreamWrapper.java
new file mode 100644
index 00000000..830a571c
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/io/OutputStreamWrapper.java
@@ -0,0 +1,33 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ *      http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.io;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.OutputStream;
+
+public class OutputStreamWrapper implements IOutputStream {
+
+    @Override
+    public OutputStream getOutputStream(File file) throws FileNotFoundException {
+        return new FileOutputStream(file);
+    }
+
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/DmaapReactiveWebClient.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/DmaapReactiveWebClient.java
index 095ba8c9..7249c083 100644
--- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/DmaapReactiveWebClient.java
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/DmaapReactiveWebClient.java
@@ -2,28 +2,29 @@
  * ============LICENSE_START======================================================================
  * Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved.
  * ===============================================================================================
- * 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
+ * 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
  *
- *      http://www.apache.org/licenses/LICENSE-2.0
+ * http://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.
+ * 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.
  * ============LICENSE_END========================================================================
  */
 
 package org.onap.dcaegen2.collectors.datafile.service;
 
+import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication;
+
 import org.onap.dcaegen2.collectors.datafile.config.DmaapCustomConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.http.HttpHeaders;
 import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
 import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.reactive.function.client.WebClient.Builder;
 
 import reactor.core.publisher.Mono;
 
@@ -35,6 +36,8 @@ public class DmaapReactiveWebClient {
     private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
     private String dmaaPContentType;
+    private String dmaaPUserName;
+    private String dmaaPUserPassword;
 
     /**
      * Creating DmaapReactiveWebClient passing to them basic DmaapConfig.
@@ -53,25 +56,29 @@ public class DmaapReactiveWebClient {
      * @return WebClient
      */
     public WebClient build() {
-        return WebClient.builder()
-            .defaultHeader(HttpHeaders.CONTENT_TYPE, dmaaPContentType)
-            .filter(logRequest())
-            .filter(logResponse())
-            .build();
+        Builder webClientBuilder = WebClient.builder().defaultHeader(HttpHeaders.CONTENT_TYPE, dmaaPContentType)
+                .filter(logRequest()).filter(logResponse());
+        if (dmaaPUserName != null && !dmaaPUserName.isEmpty() && dmaaPUserPassword != null
+                && !dmaaPUserPassword.isEmpty()) {
+            webClientBuilder.filter(basicAuthentication(dmaaPUserName, dmaaPUserPassword));
+
+        }
+        return webClientBuilder.build();
     }
 
     private ExchangeFilterFunction logResponse() {
         return ExchangeFilterFunction.ofResponseProcessor(clientResponse -> {
-            logger.info("Response Status {}", clientResponse.statusCode());
+            logger.trace("Response Status {}", clientResponse.statusCode());
             return Mono.just(clientResponse);
         });
     }
 
     private ExchangeFilterFunction logRequest() {
         return ExchangeFilterFunction.ofRequestProcessor(clientRequest -> {
-            logger.info("Request: {} {}", clientRequest.method(), clientRequest.url());
+            logger.trace("Request: {} {}", clientRequest.method(), clientRequest.url());
             clientRequest.headers()
-                .forEach((name, values) -> values.forEach(value -> logger.info("{}={}", name, value)));
+                    .forEach((name, values) -> values.forEach(value -> logger.info("{}={}", name, value)));
+            logger.trace("HTTP request headers: {}", clientRequest.headers());
             return Mono.just(clientRequest);
         });
     }
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClient.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClient.java
index 5cd9056d..b4c52693 100644
--- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClient.java
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClient.java
@@ -30,8 +30,12 @@ import java.security.NoSuchAlgorithmException;
 import org.apache.commons.codec.binary.Base64;
 import org.apache.commons.io.IOUtils;
 import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration;
+import org.onap.dcaegen2.collectors.datafile.io.FileSystemResourceWrapper;
+import org.onap.dcaegen2.collectors.datafile.io.IFileSystemResource;
 import org.onap.dcaegen2.collectors.datafile.model.CommonFunctions;
 import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
+import org.onap.dcaegen2.collectors.datafile.web.IRestTemplate;
+import org.onap.dcaegen2.collectors.datafile.web.RestTemplateWrapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.http.HttpEntity;
@@ -99,7 +103,9 @@ public class DmaapProducerReactiveHttpClient {
 
             addUserCredentialsToHead(headers);
 
-            InputStream fileInputStream = getInputStream(consumerDmaapModel.getLocation());
+            IFileSystemResource fileSystemResource = getFileSystemResource();
+            fileSystemResource.setPath(consumerDmaapModel.getLocation());
+            InputStream fileInputStream = fileSystemResource.getInputStream();
             HttpEntity<byte[]> request = addFileToRequest(fileInputStream, headers);
 
 
@@ -129,19 +135,11 @@ public class DmaapProducerReactiveHttpClient {
         metaData.getAsJsonObject().remove(LOCATION_JSON_TAG);
         headers.set(X_ATT_DR_META, metaData.toString());
     }
-
     private HttpEntity<byte[]> addFileToRequest(InputStream inputStream, HttpHeaders headers)
             throws IOException {
         return new HttpEntity<>(IOUtils.toByteArray(inputStream), headers);
     }
 
-    private InputStream getInputStream(String filePath) throws IOException {
-        if (fileResource == null) {
-            fileResource = new FileSystemResourceWrapper(filePath);
-        }
-        return fileResource.getInputStream();
-    }
-
     private IRestTemplate getRestTemplate() throws NoSuchAlgorithmException, KeyManagementException, KeyStoreException {
         if (restTemplate == null) {
             restTemplate = new RestTemplateWrapper();
@@ -155,6 +153,13 @@ public class DmaapProducerReactiveHttpClient {
                 .path(path).build();
     }
 
+    private IFileSystemResource getFileSystemResource() {
+        if (fileResource == null) {
+            fileResource = new FileSystemResourceWrapper();
+        }
+        return fileResource;
+    }
+
     protected void setFileSystemResource(IFileSystemResource fileSystemResource) {
         fileResource = fileSystemResource;
     }
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/IKeyManagerUtils.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/IKeyManagerUtils.java
new file mode 100644
index 00000000..38ea681c
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/IKeyManagerUtils.java
@@ -0,0 +1,34 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ *      http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ssl;
+
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+
+import javax.net.ssl.KeyManager;
+
+/**
+ * @author
+ *
+ */
+public interface IKeyManagerUtils {
+    public void setCredentials(String keyStorePath, String keyStorePass) throws IOException, GeneralSecurityException;
+
+    public KeyManager getClientKeyManager();
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/IKeyStore.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/IKeyStore.java
new file mode 100644
index 00000000..01fb6c5b
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/IKeyStore.java
@@ -0,0 +1,31 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ * http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ssl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.cert.CertificateException;
+
+public interface IKeyStore {
+    public void load(InputStream arg0, char[] arg1)
+            throws IOException, NoSuchAlgorithmException, CertificateException, KeyStoreException;
+
+    public KeyStore getKeyStore();
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/ITrustManagerFactory.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/ITrustManagerFactory.java
new file mode 100644
index 00000000..99e3de1f
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/ITrustManagerFactory.java
@@ -0,0 +1,30 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ *      http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ssl;
+
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+
+import javax.net.ssl.TrustManager;
+
+public interface ITrustManagerFactory {
+    public void init(KeyStore ks) throws KeyStoreException;
+
+    public TrustManager[] getTrustManagers();
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/KeyManagerUtilsWrapper.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/KeyManagerUtilsWrapper.java
new file mode 100644
index 00000000..9eeaa923
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/KeyManagerUtilsWrapper.java
@@ -0,0 +1,45 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ *      http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ssl;
+
+import java.io.File;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+
+import javax.net.ssl.KeyManager;
+
+import org.apache.commons.net.util.KeyManagerUtils;
+
+/**
+ * @author
+ *
+ */
+public class KeyManagerUtilsWrapper implements IKeyManagerUtils {
+    private KeyManager keyManager;
+
+    @Override
+    public void setCredentials(String keyStorePath, String keyStorePass) throws IOException, GeneralSecurityException {
+        keyManager = KeyManagerUtils.createClientKeyManager(new File(keyStorePath), keyStorePass);
+    }
+
+    @Override
+    public KeyManager getClientKeyManager() {
+        return keyManager;
+    }
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/KeyStoreWrapper.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/KeyStoreWrapper.java
new file mode 100644
index 00000000..0a6ff200
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/KeyStoreWrapper.java
@@ -0,0 +1,44 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ * http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ssl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.cert.CertificateException;
+
+public class KeyStoreWrapper implements IKeyStore {
+    private KeyStore keyStore;
+
+    public KeyStoreWrapper() throws KeyStoreException {
+        keyStore = KeyStore.getInstance("JKS");
+    }
+
+    @Override
+    public void load(InputStream stream, char[] password)
+            throws IOException, NoSuchAlgorithmException, CertificateException, KeyStoreException {
+        keyStore.load(stream, password);
+    }
+
+    @Override
+    public KeyStore getKeyStore() {
+        return keyStore;
+    }
+
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/TrustManagerFactoryWrapper.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/TrustManagerFactoryWrapper.java
new file mode 100644
index 00000000..f539634d
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ssl/TrustManagerFactoryWrapper.java
@@ -0,0 +1,45 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ *      http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ssl;
+
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+
+public class TrustManagerFactoryWrapper implements ITrustManagerFactory {
+    TrustManagerFactory tmf;
+
+    public TrustManagerFactoryWrapper() throws NoSuchAlgorithmException {
+        tmf = TrustManagerFactory.getInstance("SunX509");
+    }
+
+    @Override
+    public void init(KeyStore ks) throws KeyStoreException {
+        tmf.init(ks);
+    }
+
+    @Override
+    public TrustManager[] getTrustManagers() {
+        return tmf.getTrustManagers();
+    }
+
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/web/IRestTemplate.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/web/IRestTemplate.java
new file mode 100644
index 00000000..1102b54e
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/web/IRestTemplate.java
@@ -0,0 +1,32 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ * http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.web;
+
+import java.net.URI;
+
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.ResponseEntity;
+
+/**
+ * @author
+ *
+ */
+public interface IRestTemplate {
+    public ResponseEntity<String> exchange(URI url, HttpMethod method, HttpEntity<byte[]> requestEntity,
+            Class<String> responseType);
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/web/PublishRedirectStrategy.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/web/PublishRedirectStrategy.java
new file mode 100644
index 00000000..e002c284
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/web/PublishRedirectStrategy.java
@@ -0,0 +1,81 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ * http://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.
+ * ============LICENSE_END========================================================================
+ */
+package org.onap.dcaegen2.collectors.datafile.web;
+
+import java.net.URI;
+
+import org.apache.http.HttpRequest;
+import org.apache.http.HttpResponse;
+import org.apache.http.ProtocolException;
+import org.apache.http.annotation.Contract;
+import org.apache.http.annotation.ThreadingBehavior;
+import org.apache.http.client.methods.HttpDelete;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpHead;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.client.methods.RequestBuilder;
+import org.apache.http.impl.client.DefaultRedirectStrategy;
+import org.apache.http.protocol.HttpContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * PublishRedirectStrategy implementation
+ * that automatically redirects all HEAD, GET, POST, PUT, and DELETE requests.
+ * This strategy relaxes restrictions on automatic redirection of
+ * POST methods imposed by the HTTP specification.
+ *
+ */
+@Contract(threading = ThreadingBehavior.IMMUTABLE)
+public class PublishRedirectStrategy extends DefaultRedirectStrategy {
+
+    public static final PublishRedirectStrategy INSTANCE = new PublishRedirectStrategy();
+    private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+    /**
+     * Redirectable methods.
+     */
+    private static final String[] REDIRECT_METHODS = new String[] {
+        HttpPut.METHOD_NAME,
+        HttpGet.METHOD_NAME,
+        HttpPost.METHOD_NAME,
+        HttpHead.METHOD_NAME,
+        HttpDelete.METHOD_NAME
+    };
+
+    @Override
+    protected boolean isRedirectable(final String method) {
+        for (final String m: REDIRECT_METHODS) {
+            if (m.equalsIgnoreCase(method)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    @Override
+    public HttpUriRequest getRedirect(
+            final HttpRequest request,
+            final HttpResponse response,
+            final HttpContext context) throws ProtocolException {
+        final URI uri = getLocationURI(request, response, context);
+        logger.trace("getRedirect...: {}", request);
+        return RequestBuilder.copy(request).setUri(uri).build();
+    }
+
+}
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/web/RequestResponseLoggingInterceptor.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/web/RequestResponseLoggingInterceptor.java
new file mode 100644
index 00000000..15d459f8
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/web/RequestResponseLoggingInterceptor.java
@@ -0,0 +1,58 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ * http://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.
+ * ============LICENSE_END========================================================================
+ */
+package org.onap.dcaegen2.collectors.datafile.web;
+import java.io.IOException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.http.HttpRequest;
+import org.springframework.http.client.ClientHttpRequestExecution;
+import org.springframework.http.client.ClientHttpRequestInterceptor;
+import org.springframework.http.client.ClientHttpResponse;
+
+public class RequestResponseLoggingInterceptor implements ClientHttpRequestInterceptor {
+
+    private final Logger log = LoggerFactory.getLogger(this.getClass());
+
+    @Override
+    public ClientHttpResponse intercept(HttpRequest request, byte[] body, ClientHttpRequestExecution execution) throws IOException {
+        logRequest(request, body);
+        ClientHttpResponse response = execution.execute(request, body);
+        logResponse(response);
+        return response;
+    }
+
+    private void logRequest(HttpRequest request, byte[] body) throws IOException {
+        if (log.isDebugEnabled()) {
+            log.debug("===========================request begin================================================");
+            log.debug("URI         : {}", request.getURI());
+            log.debug("Method      : {}", request.getMethod());
+            log.debug("Headers     : {}", request.getHeaders());
+            log.debug("Request body: {}", new String(body, "UTF-8"));
+            log.debug("==========================request end================================================");
+        }
+    }
+
+    private void logResponse(ClientHttpResponse response) throws IOException {
+        if (log.isDebugEnabled()) {
+            log.debug("============================response begin==========================================");
+            log.debug("Status code  : {}", response.getStatusCode());
+            log.debug("Status text  : {}", response.getStatusText());
+            log.debug("Headers      : {}", response.getHeaders());
+            log.debug("=======================response end=================================================");
+        }
+    }
+}
\ No newline at end of file
diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/web/RestTemplateWrapper.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/web/RestTemplateWrapper.java
new file mode 100644
index 00000000..a1b42848
--- /dev/null
+++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/web/RestTemplateWrapper.java
@@ -0,0 +1,65 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ * http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.web;
+
+import java.net.URI;
+import java.security.KeyManagementException;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.util.Collections;
+
+import javax.net.ssl.SSLContext;
+
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.ssl.SSLContextBuilder;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.ResponseEntity;
+import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
+import org.springframework.web.client.RestTemplate;
+
+/**
+ * @author
+ *
+ */
+public class RestTemplateWrapper implements IRestTemplate {
+    private RestTemplate restTemplate;
+
+    public RestTemplateWrapper() throws KeyManagementException, NoSuchAlgorithmException, KeyStoreException {
+        SSLContext sslContext =
+                new SSLContextBuilder().loadTrustMaterial(null, (certificate, authType) -> true).build();
+        CloseableHttpClient httpClient =
+                HttpClients.custom().setSSLContext(sslContext).setSSLHostnameVerifier(new NoopHostnameVerifier())
+                        .setRedirectStrategy(new PublishRedirectStrategy()).build();
+
+        HttpComponentsClientHttpRequestFactory requestFactory = new HttpComponentsClientHttpRequestFactory();
+        requestFactory.setHttpClient(httpClient);
+
+        restTemplate = new RestTemplate(requestFactory);
+        restTemplate.setInterceptors(Collections.singletonList(new RequestResponseLoggingInterceptor()));
+
+    }
+
+    @Override
+    public ResponseEntity<String> exchange(URI url, HttpMethod method, HttpEntity<byte[]> requestEntity,
+            Class<String> responseType) {
+        return restTemplate.exchange(url,  method, requestEntity, responseType);
+    }
+
+}
diff --git a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClientTest.java b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClientTest.java
new file mode 100644
index 00000000..5d716a9b
--- /dev/null
+++ b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClientTest.java
@@ -0,0 +1,234 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ * http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ftp;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.security.GeneralSecurityException;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.TrustManager;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.collectors.datafile.io.IFile;
+import org.onap.dcaegen2.collectors.datafile.io.IFileSystemResource;
+import org.onap.dcaegen2.collectors.datafile.io.IOutputStream;
+import org.onap.dcaegen2.collectors.datafile.ssl.IKeyManagerUtils;
+import org.onap.dcaegen2.collectors.datafile.ssl.IKeyStore;
+import org.onap.dcaegen2.collectors.datafile.ssl.ITrustManagerFactory;
+import org.springframework.http.HttpStatus;
+
+public class FtpsClientTest {
+
+    private static final String REMOTE_FILE_PATH = "/dir/sample.txt";
+    private static final String LOCAL_FILE_PATH = "target/sample.txt";
+    private static final String XNF_ADDRESS = "127.0.0.1";
+    private static final int PORT = 8021;
+    private static final String FTP_KEY_PATH = "ftpKeyPath";
+    private static final String FTP_KEY_PASSWORD = "ftpKeyPassword";
+    private static final String TRUSTED_CA_PATH = "trustedCAPath";
+    private static final String TRUSTED_CA_PASSWORD = "trustedCAPassword";
+
+    private static final String USERNAME = "bob";
+    private static final String PASSWORD = "123";
+
+    private IFTPSClient ftpsClientMock = mock(IFTPSClient.class);
+    private IKeyManagerUtils keyManagerUtilsMock = mock(IKeyManagerUtils.class);
+    private KeyManager keyManagerMock = mock(KeyManager.class);
+    private IKeyStore keyStoreWrapperMock = mock(IKeyStore.class);
+    private KeyStore keyStoreMock = mock(KeyStore.class);
+    private ITrustManagerFactory trustManagerFactoryMock = mock(ITrustManagerFactory.class);
+    private TrustManager trustManagerMock = mock(TrustManager.class);
+    private IFile localFileMock = mock(IFile.class);
+    private IFileSystemResource fileResourceMock = mock(IFileSystemResource.class);
+    private IOutputStream outputStreamMock = mock(IOutputStream.class);
+    private InputStream inputStreamMock = mock(InputStream.class);
+
+    FtpsClient clientUnderTest = new FtpsClient();
+
+    @BeforeEach
+    protected void setUp() throws Exception {
+        clientUnderTest.setFtpsClient(ftpsClientMock);
+        clientUnderTest.setKeyManagerUtils(keyManagerUtilsMock);
+        clientUnderTest.setKeyStore(keyStoreWrapperMock);
+        clientUnderTest.setTrustManagerFactory(trustManagerFactoryMock);
+        clientUnderTest.setFile(localFileMock);
+        clientUnderTest.setFileSystemResource(fileResourceMock);
+        clientUnderTest.setOutputStream(outputStreamMock);
+
+        clientUnderTest.setKeyCertPath(FTP_KEY_PATH);
+        clientUnderTest.setKeyCertPassword(FTP_KEY_PASSWORD);
+        clientUnderTest.setTrustedCAPath(TRUSTED_CA_PATH);
+        clientUnderTest.setTrustedCAPassword(TRUSTED_CA_PASSWORD);
+}
+
+    @Test
+    public void collectFile_allOk() throws Exception {
+        when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
+        when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
+        when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
+        when(trustManagerFactoryMock.getTrustManagers()).thenReturn(new TrustManager[] {trustManagerMock});
+        when(ftpsClientMock.login(USERNAME, PASSWORD)).thenReturn(true);
+        when(ftpsClientMock.getReplyCode()).thenReturn(HttpStatus.OK.value());
+        File fileMock = mock(File.class);
+        when(localFileMock.getFile()).thenReturn(fileMock);
+        OutputStream osMock = mock(OutputStream.class);
+        when(outputStreamMock.getOutputStream(fileMock)).thenReturn(osMock);
+
+        ImmutableFileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(XNF_ADDRESS)
+                .userId(USERNAME).password(PASSWORD).port(PORT).build();
+
+        boolean result = clientUnderTest.collectFile(fileServerData, REMOTE_FILE_PATH, LOCAL_FILE_PATH);
+
+        assertTrue(result);
+        verify(ftpsClientMock).setNeedClientAuth(true);
+        verify(keyManagerUtilsMock).setCredentials(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+        verify(ftpsClientMock).setKeyManager(keyManagerMock);
+        verify(fileResourceMock).setPath(TRUSTED_CA_PATH);
+        verify(keyStoreWrapperMock).load(inputStreamMock, TRUSTED_CA_PASSWORD.toCharArray());
+        verify(inputStreamMock, times(1)).close();
+        verify(trustManagerFactoryMock).init(keyStoreMock);
+        verify(ftpsClientMock).setTrustManager(trustManagerMock);
+        verify(ftpsClientMock).connect(XNF_ADDRESS, PORT);
+        verify(ftpsClientMock).login(USERNAME, PASSWORD);
+        verify(ftpsClientMock).getReplyCode();
+        verify(ftpsClientMock, times(1)).enterLocalPassiveMode();
+        verify(ftpsClientMock).execPBSZ(0);
+        verify(ftpsClientMock).execPROT("P");
+        verify(localFileMock).setPath(LOCAL_FILE_PATH);
+        verify(localFileMock, times(1)).createNewFile();
+        verify(ftpsClientMock).retrieveFile(REMOTE_FILE_PATH, osMock);
+        verify(osMock, times(1)).close();
+        verify(ftpsClientMock, times(1)).logout();
+        verify(ftpsClientMock, times(1)).disconnect();
+        verifyNoMoreInteractions(ftpsClientMock);
+    }
+
+    @Test
+    public void collectFileFaultyOwnKey_shouldFail() throws Exception {
+        doThrow(new GeneralSecurityException())
+                .when(keyManagerUtilsMock).setCredentials(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+
+        ImmutableFileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(XNF_ADDRESS)
+                .userId(USERNAME).password(PASSWORD).port(PORT).build();
+
+        boolean result = clientUnderTest.collectFile(fileServerData, REMOTE_FILE_PATH, LOCAL_FILE_PATH);
+
+        assertFalse(result);
+    }
+
+    @Test
+    public void collectFileFaultTrustedCA_shouldFail() throws Exception {
+        when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
+        when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
+        when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
+
+        doThrow(new KeyStoreException()).when(trustManagerFactoryMock).init(keyStoreMock);
+
+        ImmutableFileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(XNF_ADDRESS)
+                .userId(USERNAME).password(PASSWORD).port(PORT).build();
+
+        boolean result = clientUnderTest.collectFile(fileServerData, REMOTE_FILE_PATH, LOCAL_FILE_PATH);
+
+        assertFalse(result);
+    }
+
+    @Test
+    public void collectFileFaultyLogin_shouldFail() throws Exception {
+        when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
+        when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
+        when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
+        when(trustManagerFactoryMock.getTrustManagers()).thenReturn(new TrustManager[] {trustManagerMock});
+        when(ftpsClientMock.login(USERNAME, PASSWORD)).thenReturn(false);
+
+        ImmutableFileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(XNF_ADDRESS)
+                .userId(USERNAME).password(PASSWORD).port(PORT).build();
+
+        boolean result = clientUnderTest.collectFile(fileServerData, REMOTE_FILE_PATH, LOCAL_FILE_PATH);
+
+        verify(ftpsClientMock, times(1)).logout();
+        assertFalse(result);
+    }
+
+    @Test
+    public void collectFileBadRequestResponse_shouldFail() throws Exception {
+        when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
+        when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
+        when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
+        when(trustManagerFactoryMock.getTrustManagers()).thenReturn(new TrustManager[] {trustManagerMock});
+        when(ftpsClientMock.login(USERNAME, PASSWORD)).thenReturn(true);
+        when(ftpsClientMock.getReplyCode()).thenReturn(HttpStatus.BAD_REQUEST.value());
+
+        ImmutableFileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(XNF_ADDRESS)
+                .userId(USERNAME).password(PASSWORD).port(PORT).build();
+
+        boolean result = clientUnderTest.collectFile(fileServerData, REMOTE_FILE_PATH, LOCAL_FILE_PATH);
+
+        verify(ftpsClientMock, times(1)).disconnect();
+        assertFalse(result);
+    }
+
+    @Test
+    public void collectFileFaultyConnection_shouldFail() throws Exception {
+        when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
+        when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
+        when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
+        when(trustManagerFactoryMock.getTrustManagers()).thenReturn(new TrustManager[] {trustManagerMock});
+
+        doThrow(new IOException()).when(ftpsClientMock).connect(XNF_ADDRESS, PORT);
+
+        ImmutableFileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(XNF_ADDRESS)
+                .userId(USERNAME).password(PASSWORD).port(PORT).build();
+
+        boolean result = clientUnderTest.collectFile(fileServerData, REMOTE_FILE_PATH, LOCAL_FILE_PATH);
+
+        assertFalse(result);
+    }
+
+    @Test
+    public void collectFileFailingFileCollect_shouldFail() throws Exception {
+        when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
+        when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
+        when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
+        when(trustManagerFactoryMock.getTrustManagers()).thenReturn(new TrustManager[] {trustManagerMock});
+        when(ftpsClientMock.login(USERNAME, PASSWORD)).thenReturn(true);
+        when(ftpsClientMock.getReplyCode()).thenReturn(HttpStatus.OK.value());
+
+        doThrow(new IOException()).when(localFileMock).createNewFile();
+
+        ImmutableFileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(XNF_ADDRESS)
+                .userId(USERNAME).password(PASSWORD).port(PORT).build();
+
+        boolean result = clientUnderTest.collectFile(fileServerData, REMOTE_FILE_PATH, LOCAL_FILE_PATH);
+
+        assertFalse(result);
+    }
+}
\ No newline at end of file
diff --git a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClientTest.java b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClientTest.java
new file mode 100644
index 00000000..13f1fbb9
--- /dev/null
+++ b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClientTest.java
@@ -0,0 +1,103 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * 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
+ *
+ *      http://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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ftp;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.commons.io.IOUtils.toByteArray;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.github.stefanbirkner.fakesftpserver.rule.FakeSftpServerRule;
+import com.jcraft.jsch.ChannelSftp;
+import com.jcraft.jsch.JSch;
+import com.jcraft.jsch.JSchException;
+import com.jcraft.jsch.Session;
+import com.jcraft.jsch.SftpException;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+
+import org.junit.Rule;
+import org.junit.Test;
+
+public class SftpClientTest {
+    private static final String USERNAME = "bob";
+    private static final String PASSWORD = "123";
+    private static final String DUMMY_CONTENT = "dummy content";
+    private static final String LOCAL_DUMMY_FILE = "target/dummy.txt";
+    private static final String REMOTE_DUMMY_FILE = "/dummy_directory/dummy_file.txt";
+    private static final JSch JSCH = new JSch();
+    private static final int TIMEOUT = 2000;
+
+    @Rule
+    public final FakeSftpServerRule sftpServer = new FakeSftpServerRule().addUser(USERNAME, PASSWORD);
+
+    @Test
+    public void collectFile_withOKresponse() throws IOException, JSchException, SftpException {
+        SftpClient sftpClient = new SftpClient();
+        sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
+        byte[] file = downloadFile(sftpServer, REMOTE_DUMMY_FILE);
+        FileServerData expectedFileServerData = ImmutableFileServerData.builder().serverAddress("127.0.0.1")
+                .userId(USERNAME).password(PASSWORD).port(sftpServer.getPort()).build();
+        sftpClient.collectFile(expectedFileServerData, REMOTE_DUMMY_FILE,
+                LOCAL_DUMMY_FILE);
+        byte[] localFile = Files.readAllBytes(new File(LOCAL_DUMMY_FILE).toPath());
+        assertThat(new String(file, UTF_8)).isEqualTo(DUMMY_CONTENT);
+        assertThat(new String(localFile, UTF_8)).isEqualTo(DUMMY_CONTENT);
+    }
+
+    private static Session connectToServer(FakeSftpServerRule sftpServer) throws JSchException {
+        return connectToServerAtPort(sftpServer.getPort());
+    }
+
+    private static Session connectToServerAtPort(int port) throws JSchException {
+        Session session = createSessionWithCredentials(USERNAME, PASSWORD, port);
+        session.connect(TIMEOUT);
+        return session;
+    }
+
+    private static ChannelSftp connectSftpChannel(Session session) throws JSchException {
+        ChannelSftp channel = (ChannelSftp) session.openChannel("sftp");
+        channel.connect();
+        return channel;
+    }
+
+    private static Session createSessionWithCredentials(String username, String password, int port)
+            throws JSchException {
+        Session session = JSCH.getSession(username, "127.0.0.1", port);
+        session.setConfig("StrictHostKeyChecking", "no");
+        session.setPassword(password);
+        return session;
+    }
+
+    private static byte[] downloadFile(FakeSftpServerRule server, String path)
+            throws JSchException, SftpException, IOException {
+        Session session = connectToServer(server);
+        ChannelSftp channel = connectSftpChannel(session);
+        try {
+            InputStream is = channel.get(path);
+            return toByteArray(is);
+        } finally {
+            channel.disconnect();
+            session.disconnect();
+        }
+    }
+
+}
diff --git a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClientTest.java b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClientTest.java
index 7e8ec734..ba424626 100644
--- a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClientTest.java
+++ b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClientTest.java
@@ -36,9 +36,11 @@ import org.apache.commons.io.IOUtils;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration;
+import org.onap.dcaegen2.collectors.datafile.io.IFileSystemResource;
 import org.onap.dcaegen2.collectors.datafile.model.CommonFunctions;
 import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
 import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModelForUnitTest;
+import org.onap.dcaegen2.collectors.datafile.web.IRestTemplate;
 import org.springframework.http.HttpEntity;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
@@ -123,6 +125,7 @@ class DmaapProducerReactiveHttpClientTest {
         fileStream.reset();
 
         HttpEntity<byte[]> requestEntity = new HttpEntity<>(IOUtils.toByteArray(fileStream), headers);
+        verify(fileSystemResourceMock).setPath("target/" + FILE_NAME);
         verify(restTemplateMock).exchange(expectedUri, HttpMethod.PUT, requestEntity, String.class);
         verifyNoMoreInteractions(restTemplateMock);
     }