Deliver first version of Datafile

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

18 files changed, 858 insertions, 1024 deletions

diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/DatafileAppConfigTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/DatafileAppConfigTest.java
index 7d54b4d5..12c8c7e7 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/DatafileAppConfigTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/DatafileAppConfigTest.java
@@ -1,21 +1,17 @@
 /*
- * ============LICENSE_START=======================================================
- * Datafile Collector Service
- * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. 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) 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
+ * 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.
+ * ============LICENSE_END========================================================================
  */
 
 package org.onap.dcaegen2.collectors.datafile.configuration;
@@ -31,154 +27,122 @@ import static org.mockito.Mockito.when;
 
 import com.google.gson.JsonElement;
 import com.google.gson.JsonParser;
+
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
 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.configuration.AppConfig;
-import org.onap.dcaegen2.collectors.datafile.configuration.DatafileAppConfig;
 import org.onap.dcaegen2.collectors.datafile.integration.junit5.mockito.MockitoExtension;
 
 /**
  * @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 DatafileAppConfigTest {
 
     private static final String DATAFILE_ENDPOINTS = "datafile_endpoints.json";
-    private static final String jsonString = "{\"configs\":{\"aai\":{\"aaiClientConfiguration\":{\"aaiHost\":"
-        + "\"localhost\",\"aaiPort\":8080,\"aaiIgnoreSslCertificateErrors\":true,\"aaiProtocol\":"
-        + "\"https\",\"aaiUserName\":\"admin\",\"aaiUserPassword\":\"admin\",\"aaiBasePath\":\"/aai/v11\","
-        + "\"aaiPnfPath\":\"/network/pnfs/pnf\",\"aaiHeaders\":{\"X-FromAppId\":\"datafile\",\"X-TransactionId\":\"9999\","
-        + "\"Accept\":\"application/json\",\"Real-Time\":\"true\",\"Content-Type\":\"application/merge-patch+json\","
-        + "\"Authorization\":\"Basic QUFJOkFBSQ==\"}}},"
-        + "\"dmaap\":{\"dmaapConsumerConfiguration\":{\"consumerGroup\":\"other\",\"consumerId\":\"1\","
-        + "\"dmaapContentType\":\"application/json\",\"dmaapHostName\":\"localhost\",\"dmaapPortNumber\":2222,"
-        + "\"dmaapProtocol\":\"http\",\"dmaapTopicName\":\"temp\",\"dmaapUserName\":\"admin\",\"dmaapUserPassword\""
-        + ":\"admin\",\"messageLimit\":1000,\"timeoutMs\":1000},\"dmaapProducerConfiguration\":{\"dmaapContentType\":"
-        + "\"application/json\",\"dmaapHostName\":\"localhost\",\"dmaapPortNumber\":2223,\"dmaapProtocol\":\"http\","
-        + "\"dmaapTopicName\":\"temp\",\"dmaapUserName\":\"admin\",\"dmaapUserPassword\":\"admin\"}}}}";
-
-    private static final String incorrectJsonString = "{\"configs\":{\"aai\":{\"aaiClientConfiguration\":{\"aaiHost\":"
-        + "\"localhost\",\"aaiPort\":8080,\"aaiIgnoreSslCertificateErrors\":true,\"aaiProtocol\":\"https\","
-        + "\"aaiUserName\":\"admin\",\"aaiUserPassword\":\"admin\",\"aaiBasePath\":\"/aai/v11\",\"aaiPnfPath\":"
-        + "\"/network/pnfs/pnf\",\"aaiHeaders\":{\"X-FromAppId\":\"datafile\",\"X-TransactionId\":\"9999\",\"Accept\":"
-        + "\"application/json\",\"Real-Time\":\"true\",\"Content-Type\":\"application/merge-patch+json\","
-        + "\"Authorization\":\"Basic QUFJOkFBSQ==\"}}},\"dmaap\""
-        + ":{\"dmaapConsumerConfiguration\":{\"consumerGroup\":\"other\",\"consumerId\":\"1\",\"dmaapContentType\""
-        + ":\"application/json\",\"dmaapHostName\":\"localhost\",\"dmaapPortNumber\":2222,\"dmaapProtocol\":\"http\""
-        + ",\"dmaapTopicName\":\"temp\",\"dmaapUserName\":\"admin\",\"dmaapUserPassword\":\"admin\",\"messageLimit\""
-        + ":1000,\"timeoutMs\":1000},\"dmaapProducerConfiguration\":{\"dmaapContentType\":\"application/json\","
-        + "\"dmaapHostName\":\"localhost\",\"dmaapPortNumber\":2223,\"dmaapProtocol\":\"http\",\"dmaaptopicName\""
-        + ":\"temp\",\"dmaapuserName\":\"admin\",\"dmaapuserPassword\":\"admin\"}}}}";
+    private static final String JSON_STRING =
+            "{\"configs\":{\"dmaap\":{\"dmaapConsumerConfiguration\":{\"consumerGroup\":\"other\",\"consumerId\":\"1\","
+                    + "\"dmaapContentType\":\"application/json\",\"dmaapHostName\":\"localhost\","
+                    + "\"dmaapPortNumber\":2222,\"dmaapProtocol\":\"http\",\"dmaapTopicName\":\"temp\","
+                    + "\"dmaapUserName\":\"admin\",\"dmaapUserPassword\":\"admin\",\"messageLimit\":1000,"
+                    + "\"timeoutMS\":1000},\"dmaapProducerConfiguration\":{\"dmaapContentType\":\"application/json\","
+                    + "\"dmaapHostName\":\"localhost\",\"dmaapPortNumber\":2223,\"dmaapProtocol\":\"http\","
+                    + "\"dmaapTopicName\":\"temp\",\"dmaapUserName\":\"admin\",\"dmaapUserPassword\":\"admin\"}}}}";
+    private static final String INCORRECT_JSON_STRING =
+            "{\"configs\":{\"dmaap\":{\"dmaapConsumerConfiguration\":{\"consumerGroup\":\"other\",\"consumerId\":\"1\","
+                    + "\"dmaapContentType\":\"application/json\",\"dmaapHostName\":\"localhost\","
+                    + "\"dmaapPortNumber\":2222,\"dmaapProtocol\":\"http\",\"dmaapTopicName\":\"temp\","
+                    + "\"dmaapUserName\":\"admin\",\"dmaapUserPassword\":\"admin\",\"messageLimit\":1000,"
+                    + "\"timeoutMS\":1000},\"dmaapProducerConfiguration\":{\"dmaapContentType\":\"application/json\","
+                    + "\"dmaapHostName\":\"localhost\",\"dmaapPortNumber\":2223,\"dmaapProtocol\":\"http\","
+                    + "\"FAULTY_PARAMETER_NAME\":\"temp\","
+                    + "\"dmaapUserName\":\"admin\",\"dmaapUserPassword\":\"admin\"}}}}";
 
     private static DatafileAppConfig datafileAppConfig;
     private static AppConfig appConfig;
 
     private static String filePath = Objects
-        .requireNonNull(DatafileAppConfigTest.class.getClassLoader().getResource(DATAFILE_ENDPOINTS)).getFile();
+            .requireNonNull(DatafileAppConfigTest.class.getClassLoader().getResource(DATAFILE_ENDPOINTS)).getFile();
 
     @BeforeEach
-    void setUp() {
+    public void setUp() {
         datafileAppConfig = spy(DatafileAppConfig.class);
         appConfig = spy(new AppConfig());
     }
 
     @Test
-    void whenApplicationWasStarted_FilePathIsSet() {
-        //
+    public void whenApplicationWasStarted_FilePathIsSet() {
         // When
-        //
         datafileAppConfig.setFilepath(filePath);
-        //
+
         // Then
-        //
         verify(datafileAppConfig, times(1)).setFilepath(anyString());
         verify(datafileAppConfig, times(0)).initFileStreamReader();
         Assertions.assertEquals(filePath, datafileAppConfig.getFilepath());
     }
 
     @Test
-    void whenTheConfigurationFits_GetAaiAndDmaapObjectRepresentationConfiguration()
-        throws IOException {
-        //
+    public void whenTheConfigurationFits_GetAaiAndDmaapObjectRepresentationConfiguration() throws IOException {
         // Given
-        //
-        InputStream inputStream = new ByteArrayInputStream((jsonString.getBytes(
-            StandardCharsets.UTF_8)));
-        //
+        InputStream inputStream = new ByteArrayInputStream((JSON_STRING.getBytes(StandardCharsets.UTF_8)));
+
         // When
-        //
         datafileAppConfig.setFilepath(filePath);
         doReturn(inputStream).when(datafileAppConfig).getInputStream(any());
         datafileAppConfig.initFileStreamReader();
         appConfig.dmaapConsumerConfiguration = datafileAppConfig.getDmaapConsumerConfiguration();
         appConfig.dmaapPublisherConfiguration = datafileAppConfig.getDmaapPublisherConfiguration();
-        appConfig.aaiClientConfiguration = datafileAppConfig.getAaiClientConfiguration();
-        //
+
         // Then
-        //
         verify(datafileAppConfig, times(1)).setFilepath(anyString());
         verify(datafileAppConfig, times(1)).initFileStreamReader();
-        Assertions.assertNotNull(datafileAppConfig.getAaiClientConfiguration());
         Assertions.assertNotNull(datafileAppConfig.getDmaapConsumerConfiguration());
         Assertions.assertNotNull(datafileAppConfig.getDmaapPublisherConfiguration());
-        Assertions
-            .assertEquals(appConfig.getDmaapPublisherConfiguration(), datafileAppConfig.getDmaapPublisherConfiguration());
-        Assertions
-            .assertEquals(appConfig.getDmaapConsumerConfiguration(), datafileAppConfig.getDmaapConsumerConfiguration());
-        Assertions
-            .assertEquals(appConfig.getAaiClientConfiguration(), datafileAppConfig.getAaiClientConfiguration());
+        Assertions.assertEquals(appConfig.getDmaapPublisherConfiguration(),
+                datafileAppConfig.getDmaapPublisherConfiguration());
+        Assertions.assertEquals(appConfig.getDmaapConsumerConfiguration(),
+                datafileAppConfig.getDmaapConsumerConfiguration());
 
     }
 
     @Test
-    void whenFileIsNotExist_ThrowIoException() {
-        //
+    public void whenFileIsNotExist_ThrowIoException() {
         // Given
-        //
         filePath = "/temp.json";
         datafileAppConfig.setFilepath(filePath);
-        //
+
         // When
-        //
         datafileAppConfig.initFileStreamReader();
-        //
+
         // Then
-        //
         verify(datafileAppConfig, times(1)).setFilepath(anyString());
         verify(datafileAppConfig, times(1)).initFileStreamReader();
-        Assertions.assertNull(datafileAppConfig.getAaiClientConfiguration());
         Assertions.assertNull(datafileAppConfig.getDmaapConsumerConfiguration());
         Assertions.assertNull(datafileAppConfig.getDmaapPublisherConfiguration());
 
     }
 
     @Test
-    void whenFileIsExistsButJsonIsIncorrect() throws IOException {
-        //
+    public void whenFileIsExistsButJsonIsIncorrect() throws IOException {
         // Given
-        //
-        InputStream inputStream = new ByteArrayInputStream((incorrectJsonString.getBytes(
-            StandardCharsets.UTF_8)));
-        //
+        InputStream inputStream = new ByteArrayInputStream((INCORRECT_JSON_STRING.getBytes(StandardCharsets.UTF_8)));
+
         // When
-        //
         datafileAppConfig.setFilepath(filePath);
         doReturn(inputStream).when(datafileAppConfig).getInputStream(any());
         datafileAppConfig.initFileStreamReader();
 
-        //
         // Then
-        //
         verify(datafileAppConfig, times(1)).setFilepath(anyString());
         verify(datafileAppConfig, times(1)).initFileStreamReader();
-        Assertions.assertNotNull(datafileAppConfig.getAaiClientConfiguration());
         Assertions.assertNotNull(datafileAppConfig.getDmaapConsumerConfiguration());
         Assertions.assertNull(datafileAppConfig.getDmaapPublisherConfiguration());
 
@@ -186,11 +150,9 @@ class DatafileAppConfigTest {
 
 
     @Test
-    void whenTheConfigurationFits_ButRootElementIsNotAJsonObject()
-        throws IOException {
+    public void whenTheConfigurationFits_ButRootElementIsNotAJsonObject() throws IOException {
         // Given
-        InputStream inputStream = new ByteArrayInputStream((jsonString.getBytes(
-            StandardCharsets.UTF_8)));
+        InputStream inputStream = new ByteArrayInputStream((JSON_STRING.getBytes(StandardCharsets.UTF_8)));
         // When
         datafileAppConfig.setFilepath(filePath);
         doReturn(inputStream).when(datafileAppConfig).getInputStream(any());
@@ -200,13 +162,11 @@ class DatafileAppConfigTest {
         datafileAppConfig.initFileStreamReader();
         appConfig.dmaapConsumerConfiguration = datafileAppConfig.getDmaapConsumerConfiguration();
         appConfig.dmaapPublisherConfiguration = datafileAppConfig.getDmaapPublisherConfiguration();
-        appConfig.aaiClientConfiguration = datafileAppConfig.getAaiClientConfiguration();
 
         // Then
         verify(datafileAppConfig, times(1)).setFilepath(anyString());
         verify(datafileAppConfig, times(1)).initFileStreamReader();
-        Assertions.assertNull(datafileAppConfig.getAaiClientConfiguration());
         Assertions.assertNull(datafileAppConfig.getDmaapConsumerConfiguration());
         Assertions.assertNull(datafileAppConfig.getDmaapPublisherConfiguration());
     }
-}
\ No newline at end of file
+}
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FileCollectorTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FileCollectorTest.java
new file mode 100644
index 00000000..5b9d0aaf
--- /dev/null
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FileCollectorTest.java
@@ -0,0 +1,116 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ftp;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
+import org.onap.dcaegen2.collectors.datafile.service.FileData;
+import org.onap.dcaegen2.collectors.datafile.service.ImmutableFileData;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
+ *
+ */
+public class FileCollectorTest {
+
+    private static final String PM_MEAS_CHANGE_IDINTIFIER = "PM_MEAS_FILES";
+    private static final String FILE_READY_CHANGE_TYPE = "FileReady";
+    private static final String FTPES_SCHEME = "ftpes://";
+    private static final String SFTP_SCHEME = "sftp://";
+    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 String LOCAL_FILE_LOCATION = "target/" + PM_FILE_NAME;
+    private static final String FTPES_LOCATION = FTPES_SCHEME + SERVER_ADDRESS + ":" + PORT_22 + REMOTE_FILE_LOCATION;
+    private static final String SFTP_LOCATION = SFTP_SCHEME + SERVER_ADDRESS + ":" + PORT_22 + REMOTE_FILE_LOCATION;
+    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 FtpsClient ftpsClientMock = mock(FtpsClient.class);
+
+    private SftpClient sftpClientMock = mock(SftpClient.class);
+
+    private FileCollector fileCollectorUndetTest = new FileCollector(ftpsClientMock, sftpClientMock);
+
+    @Test
+    public void whenSingleFtpesFile_returnCorrectResponse() {
+        List<FileData> listOfFileData = new ArrayList<FileData>();
+        listOfFileData.add(ImmutableFileData.builder().changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
+                .changeType(FILE_READY_CHANGE_TYPE).location(FTPES_LOCATION).compression(GZIP_COMPRESSION)
+                .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build());
+
+        FileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(SERVER_ADDRESS).port(PORT_22)
+                .userId("").password("").build();
+        when(ftpsClientMock.collectFile(fileServerData, REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION)).thenReturn(true);
+
+        Mono<List<ConsumerDmaapModel>> consumerModelsMono =
+                fileCollectorUndetTest.getFilesFromSender(listOfFileData);
+
+        List<ConsumerDmaapModel> consumerModels = consumerModelsMono.block();
+        assertEquals(1, consumerModels.size());
+        ConsumerDmaapModel consumerDmaapModel = consumerModels.get(0);
+        assertEquals(GZIP_COMPRESSION, consumerDmaapModel.getCompression());
+        assertEquals(MEAS_COLLECT_FILE_FORMAT_TYPE, consumerDmaapModel.getFileFormatType());
+        assertEquals(FILE_FORMAT_VERSION, consumerDmaapModel.getFileFormatVersion());
+        assertEquals(LOCAL_FILE_LOCATION, consumerDmaapModel.getLocation());
+        FileServerData expectedFileServerData = ImmutableFileServerData.builder().serverAddress(SERVER_ADDRESS)
+                .userId("").password("").port(PORT_22).build();
+        verify(ftpsClientMock, times(1)).collectFile(expectedFileServerData, REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION);
+        verifyNoMoreInteractions(ftpsClientMock);
+    }
+
+    @Test
+    public void whenSingleSftpFile_returnCorrectResponse() {
+        List<FileData> listOfFileData = new ArrayList<FileData>();
+        listOfFileData.add(ImmutableFileData.builder().changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
+                .changeType(FILE_READY_CHANGE_TYPE).location(SFTP_LOCATION).compression(GZIP_COMPRESSION)
+                .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build());
+
+        FileServerData fileServerData = ImmutableFileServerData.builder().serverAddress(SERVER_ADDRESS).port(PORT_22)
+                .userId("").password("").build();
+        when(sftpClientMock.collectFile(fileServerData, REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION)).thenReturn(true);
+
+        Mono<List<ConsumerDmaapModel>> consumerModelsMono =
+                fileCollectorUndetTest.getFilesFromSender(listOfFileData);
+
+        List<ConsumerDmaapModel> consumerModels = consumerModelsMono.block();
+        assertEquals(1, consumerModels.size());
+        ConsumerDmaapModel consumerDmaapModel = consumerModels.get(0);
+        assertEquals(GZIP_COMPRESSION, consumerDmaapModel.getCompression());
+        assertEquals(MEAS_COLLECT_FILE_FORMAT_TYPE, consumerDmaapModel.getFileFormatType());
+        assertEquals(FILE_FORMAT_VERSION, consumerDmaapModel.getFileFormatVersion());
+        assertEquals(LOCAL_FILE_LOCATION, consumerDmaapModel.getLocation());
+        FileServerData expectedFileServerData = ImmutableFileServerData.builder().serverAddress(SERVER_ADDRESS)
+                .userId("").password("").port(PORT_22).build();
+        verify(sftpClientMock, times(1)).collectFile(expectedFileServerData, REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION);
+        verifyNoMoreInteractions(ftpsClientMock);
+    }
+}
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpClientTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpClientTest.java
new file mode 100644
index 00000000..a95b80ed
--- /dev/null
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpClientTest.java
@@ -0,0 +1,57 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ftp;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.mockftpserver.fake.FakeFtpServer;
+import org.mockftpserver.fake.UserAccount;
+import org.mockftpserver.fake.filesystem.FileEntry;
+import org.mockftpserver.fake.filesystem.FileSystem;
+import org.mockftpserver.fake.filesystem.UnixFakeFileSystem;
+
+public class FtpClientTest {
+
+    private static final String HOME_DIR = "/";
+    private static final String FILE = "/dir/sample.txt";
+    private static final String CONTENTS = "abcdef 1234567890";
+    private static final int PORT = 8021;
+
+    private static final String USERNAME = "bob";
+    private static final String PASSWORD = "123";
+
+    private FakeFtpServer fakeFtpServer;
+
+    @BeforeAll
+    protected void setUp() throws Exception {
+        fakeFtpServer = new FakeFtpServer();
+        fakeFtpServer.setServerControlPort(PORT);
+
+        FileSystem fileSystem = new UnixFakeFileSystem();
+        fileSystem.add(new FileEntry(FILE, CONTENTS));
+        fakeFtpServer.setFileSystem(fileSystem);
+        UserAccount userAccount = new UserAccount(USERNAME, PASSWORD, HOME_DIR);
+        fakeFtpServer.addUserAccount(userAccount);
+
+        fakeFtpServer.start();
+    }
+
+    @AfterAll
+    protected void tearDown() throws Exception {
+        fakeFtpServer.stop();
+    }
+}
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..13f1fbb9
--- /dev/null
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClientTest.java
@@ -0,0 +1,103 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.ftp;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.commons.io.IOUtils.toByteArray;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.github.stefanbirkner.fakesftpserver.rule.FakeSftpServerRule;
+import com.jcraft.jsch.ChannelSftp;
+import com.jcraft.jsch.JSch;
+import com.jcraft.jsch.JSchException;
+import com.jcraft.jsch.Session;
+import com.jcraft.jsch.SftpException;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+
+import org.junit.Rule;
+import org.junit.Test;
+
+public class SftpClientTest {
+    private static final String USERNAME = "bob";
+    private static final String PASSWORD = "123";
+    private static final String DUMMY_CONTENT = "dummy content";
+    private static final String LOCAL_DUMMY_FILE = "target/dummy.txt";
+    private static final String REMOTE_DUMMY_FILE = "/dummy_directory/dummy_file.txt";
+    private static final JSch JSCH = new JSch();
+    private static final int TIMEOUT = 2000;
+
+    @Rule
+    public final FakeSftpServerRule sftpServer = new FakeSftpServerRule().addUser(USERNAME, PASSWORD);
+
+    @Test
+    public void collectFile_withOKresponse() throws IOException, JSchException, SftpException {
+        SftpClient sftpClient = new SftpClient();
+        sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
+        byte[] file = downloadFile(sftpServer, REMOTE_DUMMY_FILE);
+        FileServerData expectedFileServerData = ImmutableFileServerData.builder().serverAddress("127.0.0.1")
+                .userId(USERNAME).password(PASSWORD).port(sftpServer.getPort()).build();
+        sftpClient.collectFile(expectedFileServerData, REMOTE_DUMMY_FILE,
+                LOCAL_DUMMY_FILE);
+        byte[] localFile = Files.readAllBytes(new File(LOCAL_DUMMY_FILE).toPath());
+        assertThat(new String(file, UTF_8)).isEqualTo(DUMMY_CONTENT);
+        assertThat(new String(localFile, UTF_8)).isEqualTo(DUMMY_CONTENT);
+    }
+
+    private static Session connectToServer(FakeSftpServerRule sftpServer) throws JSchException {
+        return connectToServerAtPort(sftpServer.getPort());
+    }
+
+    private static Session connectToServerAtPort(int port) throws JSchException {
+        Session session = createSessionWithCredentials(USERNAME, PASSWORD, port);
+        session.connect(TIMEOUT);
+        return session;
+    }
+
+    private static ChannelSftp connectSftpChannel(Session session) throws JSchException {
+        ChannelSftp channel = (ChannelSftp) session.openChannel("sftp");
+        channel.connect();
+        return channel;
+    }
+
+    private static Session createSessionWithCredentials(String username, String password, int port)
+            throws JSchException {
+        Session session = JSCH.getSession(username, "127.0.0.1", port);
+        session.setConfig("StrictHostKeyChecking", "no");
+        session.setPassword(password);
+        return session;
+    }
+
+    private static byte[] downloadFile(FakeSftpServerRule server, String path)
+            throws JSchException, SftpException, IOException {
+        Session session = connectToServer(server);
+        ChannelSftp channel = connectSftpChannel(session);
+        try {
+            InputStream is = channel.get(path);
+            return toByteArray(is);
+        } finally {
+            channel.disconnect();
+            session.disconnect();
+        }
+    }
+
+}
diff --git a/datafile-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 b9aa2f78..b5f05a71 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,9 +1,7 @@
 /*
- * ============LICENSE_START=======================================================
- * PROJECT
- * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
- * ================================================================================
+ * ============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
@@ -15,7 +13,7 @@
  * 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=========================================================
+ * ============LICENSE_END========================================================================
  */
 
 package org.onap.dcaegen2.collectors.datafile.integration;
@@ -26,6 +24,7 @@ import static org.mockito.Mockito.verify;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 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;
@@ -39,6 +38,7 @@ import org.springframework.test.context.testng.AbstractTestNGSpringContextTests;
 
 /**
  * @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
@@ -62,5 +62,3 @@ class ScheduledXmlContextITest extends AbstractTestNGSpringContextTests {
         verify(scheduledTask, atLeast(1)).scheduleMainDatafileEventTask();
     }
 }
-
-
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
index 47107588..05a4f515 100644
--- 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
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * PROJECT
  * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
+ * 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.
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
index df167425..bc4e6401 100644
--- 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
@@ -1,9 +1,7 @@
 /*
- * ============LICENSE_START=======================================================
- * PROJECT
- * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
- * ================================================================================
+ * ============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
@@ -15,7 +13,7 @@
  * 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=========================================================
+ * ============LICENSE_END========================================================================
  */
 
 package org.onap.dcaegen2.collectors.datafile.integration.junit5.mockito;
@@ -36,9 +34,10 @@ 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
- *      viaMockito 2.x's {@link Mock @Mock} annotation.
+ *         {@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 {
 
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/DmaapConsumerJsonParserTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/DmaapConsumerJsonParserTest.java
index d7ceee82..dc51343d 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/DmaapConsumerJsonParserTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/DmaapConsumerJsonParserTest.java
@@ -1,21 +1,17 @@
 /*
- * ============LICENSE_START=======================================================
- * Datafile Collector Service
- * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. 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) 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
+ * 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.
+ * ============LICENSE_END========================================================================
  */
 
 package org.onap.dcaegen2.collectors.datafile.service;
@@ -24,250 +20,161 @@ import static org.mockito.Mockito.spy;
 
 import com.google.gson.JsonElement;
 import com.google.gson.JsonParser;
+
+import java.util.List;
 import java.util.Optional;
+
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 import org.onap.dcaegen2.collectors.datafile.exceptions.DmaapNotFoundException;
-import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.collectors.datafile.service.DmaapConsumerJsonParser;
-import org.onap.dcaegen2.collectors.datafile.model.ImmutableConsumerDmaapModel;
+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;
 
 /**
  * @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>
  */
 class DmaapConsumerJsonParserTest {
 
     @Test
-    void whenPassingCorrectJson_validationNotThrowingAnException() {
-        //given
-        String message =
-            "[{\"event\":{\"commonEventHeader\":{\"domain\":\"other\",\"eventId\":\"<<SerialNumber>>-reg\","
-                + "\"eventName\":\"pnfRegistration_5GDU\",\"eventType\":\"pnfRegistration\",\"internalHeaderFields\""
-                + ":{},\"lastEpochMicrosec\":1519837825682,\"nfNamingCode\":\"5GRAN\",\"nfcNamingCode\":\"5DU\","
-                + "\"priority\":\"Normal\",\"reportingEntityName\":\"5GRAN_DU\",\"sequence\":0,\"sourceId\":"
-                + "\"<<SerialNumber>>\",\"sourceName\":\"5GRAN_DU\",\"startEpochMicrosec\":1519837825682,\"version\":"
-                + "3},\"otherFields\":{\"otherFieldsVersion\":1,\"pnfFamily\":\"BBU\",\"pnfLastServiceDate\":1517206400"
-                + ",\"pnfManufactureDate\":1516406400,\"pnfModelNumber\":\"AJ02\",\"pnfOamIpv4Address\":"
-                + "\"10.16.123.234\",\"pnfOamIpv6Address\":\"0:0:0:0:0:FFFF:0A10:7BEA\",\"pnfSerialNumber\":"
-                + "\"QTFCOC540002E\",\"pnfSoftwareVersion\":\"v4.5.0.1\",\"pnfType\":\"AirScale\",\"pnfVendorName\":"
-                + "\"Nokia\"}}}]";
-
-        String parsed =
-            "{\"event\":{\"commonEventHeader\":{\"domain\":\"other\",\"eventId\":\"<<SerialNumber>>-reg\",\"eventName\""
-                + ":\"pnfRegistration_5GDU\",\"eventType\":\"pnfRegistration\",\"internalHeaderFields\":{},"
-                + "\"lastEpochMicrosec\":1519837825682,\"nfNamingCode\":\"5GRAN\",\"nfcNamingCode\":\"5DU\","
-                + "\"priority\":\"Normal\",\"reportingEntityName\":\"5GRAN_DU\",\"sequence\":0,\"sourceId\":"
-                + "\"<<SerialNumber>>\",\"sourceName\":\"5GRAN_DU\",\"startEpochMicrosec\":1519837825682,\"version\":"
-                + "3},\"otherFields\":{\"otherFieldsVersion\":1,\"pnfFamily\":\"BBU\",\"pnfLastServiceDate\":"
-                + "1517206400,\"pnfManufactureDate\":1516406400,\"pnfModelNumber\":\"AJ02\",\"pnfOamIpv4Address\":"
-                + "\"10.16.123.234\",\"pnfOamIpv6Address\":\"0:0:0:0:0:FFFF:0A10:7BEA\",\"pnfSerialNumber\":"
-                + "\"QTFCOC540002E\",\"pnfSoftwareVersion\":\"v4.5.0.1\",\"pnfType\":\"AirScale\",\"pnfVendorName\":"
-                + "\"Nokia\"}}}";
-        ConsumerDmaapModel expectedObject = ImmutableConsumerDmaapModel.builder().ipv4("10.16.123.234")
-            .ipv6("0:0:0:0:0:FFFF:0A10:7BEA")
-            .pnfName("NOKQTFCOC540002E").build();
-        //when
+    void whenPassingCorrectJson_validationNotThrowingAnException() throws DmaapNotFoundException {
+        // given
+        AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder()
+                .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();
+        JsonMessage message = new JsonMessage.JsonMessageBuilder().changeIdentifier("PM_MEAS_FILES")
+                .changeType("FileReady").notificationFieldsVersion("1.0").addAdditionalField(additionalField).build();
+
+        String messageString = message.toString();
+
+        String parsedString = message.getParsed();
+
+        FileData expectedFileData = ImmutableFileData.builder().changeIdentifier("PM_MEAS_FILES")
+                .changeType("FileReady").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();
+        // when
         DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
-        JsonElement jsonElement = new JsonParser().parse(parsed);
-        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
-            .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
-        ConsumerDmaapModel consumerDmaapModel = dmaapConsumerJsonParser
-            .getJsonObject(Mono.just((message))).block();
-        //then
-        Assertions.assertNotNull(consumerDmaapModel);
-        Assertions.assertEquals(expectedObject, consumerDmaapModel);
+        JsonElement jsonElement = new JsonParser().parse(parsedString);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(dmaapConsumerJsonParser)
+                .getJsonObjectFromAnArray(jsonElement);
+        List<FileData> listOfFileData = dmaapConsumerJsonParser.getJsonObject(Mono.just((messageString))).block();
+        // then
+        Assertions.assertNotNull(listOfFileData);
+        Assertions.assertEquals(expectedFileData, listOfFileData.get(0));
     }
 
     @Test
-    void whenPassingCorrectJsonWithoutIpv4_validationNotThrowingAnException() {
-        //given
-        String message =
-            "[{\"event\":{\"commonEventHeader\":{\"domain\":\"other\",\"eventId\":\"<<SerialNumber>>-reg\","
-                + "\"eventName\":\"pnfRegistration_5GDU\",\"eventType\":\"pnfRegistration\",\"internalHeaderFields\":"
-                + "{},\"lastEpochMicrosec\":1519837825682,\"nfNamingCode\":\"5GRAN\",\"nfcNamingCode\":\"5DU\","
-                + "\"priority\":\"Normal\",\"reportingEntityName\":\"5GRAN_DU\",\"sequence\":0,\"sourceId\":"
-                + "\"<<SerialNumber>>\",\"sourceName\":\"5GRAN_DU\",\"startEpochMicrosec\":1519837825682,\"version\":3}"
-                + ",\"otherFields\":{\"otherFieldsVersion\":1,\"pnfFamily\":\"BBU\",\"pnfLastServiceDate\":1517206400,"
-                + "\"pnfManufactureDate\":1516406400,\"pnfModelNumber\":\"AJ02\",\"pnfOamIpv6Address\":"
-                + "\"0:0:0:0:0:FFFF:0A10:7BEA\",\"pnfSerialNumber\":\"QTFCOC540002E\",\"pnfSoftwareVersion\""
-                + ":\"v4.5.0.1\",\"pnfType\":\"AirScale\",\"pnfVendorName\":\"Nokia\"}}}]";
-
-        String parsed =
-            "{\"event\":{\"commonEventHeader\":{\"domain\":\"other\",\"eventId\":\"<<SerialNumber>>-reg\",\"eventName\""
-                + ":\"pnfRegistration_5GDU\",\"eventType\":\"pnfRegistration\",\"internalHeaderFields\":{},"
-                + "\"lastEpochMicrosec\":1519837825682,\"nfNamingCode\":\"5GRAN\",\"nfcNamingCode\":\"5DU\","
-                + "\"priority\":\"Normal\",\"reportingEntityName\":\"5GRAN_DU\",\"sequence\":0,\"sourceId\":"
-                + "\"<<SerialNumber>>\",\"sourceName\":\"5GRAN_DU\",\"startEpochMicrosec\":1519837825682,\"version\":3}"
-                + ",\"otherFields\":{\"otherFieldsVersion\":1,\"pnfFamily\":\"BBU\",\"pnfLastServiceDate\":1517206400,"
-                + "\"pnfManufactureDate\":1516406400,\"pnfModelNumber\":\"AJ02\",\"pnfOamIpv6Address\":"
-                + "\"0:0:0:0:0:FFFF:0A10:7BEA\",\"pnfSerialNumber\":\"QTFCOC540002E\",\"pnfSoftwareVersion\""
-                + ":\"v4.5.0.1\",\"pnfType\":\"AirScale\",\"pnfVendorName\":\"Nokia\"}}}";
-
-        //when
+    void whenPassingCorrectJsonWihoutLocation_validationThrowingAnException() {
+        AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder().compression("gzip")
+                .fileFormatType("org.3GPP.32.435#measCollec").fileFormatVersion("V10").build();
+        JsonMessage message = new JsonMessage.JsonMessageBuilder().changeIdentifier("PM_MEAS_FILES")
+                .changeType("FileReady").notificationFieldsVersion("1.0").addAdditionalField(additionalField).build();
+
+        String messageString = message.toString();
+
+        String parsedString = message.getParsed();
+
         DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
-        JsonElement jsonElement = new JsonParser().parse(parsed);
-        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
-            .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
-        dmaapConsumerJsonParser.getJsonObject(Mono.just((message)));
-        ConsumerDmaapModel consumerDmaapModel = dmaapConsumerJsonParser.getJsonObject(Mono.just((message)))
-            .block();
-        //then
-        ConsumerDmaapModel expectedObject = ImmutableConsumerDmaapModel.builder().ipv4("")
-            .ipv6("0:0:0:0:0:FFFF:0A10:7BEA")
-            .pnfName("NOKQTFCOC540002E").build();
-        Assertions.assertNotNull(consumerDmaapModel);
-        Assertions.assertEquals(expectedObject, consumerDmaapModel);
+        JsonElement jsonElement = new JsonParser().parse(parsedString);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(dmaapConsumerJsonParser)
+                .getJsonObjectFromAnArray(jsonElement);
+        StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(messageString))).expectSubscription()
+                .expectError(DmaapNotFoundException.class).verify();
     }
 
     @Test
-    void whenPassingCorrectJsonWihoutIpv6_validationNotThrowingAnException() {
-        //given
-        String message =
-            "[{\"event\":{\"commonEventHeader\":{\"domain\":\"other\",\"eventId\":\"<<SerialNumber>>-reg\","
-                + "\"eventName\":\"pnfRegistration_5GDU\",\"eventType\":\"pnfRegistration\",\"internalHeaderFields\":"
-                + "{},\"lastEpochMicrosec\":1519837825682,\"nfNamingCode\":\"5GRAN\",\"nfcNamingCode\":\"5DU\","
-                + "\"priority\":\"Normal\",\"reportingEntityName\":\"5GRAN_DU\",\"sequence\":0,\"sourceId\":"
-                + "\"<<SerialNumber>>\",\"sourceName\":\"5GRAN_DU\",\"startEpochMicrosec\":1519837825682,"
-                + "\"version\":3},\"otherFields\":{\"otherFieldsVersion\":1,\"pnfFamily\":\"BBU\",\"pnfLastServiceDate"
-                + "\":1517206400,\"pnfManufactureDate\":1516406400,\"pnfModelNumber\":\"AJ02\",\"pnfOamIpv4Address"
-                + "\":\"10.16.123.234\",\"pnfSerialNumber\":\"QTFCOC540002E\",\"pnfSoftwareVersion\":\"v4.5.0.1\","
-                + "\"pnfType\":\"AirScale\",\"pnfVendorName\":\"Nokia\"}}}]";
-        String parsed =
-            "{\"event\":{\"commonEventHeader\":{\"domain\":\"other\",\"eventId\":\"<<SerialNumber>>-reg\",\"eventName\""
-                + ":\"pnfRegistration_5GDU\",\"eventType\":\"pnfRegistration\",\"internalHeaderFields\":{},"
-                + "\"lastEpochMicrosec\":1519837825682,\"nfNamingCode\":\"5GRAN\",\"nfcNamingCode\":\"5DU\","
-                + "\"priority\":\"Normal\",\"reportingEntityName\":\"5GRAN_DU\",\"sequence\":0,\"sourceId\":"
-                + "\"<<SerialNumber>>\",\"sourceName\":\"5GRAN_DU\",\"startEpochMicrosec\":1519837825682,"
-                + "\"version\":3},\"otherFields\":{\"otherFieldsVersion\":1,\"pnfFamily\":\"BBU\",\"pnfLastServiceDate"
-                + "\":1517206400,\"pnfManufactureDate\":1516406400,\"pnfModelNumber\":\"AJ02\",\"pnfOamIpv4Address"
-                + "\":\"10.16.123.234\",\"pnfSerialNumber\":\"QTFCOC540002E\",\"pnfSoftwareVersion\":\"v4.5.0.1\","
-                + "\"pnfType\":\"AirScale\",\"pnfVendorName\":\"Nokia\"}}}";
-
-        ConsumerDmaapModel expectedObject = ImmutableConsumerDmaapModel.builder().ipv4("10.16.123.234").ipv6("")
-            .pnfName("NOKQTFCOC540002E").build();
-        //when
+    void whenPassingCorrectJsonWihoutCompression_validationThrowingAnException() {
+        AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder()
+                .location("ftpes://192.168.0.101:22/ftp/rop/A20161224.1030-1045.bin.gz")
+                .fileFormatType("org.3GPP.32.435#measCollec").fileFormatVersion("V10").build();
+        JsonMessage message = new JsonMessage.JsonMessageBuilder().changeIdentifier("PM_MEAS_FILES")
+                .changeType("FileReady").notificationFieldsVersion("1.0").addAdditionalField(additionalField).build();
+
+        String messageString = message.toString();
+
+        String parsedString = message.getParsed();
+
         DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
-        JsonElement jsonElement = new JsonParser().parse(parsed);
-        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
-            .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
-        ConsumerDmaapModel consumerDmaapModel = dmaapConsumerJsonParser.getJsonObject(Mono.just((message)))
-            .block();
-        //then
-        Assertions.assertNotNull(consumerDmaapModel);
-        Assertions.assertEquals(expectedObject, consumerDmaapModel);
+        JsonElement jsonElement = new JsonParser().parse(parsedString);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(dmaapConsumerJsonParser)
+                .getJsonObjectFromAnArray(jsonElement);
+        StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(messageString))).expectSubscription()
+                .expectError(DmaapNotFoundException.class).verify();
     }
 
     @Test
-    void whenPassingCorrectJsonWihoutIpv4andIpv6_validationThrowingAnException() {
-        String message =
-            "[{\"event\":{\"commonEventHeader\":{\"domain\":\"other\",\"eventId\":\"<<SerialNumber>>-reg\","
-                + "\"eventName\":\"pnfRegistration_5GDU\",\"eventType\":\"pnfRegistration\",\"internalHeaderFields\":"
-                + "{},\"lastEpochMicrosec\":1519837825682,\"nfNamingCode\":\"5GRAN\",\"nfcNamingCode\":\"5DU\","
-                + "\"priority\":\"Normal\",\"reportingEntityName\":\"5GRAN_DU\",\"sequence\":0,\"sourceId\":"
-                + "\"<<SerialNumber>>\",\"sourceName\":\"5GRAN_DU\",\"startEpochMicrosec\":1519837825682,\"version\""
-                + ":3},\"otherFields\":{\"otherFieldsVersion\":1,\"pnfFamily\":\"BBU\",\"pnfLastServiceDate\""
-                + ":1517206400,\"pnfManufactureDate\":1516406400,\"pnfModelNumber\":\"AJ02\",\"pnfSoftwareVersion\":"
-                + "\"v4.5.0.1\",\"pnfType\":\"AirScale\",\"pnfVendorName\":\"Nokia\"}}}]";
-        String parsed =
-            "{\"event\":{\"commonEventHeader\":{\"domain\":\"other\",\"eventId\":\"<<SerialNumber>>-reg\",\"eventName\""
-                + ":\"pnfRegistration_5GDU\",\"eventType\":\"pnfRegistration\",\"internalHeaderFields\":{},"
-                + "\"lastEpochMicrosec\":1519837825682,\"nfNamingCode\":\"5GRAN\",\"nfcNamingCode\":\"5DU\","
-                + "\"priority\":\"Normal\",\"reportingEntityName\":\"5GRAN_DU\",\"sequence\":0,\"sourceId\":"
-                + "\"<<SerialNumber>>\",\"sourceName\":\"5GRAN_DU\",\"startEpochMicrosec\":1519837825682,\"version\""
-                + ":3},\"otherFields\":{\"otherFieldsVersion\":1,\"pnfFamily\":\"BBU\",\"pnfLastServiceDate\""
-                + ":1517206400,\"pnfManufactureDate\":1516406400,\"pnfModelNumber\":\"AJ02\",\"pnfSoftwareVersion\":"
-                + "\"v4.5.0.1\",\"pnfType\":\"AirScale\",\"pnfVendorName\":\"Nokia\"}}}";
-        DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
-        JsonElement jsonElement = new JsonParser().parse(parsed);
-        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
-            .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
-        StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(message)))
-            .expectSubscription().expectError(DmaapNotFoundException.class).verify();
+    void whenPassingCorrectJsonWihoutFileFormatType_validationThrowingAnException() {
+        AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder()
+                .location("ftpes://192.168.0.101:22/ftp/rop/A20161224.1030-1045.bin.gz").compression("gzip")
+                .fileFormatVersion("V10").build();
+        JsonMessage message = new JsonMessage.JsonMessageBuilder().changeIdentifier("PM_MEAS_FILES")
+                .changeType("FileReady").notificationFieldsVersion("1.0").addAdditionalField(additionalField).build();
+
+        String messageString = message.toString();
 
+        String parsedString = message.getParsed();
+
+        DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
+        JsonElement jsonElement = new JsonParser().parse(parsedString);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(dmaapConsumerJsonParser)
+                .getJsonObjectFromAnArray(jsonElement);
+        StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(messageString))).expectSubscription()
+                .expectError(DmaapNotFoundException.class).verify();
     }
 
     @Test
-    void whenPassingJsonWithoutMandatoryHeaderInformation_validationThrowingAnException() {
-        String parsed = "{\"event\":{\"commonEventHeader\":{\"domain\":\"other\",\"eventId\":\"<<SerialNumber>>-reg\""
-            + ",\"eventName\":\"pnfRegistration_5GDU\",\"eventType\":\"pnfRegistration\",\"internalHeaderFields\":{},"
-            + "\"lastEpochMicrosec\":1519837825682,\"nfNamingCode\":\"5GRAN\",\"nfcNamingCode\":\"5DU\",\"priority\""
-            + ":\"Normal\",\"reportingEntityName\":\"5GRAN_DU\",\"sequence\":0,\"sourceId\":\"<<SerialNumber>>\","
-            + "\"sourceName\":\"5GRAN_DU\",\"startEpochMicrosec\":1519837825682,\"version\":3}}}";
+    void whenPassingCorrectJsonWihoutFileFormatVersion_validationThrowingAnException() {
+        AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder()
+                .location("ftpes://192.168.0.101:22/ftp/rop/A20161224.1030-1045.bin.gz").compression("gzip")
+                .fileFormatType("org.3GPP.32.435#measCollec").build();
+        JsonMessage message = new JsonMessage.JsonMessageBuilder().changeIdentifier("PM_MEAS_FILES")
+                .changeType("FileReady").notificationFieldsVersion("1.0").addAdditionalField(additionalField).build();
+
+        String messageString = message.toString();
+
+        String parsedString = message.getParsed();
+
         DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
-        JsonElement jsonElement = new JsonParser().parse(parsed);
-        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
-            .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
-        String incorrectMessage =
-            "[{\"event\":{\"commonEventHeader\":{\"domain\":\"other\",\"eventId\":\"<<SerialNumber>>-reg\""
-                + ",\"eventName\":\"pnfRegistration_5GDU\",\"eventType\":\"pnfRegistration\",\"internalHeaderFields\":"
-                + "{},\"lastEpochMicrosec\":1519837825682,\"nfNamingCode\":\"5GRAN\",\"nfcNamingCode\":\"5DU\","
-                + "\"priority\":\"Normal\",\"reportingEntityName\":\"5GRAN_DU\",\"sequence\":0,\"sourceId\":"
-                + "\"<<SerialNumber>>\",\"sourceName\":\"5GRAN_DU\",\"startEpochMicrosec\":1519837825682,\"version\":3"
-                + "}}}]";
-        StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(incorrectMessage)))
-            .expectSubscription().expectError(DmaapNotFoundException.class).verify();
+        JsonElement jsonElement = new JsonParser().parse(parsedString);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(dmaapConsumerJsonParser)
+                .getJsonObjectFromAnArray(jsonElement);
+        StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(messageString))).expectSubscription()
+                .expectError(DmaapNotFoundException.class).verify();
     }
 
+    // Fixed temprarily
     @Test
-    void whenPassingJsonWithoutPnfSerialNumberOrPnfVendorName_validationThrowingAnException() {
-        String parsed = "{\"event\":{\"commonEventHeader\":{\"domain\":\"other\",\"eventId\":"
-            + "\"<<SerialNumber>>-reg\",\"eventName\":\"pnfRegistration_5GDU\",\"eventType\":\"pnfRegistration\",\""
-            + "internalHeaderFields\":{},\"lastEpochMicrosec\":1519837825682,\"nfNamingCode\":\"5GRAN\","
-            + "\"nfcNamingCode\":\"5DU\",\"priority\":\"Normal\",reportingEntityName\":\"5GRAN_DU\",\"sequence\":0,"
-            + "\"sourceId\":\"<<SerialNumber>>\",\"sourceName\":\"5GRAN_DU\",startEpochMicrosec\":1519837825682,\""
-            + "version\":3},\"otherFields\":{\"otherFieldsVersion\":1,\"pnfFamily\":\"BBU\",\"pnfLastServiceDate\""
-            + ":1517206400,\"pnfManufactureDate\":1516406400,\"pnfModelNumber\":\"AJ02\",\"pnfOamIpv4Address\":"
-            + "\"10.16.123.234\",\"pnfOamIpv6Address\":\"0:0:0:0:0:FFFF:0A10:7BEA\",\"pnfSoftwareVersion\":"
-            + "\"v4.5.0.1\",\"pnfType\":\"AirScale\"}}}";
+    void whenPassingJsonWithoutMandatoryHeaderInformation_validationThrowingAnException() {
+        JsonMessage message = new JsonMessage.JsonMessageBuilder().changeIdentifier("PM_MEAS_FILES_INVALID")
+                .changeType("FileReady_INVALID").notificationFieldsVersion("1.0_INVALID").build();
+        String incorrectMessageString = message.toString();
+
+        String parsedString = message.getParsed();
         DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
-        JsonElement jsonElement = new JsonParser().parse(parsed);
-        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
-            .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
-        String jsonWithoutPnfVendorAndSerialNumber =
-            "[{\"event\":{\"commonEventHeader\":{\"domain\":\"other\",\"eventId\":"
-                + "\"<<SerialNumber>>-reg\",\"eventName\":\"pnfRegistration_5GDU\",\"eventType\":\"pnfRegistration\",\""
-                + "internalHeaderFields\":{},\"lastEpochMicrosec\":1519837825682,\"nfNamingCode\":\"5GRAN\","
-                + "\"nfcNamingCode\":\"5DU\",\"priority\":\"Normal\",reportingEntityName\":\"5GRAN_DU\",\"sequence\":0,"
-                + "\"sourceId\":\"<<SerialNumber>>\",\"sourceName\":\"5GRAN_DU\",startEpochMicrosec\":1519837825682,"
-                + "\"version\":3},\"otherFields\":{\"otherFieldsVersion\":1,\"pnfFamily\":\"BBU\","
-                + "\"pnfLastServiceDate\":1517206400,\"pnfManufactureDate\":1516406400,\"pnfModelNumber\":\"AJ02\","
-                + "\"pnfOamIpv4Address\":\"10.16.123.234\",\"pnfOamIpv6Address\":\"0:0:0:0:0:FFFF:0A10:7BEA\","
-                + "\"pnfSoftwareVersion\":\"v4.5.0.1\",\"pnfType\":\"AirScale\"}}}]";
-        StepVerifier
-            .create(dmaapConsumerJsonParser.getJsonObject(Mono.just(jsonWithoutPnfVendorAndSerialNumber)))
-            .expectSubscription().expectError(DmaapNotFoundException.class).verify();
+        JsonElement jsonElement = new JsonParser().parse(parsedString);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(dmaapConsumerJsonParser)
+                .getJsonObjectFromAnArray(jsonElement);
+        StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(incorrectMessageString)))
+                .expectSubscription().expectError(DmaapNotFoundException.class).verify();
     }
 
     @Test
-    void whenPassingJsonWithoutIpInformation_validationThrowingAnException() {
-        String parsed =
-            "{\"event\":{\"commonEventHeader\":{\"domain\":\"other\",\"eventId\":\"<<SerialNumber>>-reg\","
-                + "\"eventName\":\"pnfRegistration_5GDU\",\"eventType\":\"pnfRegistration\",\"internalHeaderFields\""
-                + ":{},\"lastEpochMicrosec\":1519837825682,\"nfNamingCode\":\"5GRAN\",\"nfcNamingCode\":\"5DU\","
-                + "\"priority\":\"Normal\",\"reportingEntityName\":\"5GRAN_DU\",\"sequence\":0,\"sourceId\":"
-                + "\"<<SerialNumber>>\",\"sourceName\":\"5GRAN_DU\",\"startEpochMicrosec\":1519837825682,\"version\""
-                + ":3},\"otherFields\":{\"otherFieldsVersion\":1,\"pnfFamily\":\"BBU\",\"pnfLastServiceDate\":"
-                + "1517206400,\"pnfManufactureDate\":1516406400,\"pnfModelNumber\":"
-                + "\"AJ02\",\"pnfSerialNumber\":\"QTFCOC540002E\",\"pnfSoftwareVersion\":\"v4.5.0.1\",\"pnfType\":"
-                + "\"AirScale\"," + "\"pnfVendorName\":\"Nokia\"}}}";
+    void whenPassingJsonWithNullJsonElement_validationThrowingAnException() {
+        JsonMessage message = new JsonMessage.JsonMessageBuilder().build();
+        String incorrectMessageString = message.toString();
+
+        String parsedString = message.getParsed();
+
         DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
-        JsonElement jsonElement = new JsonParser().parse(parsed);
-        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
-            .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
-        String jsonWithoutIpInformation =
-            "[{\"event\":{\"commonEventHeader\":{\"domain\":\"other\",\"eventId\":\"<<SerialNumber>>-reg\","
-                + "\"eventName\":\"pnfRegistration_5GDU\",\"eventType\":\"pnfRegistration\",\"internalHeaderFields\""
-                + ":{},\"lastEpochMicrosec\":1519837825682,\"nfNamingCode\":\"5GRAN\",\"nfcNamingCode\":\"5DU\","
-                + "\"priority\":\"Normal\",\"reportingEntityName\":\"5GRAN_DU\",\"sequence\":0,\"sourceId\":"
-                + "\"<<SerialNumber>>\",\"sourceName\":\"5GRAN_DU\",\"startEpochMicrosec\":1519837825682,\"version\""
-                + ":3},\"otherFields\":{\"otherFieldsVersion\":1,\"pnfFamily\":\"BBU\",\"pnfLastServiceDate\""
-                + ":1517206400,\"pnfManufactureDate\":1516406400,\"pnfModelNumber\":\"AJ02\",\"pnfSerialNumber\""
-                + ":\"QTFCOC540002E\",\"pnfSoftwareVersion\":\"v4.5.0.1\",\"pnfType\":\"AirScale\","
-                + "\"pnfVendorName\":\"Nokia\"}}}]";
-        StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(jsonWithoutIpInformation)))
-            .expectSubscription().expectError(DmaapNotFoundException.class).verify();
+        JsonElement jsonElement = new JsonParser().parse(parsedString);
+
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(dmaapConsumerJsonParser)
+                .getJsonObjectFromAnArray(jsonElement);
+        StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(incorrectMessageString)))
+                .expectSubscription().expectError(DmaapNotFoundException.class).verify();
     }
 }
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/AaiConsumerTaskImplTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/AaiConsumerTaskImplTest.java
deleted file mode 100644
index 05a99b43..00000000
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/AaiConsumerTaskImplTest.java
+++ /dev/null
@@ -1,149 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * Datafile Collector Service
- * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. 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.tasks;
-
-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.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.util.Optional;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.function.Executable;
-import org.onap.dcaegen2.collectors.datafile.config.AaiClientConfiguration;
-import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
-import org.onap.dcaegen2.collectors.datafile.exceptions.AaiNotFoundException;
-import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
-import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.collectors.datafile.service.AaiConsumerClient;
-import org.onap.dcaegen2.collectors.datafile.tasks.AaiConsumerTaskImpl;
-import org.onap.dcaegen2.collectors.datafile.config.ImmutableAaiClientConfiguration;
-import org.onap.dcaegen2.collectors.datafile.model.ImmutableConsumerDmaapModel;
-
-/**
- * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 5/17/18
- */
-class AaiConsumerTaskImplTest {
-
-    private static ConsumerDmaapModel consumerDmaapModel;
-    private static AaiConsumerTaskImpl aaiConsumerTask;
-
-    private static final String AAI_HOST = "/aai/v12/network/pnfs/pnf/NOKQTFCOC540002E";
-    private static final Integer PORT = 1234;
-    private static final String PROTOCOL = "https";
-    private static final String USER_NAME_PASSWORD = "Datafile";
-    private static final String BASE_PATH = "/aai/v12";
-    private static final String PNF_PATH = "/network/pnfs/pnf";
-
-    private static AaiClientConfiguration aaiClientConfiguration;
-    private static AaiConsumerClient aaiConsumerClient;
-    private static AppConfig appConfig;
-
-    @BeforeAll
-    static void setUp() {
-        aaiClientConfiguration = new ImmutableAaiClientConfiguration.Builder()
-            .aaiHost(AAI_HOST)
-            .aaiPort(PORT)
-            .aaiProtocol(PROTOCOL)
-            .aaiUserName(USER_NAME_PASSWORD)
-            .aaiUserPassword(USER_NAME_PASSWORD)
-            .aaiIgnoreSslCertificateErrors(true)
-            .aaiBasePath(BASE_PATH)
-            .aaiPnfPath(PNF_PATH)
-            .build();
-        consumerDmaapModel = ImmutableConsumerDmaapModel.builder().ipv4("10.16.123.234")
-            .ipv6("0:0:0:0:0:FFFF:0A10:7BEA")
-            .pnfName("NOKQTFCOC540002E").build();
-        appConfig = mock(AppConfig.class);
-
-    }
-
-    @Test
-    void whenPassedObjectDoesntFit_ThrowsDatafileTaskException() {
-        //given/when
-        when(appConfig.getAaiClientConfiguration()).thenReturn(aaiClientConfiguration);
-        aaiConsumerTask = new AaiConsumerTaskImpl(appConfig);
-        Executable executableCode = () -> aaiConsumerTask.execute(null);
-        //then
-        Assertions
-            .assertThrows(DatafileTaskException.class, executableCode, "Passing wrong object type to execute function");
-
-    }
-
-    @Test
-    void whenPassedObjectFits_ReturnsCorrectStatus() throws DatafileTaskException, IOException {
-        //given/when
-        getAaiConsumerTask_WhenMockingHttpResponseCode("200", false);
-        String response = aaiConsumerTask.execute(consumerDmaapModel);
-
-        //then
-        verify(aaiConsumerClient, times(1)).getHttpResponse(any(ConsumerDmaapModel.class));
-        verifyNoMoreInteractions(aaiConsumerClient);
-        Assertions.assertEquals("200", response);
-    }
-
-    @Test
-    void whenPassedObjectFits_butIncorrectResponseReturns() throws IOException, AaiNotFoundException {
-        //given/when
-        getAaiConsumerTask_WhenMockingHttpResponseCode("400", false);
-        String response = aaiConsumerTask.execute(consumerDmaapModel);
-
-        //then
-        verify(aaiConsumerClient, times(1)).getHttpResponse(any(ConsumerDmaapModel.class));
-        verifyNoMoreInteractions(aaiConsumerClient);
-        Assertions.assertEquals("400", response);
-    }
-
-    @Test
-    void whenPassedObjectFits_ThrowsIoExceptionAndHandleIt() throws IOException {
-        //given/when
-        getAaiConsumerTask_WhenMockingHttpResponseCode(null, true);
-        Executable executableCode = () -> aaiConsumerTask.execute(any(ConsumerDmaapModel.class));
-        Assertions
-            .assertThrows(DatafileTaskException.class, executableCode, "HttpClient throws IOException");
-
-        //then
-        verifyNoMoreInteractions(aaiConsumerClient);
-    }
-
-
-    private static void getAaiConsumerTask_WhenMockingHttpResponseCode(String httpResponseCode, boolean throwsException)
-        throws IOException {
-        aaiConsumerClient = mock(AaiConsumerClient.class);
-        if (throwsException) {
-            when(aaiConsumerClient.getHttpResponse(consumerDmaapModel)).thenThrow(IOException.class);
-        } else {
-            when(aaiConsumerClient.getHttpResponse(consumerDmaapModel)).thenReturn(Optional.of(httpResponseCode));
-        }
-        when(appConfig.getAaiClientConfiguration()).thenReturn(aaiClientConfiguration);
-        aaiConsumerTask = spy(new AaiConsumerTaskImpl(appConfig));
-        when(aaiConsumerTask.resolveConfiguration()).thenReturn(aaiClientConfiguration);
-        doReturn(aaiConsumerClient).when(aaiConsumerTask).resolveClient();
-    }
-
-}
\ No newline at end of file
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/AaiConsumerTaskSpy.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/AaiConsumerTaskSpy.java
deleted file mode 100644
index 9e1842b0..00000000
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/AaiConsumerTaskSpy.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * PROJECT
- * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. 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.tasks;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.when;
-
-import org.onap.dcaegen2.collectors.datafile.config.AaiClientConfiguration;
-import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
-import org.onap.dcaegen2.collectors.datafile.tasks.AaiConsumerTask;
-import org.onap.dcaegen2.collectors.datafile.tasks.AaiConsumerTaskImpl;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Primary;
-
-@Configuration
-public class AaiConsumerTaskSpy {
-
-    /**
-     * Mocking bean for tests.
-     *
-     * @return A&AI ConsumerTask spy
-     */
-    @Bean
-    @Primary
-    public AaiConsumerTask registerSimpleAaiPublisherTask() {
-        AppConfig appConfig = mock(AppConfig.class);
-        when(appConfig.getAaiClientConfiguration()).thenReturn(mock(AaiClientConfiguration.class));
-        return spy(new AaiConsumerTaskImpl(appConfig));
-    }
-}
\ No newline at end of file
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/AaiProducerTaskImplTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/AaiProducerTaskImplTest.java
deleted file mode 100644
index f33fc931..00000000
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/AaiProducerTaskImplTest.java
+++ /dev/null
@@ -1,133 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * Datafile Collector Service
- * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. 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.tasks;
-
-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.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
-import static org.mockito.Mockito.when;
-
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.function.Executable;
-import org.onap.dcaegen2.collectors.datafile.config.AaiClientConfiguration;
-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.service.producer.AaiProducerReactiveHttpClient;
-import org.onap.dcaegen2.collectors.datafile.tasks.AaiProducerTaskImpl;
-import org.onap.dcaegen2.collectors.datafile.config.ImmutableAaiClientConfiguration;
-import org.onap.dcaegen2.collectors.datafile.model.ImmutableConsumerDmaapModel;
-
-import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;
-
-/**
- * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 5/14/18
- */
-class AaiProducerTaskImplTest {
-
-
-    private static final String AAI_HOST = "/aai/v11/network/pnfs/pnf/NOKQTFCOC540002E";
-    private static final Integer PORT = 1234;
-    private static final String PROTOCOL = "https";
-    private static final String USER_NAME_PASSWORD = "Datafile";
-    private static final String BASE_PATH = "/aai/v11";
-    private static final String PNF_PATH = "/network/pnfs/pnf";
-
-    private static ConsumerDmaapModel consumerDmaapModel;
-    private static AaiProducerTaskImpl aaiProducerTask;
-    private static AaiClientConfiguration aaiClientConfiguration;
-    private static AaiProducerReactiveHttpClient aaiProducerReactiveHttpClient;
-    private static AppConfig appConfig;
-
-    @BeforeAll
-    static void setUp() {
-        aaiClientConfiguration = new ImmutableAaiClientConfiguration.Builder()
-            .aaiHost(AAI_HOST)
-            .aaiPort(PORT)
-            .aaiProtocol(PROTOCOL)
-            .aaiUserName(USER_NAME_PASSWORD)
-            .aaiUserPassword(USER_NAME_PASSWORD)
-            .aaiIgnoreSslCertificateErrors(true)
-            .aaiBasePath(BASE_PATH)
-            .aaiPnfPath(PNF_PATH)
-            .build();
-        consumerDmaapModel = ImmutableConsumerDmaapModel.builder().ipv4("10.16.123.234")
-            .ipv6("0:0:0:0:0:FFFF:0A10:7BEA")
-            .pnfName("NOKQTFCOC540002E").build();
-        appConfig = mock(AppConfig.class);
-
-    }
-
-    @Test
-    void whenPassedObjectDoesntFit_ThrowsDatafileTaskException() {
-        //given/when/
-        when(appConfig.getAaiClientConfiguration()).thenReturn(aaiClientConfiguration);
-        aaiProducerTask = new AaiProducerTaskImpl(appConfig);
-        Executable executableCode = () -> aaiProducerTask.execute(null);
-
-        //then
-        Assertions
-            .assertThrows(DatafileTaskException.class, executableCode, "Passing wrong object type to execute function");
-    }
-
-    @Test
-    void whenPassedObjectFits_ReturnsCorrectStatus() throws DatafileTaskException {
-        //given/when
-        getAaiProducerTask_whenMockingResponseObject(200);
-        Mono<ConsumerDmaapModel> response = aaiProducerTask.execute(Mono.just(consumerDmaapModel));
-
-        //then
-        verify(aaiProducerReactiveHttpClient, times(1)).getAaiProducerResponse(any());
-        verifyNoMoreInteractions(aaiProducerReactiveHttpClient);
-        Assertions.assertEquals(consumerDmaapModel, response.block());
-
-    }
-
-
-    @Test
-    void whenPassedObjectFits_butIncorrectResponseReturns() throws DatafileTaskException {
-        //given/when
-        getAaiProducerTask_whenMockingResponseObject(400);
-        StepVerifier.create(aaiProducerTask.execute(Mono.just(consumerDmaapModel))).expectSubscription()
-            .expectError(DatafileTaskException.class).verify();
-        //then
-        verify(aaiProducerReactiveHttpClient, times(1)).getAaiProducerResponse(any());
-        verifyNoMoreInteractions(aaiProducerReactiveHttpClient);
-    }
-
-    private static void getAaiProducerTask_whenMockingResponseObject(Integer statusCode) {
-        //given
-        aaiProducerReactiveHttpClient = mock(AaiProducerReactiveHttpClient.class);
-        when(aaiProducerReactiveHttpClient.getAaiProducerResponse(any()))
-            .thenReturn(Mono.just(statusCode));
-        when(appConfig.getAaiClientConfiguration()).thenReturn(aaiClientConfiguration);
-        aaiProducerTask = spy(new AaiProducerTaskImpl(appConfig));
-        when(aaiProducerTask.resolveConfiguration()).thenReturn(aaiClientConfiguration);
-        doReturn(aaiProducerReactiveHttpClient).when(aaiProducerTask).resolveClient();
-    }
-}
\ No newline at end of file
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/AaiPublisherTaskSpy.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/AaiPublisherTaskSpy.java
deleted file mode 100644
index 3beda94b..00000000
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/AaiPublisherTaskSpy.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * PROJECT
- * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. 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.tasks;
-
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
-
-import org.onap.dcaegen2.collectors.datafile.config.AaiClientConfiguration;
-import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
-import org.onap.dcaegen2.collectors.datafile.service.producer.AaiProducerReactiveHttpClient;
-import org.onap.dcaegen2.collectors.datafile.tasks.AaiProducerTask;
-import org.onap.dcaegen2.collectors.datafile.tasks.AaiProducerTaskImpl;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Primary;
-
-/**
- * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/13/18
- */
-@Configuration
-public class AaiPublisherTaskSpy {
-
-    /**
-     * Mocking bean for tests.
-     *
-     * @return A&AI ProducerTask spy
-     */
-    @Bean
-    @Primary
-    public AaiProducerTask registerSimpleAaiPublisherTask() {
-        AppConfig appConfig = spy(AppConfig.class);
-        doReturn(mock(AaiClientConfiguration.class)).when(appConfig).getAaiClientConfiguration();
-        AaiProducerTaskImpl aaiProducerTask = spy(new AaiProducerTaskImpl(appConfig));
-        AaiProducerReactiveHttpClient aaiProducerReactiveHttpClient = mock(AaiProducerReactiveHttpClient.class);
-        doReturn(mock(AaiClientConfiguration.class)).when(aaiProducerTask).resolveConfiguration();
-        doReturn(aaiProducerReactiveHttpClient).when(aaiProducerTask).resolveClient();
-        return aaiProducerTask;
-    }
-}
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapConsumerTaskImplTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapConsumerTaskImplTest.java
index 19ab1ab9..c21c5988 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapConsumerTaskImplTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapConsumerTaskImplTest.java
@@ -1,136 +1,190 @@
 /*
- * ============LICENSE_START=======================================================
- * Datafile Collector Service
- * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. 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) 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
+ * 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.
+ * ============LICENSE_END========================================================================
  */
 
 package org.onap.dcaegen2.collectors.datafile.tasks;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.Mockito.doReturn;
 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 com.google.gson.JsonElement;
-import com.google.gson.JsonParser;
-import java.util.Optional;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
 import org.onap.dcaegen2.collectors.datafile.config.DmaapConsumerConfiguration;
+import org.onap.dcaegen2.collectors.datafile.config.ImmutableDmaapConsumerConfiguration;
 import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
+import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
 import org.onap.dcaegen2.collectors.datafile.exceptions.DmaapEmptyResponseException;
+import org.onap.dcaegen2.collectors.datafile.ftp.FileCollector;
 import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.collectors.datafile.service.DmaapConsumerJsonParser;
-import org.onap.dcaegen2.collectors.datafile.service.consumer.DMaaPConsumerReactiveHttpClient;
-import org.onap.dcaegen2.collectors.datafile.tasks.DmaapConsumerTaskImpl;
-import org.onap.dcaegen2.collectors.datafile.config.ImmutableDmaapConsumerConfiguration;
 import org.onap.dcaegen2.collectors.datafile.model.ImmutableConsumerDmaapModel;
+import org.onap.dcaegen2.collectors.datafile.service.DmaapConsumerJsonParser;
+import org.onap.dcaegen2.collectors.datafile.service.FileData;
+import org.onap.dcaegen2.collectors.datafile.service.ImmutableFileData;
+import org.onap.dcaegen2.collectors.datafile.service.consumer.DmaapConsumerReactiveHttpClient;
+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;
 
 /**
  * @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>
  */
 class DmaapConsumerTaskImplTest {
+    private static final String PM_MEAS_CHANGE_IDINTIFIER = "PM_MEAS_FILES";
+    private static final String FILE_READY_CHANGE_TYPE = "FileReady";
+    private static final String FTPES_SCHEME = "ftpes://";
+    private static final String SFTP_SCHEME = "sftp://";
+    private static final String SERVER_ADDRESS = "192.168.0.101";
+    private static final String 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 String LOCAL_FILE_LOCATION = "target/" + PM_FILE_NAME;
+    private static final String FTPES_LOCATION = FTPES_SCHEME + SERVER_ADDRESS + ":" + PORT_22 + REMOTE_FILE_LOCATION;
+    private static final String SFTP_LOCATION = SFTP_SCHEME + SERVER_ADDRESS + ":" + PORT_22 + REMOTE_FILE_LOCATION;
+    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 ConsumerDmaapModel consumerDmaapModel;
-    private static DmaapConsumerTaskImpl dmaapConsumerTask;
-    private static DMaaPConsumerReactiveHttpClient dMaaPConsumerReactiveHttpClient;
     private static AppConfig appConfig;
     private static DmaapConsumerConfiguration dmaapConsumerConfiguration;
-    private static String message;
-    private static String parsed;
+    private DmaapConsumerTaskImpl dmaapConsumerTask;
+    private DmaapConsumerReactiveHttpClient dmaapConsumerReactiveHttpClient;
+
+    private static FileCollector fileCollectorMock;
+
+    private static String ftpesMessage;
+    private static List<FileData> ftpesFileDataAfterConsume = new ArrayList<FileData>();
+
+    private static String sftpMessage;
+    private static List<FileData> sftpFileDataAfterConsume = new ArrayList<FileData>();
 
     @BeforeAll
-    static void setUp() {
+    public static void setUp() {
         dmaapConsumerConfiguration = new ImmutableDmaapConsumerConfiguration.Builder().consumerGroup("OpenDCAE-c12")
-            .consumerId("c12").dmaapContentType("application/json").dmaapHostName("54.45.33.2").dmaapPortNumber(1234)
-            .dmaapProtocol("https").dmaapUserName("Datafile").dmaapUserPassword("Datafile")
-            .dmaapTopicName("unauthenticated.SEC_OTHER_OUTPUT").timeoutMs(-1).messageLimit(-1).build();
+                .consumerId("c12").dmaapContentType("application/json").dmaapHostName("54.45.33.2")
+                .dmaapPortNumber(1234).dmaapProtocol("https").dmaapUserName("Datafile").dmaapUserPassword("Datafile")
+                .dmaapTopicName("unauthenticated.NOTIFICATION").timeoutMS(-1).messageLimit(-1).build();
 
-        consumerDmaapModel = ImmutableConsumerDmaapModel.builder().ipv4("10.16.123.234")
-            .ipv6("0:0:0:0:0:FFFF:0A10:7BEA")
-            .pnfName("NOKQTFCOC540002E").build();
         appConfig = mock(AppConfig.class);
-        message =
-            "[{\"event\":{\"commonEventHeader\":{\"domain\":\"other\",\"eventId\":\"<<SerialNumber>>-reg\","
-                + "\"eventName\":\"pnfRegistration_5GDU\",\"eventType\":\"pnfRegistration\",\"internalHeaderFields\""
-                + ":{},\"lastEpochMicrosec\":1519837825682,\"nfNamingCode\":\"5GRAN\",\"nfcNamingCode\":\"5DU\","
-                + "\"priority\":\"Normal\",\"reportingEntityName\":\"5GRAN_DU\",\"sequence\":0,\"sourceId\":"
-                + "\"<<SerialNumber>>\",\"sourceName\":\"5GRAN_DU\",\"startEpochMicrosec\":1519837825682,\"version\""
-                + ":3},\"otherFields\":{\"otherFieldsVersion\":1,\"pnfFamily\":\"BBU\",\"pnfLastServiceDate\":"
-                + "1517206400,\"pnfManufactureDate\":1516406400,\"pnfModelNumber\":\"AJ02\",\"pnfOamIpv4Address\":"
-                + "\"10.16.123.234\",\"pnfOamIpv6Address\":\"0:0:0:0:0:FFFF:0A10:7BEA\",\"pnfSerialNumber\":"
-                + "\"QTFCOC540002E\",\"pnfSoftwareVersion\":\"v4.5.0.1\",\"pnfType\":\"AirScale\",\"pnfVendorName\":"
-                + "\"Nokia\"}}}]";
-        parsed =
-            "{\"event\":{\"commonEventHeader\":{\"domain\":\"other\",\"eventId\":\"<<SerialNumber>>-reg\",\"eventName\""
-                + ":\"pnfRegistration_5GDU\",\"eventType\":\"pnfRegistration\",\"internalHeaderFields\":{},"
-                + "\"lastEpochMicrosec\":1519837825682,\"nfNamingCode\":\"5GRAN\",\"nfcNamingCode\":\"5DU\","
-                + "\"priority\":\"Normal\",\"reportingEntityName\":\"5GRAN_DU\",\"sequence\":0,\"sourceId\":"
-                + "\"<<SerialNumber>>\",\"sourceName\":\"5GRAN_DU\",\"startEpochMicrosec\":1519837825682,"
-                + "\"version\":3},\"otherFields\":{\"otherFieldsVersion\":1,\"pnfFamily\":\"BBU\","
-                + "\"pnfLastServiceDate\":1517206400,\"pnfManufactureDate\":1516406400,\"pnfModelNumber\":\"AJ02\","
-                + "\"pnfOamIpv4Address\":\"10.16.123.234\",\"pnfOamIpv6Address\":\"0:0:0:0:0:FFFF:0A10:7BEA\","
-                + "\"pnfSerialNumber\":\"QTFCOC540002E\",\"pnfSoftwareVersion\":\"v4.5.0.1\",\"pnfType\":\"AirScale\","
-                + "\"pnfVendorName\":\"Nokia\"}}}";
+
+        AdditionalField ftpesAdditionalField =
+                new JsonMessage.AdditionalFieldBuilder().location(FTPES_LOCATION).compression(GZIP_COMPRESSION)
+                        .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
+        JsonMessage ftpesJsonMessage = new JsonMessage.JsonMessageBuilder().changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
+                .changeType(FILE_READY_CHANGE_TYPE).notificationFieldsVersion("1.0")
+                .addAdditionalField(ftpesAdditionalField).build();
+        ftpesMessage = ftpesJsonMessage.toString();
+        FileData ftpesFileData = ImmutableFileData.builder().changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
+                .changeType(FILE_READY_CHANGE_TYPE).location(FTPES_LOCATION).compression(GZIP_COMPRESSION)
+                .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
+        ftpesFileDataAfterConsume.add(ftpesFileData);
+
+        AdditionalField sftpAdditionalField =
+                new JsonMessage.AdditionalFieldBuilder().location(SFTP_LOCATION).compression(GZIP_COMPRESSION)
+                        .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
+        JsonMessage sftpJsonMessage = new JsonMessage.JsonMessageBuilder().changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
+                .changeType(FILE_READY_CHANGE_TYPE).notificationFieldsVersion("1.0")
+                .addAdditionalField(sftpAdditionalField).build();
+        sftpMessage = sftpJsonMessage.toString();
+        FileData sftpFileData = ImmutableFileData.builder().changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
+                .changeType(FILE_READY_CHANGE_TYPE).location(SFTP_LOCATION).compression(GZIP_COMPRESSION)
+                .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
+        sftpFileDataAfterConsume.add(sftpFileData);
+
+
+        ImmutableConsumerDmaapModel consumerDmaapModel =
+                ImmutableConsumerDmaapModel.builder().location(LOCAL_FILE_LOCATION).compression(GZIP_COMPRESSION)
+                        .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
+        listOfConsumerDmaapModel.add(consumerDmaapModel);
+
+        fileCollectorMock = mock(FileCollector.class);
     }
 
     @Test
-    void whenPassedObjectDoesntFit_DoesNotThrowDatafileTaskException() {
-        //given
-        prepareMocksForDmaapConsumer(Optional.empty());
+    public void whenPassedObjectDoesntFit_ThrowsDatafileTaskException() {
+        // given
+        prepareMocksForDmaapConsumer("", new ArrayList<FileData>());
 
-        //then
+        // then
         StepVerifier.create(dmaapConsumerTask.execute("Sample input")).expectSubscription()
-            .expectError(DmaapEmptyResponseException.class).verify();
+                .expectError(DmaapEmptyResponseException.class).verify();
 
-        verify(dMaaPConsumerReactiveHttpClient, times(1)).getDMaaPConsumerResponse();
+        verify(dmaapConsumerReactiveHttpClient, times(1)).getDmaapConsumerResponse();
     }
 
     @Test
-    void whenPassedObjectFits_ReturnsCorrectResponse() {
-        //given
-        prepareMocksForDmaapConsumer(Optional.of(message));
-        //when
-        Mono<ConsumerDmaapModel> response = dmaapConsumerTask.execute("Sample input");
+    public void whenFtpes_ReturnsCorrectResponse() throws DatafileTaskException {
+        // given
+        prepareMocksForDmaapConsumer(ftpesMessage, ftpesFileDataAfterConsume);
+        // when
+        final List<ConsumerDmaapModel> arrayOfResponse = dmaapConsumerTask.execute("Sample input").block();
+        // then
+        verify(dmaapConsumerReactiveHttpClient, times(1)).getDmaapConsumerResponse();
+        verifyNoMoreInteractions(dmaapConsumerReactiveHttpClient);
+        verify(fileCollectorMock, times(1)).getFilesFromSender(ftpesFileDataAfterConsume);
+        verifyNoMoreInteractions(fileCollectorMock);
+        Assertions.assertEquals(listOfConsumerDmaapModel, arrayOfResponse);
 
-        //then
-        verify(dMaaPConsumerReactiveHttpClient, times(1)).getDMaaPConsumerResponse();
-        assertEquals(consumerDmaapModel, response.block());
+    }
 
+    @Test
+    public void whenSftp_ReturnsCorrectResponse() throws DatafileTaskException {
+        // given
+        prepareMocksForDmaapConsumer(sftpMessage, sftpFileDataAfterConsume);
+        // when
+        final List<ConsumerDmaapModel> arrayOfResponse = dmaapConsumerTask.execute("Sample input").block();
+        // then
+        verify(dmaapConsumerReactiveHttpClient, times(1)).getDmaapConsumerResponse();
+        verifyNoMoreInteractions(dmaapConsumerReactiveHttpClient);
+        verify(fileCollectorMock, times(1)).getFilesFromSender(sftpFileDataAfterConsume);
+        verifyNoMoreInteractions(fileCollectorMock);
+        Assertions.assertEquals(listOfConsumerDmaapModel, arrayOfResponse);
 
     }
 
-    private void prepareMocksForDmaapConsumer(Optional<String> message) {
-        DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
-        JsonElement jsonElement = new JsonParser().parse(parsed);
-        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
-            .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
-        dMaaPConsumerReactiveHttpClient = mock(DMaaPConsumerReactiveHttpClient.class);
-        when(dMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse()).thenReturn(Mono.just(message.orElse("")));
-        when(appConfig.getDmaapConsumerConfiguration()).thenReturn(dmaapConsumerConfiguration);
-        dmaapConsumerTask = spy(new DmaapConsumerTaskImpl(appConfig, dmaapConsumerJsonParser));
+    private void prepareMocksForDmaapConsumer(String message, List<FileData> fileDataAfterConsume) {
+        Mono<String> messageAsMono = Mono.just(message);
+        DmaapConsumerJsonParser dmaapConsumerJsonParserMock = mock(DmaapConsumerJsonParser.class);
+        dmaapConsumerReactiveHttpClient = mock(DmaapConsumerReactiveHttpClient.class);
+        when(dmaapConsumerReactiveHttpClient.getDmaapConsumerResponse()).thenReturn(messageAsMono);
+
+        if (!message.isEmpty()) {
+            when(dmaapConsumerJsonParserMock.getJsonObject(messageAsMono)).thenReturn(Mono.just(fileDataAfterConsume));
+        } else {
+            when(dmaapConsumerJsonParserMock.getJsonObject(messageAsMono))
+                    .thenReturn(Mono.error(new DmaapEmptyResponseException()));
+        }
+        when(fileCollectorMock.getFilesFromSender(fileDataAfterConsume))
+                .thenReturn(Mono.just(listOfConsumerDmaapModel));
+
+        dmaapConsumerTask = spy(new DmaapConsumerTaskImpl(appConfig, dmaapConsumerReactiveHttpClient,
+                dmaapConsumerJsonParserMock, fileCollectorMock));
         when(dmaapConsumerTask.resolveConfiguration()).thenReturn(dmaapConsumerConfiguration);
-        doReturn(dMaaPConsumerReactiveHttpClient).when(dmaapConsumerTask).resolveClient();
+        doReturn(dmaapConsumerReactiveHttpClient).when(dmaapConsumerTask).resolveClient();
     }
-}
\ No newline at end of file
+}
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapConsumerTaskSpy.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapConsumerTaskSpy.java
deleted file mode 100644
index de5c8535..00000000
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapConsumerTaskSpy.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * PROJECT
- * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. 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.tasks;
-
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
-
-import org.onap.dcaegen2.collectors.datafile.config.DmaapConsumerConfiguration;
-import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
-import org.onap.dcaegen2.collectors.datafile.service.consumer.DMaaPConsumerReactiveHttpClient;
-import org.onap.dcaegen2.collectors.datafile.tasks.DmaapConsumerTask;
-import org.onap.dcaegen2.collectors.datafile.tasks.DmaapConsumerTaskImpl;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Primary;
-
-/**
- * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/27/18
- */
-@Configuration
-public class DmaapConsumerTaskSpy {
-
-    /**
-     * Mocking bean for tests.
-     *
-     * @return DMaaP ConsumerTask spy
-     */
-    @Bean
-    @Primary
-    public DmaapConsumerTask registerSimpleDmaapConsumerTask() {
-        AppConfig appConfig = spy(AppConfig.class);
-        doReturn(mock(DmaapConsumerConfiguration.class)).when(appConfig).getDmaapConsumerConfiguration();
-        DmaapConsumerTaskImpl dmaapConsumerTask = spy(new DmaapConsumerTaskImpl(appConfig));
-        DMaaPConsumerReactiveHttpClient dmaapConsumerReactiveHttpClient = mock(
-            DMaaPConsumerReactiveHttpClient.class);
-        doReturn(mock(DmaapConsumerConfiguration.class)).when(dmaapConsumerTask).resolveConfiguration();
-        doReturn(dmaapConsumerReactiveHttpClient).when(dmaapConsumerTask).resolveClient();
-        return dmaapConsumerTask;
-    }
-}
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapProducerTaskSpy.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapProducerTaskSpy.java
deleted file mode 100644
index 870c712e..00000000
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapProducerTaskSpy.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * PROJECT
- * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. 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.tasks;
-
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
-
-import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration;
-import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
-import org.onap.dcaegen2.collectors.datafile.service.producer.DMaaPProducerReactiveHttpClient;
-import org.onap.dcaegen2.collectors.datafile.tasks.DmaapPublisherTask;
-import org.onap.dcaegen2.collectors.datafile.tasks.DmaapPublisherTaskImpl;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Primary;
-
-/**
- * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/13/18
- */
-@Configuration
-public class DmaapProducerTaskSpy {
-
-    /**
-     * Mocking bean for tests.
-     *
-     * @return DMaaP PublisherTask spy
-     */
-    @Bean
-    @Primary
-    public DmaapPublisherTask registerSimpleDmaapPublisherTask() {
-        AppConfig appConfig = spy(AppConfig.class);
-        doReturn(mock(DmaapPublisherConfiguration.class)).when(appConfig).getDmaapPublisherConfiguration();
-        DmaapPublisherTaskImpl dmaapPublisherTask = spy(new DmaapPublisherTaskImpl(appConfig));
-        DMaaPProducerReactiveHttpClient extendedDmaapProducerHttpClient = mock(
-            DMaaPProducerReactiveHttpClient.class);
-        doReturn(mock(DmaapPublisherConfiguration.class)).when(dmaapPublisherTask).resolveConfiguration();
-        doReturn(extendedDmaapProducerHttpClient).when(dmaapPublisherTask).resolveClient();
-        return dmaapPublisherTask;
-    }
-}
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapPublisherTaskImplTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapPublisherTaskImplTest.java
index 2b79bc40..4f7787e9 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapPublisherTaskImplTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapPublisherTaskImplTest.java
@@ -1,9 +1,7 @@
 /*
- * ============LICENSE_START=======================================================
- * PROJECT
- * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
- * ================================================================================
+ * ============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
@@ -15,12 +13,11 @@
  * 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=========================================================
+ * ============LICENSE_END========================================================================
  */
 
 package org.onap.dcaegen2.collectors.datafile.tasks;
 
-import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
@@ -30,95 +27,96 @@ import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
 
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.function.Executable;
 import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration;
+import org.onap.dcaegen2.collectors.datafile.config.ImmutableDmaapPublisherConfiguration;
 import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
-import org.onap.dcaegen2.collectors.datafile.exceptions.DmaapNotFoundException;
 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
 import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.collectors.datafile.service.producer.DMaaPProducerReactiveHttpClient;
-import org.onap.dcaegen2.collectors.datafile.tasks.DmaapPublisherTaskImpl;
-import org.onap.dcaegen2.collectors.datafile.config.ImmutableDmaapPublisherConfiguration;
 import org.onap.dcaegen2.collectors.datafile.model.ImmutableConsumerDmaapModel;
+import org.onap.dcaegen2.collectors.datafile.service.producer.DmaapProducerReactiveHttpClient;
 import org.springframework.http.HttpStatus;
+
 import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;
 
 /**
  * @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>
  */
 class DmaapPublisherTaskImplTest {
 
     private static ConsumerDmaapModel consumerDmaapModel;
+    private static List<ConsumerDmaapModel> listOfConsumerDmaapModel;
     private static DmaapPublisherTaskImpl dmaapPublisherTask;
-    private static DMaaPProducerReactiveHttpClient dMaaPProducerReactiveHttpClient;
+    private static DmaapProducerReactiveHttpClient dMaaPProducerReactiveHttpClient;
     private static AppConfig appConfig;
     private static DmaapPublisherConfiguration dmaapPublisherConfiguration;
 
     @BeforeAll
-    static void setUp() {
-        dmaapPublisherConfiguration = new ImmutableDmaapPublisherConfiguration.Builder()
-            .dmaapContentType("application/json").dmaapHostName("54.45.33.2").dmaapPortNumber(1234)
-            .dmaapProtocol("https").dmaapUserName("Datafile").dmaapUserPassword("Datafile")
-            .dmaapTopicName("unauthenticated.SEC_OTHER_OUTPUT").build();
-        consumerDmaapModel = ImmutableConsumerDmaapModel.builder().ipv4("10.16.123.234")
-            .ipv6("0:0:0:0:0:FFFF:0A10:7BEA")
-            .pnfName("NOKQTFCOC540002E").build();
+    public static void setUp() {
+        dmaapPublisherConfiguration =
+                new ImmutableDmaapPublisherConfiguration.Builder().dmaapContentType("application/json")
+                        .dmaapHostName("54.45.33.2").dmaapPortNumber(1234).dmaapProtocol("https").dmaapUserName("DFC")
+                        .dmaapUserPassword("DFC").dmaapTopicName("unauthenticated.VES_NOTIFICATION_OUTPUT").build();
+        consumerDmaapModel = ImmutableConsumerDmaapModel.builder().location("target/A20161224.1030-1045.bin.gz")
+                .compression("gzip").fileFormatType("org.3GPP.32.435#measCollec").fileFormatVersion("V10").build();
+        listOfConsumerDmaapModel = new ArrayList<ConsumerDmaapModel>();
+        listOfConsumerDmaapModel.add(consumerDmaapModel);
         appConfig = mock(AppConfig.class);
     }
 
     @Test
-    void whenPassedObjectDoesntFit_ThrowsDatafileTaskException() {
-        //given
+    public void whenPassedObjectDoesntFit_ThrowsDatafileTaskException() {
+        // given
         when(appConfig.getDmaapPublisherConfiguration()).thenReturn(dmaapPublisherConfiguration);
         dmaapPublisherTask = new DmaapPublisherTaskImpl(appConfig);
 
-        //when
+        // when
         Executable executableFunction = () -> dmaapPublisherTask.execute(null);
 
-        //then
-        assertThrows(DatafileTaskException.class, executableFunction, "The specified parameter is incorrect");
+        // then
+        Assertions.assertThrows(DatafileTaskException.class, executableFunction,
+                "The specified parameter is incorrect");
     }
 
     @Test
-    void whenPassedObjectFits_ReturnsCorrectStatus() throws DatafileTaskException {
-        //given
+    public void whenPassedObjectFits_ReturnsCorrectStatus() throws DatafileTaskException {
+        // given
         prepareMocksForTests(HttpStatus.OK.value());
 
-        //when
-        StepVerifier.create(dmaapPublisherTask.execute(Mono.just(consumerDmaapModel))).expectSubscription()
-            .expectNext(HttpStatus.OK.toString()).verifyComplete();
+        // when
+        dmaapPublisherTask.execute(Mono.just(listOfConsumerDmaapModel));
 
-        //then
-        verify(dMaaPProducerReactiveHttpClient, times(1))
-            .getDMaaPProducerResponse(any(Mono.class));
+        // then
+        verify(dMaaPProducerReactiveHttpClient, times(1)).getDmaapProducerResponse(any());
         verifyNoMoreInteractions(dMaaPProducerReactiveHttpClient);
     }
 
-
     @Test
-    void whenPassedObjectFits_butIncorrectResponseReturns() throws DmaapNotFoundException {
-        //given
-        prepareMocksForTests(HttpStatus.UNAUTHORIZED.value());
+    public void whenPassedObjectFits_ReturnsNoContent() throws DatafileTaskException {
+        // given
+        prepareMocksForTests(HttpStatus.NO_CONTENT.value());
 
-        //when
-        StepVerifier.create(dmaapPublisherTask.execute(Mono.just(consumerDmaapModel))).expectSubscription()
-            .expectNext(String.valueOf(HttpStatus.UNAUTHORIZED.value())).verifyComplete();
+        dmaapPublisherTask.execute(Mono.just(listOfConsumerDmaapModel));
 
-        //then
-        verify(dMaaPProducerReactiveHttpClient, times(1)).getDMaaPProducerResponse(any(Mono.class));
+        // then
+        verify(dMaaPProducerReactiveHttpClient, times(1)).getDmaapProducerResponse(any());
         verifyNoMoreInteractions(dMaaPProducerReactiveHttpClient);
     }
 
-
     private void prepareMocksForTests(Integer httpResponseCode) {
-        dMaaPProducerReactiveHttpClient = mock(DMaaPProducerReactiveHttpClient.class);
-        when(dMaaPProducerReactiveHttpClient.getDMaaPProducerResponse(any(Mono.class)))
-            .thenReturn(Mono.just(httpResponseCode.toString()));
+        dMaaPProducerReactiveHttpClient = mock(DmaapProducerReactiveHttpClient.class);
+        when(dMaaPProducerReactiveHttpClient.getDmaapProducerResponse(any()))
+                .thenReturn(Mono.just(httpResponseCode.toString()));
+        when(appConfig.getDmaapPublisherConfiguration()).thenReturn(dmaapPublisherConfiguration);
         dmaapPublisherTask = spy(new DmaapPublisherTaskImpl(appConfig));
         when(dmaapPublisherTask.resolveConfiguration()).thenReturn(dmaapPublisherConfiguration);
         doReturn(dMaaPProducerReactiveHttpClient).when(dmaapPublisherTask).resolveClient();
     }
-}
\ No newline at end of file
+}
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduleControllerSpy.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduleControllerSpy.java
deleted file mode 100644
index d47d31d1..00000000
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduleControllerSpy.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * Datafile Collector Service
- * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. 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.tasks;
-
-import static org.mockito.Mockito.spy;
-
-import org.onap.dcaegen2.collectors.datafile.tasks.AaiProducerTask;
-import org.onap.dcaegen2.collectors.datafile.tasks.DmaapConsumerTask;
-import org.onap.dcaegen2.collectors.datafile.tasks.DmaapPublisherTask;
-import org.onap.dcaegen2.collectors.datafile.tasks.ScheduledTasks;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Primary;
-
-/**
- * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/5/18
- */
-@Configuration
-public class ScheduleControllerSpy {
-
-    @Autowired
-    private DmaapConsumerTask dmaapConsumerTaskImplSpy;
-
-    @Autowired
-    private DmaapPublisherTask dmaapPublisherTaskImplSpy;
-
-    @Autowired
-    private AaiProducerTask aaiPublisherTaskImplSpy;
-
-    @Bean
-    @Primary
-    public ScheduledTasks registerSimpleScheduledTask() {
-        return spy(new ScheduledTasks(dmaapConsumerTaskImplSpy, dmaapPublisherTaskImplSpy, aaiPublisherTaskImplSpy));
-    }
-}
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
new file mode 100644
index 00000000..264a9945
--- /dev/null
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/utils/JsonMessage.java
@@ -0,0 +1,204 @@
+/*
+ * ============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.utils;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * Utility class to produce correctly formatted fileReady event Json messages.
+ *
+ * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a> on 7/25/18
+ *
+ */
+public class JsonMessage {
+    private String changeIdentifier;
+    private String changeType;
+    private String notificationFieldsVersion;
+    private List<AdditionalField> arrayOfAdditionalFields;
+
+
+    @Override
+    public String toString() {
+        return "[" + getParsed() + "]";
+    }
+
+    /**
+     * Gets the message in parsed format.
+     * @return the massage in parsed format.
+     */
+    public String getParsed() {
+        StringBuffer additionalFieldsString = new StringBuffer();
+        if (arrayOfAdditionalFields.size() > 0) {
+            additionalFieldsString.append("\"arrayOfAdditionalFields\": [");
+            for (Iterator<AdditionalField> iterator = arrayOfAdditionalFields.iterator(); iterator.hasNext();) {
+                AdditionalField additionalField = iterator.next();
+                additionalFieldsString.append(additionalField.toString());
+                if (iterator.hasNext()) {
+                    additionalFieldsString.append(",");
+                }
+            }
+            additionalFieldsString.append("]");
+        }
+
+        return "{" + "\"event\":{" + "\"commonEventHeader\":{" + "\"domain\":\"notification\","
+                + "\"eventId\":\"<<SerialNumber>>-reg\"," + "\"eventName\":\"EriNoti_RnNode_FileReady\","
+                + "\"eventType\":\"fileReady\"," + "\"internalHeaderFields\":{},"
+                + "\"lastEpochMicrosec\":1519837825682," + "\"nfNamingCode\":\"5GRAN\"," + "\"nfcNamingCode\":\"5DU\","
+                + "\"priority\":\"Normal\"," + "\"reportingEntityName\":\"5GRAN_DU\"," + "\"sequence\":0,"
+                + "\"sourceId\":\"<<SerialNumber>>\"," + "\"sourceName\":\"5GRAN_DU\","
+                + "\"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() + "}" + "}" + "}";
+    }
+
+    private JsonMessage(final JsonMessageBuilder builder) {
+        this.changeIdentifier = builder.changeIdentifier;
+        this.changeType = builder.changeType;
+        this.notificationFieldsVersion = builder.notificationFieldsVersion;
+        this.arrayOfAdditionalFields = builder.arrayOfAdditionalFields;
+    }
+
+    public static class AdditionalField {
+        private String location;
+        private String compression;
+        private String fileFormatType;
+        private String fileFormatVersion;
+
+        @Override
+        public String toString() {
+            return "{"
+                    + 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) + "}";
+        }
+
+
+        private AdditionalField(AdditionalFieldBuilder builder) {
+            this.location = builder.location;
+            this.compression = builder.compression;
+            this.fileFormatType = builder.fileFormatType;
+            this.fileFormatVersion = builder.fileFormatVersion;
+        }
+
+    }
+
+    public static class AdditionalFieldBuilder {
+        private String location;
+        private String compression;
+        private String fileFormatType;
+        private String fileFormatVersion;
+
+        public AdditionalFieldBuilder location(String location) {
+            this.location = location;
+            return this;
+        }
+
+        public AdditionalFieldBuilder compression(String compression) {
+            this.compression = compression;
+            return this;
+        }
+
+        public AdditionalFieldBuilder fileFormatType(String fileFormatType) {
+            this.fileFormatType = fileFormatType;
+            return this;
+        }
+
+        public AdditionalFieldBuilder fileFormatVersion(String fileFormatVersion) {
+            this.fileFormatVersion = fileFormatVersion;
+            return this;
+        }
+
+        public AdditionalField build() {
+            return new AdditionalField(this);
+        }
+    }
+
+    public static class JsonMessageBuilder {
+        private String changeIdentifier;
+        private String changeType;
+        private String notificationFieldsVersion;
+        private List<AdditionalField> arrayOfAdditionalFields = new ArrayList<AdditionalField>();
+
+        public JsonMessageBuilder changeIdentifier(String changeIdentifier) {
+            this.changeIdentifier = changeIdentifier;
+            return this;
+        }
+
+        public JsonMessageBuilder changeType(String changeType) {
+            this.changeType = changeType;
+            return this;
+        }
+
+        public JsonMessageBuilder notificationFieldsVersion(String notificationFieldsVersion) {
+            this.notificationFieldsVersion = notificationFieldsVersion;
+            return this;
+        }
+
+        public JsonMessageBuilder addAdditionalField(AdditionalField additionalField) {
+            this.arrayOfAdditionalFields.add(additionalField);
+            return this;
+        }
+
+        public JsonMessage build() {
+            return new JsonMessage(this);
+        }
+    }
+
+    private static String getAsStringIfParameterIsSet(String parameterName, String parameterValue,
+            boolean withSeparator) {
+        String result = "";
+        if (parameterValue != null) {
+            result = "\"" + parameterName + "\":\"" + parameterValue + "\"";
+
+            if (withSeparator) {
+                result = result + ",";
+            }
+        }
+        return result;
+    }
+
+    /**
+     * Can be used to produce a correct test Json message. Tip! Check the formatting with
+     * <a href="https://jsonformatter.org/">Json fomatter</a>
+     *
+     * @param args Not used
+     */
+    public static void main(String[] args) {
+        AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder()
+                .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()
+                .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().changeIdentifier("PM_MEAS_FILES")
+                .changeType("FileReady").notificationFieldsVersion("1.0").addAdditionalField(additionalField)
+                .addAdditionalField(secondAdditionalField).build();
+        System.out.println(message.toString());
+    }
+}