Housekeeping

No functional changes made in this commit.

Removed CheckStyle and Sonar warnings.
Formatted code.
Renamed methods and classes for better understanding.
Removed unnecessary classes.
Moved all code to single project.

Change-Id: Ie3feb6c6a985e94a382812aa083dcf57bc46c7b3
Issue-ID: DCAEGEN2-1367
Signed-off-by: elinuxhenrik <henrik.b.andersson@est.tech>

24 files changed, 1320 insertions, 440 deletions

diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/AppConfigTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/AppConfigTest.java
index 1847e3b8..5be75ab3 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/AppConfigTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/AppConfigTest.java
@@ -1,4 +1,4 @@
-/*
+/*-
  * ============LICENSE_START======================================================================
  * Copyright (C) 2018 NOKIA Intellectual Property, 2018-2019 Nordix Foundation. All rights reserved.
  * ===============================================================================================
@@ -35,14 +35,13 @@ import java.util.Objects;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.onap.dcaegen2.collectors.datafile.integration.junit5.mockito.MockitoExtension;
 
 /**
+ * Tests the AppConfig.
+ *
  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/9/18
  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  */
-@ExtendWith({MockitoExtension.class})
 class AppConfigTest {
 
     private static final String DATAFILE_ENDPOINTS = "datafile_endpoints.json";
@@ -185,8 +184,8 @@ class AppConfigTest {
         JsonObject ftpesConfigData = new JsonObject();
         ftpesConfigData.addProperty("keyCert", "config/dfc.jks");
         ftpesConfigData.addProperty("keyPassword", "secret");
-        ftpesConfigData.addProperty("trustedCA", "config/ftp.jks");
-        ftpesConfigData.addProperty("trustedCAPassword", "secret");
+        ftpesConfigData.addProperty("trustedCa", "config/ftp.jks");
+        ftpesConfigData.addProperty("trustedCaPassword", "secret");
 
         JsonObject security = new JsonObject();
         security.addProperty("trustStorePath", "trustStorePath");
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/CloudConfigParserTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/CloudConfigParserTest.java
index 1adb3709..07233d95 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/CloudConfigParserTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/CloudConfigParserTest.java
@@ -1,4 +1,4 @@
-/*
+/*-
  * ============LICENSE_START======================================================================
  * Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved.
  * ===============================================================================================
@@ -66,8 +66,8 @@ class CloudConfigParserTest {
             new ImmutableFtpesConfig.Builder() //
                     .keyCert("/config/dfc.jks") //
                     .keyPassword("secret") //
-                    .trustedCA("config/ftp.jks") //
-                    .trustedCAPassword("secret") //
+                    .trustedCa("config/ftp.jks") //
+                    .trustedCaPassword("secret") //
                     .build();
 
     private CloudConfigParser cloudConfigParser = new CloudConfigParser(getCloudConfigJsonObject());
@@ -119,8 +119,8 @@ class CloudConfigParserTest {
         config.addProperty("dmaap.dmaapProducerConfiguration.dmaapUserPassword", "dradmin");
         config.addProperty("dmaap.ftpesConfig.keyCert", "/config/dfc.jks");
         config.addProperty("dmaap.ftpesConfig.keyPassword", "secret");
-        config.addProperty("dmaap.ftpesConfig.trustedCA", "config/ftp.jks");
-        config.addProperty("dmaap.ftpesConfig.trustedCAPassword", "secret");
+        config.addProperty("dmaap.ftpesConfig.trustedCa", "config/ftp.jks");
+        config.addProperty("dmaap.ftpesConfig.trustedCaPassword", "secret");
 
         config.addProperty("dmaap.security.trustStorePath", "trustStorePath");
         config.addProperty("dmaap.security.trustStorePasswordPath", "trustStorePasswordPath");
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/SchedulerConfigTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/SchedulerConfigTest.java
new file mode 100644
index 00000000..6e2140b4
--- /dev/null
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/SchedulerConfigTest.java
@@ -0,0 +1,124 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2019 Nordix Foundation.
+ * ================================================================================
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.configuration;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ScheduledFuture;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.onap.dcaegen2.collectors.datafile.tasks.ScheduledTasks;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.scheduling.TaskScheduler;
+import reactor.test.StepVerifier;
+
+public class SchedulerConfigTest {
+
+    @Test
+    public void getResponseFromCancellationOfTasks_success() {
+        List<ScheduledFuture<?>> scheduledFutureList = new ArrayList<>();
+        ScheduledFuture<?> scheduledFutureMock = mock(ScheduledFuture.class);
+        scheduledFutureList.add(scheduledFutureMock);
+
+        SchedulerConfig.setScheduledFutureList(scheduledFutureList);
+
+        SchedulerConfig schedulerUnderTest = new SchedulerConfig(null, null, null);
+
+        String msg = "Datafile Service has already been stopped!";
+        StepVerifier.create(schedulerUnderTest.getResponseFromCancellationOfTasks())
+                .expectNext(new ResponseEntity<String>(msg, HttpStatus.CREATED)) //
+                .verifyComplete();
+
+        verify(scheduledFutureMock).cancel(false);
+        verifyNoMoreInteractions(scheduledFutureMock);
+
+        assertEquals(0, scheduledFutureList.size());
+    }
+
+    @Test
+    public void tryToStartTaskWhenNotStarted_success() {
+        TaskScheduler taskSchedulerMock = mock(TaskScheduler.class);
+        ScheduledTasks scheduledTasksMock = mock(ScheduledTasks.class);
+        CloudConfiguration cloudConfigurationMock = mock(CloudConfiguration.class);
+        List<ScheduledFuture<?>> scheduledFutureList = new ArrayList<>();
+
+        SchedulerConfig.setScheduledFutureList(scheduledFutureList);
+
+        SchedulerConfig schedulerUnderTestSpy =
+                spy(new SchedulerConfig(taskSchedulerMock, scheduledTasksMock, cloudConfigurationMock));
+
+        boolean actualResult = schedulerUnderTestSpy.tryToStartTask();
+
+        assertTrue(actualResult);
+
+        ArgumentCaptor<Runnable> runTaskRunnableCaptor = ArgumentCaptor.forClass(Runnable.class);
+        verify(taskSchedulerMock).scheduleAtFixedRate(runTaskRunnableCaptor.capture(), any(Instant.class),
+                eq(Duration.ofMinutes(5)));
+
+        ArgumentCaptor<Runnable> scheduleMainDatafileEventTaskCaptor = ArgumentCaptor.forClass(Runnable.class);
+        verify(taskSchedulerMock).scheduleWithFixedDelay(scheduleMainDatafileEventTaskCaptor.capture(),
+                eq(Duration.ofSeconds(15)));
+        ArgumentCaptor<Runnable> purgeCachedInformationCaptor = ArgumentCaptor.forClass(Runnable.class);
+        verify(taskSchedulerMock).scheduleWithFixedDelay(purgeCachedInformationCaptor.capture(),
+                eq(Duration.ofHours(1)));
+        verifyNoMoreInteractions(taskSchedulerMock);
+
+        scheduleMainDatafileEventTaskCaptor.getValue().run();
+        purgeCachedInformationCaptor.getValue().run();
+        verify(scheduledTasksMock).purgeCachedInformation(any(Instant.class));
+        verify(scheduledTasksMock).executeDatafileMainTask();
+        verifyNoMoreInteractions(scheduledTasksMock);
+
+        runTaskRunnableCaptor.getValue().run();
+        verify(cloudConfigurationMock).runTask();
+        verifyNoMoreInteractions(cloudConfigurationMock);
+
+        assertEquals(3, scheduledFutureList.size());
+    }
+
+    @Test
+    public void tryToStartTaskWhenAlreadyStarted_shouldReturnFalse() {
+        List<ScheduledFuture<?>> scheduledFutureList = new ArrayList<>();
+        ScheduledFuture<?> scheduledFutureMock = mock(ScheduledFuture.class);
+        scheduledFutureList.add(scheduledFutureMock);
+
+        SchedulerConfig.setScheduledFutureList(scheduledFutureList);
+
+        SchedulerConfig schedulerUnderTest = new SchedulerConfig(null, null, null);
+
+        boolean actualResult = schedulerUnderTest.tryToStartTask();
+
+        assertFalse(actualResult);
+    }
+}
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClientTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClientTest.java
new file mode 100644
index 00000000..e0182560
--- /dev/null
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClientTest.java
@@ -0,0 +1,234 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018-2019 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.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;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.TrustManager;
+import org.apache.commons.net.ftp.FTP;
+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.springframework.http.HttpStatus;
+
+public class FtpsClientTest {
+
+    private static final String REMOTE_FILE_PATH = "/dir/sample.txt";
+    private static final Path LOCAL_FILE_PATH = Paths.get("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 Path TRUSTED_CA_PATH = Paths.get("trustedCaPath");
+    private static final String TRUSTED_CA_PASSWORD = "trustedCaPassword";
+
+    private static final String USERNAME = "bob";
+    private static final String PASSWORD = "123";
+
+    private FTPSClient ftpsClientMock = mock(FTPSClient.class);
+    private KeyManager keyManagerMock = mock(KeyManager.class);
+    private TrustManager trustManagerMock = mock(TrustManager.class);
+    private InputStream inputStreamMock = mock(InputStream.class);
+    private OutputStream outputStreamMock = mock(OutputStream.class);
+
+    FtpsClient clientUnderTestSpy;
+
+    private ImmutableFileServerData createFileServerData() {
+        return ImmutableFileServerData.builder() //
+            .serverAddress(XNF_ADDRESS) //
+            .userId(USERNAME).password(PASSWORD) //
+            .port(PORT) //
+            .build();
+    }
+
+    @BeforeEach
+    protected void setUp() throws Exception {
+        clientUnderTestSpy = spy(new FtpsClient(createFileServerData(), FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH,
+            TRUSTED_CA_PASSWORD));
+        clientUnderTestSpy.realFtpsClient = ftpsClientMock;
+    }
+
+    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);
+        verify(ftpsClientMock).setNeedClientAuth(true);
+        verify(ftpsClientMock).setKeyManager(keyManagerMock);
+        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);
+    }
+
+    @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();
+
+        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(1)).retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH), any());
+        verifyNoMoreInteractions(ftpsClientMock);
+    }
+
+    @Test
+    public void collectFileFaultyOwnKey_shouldFail() throws Exception {
+
+        doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH);
+        assertThatThrownBy(() -> clientUnderTestSpy.open())
+            .hasMessageContaining("Could not open connection: java.io.FileNotFoundException:");
+
+        verify(ftpsClientMock).setNeedClientAuth(true);
+
+        doReturn(false).when(ftpsClientMock).isConnected();
+        clientUnderTestSpy.close();
+        verify(ftpsClientMock).isConnected();
+        verifyNoMoreInteractions(ftpsClientMock);
+    }
+
+    @Test
+    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())
+            .hasMessage("Could not open connection: java.io.IOException: problem");
+    }
+
+    @Test
+    public void collectFileFaultTrustedCA_shouldFail_empty_trustedCA_file() throws Exception {
+
+        doReturn(keyManagerMock).when(clientUnderTestSpy).createKeyManager(FTP_KEY_PATH, FTP_KEY_PASSWORD);
+        doReturn(inputStreamMock).when(clientUnderTestSpy).createInputStream(TRUSTED_CA_PATH);
+
+        assertThatThrownBy(() -> clientUnderTestSpy.open())
+            .hasMessage("Could not open connection: java.io.EOFException");
+    }
+
+    @Test
+    public void collectFileFaultyLogin_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(false).when(ftpsClientMock).login(USERNAME, PASSWORD);
+
+        assertThatThrownBy(() -> clientUnderTestSpy.open()).hasMessage("Unable to log in to xNF. 127.0.0.1");
+
+        verify(ftpsClientMock).setNeedClientAuth(true);
+        verify(ftpsClientMock).setKeyManager(keyManagerMock);
+        verify(ftpsClientMock).setTrustManager(trustManagerMock);
+        verify(ftpsClientMock).connect(XNF_ADDRESS, PORT);
+        verify(ftpsClientMock).login(USERNAME, PASSWORD);
+    }
+
+    @Test
+    public void collectFileBadRequestResponse_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(503).when(ftpsClientMock).getReplyCode();
+
+        assertThatThrownBy(() -> clientUnderTestSpy.open())
+            .hasMessage("Unable to connect to xNF. 127.0.0.1 xNF reply code: 503");
+
+        verify(ftpsClientMock).setNeedClientAuth(true);
+        verify(ftpsClientMock).setKeyManager(keyManagerMock);
+        verify(ftpsClientMock).setTrustManager(trustManagerMock);
+        verify(ftpsClientMock).connect(XNF_ADDRESS, PORT);
+        verify(ftpsClientMock).login(USERNAME, PASSWORD);
+        verify(ftpsClientMock, times(2)).getReplyCode();
+        verifyNoMoreInteractions(ftpsClientMock);
+    }
+
+    @Test
+    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();
+
+        doReturn(false).when(ftpsClientMock).retrieveFile(REMOTE_FILE_PATH, outputStreamMock);
+
+        assertThatThrownBy(() -> clientUnderTestSpy.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH))
+            .hasMessage("Could not retrieve file /dir/sample.txt");
+
+        verifyFtpsClientMock_openOk();
+        verify(ftpsClientMock, times(1)).retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH), any());
+        verifyNoMoreInteractions(ftpsClientMock);
+    }
+
+    @Test
+    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();
+        when(ftpsClientMock.isConnected()).thenReturn(false);
+
+        doThrow(new IOException("problem")).when(ftpsClientMock).retrieveFile(REMOTE_FILE_PATH, outputStreamMock);
+
+        assertThatThrownBy(() -> clientUnderTestSpy.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH))
+            .hasMessage("Could not fetch file: java.io.IOException: problem");
+
+        verifyFtpsClientMock_openOk();
+        verify(ftpsClientMock, times(1)).retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH), any());
+        verifyNoMoreInteractions(ftpsClientMock);
+    }
+}
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SchemeTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SchemeTest.java
new file mode 100644
index 00000000..82b5b229
--- /dev/null
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SchemeTest.java
@@ -0,0 +1,47 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2019 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.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
+
+public class SchemeTest {
+    @Test
+    public void getSchemeFromString_properScheme() throws DatafileTaskException {
+
+        Scheme actualScheme = Scheme.getSchemeFromString("FTPES");
+        assertEquals(Scheme.FTPS, actualScheme);
+
+        actualScheme = Scheme.getSchemeFromString("FTPS");
+        assertEquals(Scheme.FTPS, actualScheme);
+
+        actualScheme = Scheme.getSchemeFromString("SFTP");
+        assertEquals(Scheme.SFTP, actualScheme);
+    }
+
+    @Test
+    public void getSchemeFromString_invalidScheme() {
+        assertTrue(assertThrows(DatafileTaskException.class, () -> Scheme.getSchemeFromString("invalid")).getMessage()
+            .startsWith("DFC does not support protocol invalid"));
+    }
+}
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClientTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClientTest.java
new file mode 100644
index 00000000..9a4d045a
--- /dev/null
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClientTest.java
@@ -0,0 +1,144 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018-2019 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 static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+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.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import org.junit.Rule;
+import org.junit.Test;
+import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
+
+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 Path LOCAL_DUMMY_FILE = Paths.get("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 DatafileTaskException, IOException, JSchException, SftpException, Exception {
+        FileServerData expectedFileServerData = ImmutableFileServerData.builder() //
+                .serverAddress("127.0.0.1") //
+                .userId(USERNAME) //
+                .password(PASSWORD) //
+                .port(sftpServer.getPort()) //
+                .build();
+        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 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.open())
+                .hasMessageContaining("Could not open Sftp clientcom.jcraft.jsch.JSchException: Auth fail");
+        }
+    }
+
+    @Test
+    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();
+        try (SftpClient sftpClient = new SftpClient(expectedFileServerData)) {
+            sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
+            sftpClient.open();
+
+            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=");
+        }
+    }
+
+    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-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/integration/ScheduledXmlContextITest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/integration/ScheduledXmlContextITest.java
index 7059a7fe..7f6b8c51 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/integration/ScheduledXmlContextITest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/integration/ScheduledXmlContextITest.java
@@ -1,4 +1,4 @@
-/*
+/*-
  * ============LICENSE_START======================================================================
  * Copyright (C) 2018 NOKIA Intellectual Property, 2018-2019 Nordix Foundation. All rights reserved.
  * ===============================================================================================
@@ -25,7 +25,6 @@ import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.onap.dcaegen2.collectors.datafile.integration.junit5.mockito.MockitoExtension;
 import org.onap.dcaegen2.collectors.datafile.tasks.ScheduledTasks;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.ComponentScan;
@@ -35,14 +34,16 @@ import org.springframework.test.context.junit.jupiter.SpringExtension;
 import org.springframework.test.context.testng.AbstractTestNGSpringContextTests;
 
 /**
+ * Integration test for the ScheduledXmlContext.
+ *
  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/27/18
  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  */
 
 @Configuration
 @ComponentScan
-@ExtendWith({MockitoExtension.class, SpringExtension.class})
-@ContextConfiguration(locations = {"classpath:scheduled-context.xml"})
+@ExtendWith({ SpringExtension.class })
+@ContextConfiguration(locations = { "classpath:scheduled-context.xml" })
 class ScheduledXmlContextITest extends AbstractTestNGSpringContextTests {
 
     private static final int WAIT_FOR_SCHEDULING = 1;
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/integration/ServiceMockProvider.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/integration/ServiceMockProvider.java
deleted file mode 100644
index 0d5ea003..00000000
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/integration/ServiceMockProvider.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * PROJECT
- * ================================================================================
- * Copyright (C) 2018-2019 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.integration;
-
-import static org.mockito.Mockito.mock;
-
-import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
-import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-/**
- * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 7/10/18
- */
-@Configuration
-class ServiceMockProvider {
-
-    @Bean
-    public AppConfig getDatafileAppConfig() {
-        return mock(AppConfig.class);
-    }
-
-    @Bean
-    public ConsumerDmaapModel getRequestDetails() {
-        return mock(ConsumerDmaapModel.class);
-    }
-}
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/integration/junit5/mockito/MockitoExtension.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/integration/junit5/mockito/MockitoExtension.java
deleted file mode 100644
index bc4e6401..00000000
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/integration/junit5/mockito/MockitoExtension.java
+++ /dev/null
@@ -1,82 +0,0 @@
-/*
- * ============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.integration.junit5.mockito;
-
-import static org.mockito.Mockito.mock;
-
-import java.lang.reflect.Parameter;
-
-import org.junit.jupiter.api.extension.ExtensionContext;
-import org.junit.jupiter.api.extension.ExtensionContext.Namespace;
-import org.junit.jupiter.api.extension.ExtensionContext.Store;
-import org.junit.jupiter.api.extension.ParameterContext;
-import org.junit.jupiter.api.extension.ParameterResolver;
-import org.junit.jupiter.api.extension.TestInstancePostProcessor;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
-
-/**
- * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/27/18
- *
- *         {@code MockitoExtension} showcases the {@link TestInstancePostProcessor} and
- *         {@link ParameterResolver} extension APIs of JUnit 5 by providing dependency injection
- *         support at the field level and at the method parameter level via Mockito 2.x's
- *         {@link Mock @Mock} annotation.
- */
-public class MockitoExtension implements TestInstancePostProcessor, ParameterResolver {
-
-    @Override
-    public void postProcessTestInstance(Object testInstance, ExtensionContext context) {
-        MockitoAnnotations.initMocks(testInstance);
-    }
-
-    @Override
-    public boolean supportsParameter(ParameterContext parameterContext, ExtensionContext extensionContext) {
-        return parameterContext.getParameter().isAnnotationPresent(Mock.class);
-    }
-
-    @Override
-    public Object resolveParameter(ParameterContext parameterContext, ExtensionContext extensionContext) {
-        return getMock(parameterContext.getParameter(), extensionContext);
-    }
-
-    private Object getMock(Parameter parameter, ExtensionContext extensionContext) {
-        Class<?> mockType = parameter.getType();
-        Store mocks = extensionContext.getStore(Namespace.create(MockitoExtension.class, mockType));
-        String mockName = getMockName(parameter);
-
-        if (mockName != null) {
-            return mocks.getOrComputeIfAbsent(mockName, key -> mock(mockType, mockName));
-        } else {
-            return mocks.getOrComputeIfAbsent(mockType.getCanonicalName(), key -> mock(mockType));
-        }
-    }
-
-    private String getMockName(Parameter parameter) {
-        String explicitMockName = parameter.getAnnotation(Mock.class).name().trim();
-        if (!explicitMockName.isEmpty()) {
-            return explicitMockName;
-        } else if (parameter.isNamePresent()) {
-            return parameter.getName();
-        }
-        return null;
-    }
-
-
-}
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/model/CommonFunctionsTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/model/CommonFunctionsTest.java
new file mode 100644
index 00000000..5fdf30fc
--- /dev/null
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/model/CommonFunctionsTest.java
@@ -0,0 +1,50 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2019 Nordix Foundation.
+ * ================================================================================
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.model;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.file.Paths;
+import org.junit.jupiter.api.Test;
+
+public class CommonFunctionsTest {
+    @Test
+    public void createJsonBody_success() {
+        ImmutableFilePublishInformation filePublishInformation = ImmutableFilePublishInformation //
+            .builder() //
+            .productName("") //
+            .vendorName("") //
+            .lastEpochMicrosec("") //
+            .sourceName("") //
+            .startEpochMicrosec("") //
+            .timeZoneOffset("") //
+            .name("") //
+            .location("") //
+            .internalLocation(Paths.get("internalLocation")) //
+            .compression("") //
+            .fileFormatType("") //
+            .fileFormatVersion("") //
+            .build();
+        String actualBody = CommonFunctions.createJsonBody(filePublishInformation);
+
+        assertTrue(actualBody.contains("\"internalLocation\":\"internalLocation\""));
+    }
+}
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/model/FileDataTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/model/FileDataTest.java
index 84c5e07b..79666f72 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/model/FileDataTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/model/FileDataTest.java
@@ -27,10 +27,6 @@ import org.onap.dcaegen2.collectors.datafile.ftp.FileServerData;
 import org.onap.dcaegen2.collectors.datafile.ftp.ImmutableFileServerData;
 import org.onap.dcaegen2.collectors.datafile.ftp.Scheme;
 
-/**
- * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
- *
- */
 public class FileDataTest {
     private static final String FTPES_SCHEME = "ftpes://";
     private static final String PM_FILE_NAME = "A20161224.1030-1045.bin.gz";
@@ -46,56 +42,50 @@ public class FileDataTest {
 
 
     private MessageMetaData messageMetaData() {
-        return ImmutableMessageMetaData.builder()
-            .productName("PRODUCT_NAME")
-            .vendorName("VENDOR_NAME")
-            .lastEpochMicrosec("LAST_EPOCH_MICROSEC")
-            .sourceName("SOURCE_NAME")
-            .startEpochMicrosec("START_EPOCH_MICROSEC")
-            .timeZoneOffset("TIME_ZONE_OFFSET")
-            .changeIdentifier("PM_MEAS_CHANGE_IDENTIFIER")
-            .changeType("FILE_READY_CHANGE_TYPE")
-            .build();
+        return ImmutableMessageMetaData.builder() //
+                .productName("PRODUCT_NAME") //
+                .vendorName("VENDOR_NAME") //
+                .lastEpochMicrosec("LAST_EPOCH_MICROSEC") //
+                .sourceName("SOURCE_NAME") //
+                .startEpochMicrosec("START_EPOCH_MICROSEC") //
+                .timeZoneOffset("TIME_ZONE_OFFSET") //
+                .changeIdentifier("PM_MEAS_CHANGE_IDENTIFIER") //
+                .changeType("FILE_READY_CHANGE_TYPE") //
+                .build();
     }
 
     private FileData properFileDataWithUser() {
-        // @formatter:off
-        return ImmutableFileData.builder()
-         .name("name")
-         .location(LOCATION_WITH_USER)
-         .compression("comp")
-         .fileFormatType("type")
-         .fileFormatVersion("version")
-         .scheme(Scheme.FTPS)
-         .messageMetaData(messageMetaData())
-         .build();
-        // @formatter:on
+        return ImmutableFileData.builder() //
+                .name("name") //
+                .location(LOCATION_WITH_USER) //
+                .compression("comp") //
+                .fileFormatType("type") //
+                .fileFormatVersion("version") //
+                .scheme(Scheme.FTPS) //
+                .messageMetaData(messageMetaData()) //
+                .build();
     }
 
     private FileData properFileDataWithoutUser() {
-        // @formatter:off
-        return ImmutableFileData.builder()
-            .name("name")
-            .location(LOCATION_WITHOUT_USER)
-            .compression("comp")
-            .fileFormatType("type")
-            .fileFormatVersion("version")
-            .scheme(Scheme.FTPS)
-            .messageMetaData(messageMetaData())
-            .build();
-        // @formatter:on
+        return ImmutableFileData.builder() //
+                .name("name") //
+                .location(LOCATION_WITHOUT_USER) //
+                .compression("comp") //
+                .fileFormatType("type") //
+                .fileFormatVersion("version") //
+                .scheme(Scheme.FTPS) //
+                .messageMetaData(messageMetaData()) //
+                .build();
     }
 
     @Test
     public void fileServerData_properLocationWithUser() {
-        // @formatter:off
-        ImmutableFileServerData expectedFileServerData = ImmutableFileServerData.builder()
-                .serverAddress(SERVER_ADDRESS)
-                .port(PORT_22)
-                .userId(USER)
-                .password(PWD)
+        ImmutableFileServerData expectedFileServerData = ImmutableFileServerData.builder() //
+                .serverAddress(SERVER_ADDRESS) //
+                .port(PORT_22) //
+                .userId(USER) //
+                .password(PWD) //
                 .build();
-        // @formatter:on
 
         FileServerData actualFileServerData = properFileDataWithUser().fileServerData();
         assertEquals(expectedFileServerData, actualFileServerData);
@@ -103,14 +93,12 @@ public class FileDataTest {
 
     @Test
     public void fileServerData_properLocationWithoutUser() {
-        // @formatter:off
-        ImmutableFileServerData expectedFileServerData = ImmutableFileServerData.builder()
-                .serverAddress(SERVER_ADDRESS)
-                .port(PORT_22)
-                .userId("")
-                .password("")
+        ImmutableFileServerData expectedFileServerData = ImmutableFileServerData.builder() //
+                .serverAddress(SERVER_ADDRESS) //
+                .port(PORT_22) //
+                .userId("") //
+                .password("") //
                 .build();
-        // @formatter:on
 
         FileServerData actualFileServerData = properFileDataWithoutUser().fileServerData();
         assertEquals(expectedFileServerData, actualFileServerData);
@@ -125,17 +113,13 @@ public class FileDataTest {
 
     @Test
     public void fileServerData_properLocationWithoutPort() {
-        // @formatter:off
-        ImmutableFileServerData fileServerData = ImmutableFileServerData.builder()
-                .serverAddress(SERVER_ADDRESS)
-                .userId("")
-                .password("")
+        ImmutableFileServerData fileServerData = ImmutableFileServerData.builder() //
+                .serverAddress(SERVER_ADDRESS) //
+                .userId("") //
+                .password("") //
                 .build();
-        // @formatter:on
 
         assertFalse(fileServerData.port().isPresent());
     }
-
-
 }
 
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/model/FilePublishInformationTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/model/FilePublishInformationTest.java
new file mode 100644
index 00000000..950b9a6f
--- /dev/null
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/model/FilePublishInformationTest.java
@@ -0,0 +1,69 @@
+/*-
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property, 2018-2019 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.model;
+
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class FilePublishInformationTest {
+    private static final String PRODUCT_NAME = "NrRadio";
+    private static final String VENDOR_NAME = "Ericsson";
+    private static final String LAST_EPOCH_MICROSEC = "8745745764578";
+    private static final String SOURCE_NAME = "oteNB5309";
+    private static final String START_EPOCH_MICROSEC = "8745745764578";
+    private static final String TIME_ZONE_OFFSET = "UTC+05:00";
+    private static final String NAME = "A20161224.1030-1045.bin.gz";
+    private static final String LOCATION = "ftpes://192.168.0.101:22/ftp/rop/A20161224.1030-1145.bin.gz";
+    private static final Path INTERNAL_LOCATION = Paths.get("target/A20161224.1030-1045.bin.gz");
+    private static final String COMPRESSION = "gzip";
+    private static final String FILE_FORMAT_TYPE = "org.3GPP.32.435#measCollec";
+    private static final String FILE_FORMAT_VERSION = "V10";
+
+    @Test
+    public void filePublishInformationBuilder_shouldBuildAnObject() {
+        FilePublishInformation filePublishInformation = ImmutableFilePublishInformation.builder() //
+            .productName(PRODUCT_NAME) //
+            .vendorName(VENDOR_NAME) //
+            .lastEpochMicrosec(LAST_EPOCH_MICROSEC) //
+            .sourceName(SOURCE_NAME) //
+            .startEpochMicrosec(START_EPOCH_MICROSEC) //
+            .timeZoneOffset(TIME_ZONE_OFFSET) //
+            .name(NAME) //
+            .location(LOCATION) //
+            .internalLocation(INTERNAL_LOCATION) //
+            .compression(COMPRESSION) //
+            .fileFormatType(FILE_FORMAT_TYPE) //
+            .fileFormatVersion(FILE_FORMAT_VERSION) //
+            .build();
+
+        Assertions.assertNotNull(filePublishInformation);
+        Assertions.assertEquals(PRODUCT_NAME, filePublishInformation.getProductName());
+        Assertions.assertEquals(VENDOR_NAME, filePublishInformation.getVendorName());
+        Assertions.assertEquals(LAST_EPOCH_MICROSEC, filePublishInformation.getLastEpochMicrosec());
+        Assertions.assertEquals(SOURCE_NAME, filePublishInformation.getSourceName());
+        Assertions.assertEquals(START_EPOCH_MICROSEC, filePublishInformation.getStartEpochMicrosec());
+        Assertions.assertEquals(TIME_ZONE_OFFSET, filePublishInformation.getTimeZoneOffset());
+        Assertions.assertEquals(NAME, filePublishInformation.getName());
+        Assertions.assertEquals(LOCATION, filePublishInformation.getLocation());
+        Assertions.assertEquals(INTERNAL_LOCATION, filePublishInformation.getInternalLocation());
+        Assertions.assertEquals(COMPRESSION, filePublishInformation.getCompression());
+        Assertions.assertEquals(FILE_FORMAT_TYPE, filePublishInformation.getFileFormatType());
+        Assertions.assertEquals(FILE_FORMAT_VERSION, filePublishInformation.getFileFormatVersion());
+    }
+}
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/DmaapWebClientTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/DmaapWebClientTest.java
new file mode 100644
index 00000000..9aaca4bf
--- /dev/null
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/DmaapWebClientTest.java
@@ -0,0 +1,51 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018-2019 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.service;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
+import org.springframework.web.reactive.function.client.WebClient;
+
+class DmaapWebClientTest {
+
+    @Mock
+    private DmaapConsumerConfiguration dmaapConsumerConfiguration;
+
+    @BeforeEach
+    void setUp() {
+        dmaapConsumerConfiguration = mock(DmaapConsumerConfiguration.class);
+    }
+
+    @Test
+    void buildsDMaaPReactiveWebClientProperly() {
+        when(dmaapConsumerConfiguration.dmaapContentType()).thenReturn("*/*");
+        WebClient dmaapWebClient = new DmaapWebClient() //
+            .fromConfiguration(dmaapConsumerConfiguration) //
+            .build();
+
+        verify(dmaapConsumerConfiguration, times(1)).dmaapContentType();
+        assertNotNull(dmaapWebClient);
+    }
+}
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/HttpUtilsTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/HttpUtilsTest.java
new file mode 100644
index 00000000..a95bfe2b
--- /dev/null
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/HttpUtilsTest.java
@@ -0,0 +1,35 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018-2019 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.service;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.jupiter.api.Test;
+
+class HttpUtilsTest {
+
+    @Test
+    public void shouldReturnSuccessfulResponse() {
+        assertTrue(HttpUtils.isSuccessfulResponseCode(200));
+    }
+
+    @Test
+    public void shouldReturnBadResponse() {
+        assertFalse(HttpUtils.isSuccessfulResponseCode(404));
+    }
+}
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParserTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParserTest.java
index b8aa7da2..becfba31 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParserTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParserTest.java
@@ -22,11 +22,9 @@ import static org.mockito.Mockito.spy;
 
 import com.google.gson.JsonElement;
 import com.google.gson.JsonParser;
-
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
-
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 import org.onap.dcaegen2.collectors.datafile.ftp.Scheme;
@@ -38,11 +36,12 @@ import org.onap.dcaegen2.collectors.datafile.model.ImmutableMessageMetaData;
 import org.onap.dcaegen2.collectors.datafile.model.MessageMetaData;
 import org.onap.dcaegen2.collectors.datafile.utils.JsonMessage;
 import org.onap.dcaegen2.collectors.datafile.utils.JsonMessage.AdditionalField;
-
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 /**
+ * Tests the JsonMessageParser.
+ *
  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 5/8/18
  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  */
@@ -99,11 +98,11 @@ class JsonMessageParserTest {
                 .compression(GZIP_COMPRESSION) //
                 .fileFormatType(FILE_FORMAT_TYPE) //
                 .fileFormatVersion(FILE_FORMAT_VERSION) //
-                .messageMetaData(messageMetaData)
+                .messageMetaData(messageMetaData) //
                 .build();
         List<FileData> files = new ArrayList<>();
         files.add(expectedFileData);
-        FileReadyMessage expectedMessage = ImmutableFileReadyMessage.builder()
+        FileReadyMessage expectedMessage = ImmutableFileReadyMessage.builder() //
                 .files(files) //
                 .build();
 
@@ -152,7 +151,7 @@ class JsonMessageParserTest {
                 .compression(GZIP_COMPRESSION) //
                 .fileFormatType(FILE_FORMAT_TYPE) //
                 .fileFormatVersion(FILE_FORMAT_VERSION) //
-                .messageMetaData(messageMetaData)
+                .messageMetaData(messageMetaData) //
                 .build();
         List<FileData> files = new ArrayList<>();
         files.add(expectedFileData);
@@ -232,7 +231,7 @@ class JsonMessageParserTest {
                 .compression(GZIP_COMPRESSION) //
                 .fileFormatType(FILE_FORMAT_TYPE) //
                 .fileFormatVersion(FILE_FORMAT_VERSION) //
-                .messageMetaData(messageMetaData)
+                .messageMetaData(messageMetaData) //
                 .build();
         List<FileData> files = new ArrayList<>();
         files.add(expectedFileData);
@@ -417,7 +416,7 @@ class JsonMessageParserTest {
                 .compression(GZIP_COMPRESSION) //
                 .fileFormatType(FILE_FORMAT_TYPE) //
                 .fileFormatVersion(FILE_FORMAT_VERSION) //
-                .messageMetaData(messageMetaData)
+                .messageMetaData(messageMetaData) //
                 .build();
         List<FileData> files = new ArrayList<>();
         files.add(expectedFileData);
@@ -473,7 +472,7 @@ class JsonMessageParserTest {
         AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder() //
                 .name(PM_FILE_NAME) //
                 .location(LOCATION) //
-                .compression(GZIP_COMPRESSION)
+                .compression(GZIP_COMPRESSION) //
                 .fileFormatVersion(FILE_FORMAT_VERSION) //
                 .build();
         JsonMessage message = new JsonMessage.JsonMessageBuilder() //
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/PublishedFileCacheTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/PublishedFileCacheTest.java
index 7b38ee42..64cfb38f 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/PublishedFileCacheTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/PublishedFileCacheTest.java
@@ -1,4 +1,4 @@
-/*
+/*-
  * ============LICENSE_START======================================================================
  * Copyright (C) 2019 Nordix Foundation. All rights reserved.
  * ===============================================================================================
@@ -19,7 +19,6 @@ package org.onap.dcaegen2.collectors.datafile.service;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.time.Instant;
-
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerHttpClientTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerHttpClientTest.java
new file mode 100644
index 00000000..add47b0a
--- /dev/null
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerHttpClientTest.java
@@ -0,0 +1,202 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property, 2018-2019 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.service.producer;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.security.KeyManagementException;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Future;
+import javax.net.ssl.SSLContext;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.http.Header;
+import org.apache.http.HttpResponse;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
+import org.onap.dcaegen2.collectors.datafile.http.HttpAsyncClientBuilderWrapper;
+import org.onap.dcaegen2.collectors.datafile.web.PublishRedirectStrategy;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPublisherConfiguration;
+
+/**
+ * Test for DmaapProducerHttpClient.
+ *
+ * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 7/4/18
+ * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
+ */
+class DmaapProducerHttpClientTest {
+
+    private static final String HOST = "54.45.33.2";
+    private static final String HTTPS_SCHEME = "https";
+    private static final int PORT = 1234;
+    private static final String USER_NAME = "dradmin";
+    private static final Duration TWO_SECOND_TIMEOUT = Duration.ofSeconds(2);
+
+    private static final Map<String, String> CONTEXT_MAP = new HashMap<>();
+
+
+    private DmaapProducerHttpClient producerClientUnderTestSpy;
+
+    private DmaapPublisherConfiguration dmaapPublisherConfigurationMock = mock(DmaapPublisherConfiguration.class);
+
+    private HttpAsyncClientBuilderWrapper clientBuilderMock;
+
+    private CloseableHttpAsyncClient clientMock;
+    @SuppressWarnings("unchecked")
+    private Future<HttpResponse> futureMock = mock(Future.class);
+
+    @BeforeEach
+    void setUp() throws KeyManagementException, NoSuchAlgorithmException, KeyStoreException {
+        when(dmaapPublisherConfigurationMock.dmaapHostName()).thenReturn(HOST);
+        when(dmaapPublisherConfigurationMock.dmaapProtocol()).thenReturn(HTTPS_SCHEME);
+        when(dmaapPublisherConfigurationMock.dmaapPortNumber()).thenReturn(PORT);
+        when(dmaapPublisherConfigurationMock.dmaapUserName()).thenReturn("dradmin");
+        when(dmaapPublisherConfigurationMock.dmaapUserPassword()).thenReturn("dradmin");
+
+        producerClientUnderTestSpy = spy(new DmaapProducerHttpClient(dmaapPublisherConfigurationMock));
+
+        clientBuilderMock = mock(HttpAsyncClientBuilderWrapper.class);
+        clientMock = mock(CloseableHttpAsyncClient.class);
+    }
+
+    @Test
+    void getHttpResponseWithRederict_Success() throws Exception {
+        doReturn(clientBuilderMock).when(producerClientUnderTestSpy).getHttpClientBuilder();
+        when(clientBuilderMock.setSslContext(any(SSLContext.class))).thenReturn(clientBuilderMock);
+        when(clientBuilderMock.setSslHostnameVerifier(any(NoopHostnameVerifier.class))).thenReturn(clientBuilderMock);
+        when(clientBuilderMock.build()).thenReturn(clientMock);
+        when(clientMock.execute(any(HttpUriRequest.class), any())).thenReturn(futureMock);
+        HttpResponse responseMock = mock(HttpResponse.class);
+        when(futureMock.get()).thenReturn(responseMock);
+
+        HttpGet request = new HttpGet();
+        producerClientUnderTestSpy.getDmaapProducerResponseWithRedirect(request, CONTEXT_MAP);
+
+        verify(clientBuilderMock).setSslContext(any(SSLContext.class));
+        verify(clientBuilderMock).setSslHostnameVerifier(any(NoopHostnameVerifier.class));
+        verify(clientBuilderMock).setRedirectStrategy(PublishRedirectStrategy.INSTANCE);
+        verify(clientBuilderMock).setDefaultRequestConfig(any());
+        verify(clientBuilderMock).build();
+        verifyNoMoreInteractions(clientBuilderMock);
+
+        verify(clientMock).start();
+        verify(clientMock).close();
+
+        verify(futureMock).get();
+        verifyNoMoreInteractions(futureMock);
+    }
+
+    @Test
+    void getHttpResponseWithCustomTimeout_Success() throws Exception {
+        doReturn(clientBuilderMock).when(producerClientUnderTestSpy).getHttpClientBuilder();
+        when(clientBuilderMock.setSslContext(any(SSLContext.class))).thenReturn(clientBuilderMock);
+        when(clientBuilderMock.setDefaultRequestConfig(any(RequestConfig.class))).thenReturn(clientBuilderMock);
+        when(clientBuilderMock.build()).thenReturn(clientMock);
+        when(clientMock.execute(any(HttpUriRequest.class), any())).thenReturn(futureMock);
+        HttpResponse responseMock = mock(HttpResponse.class);
+        when(futureMock.get()).thenReturn(responseMock);
+
+        HttpGet request = new HttpGet();
+        producerClientUnderTestSpy.getDmaapProducerResponseWithCustomTimeout(request, TWO_SECOND_TIMEOUT, CONTEXT_MAP);
+
+        ArgumentCaptor<RequestConfig> requestConfigCaptor = ArgumentCaptor.forClass(RequestConfig.class);
+        verify(clientBuilderMock).setSslContext(any(SSLContext.class));
+        verify(clientBuilderMock).setSslHostnameVerifier(any(NoopHostnameVerifier.class));
+        verify(clientBuilderMock).setDefaultRequestConfig(requestConfigCaptor.capture());
+        RequestConfig requestConfig = requestConfigCaptor.getValue();
+        assertEquals(TWO_SECOND_TIMEOUT.toMillis(), requestConfig.getSocketTimeout());
+        assertEquals(TWO_SECOND_TIMEOUT.toMillis(), requestConfig.getConnectTimeout());
+        assertEquals(TWO_SECOND_TIMEOUT.toMillis(), requestConfig.getConnectionRequestTimeout());
+        verify(clientBuilderMock).build();
+        verifyNoMoreInteractions(clientBuilderMock);
+
+        verify(clientMock).start();
+        verify(clientMock).close();
+
+        verify(futureMock).get();
+        verifyNoMoreInteractions(futureMock);
+    }
+
+    @Test
+    public void getResponseWithException_throwsException() throws Exception {
+        doReturn(clientBuilderMock).when(producerClientUnderTestSpy).getHttpClientBuilder();
+        when(clientBuilderMock.setDefaultRequestConfig(any(RequestConfig.class))).thenReturn(clientBuilderMock);
+        when(clientBuilderMock.setSslContext(any(SSLContext.class))).thenReturn(clientBuilderMock);
+        when(clientBuilderMock.build()).thenReturn(clientMock);
+        HttpPut request = new HttpPut();
+        when(clientMock.execute(any(HttpPut.class), any())).thenReturn(futureMock);
+
+        try {
+            when(futureMock.get()).thenThrow(new InterruptedException("Interrupted"));
+
+            producerClientUnderTestSpy.getDmaapProducerResponseWithCustomTimeout(request, TWO_SECOND_TIMEOUT,
+                    CONTEXT_MAP);
+
+            fail("Should have got an exception.");
+        } catch (DatafileTaskException e) {
+            assertTrue(e.getCause() instanceof InterruptedException);
+            assertEquals("Interrupted", e.getCause().getMessage());
+        } catch (Exception e) {
+            fail("Wrong exception");
+        }
+
+        verify(clientMock).start();
+        verify(clientMock).close();
+    }
+
+    @Test
+    public void addCredentialsToHead_success() {
+        HttpPut request = new HttpPut();
+
+        producerClientUnderTestSpy.addUserCredentialsToHead(request);
+
+        String plainCreds = USER_NAME + ":" + USER_NAME;
+        byte[] plainCredsBytes = plainCreds.getBytes(StandardCharsets.ISO_8859_1);
+        byte[] base64CredsBytes = Base64.encodeBase64(plainCredsBytes);
+        String base64Creds = "Basic " + new String(base64CredsBytes);
+        Header[] authorizationHeaders = request.getHeaders("Authorization");
+        assertEquals(base64Creds, authorizationHeaders[0].getValue());
+    }
+
+    @Test
+    public void getBaseUri_success() {
+        URI uri = producerClientUnderTestSpy.getBaseUri().build();
+        assertEquals(HTTPS_SCHEME + "://" + HOST + ":" + PORT, uri.toString());
+    }
+}
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DMaaPMessageConsumerTaskImplTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DMaaPMessageConsumerTest.java
index afa51de1..06fa0b4e 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DMaaPMessageConsumerTaskImplTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DMaaPMessageConsumerTest.java
@@ -31,16 +31,15 @@ import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.List;
-
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
 import org.onap.dcaegen2.collectors.datafile.ftp.Scheme;
-import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
 import org.onap.dcaegen2.collectors.datafile.model.FileData;
+import org.onap.dcaegen2.collectors.datafile.model.FilePublishInformation;
 import org.onap.dcaegen2.collectors.datafile.model.FileReadyMessage;
-import org.onap.dcaegen2.collectors.datafile.model.ImmutableConsumerDmaapModel;
 import org.onap.dcaegen2.collectors.datafile.model.ImmutableFileData;
+import org.onap.dcaegen2.collectors.datafile.model.ImmutableFilePublishInformation;
 import org.onap.dcaegen2.collectors.datafile.model.ImmutableFileReadyMessage;
 import org.onap.dcaegen2.collectors.datafile.model.ImmutableMessageMetaData;
 import org.onap.dcaegen2.collectors.datafile.model.MessageMetaData;
@@ -48,15 +47,11 @@ import org.onap.dcaegen2.collectors.datafile.service.JsonMessageParser;
 import org.onap.dcaegen2.collectors.datafile.utils.JsonMessage;
 import org.onap.dcaegen2.collectors.datafile.utils.JsonMessage.AdditionalField;
 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer.DMaaPConsumerReactiveHttpClient;
-
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
-/**
- * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
- */
-public class DMaaPMessageConsumerTaskImplTest {
+public class DMaaPMessageConsumerTest {
     private static final String NR_RADIO_ERICSSON_EVENT_NAME = "Noti_NrRadio-Ericsson_FileReady";
     private static final String PRODUCT_NAME = "NrRadio";
     private static final String VENDOR_NAME = "Ericsson";
@@ -78,11 +73,11 @@ public class DMaaPMessageConsumerTaskImplTest {
     private static final String GZIP_COMPRESSION = "gzip";
     private static final String MEAS_COLLECT_FILE_FORMAT_TYPE = "org.3GPP.32.435#measCollec";
     private static final String FILE_FORMAT_VERSION = "V10";
-    private static List<ConsumerDmaapModel> listOfConsumerDmaapModel = new ArrayList<ConsumerDmaapModel>();
+    private static List<FilePublishInformation> listOfFilePublishInformation = new ArrayList<FilePublishInformation>();
 
-    private DMaaPMessageConsumerTask messageConsumerTask;
     private DMaaPConsumerReactiveHttpClient httpClientMock;
 
+    private DMaaPMessageConsumer messageConsumer;
     private static String ftpesMessageString;
     private static FileData ftpesFileData;
     private static FileReadyMessage expectedFtpesMessage;
@@ -96,7 +91,6 @@ public class DMaaPMessageConsumerTaskImplTest {
      */
     @BeforeAll
     public static void setUp() {
-
         AdditionalField ftpesAdditionalField = new JsonMessage.AdditionalFieldBuilder() //
                 .location(FTPES_LOCATION) //
                 .compression(GZIP_COMPRESSION) //
@@ -130,7 +124,7 @@ public class DMaaPMessageConsumerTaskImplTest {
                 .compression(GZIP_COMPRESSION) //
                 .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE) //
                 .fileFormatVersion(FILE_FORMAT_VERSION) //
-                .messageMetaData(messageMetaData)
+                .messageMetaData(messageMetaData) //
                 .build();
 
         List<FileData> files = new ArrayList<>();
@@ -160,10 +154,10 @@ public class DMaaPMessageConsumerTaskImplTest {
                 .compression(GZIP_COMPRESSION) //
                 .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE) //
                 .fileFormatVersion(FILE_FORMAT_VERSION) //
-                .messageMetaData(messageMetaData)
+                .messageMetaData(messageMetaData) //
                 .build();
 
-        ImmutableConsumerDmaapModel consumerDmaapModel = ImmutableConsumerDmaapModel.builder() //
+        ImmutableFilePublishInformation filePublishInformation = ImmutableFilePublishInformation.builder() //
                 .productName(PRODUCT_NAME) //
                 .vendorName(VENDOR_NAME) //
                 .lastEpochMicrosec(LAST_EPOCH_MICROSEC) //
@@ -177,7 +171,7 @@ public class DMaaPMessageConsumerTaskImplTest {
                 .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE) //
                 .fileFormatVersion(FILE_FORMAT_VERSION) //
                 .build();
-        listOfConsumerDmaapModel.add(consumerDmaapModel);
+        listOfFilePublishInformation.add(filePublishInformation);
 
         files = new ArrayList<>();
         files.add(sftpFileData);
@@ -190,7 +184,7 @@ public class DMaaPMessageConsumerTaskImplTest {
     public void whenPassedObjectDoesntFit_ThrowsDatafileTaskException() {
         prepareMocksForDmaapConsumer("", null);
 
-        StepVerifier.create(messageConsumerTask.execute()) //
+        StepVerifier.create(messageConsumer.getMessageRouterResponse()) //
                 .expectSubscription() //
                 .expectError(DatafileTaskException.class) //
                 .verify();
@@ -202,7 +196,7 @@ public class DMaaPMessageConsumerTaskImplTest {
     public void whenFtpes_ReturnsCorrectResponse() throws DatafileTaskException {
         prepareMocksForDmaapConsumer(ftpesMessageString, expectedFtpesMessage);
 
-        StepVerifier.create(messageConsumerTask.execute()) //
+        StepVerifier.create(messageConsumer.getMessageRouterResponse()) //
                 .expectNext(expectedFtpesMessage) //
                 .verifyComplete();
 
@@ -214,7 +208,7 @@ public class DMaaPMessageConsumerTaskImplTest {
     public void whenSftp_ReturnsCorrectResponse() throws DatafileTaskException {
         prepareMocksForDmaapConsumer(sftpMessageString, expectedSftpMessage);
 
-        StepVerifier.create(messageConsumerTask.execute()) //
+        StepVerifier.create(messageConsumer.getMessageRouterResponse()) //
                 .expectNext(expectedSftpMessage) //
                 .verifyComplete();
 
@@ -236,6 +230,6 @@ public class DMaaPMessageConsumerTaskImplTest {
                     .thenReturn(Flux.error(new DatafileTaskException("problemas")));
         }
 
-        messageConsumerTask = spy(new DMaaPMessageConsumerTask(httpClientMock, jsonMessageParserMock));
+        messageConsumer = spy(new DMaaPMessageConsumer(httpClientMock, jsonMessageParserMock));
     }
 }
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisherTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisherTest.java
index 22900b38..5746e0fd 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisherTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisherTest.java
@@ -47,8 +47,8 @@ import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
-import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.collectors.datafile.model.ImmutableConsumerDmaapModel;
+import org.onap.dcaegen2.collectors.datafile.model.FilePublishInformation;
+import org.onap.dcaegen2.collectors.datafile.model.ImmutableFilePublishInformation;
 import org.onap.dcaegen2.collectors.datafile.service.producer.DmaapProducerHttpClient;
 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPublisherConfiguration;
 import org.springframework.http.HttpStatus;
@@ -57,6 +57,8 @@ import org.springframework.web.util.UriBuilder;
 import reactor.test.StepVerifier;
 
 /**
+ * Tests the DataRouter publisher.
+ *
  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 5/17/18
  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  */
@@ -83,23 +85,20 @@ class DataRouterPublisherTest {
     private static final String FEED_ID = "1";
     private static final String FILE_CONTENT = "Just a string.";
 
-    private static ConsumerDmaapModel consumerDmaapModel;
+    private static FilePublishInformation filePublishInformation;
     private static DmaapProducerHttpClient httpClientMock;
     private static AppConfig appConfig;
     private static DmaapPublisherConfiguration publisherConfigurationMock = mock(DmaapPublisherConfiguration.class);
     private final Map<String, String> contextMap = new HashMap<>();
     private static DataRouterPublisher publisherTaskUnderTestSpy;
 
-    /**
-     * Sets up data for tests.
-     */
     @BeforeAll
     public static void setUp() {
         when(publisherConfigurationMock.dmaapHostName()).thenReturn(HOST);
         when(publisherConfigurationMock.dmaapProtocol()).thenReturn(HTTPS_SCHEME);
         when(publisherConfigurationMock.dmaapPortNumber()).thenReturn(PORT);
 
-        consumerDmaapModel = ImmutableConsumerDmaapModel.builder() //
+        filePublishInformation = ImmutableFilePublishInformation.builder() //
                 .productName(PRODUCT_NAME) //
                 .vendorName(VENDOR_NAME) //
                 .lastEpochMicrosec(LAST_EPOCH_MICROSEC) //
@@ -121,8 +120,9 @@ class DataRouterPublisherTest {
     public void whenPassedObjectFits_ReturnsCorrectStatus() throws Exception {
         prepareMocksForTests(null, Integer.valueOf(HttpStatus.OK.value()));
         StepVerifier
-                .create(publisherTaskUnderTestSpy.execute(consumerDmaapModel, 1, Duration.ofSeconds(0), contextMap))
-                .expectNext(consumerDmaapModel) //
+                .create(publisherTaskUnderTestSpy.publishFile(filePublishInformation, 1, Duration.ofSeconds(0),
+                        contextMap))
+                .expectNext(filePublishInformation) //
                 .verifyComplete();
 
         ArgumentCaptor<HttpUriRequest> requestCaptor = ArgumentCaptor.forClass(HttpUriRequest.class);
@@ -164,8 +164,9 @@ class DataRouterPublisherTest {
         prepareMocksForTests(new DatafileTaskException("Error"), HttpStatus.OK.value());
 
         StepVerifier
-                .create(publisherTaskUnderTestSpy.execute(consumerDmaapModel, 2, Duration.ofSeconds(0), contextMap))
-                .expectNext(consumerDmaapModel) //
+                .create(publisherTaskUnderTestSpy.publishFile(filePublishInformation, 2, Duration.ofSeconds(0),
+                        contextMap))
+                .expectNext(filePublishInformation) //
                 .verifyComplete();
     }
 
@@ -175,8 +176,9 @@ class DataRouterPublisherTest {
                 Integer.valueOf(HttpStatus.OK.value()));
 
         StepVerifier
-                .create(publisherTaskUnderTestSpy.execute(consumerDmaapModel, 1, Duration.ofSeconds(0), contextMap))
-                .expectNext(consumerDmaapModel) //
+                .create(publisherTaskUnderTestSpy.publishFile(filePublishInformation, 1, Duration.ofSeconds(0),
+                        contextMap))
+                .expectNext(filePublishInformation) //
                 .verifyComplete();
 
         verify(httpClientMock, times(2)).getBaseUri();
@@ -191,7 +193,8 @@ class DataRouterPublisherTest {
                 Integer.valueOf((HttpStatus.BAD_GATEWAY.value())));
 
         StepVerifier
-                .create(publisherTaskUnderTestSpy.execute(consumerDmaapModel, 1, Duration.ofSeconds(0), contextMap))
+                .create(publisherTaskUnderTestSpy.publishFile(filePublishInformation, 1, Duration.ofSeconds(0),
+                        contextMap))
                 .expectErrorMessage("Retries exhausted: 1/1") //
                 .verify();
 
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollectorTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollectorTest.java
index 83827177..085f5734 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollectorTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollectorTest.java
@@ -39,18 +39,14 @@ import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
 import org.onap.dcaegen2.collectors.datafile.ftp.FtpsClient;
 import org.onap.dcaegen2.collectors.datafile.ftp.Scheme;
 import org.onap.dcaegen2.collectors.datafile.ftp.SftpClient;
-import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
 import org.onap.dcaegen2.collectors.datafile.model.FileData;
-import org.onap.dcaegen2.collectors.datafile.model.ImmutableConsumerDmaapModel;
+import org.onap.dcaegen2.collectors.datafile.model.FilePublishInformation;
 import org.onap.dcaegen2.collectors.datafile.model.ImmutableFileData;
+import org.onap.dcaegen2.collectors.datafile.model.ImmutableFilePublishInformation;
 import org.onap.dcaegen2.collectors.datafile.model.ImmutableMessageMetaData;
 import org.onap.dcaegen2.collectors.datafile.model.MessageMetaData;
 import reactor.test.StepVerifier;
 
-/**
- * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
- *
- */
 public class FileCollectorTest {
     private static final String PRODUCT_NAME = "NrRadio";
     private static final String VENDOR_NAME = "Ericsson";
@@ -65,8 +61,8 @@ public class FileCollectorTest {
     private static final String SERVER_ADDRESS = "192.168.0.101";
     private static final int PORT_22 = 22;
     private static final String PM_FILE_NAME = "A20161224.1030-1045.bin.gz";
-    private static final String REMOTE_FILE_LOCATION = "/ftp/rop/" + PM_FILE_NAME;
     private static final Path LOCAL_FILE_LOCATION = Paths.get(FileData.DATAFILE_TMPDIR, PM_FILE_NAME);
+    private static final String REMOTE_FILE_LOCATION = "/ftp/rop/" + PM_FILE_NAME;
     private static final String USER = "usr";
     private static final String PWD = "pwd";
     private static final String FTPES_LOCATION =
@@ -96,60 +92,57 @@ public class FileCollectorTest {
 
 
     private MessageMetaData createMessageMetaData() {
-        // @formatter:off
-        return ImmutableMessageMetaData.builder()
-                .productName(PRODUCT_NAME)
-                .vendorName(VENDOR_NAME)
-                .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
-                .sourceName(SOURCE_NAME)
-                .startEpochMicrosec(START_EPOCH_MICROSEC)
-                .timeZoneOffset(TIME_ZONE_OFFSET)
-                .changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER)
-                .changeType(FILE_READY_CHANGE_TYPE)
+        return ImmutableMessageMetaData.builder() //
+                .productName(PRODUCT_NAME) //
+                .vendorName(VENDOR_NAME) //
+                .lastEpochMicrosec(LAST_EPOCH_MICROSEC) //
+                .sourceName(SOURCE_NAME) //
+                .startEpochMicrosec(START_EPOCH_MICROSEC) //
+                .timeZoneOffset(TIME_ZONE_OFFSET) //
+                .changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER) //
+                .changeType(FILE_READY_CHANGE_TYPE) //
                 .build();
-        // @formatter:on
     }
 
     private FileData createFileData(String location, Scheme scheme) {
-        // @formatter:off
-        return  ImmutableFileData.builder()
-            .name(PM_FILE_NAME)
-            .location(location)
-            .compression(GZIP_COMPRESSION)
-            .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
-            .fileFormatVersion(FILE_FORMAT_VERSION)
-            .scheme(scheme)
-            .messageMetaData(createMessageMetaData())
-            .build();
-        // @formatter:on
+        return ImmutableFileData.builder() //
+                .name(PM_FILE_NAME) //
+                .location(location) //
+                .compression(GZIP_COMPRESSION) //
+                .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE) //
+                .fileFormatVersion(FILE_FORMAT_VERSION) //
+                .scheme(scheme) //
+                .messageMetaData(createMessageMetaData()) //
+                .build();
     }
 
-    private ConsumerDmaapModel createExpectedConsumerDmaapModel(String location) {
-        // @formatter:off
-        return ImmutableConsumerDmaapModel.builder()
-            .productName(PRODUCT_NAME)
-            .vendorName(VENDOR_NAME)
-            .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
-            .sourceName(SOURCE_NAME)
-            .startEpochMicrosec(START_EPOCH_MICROSEC)
-            .timeZoneOffset(TIME_ZONE_OFFSET)
-            .name(PM_FILE_NAME)
-            .location(location)
-            .internalLocation(LOCAL_FILE_LOCATION)
-            .compression(GZIP_COMPRESSION)
-            .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
-            .fileFormatVersion(FILE_FORMAT_VERSION)
-            .build();
-      // @formatter:on
+    private FilePublishInformation createExpectedFilePublishInformation(String location) {
+        return ImmutableFilePublishInformation.builder() //
+                .productName(PRODUCT_NAME) //
+                .vendorName(VENDOR_NAME) //
+                .lastEpochMicrosec(LAST_EPOCH_MICROSEC) //
+                .sourceName(SOURCE_NAME) //
+                .startEpochMicrosec(START_EPOCH_MICROSEC) //
+                .timeZoneOffset(TIME_ZONE_OFFSET) //
+                .name(PM_FILE_NAME) //
+                .location(location) //
+                .internalLocation(LOCAL_FILE_LOCATION) //
+                .compression(GZIP_COMPRESSION) //
+                .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE) //
+                .fileFormatVersion(FILE_FORMAT_VERSION) //
+                .build();
     }
 
+    /**
+     * Sets up the configuration.
+     */
     @BeforeAll
     public static void setUpConfiguration() {
         when(appConfigMock.getFtpesConfiguration()).thenReturn(ftpesConfigMock);
         when(ftpesConfigMock.keyCert()).thenReturn(FTP_KEY_PATH);
         when(ftpesConfigMock.keyPassword()).thenReturn(FTP_KEY_PASSWORD);
-        when(ftpesConfigMock.trustedCA()).thenReturn(TRUSTED_CA_PATH);
-        when(ftpesConfigMock.trustedCAPassword()).thenReturn(TRUSTED_CA_PASSWORD);
+        when(ftpesConfigMock.trustedCa()).thenReturn(TRUSTED_CA_PATH);
+        when(ftpesConfigMock.trustedCaPassword()).thenReturn(TRUSTED_CA_PASSWORD);
     }
 
     @Test
@@ -159,11 +152,12 @@ public class FileCollectorTest {
 
         FileData fileData = createFileData(FTPES_LOCATION_NO_PORT, Scheme.FTPS);
 
-        ConsumerDmaapModel expectedConsumerDmaapModel = createExpectedConsumerDmaapModel(FTPES_LOCATION_NO_PORT);
+        FilePublishInformation expectedfilePublishInformation =
+                createExpectedFilePublishInformation(FTPES_LOCATION_NO_PORT);
 
-        StepVerifier.create(
-                collectorUndetTest.execute(fileData, 3, Duration.ofSeconds(0), contextMap))
-                .expectNext(expectedConsumerDmaapModel).verifyComplete();
+        StepVerifier.create(collectorUndetTest.collectFile(fileData, 3, Duration.ofSeconds(0), contextMap))
+                .expectNext(expectedfilePublishInformation) //
+                .verifyComplete();
 
         verify(ftpsClientMock, times(1)).open();
         verify(ftpsClientMock, times(1)).collectFile(REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION);
@@ -179,22 +173,19 @@ public class FileCollectorTest {
 
 
         FileData fileData = createFileData(SFTP_LOCATION_NO_PORT, Scheme.SFTP);
-        ConsumerDmaapModel expectedConsumerDmaapModel = createExpectedConsumerDmaapModel(SFTP_LOCATION_NO_PORT);
+        FilePublishInformation expectedfilePublishInformation =
+                createExpectedFilePublishInformation(SFTP_LOCATION_NO_PORT);
 
-        StepVerifier
-                .create(collectorUndetTest.execute(fileData, 3, Duration.ofSeconds(0),
-                        contextMap))
-                .expectNext(expectedConsumerDmaapModel) //
+        StepVerifier.create(collectorUndetTest.collectFile(fileData, 3, Duration.ofSeconds(0), contextMap))
+                .expectNext(expectedfilePublishInformation) //
                 .verifyComplete();
 
         // The same again, but with port
         fileData = createFileData(SFTP_LOCATION, Scheme.SFTP);
-        expectedConsumerDmaapModel = createExpectedConsumerDmaapModel(SFTP_LOCATION);
+        expectedfilePublishInformation = createExpectedFilePublishInformation(SFTP_LOCATION);
 
-        StepVerifier
-                .create(collectorUndetTest.execute(fileData, 3, Duration.ofSeconds(0),
-                        contextMap))
-                .expectNext(expectedConsumerDmaapModel) //
+        StepVerifier.create(collectorUndetTest.collectFile(fileData, 3, Duration.ofSeconds(0), contextMap))
+                .expectNext(expectedfilePublishInformation) //
                 .verifyComplete();
 
         verify(sftpClientMock, times(2)).open();
@@ -212,9 +203,9 @@ public class FileCollectorTest {
         doThrow(new DatafileTaskException("Unable to collect file.")).when(ftpsClientMock)
                 .collectFile(REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION);
 
-        StepVerifier.create(
-                collectorUndetTest.execute(fileData, 3, Duration.ofSeconds(0), contextMap))
-                .expectErrorMessage("Retries exhausted: 3/3").verify();
+        StepVerifier.create(collectorUndetTest.collectFile(fileData, 3, Duration.ofSeconds(0), contextMap))
+                .expectErrorMessage("Retries exhausted: 3/3") //
+                .verify();
 
         verify(ftpsClientMock, times(4)).collectFile(REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION);
     }
@@ -226,13 +217,14 @@ public class FileCollectorTest {
         doThrow(new DatafileTaskException("Unable to collect file.")).doNothing().when(ftpsClientMock)
                 .collectFile(REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION);
 
-        ConsumerDmaapModel expectedConsumerDmaapModel = createExpectedConsumerDmaapModel(FTPES_LOCATION_NO_PORT);
+        FilePublishInformation expectedfilePublishInformation =
+                createExpectedFilePublishInformation(FTPES_LOCATION_NO_PORT);
 
         FileData fileData = createFileData(FTPES_LOCATION_NO_PORT, Scheme.FTPS);
 
-        StepVerifier.create(
-                collectorUndetTest.execute(fileData, 3, Duration.ofSeconds(0), contextMap))
-                .expectNext(expectedConsumerDmaapModel).verifyComplete();
+        StepVerifier.create(collectorUndetTest.collectFile(fileData, 3, Duration.ofSeconds(0), contextMap))
+                .expectNext(expectedfilePublishInformation) //
+                .verifyComplete();
 
         verify(ftpsClientMock, times(2)).collectFile(REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION);
     }
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/PublishedCheckerTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/PublishedCheckerTest.java
index d5f65150..83643637 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/PublishedCheckerTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/PublishedCheckerTest.java
@@ -38,7 +38,6 @@ import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.HashMap;
 import java.util.Map;
-
 import org.apache.http.HttpEntity;
 import org.apache.http.HttpResponse;
 import org.apache.http.StatusLine;
@@ -55,10 +54,6 @@ import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPub
 import org.springframework.web.util.DefaultUriBuilderFactory;
 import org.springframework.web.util.UriBuilder;
 
-/**
- * @author <a href="mailto:maxime.bonneau@est.tech">Maxime Bonneau</a>
- *
- */
 public class PublishedCheckerTest {
     private static final String EMPTY_CONTENT = "[]";
     private static final String FEEDLOG_TOPIC = "feedlog";
@@ -95,7 +90,7 @@ public class PublishedCheckerTest {
     public void executeWhenNotPublished_returnsFalse() throws Exception {
         prepareMocksForTests(HttpUtils.SC_OK, EMPTY_CONTENT, null);
 
-        boolean isPublished = publishedCheckerUnderTestSpy.execute(LOCAL_FILE_NAME, CONTEXT_MAP);
+        boolean isPublished = publishedCheckerUnderTestSpy.isFilePublished(LOCAL_FILE_NAME, CONTEXT_MAP);
 
         assertFalse(isPublished);
 
@@ -123,7 +118,7 @@ public class PublishedCheckerTest {
     public void executeWhenDataRouterReturnsNok_returnsFalse() throws Exception {
         prepareMocksForTests(HttpUtils.SC_BAD_REQUEST, EMPTY_CONTENT, null);
 
-        boolean isPublished = publishedCheckerUnderTestSpy.execute(LOCAL_FILE_NAME, CONTEXT_MAP);
+        boolean isPublished = publishedCheckerUnderTestSpy.isFilePublished(LOCAL_FILE_NAME, CONTEXT_MAP);
 
         assertFalse(isPublished);
     }
@@ -132,7 +127,7 @@ public class PublishedCheckerTest {
     public void executeWhenPublished_returnsTrue() throws Exception {
         prepareMocksForTests(HttpUtils.SC_OK, "[" + LOCAL_FILE_NAME + "]", null);
 
-        boolean isPublished = publishedCheckerUnderTestSpy.execute(LOCAL_FILE_NAME, CONTEXT_MAP);
+        boolean isPublished = publishedCheckerUnderTestSpy.isFilePublished(LOCAL_FILE_NAME, CONTEXT_MAP);
 
         assertTrue(isPublished);
     }
@@ -141,7 +136,7 @@ public class PublishedCheckerTest {
     public void executeWhenErrorInDataRouter_returnsFalse() throws Exception {
         prepareMocksForTests(HttpUtils.SC_OK, EMPTY_CONTENT, new DatafileTaskException(""));
 
-        boolean isPublished = publishedCheckerUnderTestSpy.execute(LOCAL_FILE_NAME, CONTEXT_MAP);
+        boolean isPublished = publishedCheckerUnderTestSpy.isFilePublished(LOCAL_FILE_NAME, CONTEXT_MAP);
 
         assertFalse(isPublished);
     }
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasksTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasksTest.java
index 09908f13..24bb759f 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasksTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasksTest.java
@@ -1,17 +1,21 @@
-/*
- * ============LICENSE_START======================================================================
- * Copyright (C) 2019 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
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2019 Nordix Foundation.
+ * ================================================================================
+ * 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.
- * ============LICENSE_END========================================================================
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
  */
 
 package org.onap.dcaegen2.collectors.datafile.tasks;
@@ -34,22 +38,20 @@ import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
-
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
 import org.onap.dcaegen2.collectors.datafile.ftp.Scheme;
-import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
 import org.onap.dcaegen2.collectors.datafile.model.FileData;
+import org.onap.dcaegen2.collectors.datafile.model.FilePublishInformation;
 import org.onap.dcaegen2.collectors.datafile.model.FileReadyMessage;
-import org.onap.dcaegen2.collectors.datafile.model.ImmutableConsumerDmaapModel;
 import org.onap.dcaegen2.collectors.datafile.model.ImmutableFileData;
+import org.onap.dcaegen2.collectors.datafile.model.ImmutableFilePublishInformation;
 import org.onap.dcaegen2.collectors.datafile.model.ImmutableFileReadyMessage;
 import org.onap.dcaegen2.collectors.datafile.model.ImmutableMessageMetaData;
 import org.onap.dcaegen2.collectors.datafile.model.MessageMetaData;
 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPublisherConfiguration;
 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.ImmutableDmaapPublisherConfiguration;
-
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
@@ -62,7 +64,7 @@ public class ScheduledTasksTest {
     private ScheduledTasks testedObject = spy(new ScheduledTasks(appConfig));
 
     private int uniqueValue = 0;
-    private DMaaPMessageConsumerTask consumerMock;
+    private DMaaPMessageConsumer consumerMock;
     private PublishedChecker publishedCheckerMock;
     private FileCollector fileCollectorMock;
     private DataRouterPublisher dataRouterMock;
@@ -86,7 +88,7 @@ public class ScheduledTasksTest {
                 .build(); //
         doReturn(dmaapPublisherConfiguration).when(appConfig).getDmaapPublisherConfiguration();
 
-        consumerMock = mock(DMaaPMessageConsumerTask.class);
+        consumerMock = mock(DMaaPMessageConsumer.class);
         publishedCheckerMock = mock(PublishedChecker.class);
         fileCollectorMock = mock(FileCollector.class);
         dataRouterMock = mock(DataRouterPublisher.class);
@@ -118,7 +120,7 @@ public class ScheduledTasksTest {
                 .location("ftpes://192.168.0.101/ftp/rop/" + PM_FILE_NAME + instanceNumber) //
                 .scheme(Scheme.FTPS) //
                 .compression("") //
-                .messageMetaData(messageMetaData())
+                .messageMetaData(messageMetaData()) //
                 .build();
     }
 
@@ -145,8 +147,8 @@ public class ScheduledTasksTest {
         return Flux.fromIterable(list);
     }
 
-    private ConsumerDmaapModel consumerData() {
-        return ImmutableConsumerDmaapModel //
+    private FilePublishInformation filePublishInformation() {
+        return ImmutableFilePublishInformation //
                 .builder() //
                 .productName("") //
                 .vendorName("") //
@@ -166,12 +168,12 @@ public class ScheduledTasksTest {
     @Test
     public void notingToConsume() {
         doReturn(consumerMock).when(testedObject).createConsumerTask();
-        doReturn(Flux.empty()).when(consumerMock).execute();
+        doReturn(Flux.empty()).when(consumerMock).getMessageRouterResponse();
 
         testedObject.executeDatafileMainTask();
 
         assertEquals(0, testedObject.getCurrentNumberOfTasks());
-        verify(consumerMock, times(1)).execute();
+        verify(consumerMock, times(1)).getMessageRouterResponse();
         verifyNoMoreInteractions(consumerMock);
     }
 
@@ -182,13 +184,13 @@ public class ScheduledTasksTest {
         final int noOfFiles = noOfEvents * noOfFilesPerEvent;
 
         Flux<FileReadyMessage> fileReadyMessages = fileReadyMessageFlux(noOfEvents, noOfFilesPerEvent, true);
-        doReturn(fileReadyMessages).when(consumerMock).execute();
+        doReturn(fileReadyMessages).when(consumerMock).getMessageRouterResponse();
 
-        doReturn(false).when(publishedCheckerMock).execute(anyString(), any());
+        doReturn(false).when(publishedCheckerMock).isFilePublished(anyString(), any());
 
-        Mono<ConsumerDmaapModel> collectedFile = Mono.just(consumerData());
-        doReturn(collectedFile).when(fileCollectorMock).execute(notNull(), anyLong(), notNull(), notNull());
-        doReturn(collectedFile).when(dataRouterMock).execute(notNull(), anyLong(), notNull(), any());
+        Mono<FilePublishInformation> collectedFile = Mono.just(filePublishInformation());
+        doReturn(collectedFile).when(fileCollectorMock).collectFile(notNull(), anyLong(), notNull(), notNull());
+        doReturn(collectedFile).when(dataRouterMock).publishFile(notNull(), anyLong(), notNull(), any());
 
         StepVerifier.create(testedObject.createMainTask(contextMap)).expectSubscription() //
                 .expectNextCount(noOfFiles) //
@@ -196,9 +198,9 @@ public class ScheduledTasksTest {
                 .verify(); //
 
         assertEquals(0, testedObject.getCurrentNumberOfTasks());
-        verify(consumerMock, times(1)).execute();
-        verify(fileCollectorMock, times(noOfFiles)).execute(notNull(), anyLong(), notNull(), notNull());
-        verify(dataRouterMock, times(noOfFiles)).execute(notNull(), anyLong(), notNull(), any());
+        verify(consumerMock, times(1)).getMessageRouterResponse();
+        verify(fileCollectorMock, times(noOfFiles)).collectFile(notNull(), anyLong(), notNull(), notNull());
+        verify(dataRouterMock, times(noOfFiles)).publishFile(notNull(), anyLong(), notNull(), any());
         verifyNoMoreInteractions(dataRouterMock);
         verifyNoMoreInteractions(fileCollectorMock);
         verifyNoMoreInteractions(consumerMock);
@@ -207,20 +209,20 @@ public class ScheduledTasksTest {
     @Test
     public void consume_fetchFailedOnce() {
         Flux<FileReadyMessage> fileReadyMessages = fileReadyMessageFlux(2, 2, true); // 4 files
-        doReturn(fileReadyMessages).when(consumerMock).execute();
+        doReturn(fileReadyMessages).when(consumerMock).getMessageRouterResponse();
 
-        doReturn(false).when(publishedCheckerMock).execute(anyString(), any());
+        doReturn(false).when(publishedCheckerMock).isFilePublished(anyString(), any());
 
-        Mono<ConsumerDmaapModel> collectedFile = Mono.just(consumerData());
+        Mono<FilePublishInformation> collectedFile = Mono.just(filePublishInformation());
         Mono<Object> error = Mono.error(new Exception("problem"));
 
         // First file collect will fail, 3 will succeed
         doReturn(error, collectedFile, collectedFile, collectedFile) //
                 .when(fileCollectorMock) //
-                .execute(any(FileData.class), anyLong(), any(Duration.class), notNull());
+                .collectFile(any(FileData.class), anyLong(), any(Duration.class), notNull());
 
-        doReturn(collectedFile).when(dataRouterMock).execute(notNull(), anyLong(), notNull(), any());
-        doReturn(collectedFile).when(dataRouterMock).execute(notNull(), anyLong(), notNull(), any());
+        doReturn(collectedFile).when(dataRouterMock).publishFile(notNull(), anyLong(), notNull(), any());
+        doReturn(collectedFile).when(dataRouterMock).publishFile(notNull(), anyLong(), notNull(), any());
 
         StepVerifier.create(testedObject.createMainTask(contextMap)).expectSubscription() //
                 .expectNextCount(3) //
@@ -228,9 +230,9 @@ public class ScheduledTasksTest {
                 .verify(); //
 
         assertEquals(0, testedObject.getCurrentNumberOfTasks());
-        verify(consumerMock, times(1)).execute();
-        verify(fileCollectorMock, times(4)).execute(notNull(), anyLong(), notNull(), notNull());
-        verify(dataRouterMock, times(3)).execute(notNull(), anyLong(), notNull(), any());
+        verify(consumerMock, times(1)).getMessageRouterResponse();
+        verify(fileCollectorMock, times(4)).collectFile(notNull(), anyLong(), notNull(), notNull());
+        verify(dataRouterMock, times(3)).publishFile(notNull(), anyLong(), notNull(), any());
         verifyNoMoreInteractions(dataRouterMock);
         verifyNoMoreInteractions(fileCollectorMock);
         verifyNoMoreInteractions(consumerMock);
@@ -240,18 +242,18 @@ public class ScheduledTasksTest {
     public void consume_publishFailedOnce() {
 
         Flux<FileReadyMessage> fileReadyMessages = fileReadyMessageFlux(2, 2, true); // 4 files
-        doReturn(fileReadyMessages).when(consumerMock).execute();
+        doReturn(fileReadyMessages).when(consumerMock).getMessageRouterResponse();
 
-        doReturn(false).when(publishedCheckerMock).execute(anyString(), any());
+        doReturn(false).when(publishedCheckerMock).isFilePublished(anyString(), any());
 
-        Mono<ConsumerDmaapModel> collectedFile = Mono.just(consumerData());
-        doReturn(collectedFile).when(fileCollectorMock).execute(notNull(), anyLong(), notNull(), notNull());
+        Mono<FilePublishInformation> collectedFile = Mono.just(filePublishInformation());
+        doReturn(collectedFile).when(fileCollectorMock).collectFile(notNull(), anyLong(), notNull(), notNull());
 
         Mono<Object> error = Mono.error(new Exception("problem"));
         // One publish will fail, the rest will succeed
         doReturn(collectedFile, error, collectedFile, collectedFile) //
                 .when(dataRouterMock) //
-                .execute(notNull(), anyLong(), notNull(), any());
+                .publishFile(notNull(), anyLong(), notNull(), any());
 
         StepVerifier.create(testedObject.createMainTask(contextMap)).expectSubscription() //
                 .expectNextCount(3) // 3 completed files
@@ -259,9 +261,9 @@ public class ScheduledTasksTest {
                 .verify(); //
 
         assertEquals(0, testedObject.getCurrentNumberOfTasks());
-        verify(consumerMock, times(1)).execute();
-        verify(fileCollectorMock, times(4)).execute(notNull(), anyLong(), notNull(), notNull());
-        verify(dataRouterMock, times(4)).execute(notNull(), anyLong(), notNull(), any());
+        verify(consumerMock, times(1)).getMessageRouterResponse();
+        verify(fileCollectorMock, times(4)).collectFile(notNull(), anyLong(), notNull(), notNull());
+        verify(dataRouterMock, times(4)).publishFile(notNull(), anyLong(), notNull(), any());
         verifyNoMoreInteractions(dataRouterMock);
         verifyNoMoreInteractions(fileCollectorMock);
         verifyNoMoreInteractions(consumerMock);
@@ -274,13 +276,13 @@ public class ScheduledTasksTest {
 
         // 100 files with the same name
         Flux<FileReadyMessage> fileReadyMessages = fileReadyMessageFlux(noOfEvents, noOfFilesPerEvent, false);
-        doReturn(fileReadyMessages).when(consumerMock).execute();
+        doReturn(fileReadyMessages).when(consumerMock).getMessageRouterResponse();
 
-        doReturn(false).when(publishedCheckerMock).execute(anyString(), any());
+        doReturn(false).when(publishedCheckerMock).isFilePublished(anyString(), any());
 
-        Mono<ConsumerDmaapModel> collectedFile = Mono.just(consumerData());
-        doReturn(collectedFile).when(fileCollectorMock).execute(notNull(), anyLong(), notNull(), notNull());
-        doReturn(collectedFile).when(dataRouterMock).execute(notNull(), anyLong(), notNull(), any());
+        Mono<FilePublishInformation> collectedFile = Mono.just(filePublishInformation());
+        doReturn(collectedFile).when(fileCollectorMock).collectFile(notNull(), anyLong(), notNull(), notNull());
+        doReturn(collectedFile).when(dataRouterMock).publishFile(notNull(), anyLong(), notNull(), any());
 
         StepVerifier.create(testedObject.createMainTask(contextMap)).expectSubscription() //
                 .expectNextCount(1) // 99 is skipped
@@ -288,13 +290,11 @@ public class ScheduledTasksTest {
                 .verify(); //
 
         assertEquals(0, testedObject.getCurrentNumberOfTasks());
-        verify(consumerMock, times(1)).execute();
-        verify(fileCollectorMock, times(1)).execute(notNull(), anyLong(), notNull(), notNull());
-        verify(dataRouterMock, times(1)).execute(notNull(), anyLong(), notNull(), any());
+        verify(consumerMock, times(1)).getMessageRouterResponse();
+        verify(fileCollectorMock, times(1)).collectFile(notNull(), anyLong(), notNull(), notNull());
+        verify(dataRouterMock, times(1)).publishFile(notNull(), anyLong(), notNull(), any());
         verifyNoMoreInteractions(dataRouterMock);
         verifyNoMoreInteractions(fileCollectorMock);
         verifyNoMoreInteractions(consumerMock);
     }
-
-
 }
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/utils/JsonMessage.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/utils/JsonMessage.java
index 733aa3e8..cc40dc67 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/utils/JsonMessage.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/utils/JsonMessage.java
@@ -1,17 +1,21 @@
-/*
- * ============LICENSE_START======================================================================
- * Copyright (C) 2018-2019 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
+/*-
+ * ============LICENSE_START=======================================================
+ * 2018-2019 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
+ *      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========================================================================
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
  */
 
 package org.onap.dcaegen2.collectors.datafile.utils;
@@ -57,35 +61,40 @@ public class JsonMessage {
             }
             additionalFieldsString.append("]");
         }
-        // @formatter:off
-        return "{" + "\"event\":{"
-                + "\"commonEventHeader\":{"
-                + "\"domain\":\"notification\","
-                + "\"eventId\":\"<<SerialNumber>>-reg\","
-                + "\"eventName\":\"" + eventName + "\","
-                + "\"eventType\":\"fileReady\","
-                + "\"internalHeaderFields\":{},"
-                + "\"lastEpochMicrosec\":1519837825682,"
-                + "\"nfNamingCode\":\"5GRAN\","
-                + "\"nfcNamingCode\":\"5DU\","
-                + "\"priority\":\"Normal\","
-                + "\"reportingEntityName\":\"5GRAN_DU\","
-                + "\"sequence\":0,"
-                + "\"sourceId\":\"<<SerialNumber>>\","
-                + "\"sourceName\":\"5GRAN_DU\","
-                + "\"timeZoneOffset\":\"UTC+05:00\","
-                + "\"startEpochMicrosec\":\"1519837825682\","
-                + "\"version\":3"
-                + "},"
-                + "\"notificationFields\":{"
+        return "{" //
+                + "\"event\":" //
+                + "{" //
+                + "\"commonEventHeader\":" //
+                + "{" //
+                + "\"domain\":\"notification\"," //
+                + "\"eventId\":\"<<SerialNumber>>-reg\"," //
+                + "\"eventName\":\"" + eventName + "\"," //
+                + "\"eventType\":\"fileReady\"," //
+                + "\"internalHeaderFields\":{}," //
+                + "\"lastEpochMicrosec\":1519837825682," //
+                + "\"nfNamingCode\":\"5GRAN\"," //
+                + "\"nfcNamingCode\":\"5DU\"," //
+                + "\"priority\":\"Normal\"," //
+                + "\"reportingEntityName\":\"5GRAN_DU\"," //
+                + "\"sequence\":0," //
+                + "\"sourceId\":\"<<SerialNumber>>\"," //
+                + "\"sourceName\":\"5GRAN_DU\"," //
+                + "\"timeZoneOffset\":\"UTC+05:00\"," //
+                + "\"startEpochMicrosec\":\"1519837825682\"," //
+                + "\"version\":3" //
+                + "}," //
+                + "\"notificationFields\":" //
+                + "{" //
                 + getAsStringIfParameterIsSet("changeIdentifier", changeIdentifier,
                         changeType != null || notificationFieldsVersion != null || arrayOfAdditionalFields.size() > 0)
                 + getAsStringIfParameterIsSet("changeType", changeType,
                         notificationFieldsVersion != null || arrayOfAdditionalFields.size() > 0)
                 + getAsStringIfParameterIsSet("notificationFieldsVersion", notificationFieldsVersion,
                         arrayOfAdditionalFields.size() > 0)
-                + additionalFieldsString.toString() + "}" + "}" + "}";
-        // @formatter:on
+                + additionalFieldsString.toString() //
+                + "}" //
+                + "}" //
+                + "}";
     }
 
     private JsonMessage(final JsonMessageBuilder builder) {
@@ -105,16 +114,20 @@ public class JsonMessage {
 
         @Override
         public String toString() {
-            return "{" + getAsStringIfParameterIsSet("name", name, true) + "\"hashMap\":{"
+            return "{" //
+                    + getAsStringIfParameterIsSet("name", name, true) //
+                    + "\"hashMap\":" //
+                    + "{"
                     + getAsStringIfParameterIsSet("location", location,
                             compression != null || fileFormatType != null || fileFormatVersion != null)
                     + getAsStringIfParameterIsSet("compression", compression,
                             fileFormatType != null || fileFormatVersion != null)
                     + getAsStringIfParameterIsSet("fileFormatType", fileFormatType, fileFormatVersion != null)
-                    + getAsStringIfParameterIsSet("fileFormatVersion", fileFormatVersion, false) + "}}";
+                    + getAsStringIfParameterIsSet("fileFormatVersion", fileFormatVersion, false) //
+                    + "}" //
+                    + "}";
         }
 
-
         private AdditionalField(AdditionalFieldBuilder builder) {
             this.name = builder.name;
             this.location = builder.location;
@@ -214,35 +227,33 @@ public class JsonMessage {
 
     /**
      * Can be used to produce a correct test Json message. Tip! Check the formatting with
-     * <a href="https://jsonformatter.org/">Json fomatter</a>
+     * <a href="https://jsonformatter.org/">Json formatter</a>
      *
      * @param args Not used
      */
     public static void main(String[] args) {
-        // @formatter:off
-        AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder()
-                .name("A20161224.1030-1045.bin.gz")
-                .location("ftpes://192.168.0.101:22/ftp/rop/A20161224.1030-1045.bin.gz")
-                .compression("gzip")
-                .fileFormatType("org.3GPP.32.435#measCollec")
-                .fileFormatVersion("V10")
+        AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder() //
+                .name("A20161224.1030-1045.bin.gz") //
+                .location("ftpes://192.168.0.101:22/ftp/rop/A20161224.1030-1045.bin.gz") //
+                .compression("gzip") //
+                .fileFormatType("org.3GPP.32.435#measCollec") //
+                .fileFormatVersion("V10") //
                 .build();
-        AdditionalField secondAdditionalField = new JsonMessage.AdditionalFieldBuilder()
-                .name("A20161224.1030-1045.bin.gz")
-                .location("sftp://192.168.0.101:22/ftp/rop/A20161224.1030-1045.bin.gz")
-                .compression("gzip")
-                .fileFormatType("org.3GPP.32.435#measCollec")
-                .fileFormatVersion("V10")
+        AdditionalField secondAdditionalField = new JsonMessage.AdditionalFieldBuilder() //
+                .name("A20161224.1030-1045.bin.gz") //
+                .location("sftp://192.168.0.101:22/ftp/rop/A20161224.1030-1045.bin.gz") //
+                .compression("gzip") //
+                .fileFormatType("org.3GPP.32.435#measCollec") //
+                .fileFormatVersion("V10") //
                 .build();
-        JsonMessage message = new JsonMessage.JsonMessageBuilder()
-                .eventName("Noti_NrRadio-Ericsson_FileReady")
-                .changeIdentifier("PM_MEAS_FILES")
-                .changeType("FileReady")
-                .notificationFieldsVersion("2.0")
-                .addAdditionalField(additionalField)
-                .addAdditionalField(secondAdditionalField)
+        JsonMessage message = new JsonMessage.JsonMessageBuilder() //
+                .eventName("Noti_NrRadio-Ericsson_FileReady") //
+                .changeIdentifier("PM_MEAS_FILES") //
+                .changeType("FileReady") //
+                .notificationFieldsVersion("2.0") //
+                .addAdditionalField(additionalField) //
+                .addAdditionalField(secondAdditionalField) //
                 .build();
-        // @formatter:on
         System.out.println(message.toString());
     }
 }
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/web/PublishRedirectStrategyTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/web/PublishRedirectStrategyTest.java
new file mode 100644
index 00000000..298656a8
--- /dev/null
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/web/PublishRedirectStrategyTest.java
@@ -0,0 +1,74 @@
+/*
+ * ============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 static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.apache.http.Header;
+import org.apache.http.HttpRequest;
+import org.apache.http.HttpResponse;
+import org.apache.http.ProtocolException;
+import org.apache.http.RequestLine;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.client.protocol.HttpClientContext;
+import org.apache.http.protocol.HttpContext;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+class PublishRedirectStrategyTest {
+
+    private static final String URI = "sftp://localhost:80/";
+
+    private static PublishRedirectStrategy publishRedirectStrategy;
+
+    @BeforeAll
+    static void setUp() {
+        publishRedirectStrategy = new PublishRedirectStrategy();
+    }
+
+    @Test
+    void isRedirectable_shouldReturnTrue() {
+        Assertions.assertTrue(publishRedirectStrategy.isRedirectable("Put"));
+    }
+
+    @Test
+    void isRedirectable_shouldReturnFalse() {
+        Assertions.assertFalse(publishRedirectStrategy.isRedirectable("not valid method"));
+    }
+
+    @Test
+    void getRedirect_shouldReturnCorrectUri() throws ProtocolException {
+        HttpRequest requestMock = mock(HttpRequest.class);
+        HttpResponse responseMock = mock(HttpResponse.class);
+        HttpContext contextMock = mock(HttpContext.class);
+        Header headerMock = mock(Header.class);
+        when(responseMock.getFirstHeader("location")).thenReturn(headerMock);
+        when(headerMock.getValue()).thenReturn(URI);
+        RequestConfig requestConfigMock = mock(RequestConfig.class);
+        when(contextMock.getAttribute(HttpClientContext.REQUEST_CONFIG)).thenReturn(requestConfigMock);
+        RequestLine requestLineMock = mock(RequestLine.class);
+        when(requestMock.getRequestLine()).thenReturn(requestLineMock);
+        when(requestLineMock.getUri()).thenReturn(URI);
+
+        HttpUriRequest actualRedirect = publishRedirectStrategy.getRedirect(requestMock, responseMock, contextMock);
+        assertEquals(URI, actualRedirect.getURI().toString());
+    }
+}