Deliver first version of Datafile

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

6 files changed, 168 insertions, 236 deletions

diff --git a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/DMaaPReactiveWebClientTest.java b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/DMaaPReactiveWebClientTest.java
deleted file mode 100644
index dea22340..00000000
--- a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/DMaaPReactiveWebClientTest.java
+++ /dev/null
@@ -1,58 +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.service;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-import org.onap.dcaegen2.collectors.datafile.config.DmaapConsumerConfiguration;
-import org.onap.dcaegen2.collectors.datafile.service.DMaaPReactiveWebClient;
-import org.springframework.web.reactive.function.client.WebClient;
-
-/**
- * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 7/5/18
- */
-class DMaaPReactiveWebClientTest {
-
-
-    @Test
-    void builder_shouldBuildDMaaPReactiveWebClient() {
-        //given
-        DmaapConsumerConfiguration dmaapConsumerConfiguration = mock(DmaapConsumerConfiguration.class);
-        String dmaaPContentType = "*/*";
-        String dmaaPUserName = "DMaaP";
-        String dmaaPUserPassword = "DMaaP";
-
-        //when
-        when(dmaapConsumerConfiguration.dmaapContentType()).thenReturn(dmaaPContentType);
-        when(dmaapConsumerConfiguration.dmaapUserName()).thenReturn(dmaaPUserName);
-        when(dmaapConsumerConfiguration.dmaapUserPassword()).thenReturn(dmaaPUserPassword);
-        WebClient dmaapreactiveWebClient = new DMaaPReactiveWebClient()
-            .fromConfiguration(dmaapConsumerConfiguration)
-            .build();
-
-        //then
-        Assertions.assertNotNull(dmaapreactiveWebClient);
-
-    }
-}
\ No newline at end of file
diff --git a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/config/DmaapConsumerConfigurationTest.java b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/config/DmaapConsumerConfigurationTest.java
index d0943709..b67946b2 100644
--- a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/config/DmaapConsumerConfigurationTest.java
+++ b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/config/DmaapConsumerConfigurationTest.java
@@ -1,9 +1,7 @@
 /*
- * ============LICENSE_START=======================================================
- * Datafile Collector Service
- * ================================================================================
- * 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.service.config;
@@ -25,10 +23,10 @@ import org.junit.jupiter.api.Test;
 import org.onap.dcaegen2.collectors.datafile.config.DmaapConsumerConfiguration;
 import org.onap.dcaegen2.collectors.datafile.config.ImmutableDmaapConsumerConfiguration;
 
-class DmaapConsumerConfigurationTest {
+public class DmaapConsumerConfigurationTest {
 
     @Test
-    void builder_shouldBuildConfigurationObject() {
+    public void builder_shouldBuildConfigurationObject() {
 
         // Given
         DmaapConsumerConfiguration configuration;
@@ -45,19 +43,11 @@ class DmaapConsumerConfigurationTest {
         Integer messageLimit = 1000;
 
         // When
-        configuration = new ImmutableDmaapConsumerConfiguration.Builder()
-            .consumerId(consumerId)
-            .dmaapHostName(dmaapHostName)
-            .dmaapPortNumber(dmaapPortNumber)
-            .dmaapTopicName(dmaapTopicName)
-            .dmaapProtocol(dmaapProtocol)
-            .dmaapUserName(dmaapUserName)
-            .dmaapUserPassword(dmaapUserPassword)
-            .dmaapContentType(dmaapContentType)
-            .consumerGroup(consumerGroup)
-            .timeoutMs(timeoutMs)
-            .messageLimit(messageLimit)
-            .build();
+        configuration = new ImmutableDmaapConsumerConfiguration.Builder().consumerId(consumerId)
+                .dmaapHostName(dmaapHostName).dmaapPortNumber(dmaapPortNumber).dmaapTopicName(dmaapTopicName)
+                .dmaapProtocol(dmaapProtocol).dmaapUserName(dmaapUserName).dmaapUserPassword(dmaapUserPassword)
+                .dmaapContentType(dmaapContentType).consumerGroup(consumerGroup).timeoutMS(timeoutMs)
+                .messageLimit(messageLimit).build();
 
         // Then
         Assertions.assertNotNull(configuration);
@@ -69,7 +59,7 @@ class DmaapConsumerConfigurationTest {
         Assertions.assertEquals(dmaapUserName, configuration.dmaapUserName());
         Assertions.assertEquals(dmaapUserPassword, configuration.dmaapUserPassword());
         Assertions.assertEquals(consumerGroup, configuration.consumerGroup());
-        Assertions.assertEquals(timeoutMs, configuration.timeoutMs());
+        Assertions.assertEquals(timeoutMs, configuration.timeoutMS());
         Assertions.assertEquals(messageLimit, configuration.messageLimit());
     }
 }
diff --git a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/config/DmaapPublisherConfigurationTest.java b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/config/DmaapPublisherConfigurationTest.java
index a188921c..fb8e8751 100644
--- a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/config/DmaapPublisherConfigurationTest.java
+++ b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/config/DmaapPublisherConfigurationTest.java
@@ -1,9 +1,7 @@
 /*
- * ============LICENSE_START=======================================================
- * Datafile Collector Service
- * ================================================================================
- * 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.service.config;
@@ -25,11 +23,11 @@ import org.junit.jupiter.api.Test;
 import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration;
 import org.onap.dcaegen2.collectors.datafile.config.ImmutableDmaapPublisherConfiguration;
 
-class DmaapPublisherConfigurationTest {
+public class DmaapPublisherConfigurationTest {
 
 
     @Test
-    void builder_shouldBuildConfigurationObject() {
+    public void builder_shouldBuildConfigurationObject() {
 
         // Given
         DmaapPublisherConfiguration configuration;
@@ -42,15 +40,10 @@ class DmaapPublisherConfigurationTest {
         String dmaapContentType = "application/json";
 
         // When
-        configuration = new ImmutableDmaapPublisherConfiguration.Builder()
-            .dmaapHostName(dmaapHostName)
-            .dmaapPortNumber(dmaapPortNumber)
-            .dmaapTopicName(dmaapTopicName)
-            .dmaapProtocol(dmaapProtocol)
-            .dmaapUserName(dmaapUserName)
-            .dmaapUserPassword(dmaapUserPassword)
-            .dmaapContentType(dmaapContentType)
-            .build();
+        configuration = new ImmutableDmaapPublisherConfiguration.Builder().dmaapHostName(dmaapHostName)
+                .dmaapPortNumber(dmaapPortNumber).dmaapTopicName(dmaapTopicName).dmaapProtocol(dmaapProtocol)
+                .dmaapUserName(dmaapUserName).dmaapUserPassword(dmaapUserPassword).dmaapContentType(dmaapContentType)
+                .build();
 
         // Then
         Assertions.assertNotNull(configuration);
diff --git a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/consumer/DMaaPConsumerReactiveHttpClientTest.java b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/consumer/DmaapConsumerReactiveHttpClientTest.java
index abac3cd5..4f96a903 100644
--- a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/consumer/DMaaPConsumerReactiveHttpClientTest.java
+++ b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/consumer/DmaapConsumerReactiveHttpClientTest.java
@@ -1,9 +1,7 @@
 /*
- * ============LICENSE_START=======================================================
- * Datafile Collector Service
- * ================================================================================
- * 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.service.consumer;
@@ -29,24 +27,25 @@ import static org.springframework.web.reactive.function.client.ExchangeFilterFun
 
 import java.net.URI;
 import java.net.URISyntaxException;
+
+import org.apache.http.HttpHeaders;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.onap.dcaegen2.collectors.datafile.config.DmaapConsumerConfiguration;
-import org.onap.dcaegen2.collectors.datafile.service.consumer.DMaaPConsumerReactiveHttpClient;
-import org.springframework.http.HttpHeaders;
 import org.springframework.web.reactive.function.client.WebClient;
 import org.springframework.web.reactive.function.client.WebClient.RequestHeadersUriSpec;
 import org.springframework.web.reactive.function.client.WebClient.ResponseSpec;
+
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 /**
  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 6/27/18
  */
-class DMaaPConsumerReactiveHttpClientTest {
+class DmaapConsumerReactiveHttpClientTest {
 
-    private DMaaPConsumerReactiveHttpClient dmaapConsumerReactiveHttpClient;
+    private DmaapConsumerReactiveHttpClient dmaapConsumerReactiveHttpClient;
 
     private DmaapConsumerConfiguration consumerConfigurationMock = mock(DmaapConsumerConfiguration.class);
     private static final String JSON_MESSAGE = "{ \"responseFromDmaap\": \"Success\"}";
@@ -61,14 +60,14 @@ class DMaaPConsumerReactiveHttpClientTest {
         when(consumerConfigurationMock.dmaapHostName()).thenReturn("54.45.33.2");
         when(consumerConfigurationMock.dmaapProtocol()).thenReturn("https");
         when(consumerConfigurationMock.dmaapPortNumber()).thenReturn(1234);
-        when(consumerConfigurationMock.dmaapUserName()).thenReturn("Datafile");
-        when(consumerConfigurationMock.dmaapUserPassword()).thenReturn("Datafile");
+        when(consumerConfigurationMock.dmaapUserName()).thenReturn("DATAFILE");
+        when(consumerConfigurationMock.dmaapUserPassword()).thenReturn("DATFILE");
         when(consumerConfigurationMock.dmaapContentType()).thenReturn("application/json");
-        when(consumerConfigurationMock.dmaapTopicName()).thenReturn("unauthenticated.SEC_OTHER_OUTPUT");
+        when(consumerConfigurationMock.dmaapTopicName()).thenReturn("unauthenticated.VES_NOTIFICATION_OUTPUT");
         when(consumerConfigurationMock.consumerGroup()).thenReturn("OpenDCAE-c12");
         when(consumerConfigurationMock.consumerId()).thenReturn("c12");
 
-        dmaapConsumerReactiveHttpClient = new DMaaPConsumerReactiveHttpClient(consumerConfigurationMock);
+        dmaapConsumerReactiveHttpClient = new DmaapConsumerReactiveHttpClient(consumerConfigurationMock);
         webClient = spy(WebClient.builder()
             .defaultHeader(HttpHeaders.CONTENT_TYPE, consumerConfigurationMock.dmaapContentType())
             .filter(basicAuthentication(consumerConfigurationMock.dmaapUserName(),
@@ -87,8 +86,8 @@ class DMaaPConsumerReactiveHttpClientTest {
         //when
         mockDependantObjects();
         doReturn(expectedResult).when(responseSpec).bodyToMono(String.class);
-        dmaapConsumerReactiveHttpClient.createDMaaPWebClient(webClient);
-        Mono<String> response = dmaapConsumerReactiveHttpClient.getDMaaPConsumerResponse();
+        dmaapConsumerReactiveHttpClient.createDmaapWebClient(webClient);
+        Mono<String> response = dmaapConsumerReactiveHttpClient.getDmaapConsumerResponse();
 
         //then
         StepVerifier.create(response).expectSubscription()
@@ -104,11 +103,11 @@ class DMaaPConsumerReactiveHttpClientTest {
         dmaapConsumerReactiveHttpClient = spy(dmaapConsumerReactiveHttpClient);
         //when
         when(webClient.get()).thenReturn(requestHeadersSpec);
-        dmaapConsumerReactiveHttpClient.createDMaaPWebClient(webClient);
+        dmaapConsumerReactiveHttpClient.createDmaapWebClient(webClient);
         when(dmaapConsumerReactiveHttpClient.getUri()).thenThrow(URISyntaxException.class);
 
         //then
-        StepVerifier.create(dmaapConsumerReactiveHttpClient.getDMaaPConsumerResponse()).expectSubscription()
+        StepVerifier.create(dmaapConsumerReactiveHttpClient.getDmaapConsumerResponse()).expectSubscription()
             .expectError(Exception.class).verify();
     }
 
diff --git a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DMaaPProducerReactiveHttpClientTest.java b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DMaaPProducerReactiveHttpClientTest.java
deleted file mode 100644
index bb1ce19d..00000000
--- a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DMaaPProducerReactiveHttpClientTest.java
+++ /dev/null
@@ -1,121 +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.service.producer;
-
-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.when;
-import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication;
-
-import java.net.URI;
-import java.net.URISyntaxException;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration;
-import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModelForUnitTest;
-import org.onap.dcaegen2.collectors.datafile.service.producer.DMaaPProducerReactiveHttpClient;
-import org.springframework.http.HttpHeaders;
-import org.springframework.web.reactive.function.client.WebClient;
-import org.springframework.web.reactive.function.client.WebClient.RequestBodyUriSpec;
-import org.springframework.web.reactive.function.client.WebClient.RequestHeadersSpec;
-import org.springframework.web.reactive.function.client.WebClient.ResponseSpec;
-import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;
-
-/**
- * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 7/4/18
- */
-class DMaaPProducerReactiveHttpClientTest {
-
-    private DMaaPProducerReactiveHttpClient dmaapProducerReactiveHttpClient;
-
-    private DmaapPublisherConfiguration dmaapPublisherConfigurationMock = mock(
-        DmaapPublisherConfiguration.class);
-    private ConsumerDmaapModel consumerDmaapModel = new ConsumerDmaapModelForUnitTest();
-    private WebClient webClient = mock(WebClient.class);
-    private RequestBodyUriSpec requestBodyUriSpec;
-    private ResponseSpec responseSpec;
-
-
-    @BeforeEach
-    void setUp() {
-        when(dmaapPublisherConfigurationMock.dmaapHostName()).thenReturn("54.45.33.2");
-        when(dmaapPublisherConfigurationMock.dmaapProtocol()).thenReturn("https");
-        when(dmaapPublisherConfigurationMock.dmaapPortNumber()).thenReturn(1234);
-        when(dmaapPublisherConfigurationMock.dmaapUserName()).thenReturn("Datafile");
-        when(dmaapPublisherConfigurationMock.dmaapUserPassword()).thenReturn("Datafile");
-        when(dmaapPublisherConfigurationMock.dmaapContentType()).thenReturn("application/json");
-        when(dmaapPublisherConfigurationMock.dmaapTopicName()).thenReturn("pnfReady");
-
-        dmaapProducerReactiveHttpClient = new DMaaPProducerReactiveHttpClient(dmaapPublisherConfigurationMock);
-
-        webClient = spy(WebClient.builder()
-            .defaultHeader(HttpHeaders.CONTENT_TYPE, dmaapPublisherConfigurationMock.dmaapContentType())
-            .filter(basicAuthentication(dmaapPublisherConfigurationMock.dmaapUserName(),
-                dmaapPublisherConfigurationMock.dmaapUserPassword()))
-            .build());
-        requestBodyUriSpec = mock(RequestBodyUriSpec.class);
-        responseSpec = mock(ResponseSpec.class);
-    }
-
-    @Test
-    void getHttpResponse_Success() {
-        //given
-        Integer responseSuccess = 200;
-        Mono<Integer> expectedResult = Mono.just(responseSuccess);
-
-        //when
-        mockWebClientDependantObject();
-        doReturn(expectedResult).when(responseSpec).bodyToMono(String.class);
-        dmaapProducerReactiveHttpClient.createDMaaPWebClient(webClient);
-        Mono<String> response = dmaapProducerReactiveHttpClient.getDMaaPProducerResponse(Mono.just(consumerDmaapModel));
-
-        //then
-        Assertions.assertEquals(response.block(), expectedResult.block());
-    }
-
-    @Test
-    void getHttpResponse_whenUriSyntaxExceptionHasBeenThrown() throws URISyntaxException {
-        //given
-        dmaapProducerReactiveHttpClient = spy(dmaapProducerReactiveHttpClient);
-        //when
-        when(webClient.post()).thenReturn(requestBodyUriSpec);
-        dmaapProducerReactiveHttpClient.createDMaaPWebClient(webClient);
-        when(dmaapProducerReactiveHttpClient.getUri()).thenThrow(URISyntaxException.class);
-
-        //then
-        StepVerifier.create(dmaapProducerReactiveHttpClient.getDMaaPProducerResponse(any())).expectSubscription()
-            .expectError(Exception.class).verify();
-    }
-
-    private void mockWebClientDependantObject() {
-        RequestHeadersSpec requestHeadersSpec = mock(RequestHeadersSpec.class);
-        when(webClient.post()).thenReturn(requestBodyUriSpec);
-        when(requestBodyUriSpec.uri((URI) any())).thenReturn(requestBodyUriSpec);
-        when(requestBodyUriSpec.body(any())).thenReturn(requestHeadersSpec);
-        doReturn(responseSpec).when(requestHeadersSpec).retrieve();
-        doReturn(responseSpec).when(responseSpec).onStatus(any(), any());
-    }
-}
\ No newline at end of file
diff --git a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClientTest.java b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClientTest.java
new file mode 100644
index 00000000..213e8d77
--- /dev/null
+++ b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClientTest.java
@@ -0,0 +1,129 @@
+/*
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved.
+ * ===============================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcaegen2.collectors.datafile.service.producer;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication;
+
+import com.google.gson.JsonElement;
+import com.google.gson.JsonParser;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.http.client.utils.URIBuilder;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration;
+import org.onap.dcaegen2.collectors.datafile.model.CommonFunctions;
+import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
+import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModelForUnitTest;
+import org.springframework.http.HttpHeaders;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.reactive.function.client.WebClient.RequestBodyUriSpec;
+import org.springframework.web.reactive.function.client.WebClient.ResponseSpec;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 7/4/18
+ */
+class DmaapProducerReactiveHttpClientTest {
+
+    private static final String LOCATION = "location";
+    private static final String X_ATT_DR_META = "X-ATT-DR-META";
+
+    private static final String HOST = "54.45.33.2";
+    private static final String HTTPS_SCHEME = "https";
+    private static final int PORT = 1234;
+    private static final String APPLICATION_OCTET_STREAM_CONTENT_TYPE = "application/octet-stream";
+    private static final String FILE_READY_TOPIC = "fileReady";
+
+    private DmaapProducerReactiveHttpClient dmaapProducerReactiveHttpClient;
+
+    private DmaapPublisherConfiguration dmaapPublisherConfigurationMock = mock(DmaapPublisherConfiguration.class);
+    private ConsumerDmaapModel consumerDmaapModel = new ConsumerDmaapModelForUnitTest();
+    private WebClient webClientMock = mock(WebClient.class);
+    private RequestBodyUriSpec requestBodyUriSpecMock;
+    private ResponseSpec responseSpecMock;
+
+
+    @BeforeEach
+    void setUp() {
+        when(dmaapPublisherConfigurationMock.dmaapHostName()).thenReturn(HOST);
+        when(dmaapPublisherConfigurationMock.dmaapProtocol()).thenReturn(HTTPS_SCHEME);
+        when(dmaapPublisherConfigurationMock.dmaapPortNumber()).thenReturn(PORT);
+        when(dmaapPublisherConfigurationMock.dmaapUserName()).thenReturn("DATAFILE");
+        when(dmaapPublisherConfigurationMock.dmaapUserPassword()).thenReturn("DATAFILE");
+        when(dmaapPublisherConfigurationMock.dmaapContentType()).thenReturn(APPLICATION_OCTET_STREAM_CONTENT_TYPE);
+        when(dmaapPublisherConfigurationMock.dmaapTopicName()).thenReturn(FILE_READY_TOPIC);
+
+        dmaapProducerReactiveHttpClient = new DmaapProducerReactiveHttpClient(dmaapPublisherConfigurationMock);
+
+        webClientMock = spy(WebClient.builder()
+                .defaultHeader(HttpHeaders.CONTENT_TYPE, dmaapPublisherConfigurationMock.dmaapContentType())
+                .filter(basicAuthentication(dmaapPublisherConfigurationMock.dmaapUserName(),
+                        dmaapPublisherConfigurationMock.dmaapUserPassword()))
+                .build());
+        requestBodyUriSpecMock = mock(RequestBodyUriSpec.class);
+        responseSpecMock = mock(ResponseSpec.class);
+    }
+
+    @Test
+    void getHttpResponse_Success() {
+        // given
+
+        // when
+        mockWebClientDependantObject();
+        dmaapProducerReactiveHttpClient.createDmaapWebClient(webClientMock);
+        List<ConsumerDmaapModel> consumerDmaapModelList = new ArrayList<ConsumerDmaapModel>();
+        consumerDmaapModelList.add(consumerDmaapModel);
+
+        dmaapProducerReactiveHttpClient.getDmaapProducerResponse(Mono.just(consumerDmaapModelList));
+
+        // then
+        verify(requestBodyUriSpecMock).header(HttpHeaders.CONTENT_TYPE, APPLICATION_OCTET_STREAM_CONTENT_TYPE);
+        JsonElement metaData = new JsonParser().parse(CommonFunctions.createJsonBody(consumerDmaapModel));
+        metaData.getAsJsonObject().remove(LOCATION);
+        verify(requestBodyUriSpecMock).header(X_ATT_DR_META, metaData.toString());
+        URI expectedUri = null;
+        try {
+            expectedUri = new URIBuilder().setScheme(HTTPS_SCHEME).setHost(HOST).setPort(1234).setPath(FILE_READY_TOPIC)
+                    .build();
+        } catch (URISyntaxException e) {
+            // Nothing
+        }
+        verify(requestBodyUriSpecMock).uri(expectedUri);
+        verify(requestBodyUriSpecMock).body(any());
+    }
+
+    private void mockWebClientDependantObject() {
+        when(webClientMock.post()).thenReturn(requestBodyUriSpecMock);
+        when(requestBodyUriSpecMock.uri((URI) any())).thenReturn(requestBodyUriSpecMock);
+
+        when(requestBodyUriSpecMock.retrieve()).thenReturn(responseSpecMock);
+        when(responseSpecMock.onStatus(any(), any())).thenReturn(responseSpecMock);
+        Mono<String> expectedResult = Mono.just("200");
+        when(responseSpecMock.bodyToMono(String.class)).thenReturn(expectedResult);
+    }
+}