Merge "Thread safety issues"

3 files changed, 151 insertions, 193 deletions

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
index 670b1bdc..e2882606 100644
--- 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
@@ -17,8 +17,11 @@
 package org.onap.dcaegen2.collectors.datafile.ftp;
 
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
@@ -29,23 +32,15 @@ import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.security.GeneralSecurityException;
-import java.security.KeyStore;
-import java.security.KeyStoreException;
 
 import javax.net.ssl.KeyManager;
 import javax.net.ssl.TrustManager;
 
 import org.apache.commons.net.ftp.FTP;
-import org.apache.commons.net.ftp.FTPReply;
 import org.apache.commons.net.ftp.FTPSClient;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentMatchers;
-import org.onap.dcaegen2.collectors.datafile.io.IFileSystemResource;
-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 {
@@ -62,18 +57,14 @@ public class FtpsClientTest {
     private static final String USERNAME = "bob";
     private static final String PASSWORD = "123";
 
+
     private FTPSClient ftpsClientMock = mock(FTPSClient.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 IFileSystemResource fileResourceMock = mock(IFileSystemResource.class);
     private InputStream inputStreamMock = mock(InputStream.class);
+    private OutputStream outputStreamMock = mock(OutputStream.class);
 
-    FtpsClient clientUnderTest = new FtpsClient(createFileServerData());
-
+    FtpsClient clientUnderTestSpy;
 
     private ImmutableFileServerData createFileServerData() {
         return ImmutableFileServerData.builder().serverAddress(XNF_ADDRESS).userId(USERNAME).password(PASSWORD)
@@ -82,40 +73,17 @@ public class FtpsClientTest {
 
     @BeforeEach
     protected void setUp() throws Exception {
-        clientUnderTest.setFtpsClient(ftpsClientMock);
-        clientUnderTest.setKeyManagerUtils(keyManagerUtilsMock);
-        clientUnderTest.setKeyStore(keyStoreWrapperMock);
-        clientUnderTest.setTrustManagerFactory(trustManagerFactoryMock);
-        clientUnderTest.setFileSystemResource(fileResourceMock);
-
-        clientUnderTest.setKeyCertPath(FTP_KEY_PATH);
-        clientUnderTest.setKeyCertPassword(FTP_KEY_PASSWORD);
-        clientUnderTest.setTrustedCAPath(TRUSTED_CA_PATH.toString());
-        clientUnderTest.setTrustedCAPassword(TRUSTED_CA_PASSWORD);
+        clientUnderTestSpy = spy(new FtpsClient(createFileServerData()));
+        clientUnderTestSpy.realFtpsClient = ftpsClientMock;
     }
 
-    @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());
-
-        when(ftpsClientMock.isConnected()).thenReturn(false, true);
+    private void verifyFtpsClientMock_openOK() throws Exception {
+        doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH);
+
         when(ftpsClientMock.retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH),
                 ArgumentMatchers.any(OutputStream.class))).thenReturn(true);
-
-        clientUnderTest.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH);
-
         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);
@@ -125,59 +93,85 @@ public class FtpsClientTest {
         verify(ftpsClientMock).execPROT("P");
         verify(ftpsClientMock).setFileType(FTP.BINARY_FILE_TYPE);
         verify(ftpsClientMock).setBufferSize(1024 * 1024);
-        verify(ftpsClientMock).retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH),
-                ArgumentMatchers.any(OutputStream.class));
+    }
+
+    @Test
+    public void collectFile_allOk() throws Exception {
+
+        doReturn(keyManagerMock).when(clientUnderTestSpy).createKeyManager(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+        doReturn(trustManagerMock).when(clientUnderTestSpy).getTrustManager(TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD);
+        doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH);
+        doReturn(true).when(ftpsClientMock).login(USERNAME, PASSWORD);
+        doReturn(HttpStatus.OK.value()).when(ftpsClientMock).getReplyCode();
+
+        clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD);
+
+        doReturn(true).when(ftpsClientMock).retrieveFile(REMOTE_FILE_PATH, outputStreamMock);
+        clientUnderTestSpy.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH);
+
+        doReturn(true).when(ftpsClientMock).isConnected();
+        clientUnderTestSpy.close();
+
+        verifyFtpsClientMock_openOK();
+        verify(ftpsClientMock, times(1)).isConnected();
         verify(ftpsClientMock, times(1)).logout();
         verify(ftpsClientMock, times(1)).disconnect();
-        verify(ftpsClientMock, times(2)).isConnected();
+        verify(ftpsClientMock, times(1)).retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH), any());
         verifyNoMoreInteractions(ftpsClientMock);
     }
 
     @Test
     public void collectFileFaultyOwnKey_shouldFail() throws Exception {
-        doThrow(new IKeyManagerUtils.KeyManagerException(new GeneralSecurityException())).when(keyManagerUtilsMock)
-                .setCredentials(FTP_KEY_PATH, FTP_KEY_PASSWORD);
-        when(ftpsClientMock.isConnected()).thenReturn(false);
 
-        assertThatThrownBy(() -> clientUnderTest.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH)).hasMessage(
-                "org.onap.dcaegen2.collectors.datafile.ssl.IKeyManagerUtils$KeyManagerException: java.security.GeneralSecurityException");
+        doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH);
+        assertThatThrownBy(() -> clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH,
+                TRUSTED_CA_PASSWORD)).hasMessageContaining(
+                        "Could not open connection: java.io.FileNotFoundException:");
 
         verify(ftpsClientMock).setNeedClientAuth(true);
-        verify(keyManagerUtilsMock).setCredentials(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+
+        doReturn(false).when(ftpsClientMock).isConnected();
+        clientUnderTestSpy.close();
         verify(ftpsClientMock).isConnected();
         verifyNoMoreInteractions(ftpsClientMock);
     }
 
     @Test
-    public void collectFileFaultTrustedCA_shouldFail() throws Exception {
-        when(keyManagerUtilsMock.getClientKeyManager()).thenReturn(keyManagerMock);
-        when(fileResourceMock.getInputStream()).thenReturn(inputStreamMock);
-        when(keyStoreWrapperMock.getKeyStore()).thenReturn(keyStoreMock);
+    public void collectFileFaultTrustedCA_shouldFail_no_trustedCA_file() throws Exception {
+
+        doReturn(keyManagerMock).when(clientUnderTestSpy).createKeyManager(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+        doThrow(new IOException("problem")).when(clientUnderTestSpy).createInputStream(TRUSTED_CA_PATH);
+
+        assertThatThrownBy(
+                () -> clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD))
+                        .hasMessage("Could not open connection: java.io.IOException: problem");
+    }
+
+    @Test
+    public void collectFileFaultTrustedCA_shouldFail_empty_trustedCA_file() throws Exception {
 
-        doThrow(new KeyStoreException()).when(trustManagerFactoryMock).init(keyStoreMock);
+        doReturn(keyManagerMock).when(clientUnderTestSpy).createKeyManager(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+        doReturn(inputStreamMock).when(clientUnderTestSpy).createInputStream(TRUSTED_CA_PATH);
 
-        assertThatThrownBy(() -> clientUnderTest.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH))
-                .hasMessage("Unable to trust xNF's CA, trustedCAPath java.security.KeyStoreException");
+        assertThatThrownBy(
+                () -> clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD))
+                        .hasMessage("Could not open connection: java.io.EOFException");
     }
 
     @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);
 
-        assertThatThrownBy(() -> clientUnderTest.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH))
-                .hasMessage("Unable to log in to xNF. 127.0.0.1");
+        doReturn(keyManagerMock).when(clientUnderTestSpy).createKeyManager(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+        doReturn(trustManagerMock).when(clientUnderTestSpy).getTrustManager(TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD);
+        doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH);
+        doReturn(false).when(ftpsClientMock).login(USERNAME, PASSWORD);
+
+        assertThatThrownBy(
+                () -> clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD))
+                        .hasMessage("Unable to log in to xNF. 127.0.0.1");
 
         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);
@@ -185,105 +179,61 @@ public class FtpsClientTest {
 
     @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(FTPReply.BAD_COMMAND_SEQUENCE);
+        doReturn(keyManagerMock).when(clientUnderTestSpy).createKeyManager(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+        doReturn(trustManagerMock).when(clientUnderTestSpy).getTrustManager(TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD);
+        doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH);
+        doReturn(true).when(ftpsClientMock).login(USERNAME, PASSWORD);
+        doReturn(503).when(ftpsClientMock).getReplyCode();
 
-        assertThatThrownBy(() -> clientUnderTest.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH))
-                .hasMessage("Unable to connect to xNF. 127.0.0.1 xNF reply code: 503");
+        assertThatThrownBy(
+                () -> clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD))
+                        .hasMessage("Unable to connect to xNF. 127.0.0.1 xNF reply code: 503");
 
         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, times(2)).getReplyCode();
-        verify(ftpsClientMock, times(2)).isConnected();
         verifyNoMoreInteractions(ftpsClientMock);
     }
 
     @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});
+    public void collectFile_shouldFail() throws Exception {
+        doReturn(keyManagerMock).when(clientUnderTestSpy).createKeyManager(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+        doReturn(trustManagerMock).when(clientUnderTestSpy).getTrustManager(TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD);
+        doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH);
+        doReturn(true).when(ftpsClientMock).login(USERNAME, PASSWORD);
+        doReturn(HttpStatus.OK.value()).when(ftpsClientMock).getReplyCode();
+        clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD);
 
-        doThrow(new IOException()).when(ftpsClientMock).connect(XNF_ADDRESS, PORT);
+        doReturn(false).when(ftpsClientMock).retrieveFile(REMOTE_FILE_PATH, outputStreamMock);
 
-        assertThatThrownBy(() -> clientUnderTest.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH))
-                .hasMessage("Could not open connection: java.io.IOException");
+        assertThatThrownBy(() -> clientUnderTestSpy.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH))
+                .hasMessage("Could not retrieve file /dir/sample.txt");
 
-        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, times(2)).isConnected();
+        verifyFtpsClientMock_openOK();
+        verify(ftpsClientMock, times(1)).retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH), any());
         verifyNoMoreInteractions(ftpsClientMock);
     }
 
     @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());
-
-        assertThatThrownBy(() -> clientUnderTest.collectFile(REMOTE_FILE_PATH, Paths.get("")))
-                .hasMessage("Could not open connection: java.io.IOException: No such file or directory");
-
-    }
-
-    @Test
-    public void collectFileFailingFileRetrieve_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());
+    public void collectFile_shouldFail_ioexception() throws Exception {
+        doReturn(keyManagerMock).when(clientUnderTestSpy).createKeyManager(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+        doReturn(trustManagerMock).when(clientUnderTestSpy).getTrustManager(TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD);
+        doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH);
+        doReturn(true).when(ftpsClientMock).login(USERNAME, PASSWORD);
+        doReturn(HttpStatus.OK.value()).when(ftpsClientMock).getReplyCode();
+        clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD);
         when(ftpsClientMock.isConnected()).thenReturn(false);
 
-        doThrow(new IOException("problemas")).when(ftpsClientMock).retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH),
-                ArgumentMatchers.any(OutputStream.class));
+        doThrow(new IOException("problem")).when(ftpsClientMock).retrieveFile(REMOTE_FILE_PATH, outputStreamMock);
 
-        assertThatThrownBy(() -> clientUnderTest.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH))
-                .hasMessage("Could not open connection: java.io.IOException: problemas");
+        assertThatThrownBy(() -> clientUnderTestSpy.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH))
+                .hasMessage("Could not fetch file: java.io.IOException: problem");
 
-        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(ftpsClientMock).setFileType(FTP.BINARY_FILE_TYPE);
-        verify(ftpsClientMock).setBufferSize(1024 * 1024);
-        verify(ftpsClientMock).retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH),
-                ArgumentMatchers.any(OutputStream.class));
-        verify(ftpsClientMock, times(2)).isConnected();
+        verifyFtpsClientMock_openOK();
+        verify(ftpsClientMock, times(1)).retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH), any());
         verifyNoMoreInteractions(ftpsClientMock);
     }
 }
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
index 90fb9336..85a0c090 100644
--- 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
@@ -20,7 +20,6 @@ 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 static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import com.github.stefanbirkner.fakesftpserver.rule.FakeSftpServerRule;
 import com.jcraft.jsch.ChannelSftp;
@@ -56,45 +55,45 @@ public class SftpClientTest {
             throws DatafileTaskException, IOException, JSchException, SftpException, Exception {
         FileServerData expectedFileServerData = ImmutableFileServerData.builder().serverAddress("127.0.0.1")
                 .userId(USERNAME).password(PASSWORD).port(sftpServer.getPort()).build();
-        SftpClient sftpClient = new SftpClient(expectedFileServerData);
-        sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
-        byte[] file = downloadFile(sftpServer, REMOTE_DUMMY_FILE);
-
-        sftpClient.collectFile(REMOTE_DUMMY_FILE, LOCAL_DUMMY_FILE);
-        byte[] localFile = Files.readAllBytes(LOCAL_DUMMY_FILE.toFile().toPath());
-        assertThat(new String(file, UTF_8)).isEqualTo(DUMMY_CONTENT);
-        assertThat(new String(localFile, UTF_8)).isEqualTo(DUMMY_CONTENT);
+        try (SftpClient sftpClient = new SftpClient(expectedFileServerData)) {
+            sftpClient.open();
+            sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
+            byte[] file = downloadFile(sftpServer, REMOTE_DUMMY_FILE);
+
+            sftpClient.collectFile(REMOTE_DUMMY_FILE, LOCAL_DUMMY_FILE);
+            byte[] localFile = Files.readAllBytes(LOCAL_DUMMY_FILE.toFile().toPath());
+            assertThat(new String(file, UTF_8)).isEqualTo(DUMMY_CONTENT);
+            assertThat(new String(localFile, UTF_8)).isEqualTo(DUMMY_CONTENT);
+        }
     }
 
     @Test
-    public void collectFile_withWrongUserName_shouldFail() throws IOException, JSchException, SftpException {
-        FileServerData expectedFileServerData =
-                ImmutableFileServerData.builder().serverAddress("127.0.0.1").userId("Wrong").password(PASSWORD).build();
-        SftpClient sftpClient = new SftpClient(expectedFileServerData);
-        sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
+    public void collectFile_withWrongUserName_shouldFail() throws DatafileTaskException, IOException {
+        FileServerData expectedFileServerData = ImmutableFileServerData.builder().serverAddress("127.0.0.1")
+                .userId("wrong").password(PASSWORD).port(sftpServer.getPort()).build();
+        try (SftpClient sftpClient = new SftpClient(expectedFileServerData)) {
+
+            sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
 
 
-        assertThatThrownBy(() -> sftpClient.collectFile(REMOTE_DUMMY_FILE, LOCAL_DUMMY_FILE))
-                .hasMessageContaining("Unable to get file from xNF");
+            assertThatThrownBy(() -> sftpClient.open())
+                    .hasMessageContaining("Could not open Sftp clientcom.jcraft.jsch.JSchException: Auth fail");
+        }
     }
 
     @Test
-    public void collectFile_withWrongFileName_shouldFail() throws IOException, JSchException, SftpException {
+    public void collectFile_withWrongFileName_shouldFail()
+            throws IOException, JSchException, SftpException, DatafileTaskException {
         FileServerData expectedFileServerData = ImmutableFileServerData.builder().serverAddress("127.0.0.1")
                 .userId(USERNAME).password(PASSWORD).port(sftpServer.getPort()).build();
-        SftpClient sftpClient = new SftpClient(expectedFileServerData);
-        sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
+        try (SftpClient sftpClient = new SftpClient(expectedFileServerData)) {
+            sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
+            sftpClient.open();
 
-        String errorMessage = "";
-        try {
-            sftpClient.collectFile("wrong", LOCAL_DUMMY_FILE);
-        } catch (Exception e) {
-            errorMessage = e.getMessage();
+            assertThatThrownBy(() -> sftpClient.collectFile("wrong", LOCAL_DUMMY_FILE)).hasMessageStartingWith(
+                    "Unable to get file from xNF. Data: FileServerData{serverAddress=127.0.0.1, "
+                            + "userId=bob, password=123, port=");
         }
-
-        String expectedErrorMessage = "Unable to get file from xNF. Data: FileServerData{serverAddress=127.0.0.1, "
-                + "userId=bob, password=123, port=";
-        assertTrue(errorMessage.startsWith(expectedErrorMessage));
     }
 
     private static Session connectToServer(FakeSftpServerRule sftpServer) throws JSchException {
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 2bbe8e1d..06ff570c 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
@@ -17,21 +17,30 @@
 package org.onap.dcaegen2.collectors.datafile.service.producer;
 
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
+
 import com.google.gson.JsonElement;
 import com.google.gson.JsonParser;
+
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Paths;
+import java.security.KeyManagementException;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
+
 import org.apache.commons.codec.binary.Base64;
 import org.apache.commons.io.IOUtils;
 import org.apache.http.HttpResponse;
@@ -50,6 +59,7 @@ import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPub
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
 import org.springframework.web.util.DefaultUriBuilderFactory;
+
 import reactor.test.StepVerifier;
 
 /**
@@ -80,12 +90,13 @@ class DmaapProducerReactiveHttpClientTest {
     private IFileSystemResource fileSystemResourceMock = mock(IFileSystemResource.class);
     private CloseableHttpAsyncClient clientMock;
     private HttpResponse responseMock = mock(HttpResponse.class);
+    @SuppressWarnings("unchecked")
     private Future<HttpResponse> futureMock = mock(Future.class);
     private StatusLine statusLine = mock(StatusLine.class);
     private InputStream fileStream;
 
     @BeforeEach
-    void setUp() {
+    void setUp() throws KeyManagementException, NoSuchAlgorithmException, KeyStoreException {
         when(dmaapPublisherConfigurationMock.dmaapHostName()).thenReturn(HOST);
         when(dmaapPublisherConfigurationMock.dmaapProtocol()).thenReturn(HTTPS_SCHEME);
         when(dmaapPublisherConfigurationMock.dmaapPortNumber()).thenReturn(PORT);
@@ -111,15 +122,15 @@ class DmaapProducerReactiveHttpClientTest {
                 .build();
         //formatter:on
 
-        dmaapProducerReactiveHttpClient = new DmaapProducerReactiveHttpClient(dmaapPublisherConfigurationMock);
+        dmaapProducerReactiveHttpClient = spy(new DmaapProducerReactiveHttpClient(dmaapPublisherConfigurationMock));
         dmaapProducerReactiveHttpClient.setFileSystemResource(fileSystemResourceMock);
         clientMock = mock(CloseableHttpAsyncClient.class);
-        dmaapProducerReactiveHttpClient.setWebClient(clientMock);
+        doReturn(clientMock).when(dmaapProducerReactiveHttpClient).createWebClient();
     }
 
     @Test
     void getHttpResponse_Success() throws Exception {
-        mockWebClientDependantObject(true);
+        mockWebClientDependantObject();
 
         HttpPut httpPut = new HttpPut();
         httpPut.addHeader(HttpHeaders.CONTENT_TYPE, APPLICATION_OCTET_STREAM_CONTENT_TYPE);
@@ -151,24 +162,22 @@ class DmaapProducerReactiveHttpClientTest {
 
     @Test
     void getHttpResponse_Fail() throws Exception {
-        mockWebClientDependantObject(false);
-        StepVerifier.create(dmaapProducerReactiveHttpClient.getDmaapProducerResponse(consumerDmaapModel, any()))
-        .expectError()
-        .verify();
+        Map<String, String> contextMap = new HashMap<>();
+        doReturn(futureMock).when(clientMock).execute(any(), any());
+        doThrow(new InterruptedException()).when(futureMock).get();
+        StepVerifier.create(dmaapProducerReactiveHttpClient.getDmaapProducerResponse(consumerDmaapModel, contextMap)) //
+           .expectError() //
+           .verify(); //
     }
 
-    private void mockWebClientDependantObject(boolean success)
+    private void mockWebClientDependantObject()
             throws IOException, InterruptedException, ExecutionException {
         fileStream = new ByteArrayInputStream(FILE_CONTENT.getBytes());
         when(fileSystemResourceMock.getInputStream()).thenReturn(fileStream);
-        if (success) {
-            when(clientMock.execute(any(HttpPut.class), any())).thenReturn(futureMock);
-            when(futureMock.get()).thenReturn(responseMock);
-            when(responseMock.getStatusLine()).thenReturn(statusLine);
-            when(statusLine.getStatusCode()).thenReturn(HttpUtils.SC_OK);
-        } else {
-            when(clientMock.execute(any(HttpPut.class), any())).thenReturn(futureMock);
-            when(futureMock.get()).thenThrow(new InterruptedException());
-        }
+        when(clientMock.execute(any(HttpPut.class), any())).thenReturn(futureMock);
+        when(futureMock.get()).thenReturn(responseMock);
+        when(responseMock.getStatusLine()).thenReturn(statusLine);
+        when(statusLine.getStatusCode()).thenReturn(HttpUtils.SC_OK);
+
     }
 }