Added reactive DMaaPClient

Extracted WebCLientBuilder for
Producer and Consumer.
Added unit test for ReactiveProducerClient.

Change-Id: I632e6928813ed9feb48982900c173f741e4483e3
Issue-ID: DCAEGEN2-563
Signed-off-by: wasala <przemyslaw.wasala@nokia.com>

3 files changed, 166 insertions, 135 deletions

diff --git a/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/consumer/DmaapConsumerReactiveHttpClientTest.java b/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/consumer/DMaaPConsumerReactiveHttpClientTest.java
index 63966602..cbc7bd62 100644
--- a/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/consumer/DmaapConsumerReactiveHttpClientTest.java
+++ b/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/consumer/DMaaPConsumerReactiveHttpClientTest.java
@@ -44,9 +44,9 @@ import reactor.test.StepVerifier;
 /**
  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 6/27/18
  */
-public class DmaapConsumerReactiveHttpClientTest {
+public class DMaaPConsumerReactiveHttpClientTest {
 
-    private static DmaapConsumerReactiveHttpClient dmaapConsumerReactiveHttpClient;
+    private static DMaaPConsumerReactiveHttpClient DMaaPConsumerReactiveHttpClient;
 
     private static DmaapConsumerConfiguration consumerConfigurationMock = mock(DmaapConsumerConfiguration.class);
     private static final String JSON_MESSAGE = "{ \"responseFromDmaap\": \"Success\"}";
@@ -68,13 +68,11 @@ public class DmaapConsumerReactiveHttpClientTest {
         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(),
                 consumerConfigurationMock.dmaapUserPassword()))
-            .filter(dmaapConsumerReactiveHttpClient.logRequest())
-            .filter(dmaapConsumerReactiveHttpClient.logResponse())
             .build());
         requestHeadersSpec = mock(RequestHeadersUriSpec.class);
         responseSpec = mock(ResponseSpec.class);
@@ -89,8 +87,8 @@ public class DmaapConsumerReactiveHttpClientTest {
         //when
         mockDependantObjects();
         doReturn(expectedResult).when(responseSpec).bodyToMono(String.class);
-        dmaapConsumerReactiveHttpClient.initWebClient(webClient);
-        Mono<String> response = dmaapConsumerReactiveHttpClient.getDmaaPConsumerResponse();
+        DMaaPConsumerReactiveHttpClient.createDMaaPWebClient(webClient);
+        Mono<String> response = DMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse();
 
         //then
         StepVerifier.create(response).expectSubscription()
@@ -108,11 +106,10 @@ public class DmaapConsumerReactiveHttpClientTest {
         mockDependantObjects();
         doAnswer(invocationOnMock -> Mono.error(new Exception("400")))
             .when(responseSpec).onStatus(HttpStatus::is4xxClientError, e -> Mono.error(new Exception("400")));
-        dmaapConsumerReactiveHttpClient.initWebClient();
-        dmaapConsumerReactiveHttpClient.initWebClient(webClient);
+        DMaaPConsumerReactiveHttpClient.createDMaaPWebClient(webClient);
 
         //then
-        StepVerifier.create(dmaapConsumerReactiveHttpClient.getDmaaPConsumerResponse()).expectSubscription()
+        StepVerifier.create(DMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse()).expectSubscription()
             .expectError(Exception.class);
 
     }
@@ -124,25 +121,24 @@ public class DmaapConsumerReactiveHttpClientTest {
         mockDependantObjects();
         doAnswer(invocationOnMock -> Mono.error(new Exception("500")))
             .when(responseSpec).onStatus(HttpStatus::is4xxClientError, e -> Mono.error(new Exception("500")));
-        dmaapConsumerReactiveHttpClient.initWebClient();
-        dmaapConsumerReactiveHttpClient.initWebClient(webClient);
+        DMaaPConsumerReactiveHttpClient.createDMaaPWebClient(webClient);
 
         //then
-        StepVerifier.create(dmaapConsumerReactiveHttpClient.getDmaaPConsumerResponse()).expectSubscription()
+        StepVerifier.create(DMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse()).expectSubscription()
             .expectError(Exception.class);
     }
 
     @Test
     public void getHttpResponse_whenURISyntaxExceptionHasBeenThrown() throws URISyntaxException {
         //given
-        dmaapConsumerReactiveHttpClient = spy(dmaapConsumerReactiveHttpClient);
+        DMaaPConsumerReactiveHttpClient = spy(DMaaPConsumerReactiveHttpClient);
         //when
         when(webClient.get()).thenReturn(requestHeadersSpec);
-        dmaapConsumerReactiveHttpClient.initWebClient(webClient);
-        when(dmaapConsumerReactiveHttpClient.getUri()).thenThrow(URISyntaxException.class);
+        DMaaPConsumerReactiveHttpClient.createDMaaPWebClient(webClient);
+        when(DMaaPConsumerReactiveHttpClient.getUri()).thenThrow(URISyntaxException.class);
 
         //then
-        StepVerifier.create(dmaapConsumerReactiveHttpClient.getDmaaPConsumerResponse()).expectSubscription()
+        StepVerifier.create(DMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse()).expectSubscription()
             .expectError(Exception.class);
     }
 
diff --git a/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/producer/DMaaPProducerReactiveHttpClientTest.java b/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/producer/DMaaPProducerReactiveHttpClientTest.java
new file mode 100644
index 00000000..c0b0c40b
--- /dev/null
+++ b/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/producer/DMaaPProducerReactiveHttpClientTest.java
@@ -0,0 +1,153 @@
+/*
+ * ============LICENSE_START=======================================================
+ * PNF-REGISTRATION-HANDLER
+ * ================================================================================
+ * 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.services.prh.service.producer;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+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.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.prh.config.DmaapPublisherConfiguration;
+import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
+import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModelForUnitTest;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
+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
+ */
+public class DMaaPProducerReactiveHttpClientTest {
+
+    private static DMaaPProducerReactiveHttpClient dMaaPProducerReactiveHttpClient;
+
+    private static DmaapPublisherConfiguration dmaapPublisherConfigurationMock = mock(
+        DmaapPublisherConfiguration.class);
+    private static final Integer RESPONSE_SUCCESS = 200;
+    private static ConsumerDmaapModel consumerDmaapModel = new ConsumerDmaapModelForUnitTest();
+    private static Mono<Integer> expectedResult = Mono.empty();
+    private static WebClient webClient = mock(WebClient.class);
+    private static RequestBodyUriSpec requestBodyUriSpec;
+    private static ResponseSpec responseSpec;
+
+
+    @BeforeAll
+    public static void setUp() {
+        when(dmaapPublisherConfigurationMock.dmaapHostName()).thenReturn("54.45.33.2");
+        when(dmaapPublisherConfigurationMock.dmaapProtocol()).thenReturn("https");
+        when(dmaapPublisherConfigurationMock.dmaapPortNumber()).thenReturn(1234);
+        when(dmaapPublisherConfigurationMock.dmaapUserName()).thenReturn("PRH");
+        when(dmaapPublisherConfigurationMock.dmaapUserPassword()).thenReturn("PRH");
+        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
+    public void getHttpResponse_Success() {
+        //given
+        expectedResult = Mono.just(RESPONSE_SUCCESS);
+
+        //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
+    public void getHttpResponse_HttpResponse4xxClientError() {
+        //when
+        mockWebClientDependantObject();
+
+        doAnswer(invocationOnMock -> Mono.error(new Exception("400")))
+            .when(responseSpec).onStatus(HttpStatus::is4xxClientError, e -> Mono.error(new Exception("400")));
+        dMaaPProducerReactiveHttpClient.createDMaaPWebClient(webClient);
+
+        //then
+        StepVerifier.create(dMaaPProducerReactiveHttpClient.getDMaaPProducerResponse(Mono.just(consumerDmaapModel)))
+            .expectSubscription()
+            .expectError(Exception.class);
+
+    }
+
+    @Test
+    public void getHttpResponse_HttpResponse5xxClientError() {
+
+        //when
+        mockWebClientDependantObject();
+        doAnswer(invocationOnMock -> Mono.error(new Exception("500")))
+            .when(responseSpec).onStatus(HttpStatus::is4xxClientError, e -> Mono.error(new Exception("500")));
+        dMaaPProducerReactiveHttpClient.createDMaaPWebClient(webClient);
+
+        //then
+        StepVerifier.create(dMaaPProducerReactiveHttpClient.getDMaaPProducerResponse(Mono.just(consumerDmaapModel)))
+            .expectSubscription()
+            .expectError(Exception.class);
+    }
+
+    @Test
+    public 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);
+    }
+
+    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/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/producer/ExtendedDmaapProducerHttpClientImplTest.java b/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/producer/ExtendedDmaapProducerHttpClientImplTest.java
deleted file mode 100644
index d9e74267..00000000
--- a/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/producer/ExtendedDmaapProducerHttpClientImplTest.java
+++ /dev/null
@@ -1,118 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * PNF-REGISTRATION-HANDLER
- * ================================================================================
- * 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.services.prh.service.producer;
-
-import org.apache.http.HttpEntity;
-import org.apache.http.HttpResponse;
-import org.apache.http.HttpStatus;
-import org.apache.http.StatusLine;
-import org.apache.http.client.ResponseHandler;
-import org.apache.http.client.methods.HttpPost;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
-import org.onap.dcaegen2.services.prh.config.DmaapPublisherConfiguration;
-import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModelForUnitTest;
-
-import java.io.IOException;
-import java.lang.reflect.Field;
-import java.util.Optional;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-
-class ExtendedDmaapProducerHttpClientImplTest {
-
-    private static ExtendedDmaapProducerHttpClientImpl objectUnderTest;
-    private static DmaapPublisherConfiguration configurationMock = mock(DmaapPublisherConfiguration.class);
-    private static CloseableHttpClient closeableHttpClientMock = mock(CloseableHttpClient.class);
-    private static ConsumerDmaapModel consumerDmaapModel = new ConsumerDmaapModelForUnitTest();
-    private static Integer expectedResult;
-    private static final Integer RESPONSE_SUCCESS = 200;
-    private static final Integer RESPONSE_FAILURE = 404;
-    private final static HttpResponse httpResponseMock = mock(HttpResponse.class);
-    private final static HttpEntity httpEntityMock = mock(HttpEntity.class);
-    private final static StatusLine statusLineMock = mock(StatusLine.class);
-
-
-    @BeforeAll
-    static void init() throws NoSuchFieldException, IllegalAccessException {
-        when(configurationMock.dmaapHostName()).thenReturn("54.45.33.2");
-        when(configurationMock.dmaapProtocol()).thenReturn("https");
-        when(configurationMock.dmaapPortNumber()).thenReturn(1234);
-        when(configurationMock.dmaapUserName()).thenReturn("PRH");
-        when(configurationMock.dmaapUserPassword()).thenReturn("PRH");
-        when(configurationMock.dmaapContentType()).thenReturn("application/json");
-        when(configurationMock.dmaapTopicName()).thenReturn("pnfReady");
-        objectUnderTest = new ExtendedDmaapProducerHttpClientImpl(configurationMock);
-        setField();
-    }
-
-
-    @Test
-    void getHttpResponsePost_success() throws IOException {
-        expectedResult = RESPONSE_SUCCESS;
-        when(closeableHttpClientMock.execute(any(HttpPost.class), any(ResponseHandler.class)))
-            .thenReturn(Optional.of(expectedResult));
-        Optional<Integer> actualResult = objectUnderTest.getHttpProducerResponse(consumerDmaapModel);
-        Assertions.assertEquals(expectedResult, actualResult.get());
-    }
-
-    @Test
-    void getExtendedDetails_returnsFailure() throws IOException {
-        expectedResult = RESPONSE_FAILURE;
-        when(closeableHttpClientMock.execute(any(HttpPost.class), any(ResponseHandler.class)))
-            .thenReturn(Optional.of(expectedResult));
-        Optional<Integer> actualResult = objectUnderTest.getHttpProducerResponse(consumerDmaapModel);
-        Assertions.assertEquals(expectedResult, actualResult.get());
-    }
-
-    @Test
-    void handleResponse_shouldReturn200() throws IOException {
-        // When
-        when(httpResponseMock.getEntity()).thenReturn(httpEntityMock);
-        when(httpResponseMock.getStatusLine()).thenReturn(statusLineMock);
-        when(httpResponseMock.getStatusLine().getStatusCode()).thenReturn(HttpStatus.SC_OK);
-        // Then
-        assertEquals(Optional.of(HttpStatus.SC_OK), objectUnderTest.handleResponse(httpResponseMock));
-    }
-
-    @Test
-    void handleResponse_shouldReturn300() throws IOException {
-        // When
-        when(httpResponseMock.getEntity()).thenReturn(httpEntityMock);
-        when(httpResponseMock.getStatusLine()).thenReturn(statusLineMock);
-        when(httpResponseMock.getStatusLine().getStatusCode()).thenReturn(HttpStatus.SC_BAD_REQUEST);
-        // Then
-        assertEquals(Optional.of(HttpStatus.SC_BAD_REQUEST), objectUnderTest.handleResponse(httpResponseMock));
-    }
-
-    private static void setField() throws NoSuchFieldException, IllegalAccessException {
-        Field field = objectUnderTest.getClass().getDeclaredField("closeableHttpClient");
-        field.setAccessible(true);
-        field.set(objectUnderTest, closeableHttpClientMock);
-    }
-}