Align PRH to El Alto SDK

Change-Id: I65c445d76092e11084fb60c68740e1321b35708c
Issue-ID: DCAEGEN2-1501
Signed-off-by: pwielebs <piotr.wielebski@nokia.com>

5 files changed, 84 insertions, 279 deletions

diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/AaiPublisherTaskSpy.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/AaiPublisherTaskSpy.java
index 18e1a27a..04388fb7 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/AaiPublisherTaskSpy.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/AaiPublisherTaskSpy.java
@@ -20,12 +20,6 @@
 
 package org.onap.dcaegen2.services.prh.tasks;
 
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
-
-import javax.net.ssl.SSLException;
-
 import org.onap.dcaegen2.services.prh.configuration.CbsConfiguration;
 import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
 import org.onap.dcaegen2.services.sdk.rest.services.aai.client.config.AaiClientConfiguration;
@@ -34,6 +28,10 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Primary;
 
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+
 
 
 /**
@@ -49,7 +47,7 @@ public class AaiPublisherTaskSpy {
      */
     @Bean
     @Primary
-    public AaiProducerTask registerSimpleAaiPublisherTask() throws SSLException {
+    public AaiProducerTask registerSimpleAaiPublisherTask() {
         CbsConfiguration cbsConfiguration = spy(CbsConfiguration.class);
         ConsumerDmaapModel consumerDmaapModel = spy(ConsumerDmaapModel.class);
         doReturn(mock(AaiClientConfiguration.class)).when(cbsConfiguration).getAaiClientConfiguration();
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImplTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImplTest.java
deleted file mode 100644
index 9afa7671..00000000
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImplTest.java
+++ /dev/null
@@ -1,150 +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.tasks;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNull;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-import static org.onap.dcaegen2.services.prh.TestAppConfiguration.createDefaultDmaapConsumerConfiguration;
-
-import com.google.gson.JsonArray;
-import com.google.gson.JsonObject;
-import com.google.gson.JsonParser;
-import java.util.Optional;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
-import org.onap.dcaegen2.services.prh.configuration.CbsConfiguration;
-import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.services.prh.model.ImmutableConsumerDmaapModel;
-import org.onap.dcaegen2.services.prh.service.DmaapConsumerJsonParser;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer.ConsumerReactiveHttpClientFactory;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer.DMaaPConsumerReactiveHttpClient;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
-
-/**
- * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 5/17/18
- */
-class DmaapConsumerTaskImplTest {
-
-    private static ConsumerDmaapModel consumerDmaapModel;
-    private static DmaapConsumerTaskImpl dmaapConsumerTask;
-    private static DMaaPConsumerReactiveHttpClient dMaaPConsumerReactiveHttpClient;
-    private static DmaapConsumerConfiguration dmaapConsumerConfiguration;
-    private static String message;
-    private static String messageContentEmpty;
-    private static JsonArray jsonArray;
-    private static JsonArray jsonArrayWrongContent;
-
-    private static CbsConfiguration cbsConfiguration;
-
-    @BeforeAll
-    static void setUp() {
-        dmaapConsumerConfiguration = createDefaultDmaapConsumerConfiguration();
-
-        JsonObject jsonObject = new JsonParser().parse("{\n"
-            + "        \"attachmentPoint\": \"bla-bla-30-3\",\n"
-            + "        \"cvlan\": \"678\",\n"
-            + "        \"svlan\": \"1005\"\n"
-            + "      }").getAsJsonObject();
-
-        consumerDmaapModel = ImmutableConsumerDmaapModel.builder()
-            .ipv4("10.16.123.234")
-            .ipv6("0:0:0:0:0:FFFF:0A10:7BEA")
-            .correlationId("NOKQTFCOC540002E")
-            .serialNumber("QTFCOC540002E")
-            .equipVendor("nokia")
-            .equipModel("3310")
-            .equipType("type")
-            .nfRole("gNB")
-            .swVersion("v4.5.0.1")
-            .additionalFields(jsonObject)
-            .build();
-        cbsConfiguration = mock(CbsConfiguration.class);
-
-        message = "[{\"event\": {"
-            + "\"commonEventHeader\": { "
-            + " \"sourceName\":\"NOKQTFCOC540002E\","
-            + " \"nfNamingCode\":\"gNB\" "
-            + "},"
-            + "\"pnfRegistrationFields\": {"
-            + " \"vendorName\": \"nokia\","
-            + " \"serialNumber\": \"QTFCOC540002E\","
-            + " \"pnfRegistrationFieldsVersion\": \"2.0\","
-            + " \"modelNumber\": \"3310\","
-            + " \"unitType\": \"type\",\n"
-            + " \"unitFamily\": \"BBU\","
-            + " \"oamV4IpAddress\": \"10.16.123.234\","
-            + " \"softwareVersion\": \"v4.5.0.1\","
-            + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\","
-            + " \"additionalFields\": {\"attachmentPoint\": \"bla-bla-30-3\",\"cvlan\": \"678\",\"svlan\": \"1005\"}"
-            + "}}}]";
-
-        messageContentEmpty = "[]";
-        JsonParser jsonParser = new JsonParser();
-        jsonArray = (JsonArray) jsonParser.parse(message);
-        jsonArrayWrongContent = (JsonArray) jsonParser.parse(messageContentEmpty);
-
-    }
-
-    @Test
-    void whenPassedObjectDoesNotFit_DoesNotThrowPrhTaskException() throws Exception {
-        //given
-        prepareMocksForDmaapConsumer(Optional.of(jsonArrayWrongContent));
-
-        //when
-        Flux<ConsumerDmaapModel> response = dmaapConsumerTask.execute("Sample input");
-
-        //then
-        verify(dMaaPConsumerReactiveHttpClient).getDMaaPConsumerResponse(Optional.empty());
-        assertNull(response.blockFirst());
-    }
-
-    @Test
-    void whenPassedObjectFits_ReturnsCorrectResponse() throws Exception {
-        //given
-        prepareMocksForDmaapConsumer(Optional.of(jsonArray));
-
-        //when
-        Flux<ConsumerDmaapModel> response = dmaapConsumerTask.execute("Sample input");
-
-        //then
-        verify(dMaaPConsumerReactiveHttpClient).getDMaaPConsumerResponse(Optional.empty());
-        assertEquals(consumerDmaapModel, response.blockFirst());
-    }
-
-
-
-    private void prepareMocksForDmaapConsumer(Optional<JsonArray> message) throws Exception {
-        dMaaPConsumerReactiveHttpClient = mock(DMaaPConsumerReactiveHttpClient.class);
-        when(dMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse(Optional.empty()))
-            .thenReturn(Mono.just(message.get()));
-        when(cbsConfiguration.getDmaapConsumerConfiguration()).thenReturn(dmaapConsumerConfiguration);
-        ConsumerReactiveHttpClientFactory httpClientFactory = mock(ConsumerReactiveHttpClientFactory.class);
-        doReturn(dMaaPConsumerReactiveHttpClient).when(httpClientFactory).create(dmaapConsumerConfiguration);
-        dmaapConsumerTask = new DmaapConsumerTaskImpl(cbsConfiguration, new DmaapConsumerJsonParser(), httpClientFactory);
-    }
-}
\ No newline at end of file
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskSpy.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskSpy.java
index 594575e5..4c95c717 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskSpy.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskSpy.java
@@ -20,19 +20,16 @@
 
 package org.onap.dcaegen2.services.prh.tasks;
 
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
-
-import javax.net.ssl.SSLException;
-
 import org.onap.dcaegen2.services.prh.configuration.CbsConfiguration;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer.DMaaPConsumerReactiveHttpClient;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Primary;
 
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+
 
 /**
  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/27/18
@@ -47,13 +44,10 @@ public class DmaapConsumerTaskSpy {
      */
     @Bean
     @Primary
-    public DmaapConsumerTask registerSimpleDmaapConsumerTask() throws SSLException {
+    public DmaapConsumerTask registerSimpleDmaapConsumerTask() {
         CbsConfiguration cbsConfiguration = spy(CbsConfiguration.class);
-        doReturn(mock(DmaapConsumerConfiguration.class)).when(cbsConfiguration).getDmaapConsumerConfiguration();
+        doReturn(mock(MessageRouterPublishRequest.class)).when(cbsConfiguration).getMessageRouterPublishRequest();
         DmaapConsumerTaskImpl dmaapConsumerTask = spy(new DmaapConsumerTaskImpl(cbsConfiguration));
-        DMaaPConsumerReactiveHttpClient dmaapConsumerReactiveHttpClient = mock(
-            DMaaPConsumerReactiveHttpClient.class);
-        doReturn(dmaapConsumerReactiveHttpClient).when(dmaapConsumerTask).resolveClient();
         return dmaapConsumerTask;
     }
 }
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapProducerTaskSpy.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapProducerTaskSpy.java
index 77028a34..7a68bc8c 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapProducerTaskSpy.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapProducerTaskSpy.java
@@ -20,20 +20,18 @@
 
 package org.onap.dcaegen2.services.prh.tasks;
 
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
-
-import javax.net.ssl.SSLException;
 import org.onap.dcaegen2.services.prh.configuration.CbsConfiguration;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPublisherConfiguration;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.producer.DMaaPPublisherReactiveHttpClient;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Primary;
 
 import java.util.function.Supplier;
 
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+
 /**
  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/13/18
  */
@@ -47,14 +45,10 @@ public class DmaapProducerTaskSpy {
      */
     @Bean
     @Primary
-    public DmaapPublisherTask registerSimpleDmaapPublisherTask() throws SSLException {
-        final CbsConfiguration cbsConfiguration = spy(CbsConfiguration.class);
-        final Supplier<DmaapPublisherConfiguration> configSupplier = () -> cbsConfiguration.getDmaapPublisherConfiguration();
-        doReturn(mock(DmaapPublisherConfiguration.class)).when(cbsConfiguration).getDmaapPublisherConfiguration();
-        final DmaapPublisherTaskImpl dmaapPublisherTask = spy(new DmaapPublisherTaskImpl(configSupplier));
-        final DMaaPPublisherReactiveHttpClient extendedDmaapProducerHttpClient = mock(
-            DMaaPPublisherReactiveHttpClient.class);
-        doReturn(extendedDmaapProducerHttpClient).when(dmaapPublisherTask).resolveClient();
-        return dmaapPublisherTask;
+    public DmaapPublisherTask registerSimpleDmaapPublisherTask() {
+        final CbsConfiguration cbsConfiguration = mock(CbsConfiguration.class);
+        final Supplier<MessageRouterPublishRequest> configSupplier = cbsConfiguration::getMessageRouterPublishRequest;
+        doReturn(mock(MessageRouterPublishRequest.class)).when(cbsConfiguration).getMessageRouterPublishRequest();
+        return spy(new DmaapPublisherTaskImpl(configSupplier, new MessageRouterPublisherResolver()));
     }
 }
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImplTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImplTest.java
index fb4a50ea..6347ad3d 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImplTest.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImplTest.java
@@ -20,136 +20,105 @@
 
 package org.onap.dcaegen2.services.prh.tasks;
 
-import io.netty.handler.codec.http.HttpResponseStatus;
+import com.google.gson.JsonPrimitive;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.function.Executable;
-import org.onap.dcaegen2.services.prh.configuration.CbsConfiguration;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
 import org.onap.dcaegen2.services.prh.exceptions.DmaapNotFoundException;
 import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException;
-import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPublisherConfiguration;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.producer.DMaaPPublisherReactiveHttpClient;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.producer.PublisherReactiveHttpClientFactory;
-import org.onap.dcaegen2.services.sdk.rest.services.model.DmaapModel;
-import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
-import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;
-
-import javax.net.ssl.SSLException;
-import java.util.Optional;
+import org.onap.dcaegen2.services.prh.integration.junit5.mockito.MockitoExtension;
+import org.onap.dcaegen2.services.prh.model.ImmutableConsumerDmaapModel;
+import org.onap.dcaegen2.services.sdk.model.streams.dmaap.ImmutableMessageRouterSink;
+import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSink;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterPublisher;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest;
+import reactor.core.publisher.Flux;
+
 import java.util.function.Supplier;
 
-import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.mockito.Mockito.*;
-import static org.onap.dcaegen2.services.prh.TestAppConfiguration.createDefaultDmaapPublisherConfiguration;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 /**
  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 5/17/18
  */
+@ExtendWith(MockitoExtension.class)
 class DmaapPublisherTaskImplTest {
 
-    private static ConsumerDmaapModel consumerDmaapModel;
     private static DmaapPublisherTaskImpl dmaapPublisherTask;
-    private static DMaaPPublisherReactiveHttpClient dMaaPPublisherReactiveHttpClient;
-    private static CbsConfiguration cbsConfiguration;
-    private static DmaapPublisherConfiguration dmaapPublisherConfiguration;
-    private Optional<RequestDiagnosticContext> requestDiagnosticContextOptionalMock;
-    private DmaapModel dmaapModel;
-    private PublisherReactiveHttpClientFactory publisherReactiveHttpClientFactory;
-    private Supplier<DmaapPublisherConfiguration> configSupplier;
+
+    @Mock
+    private static MessageRouterPublisherResolver messageRouterPublisherClientResolver;
+    @Mock
+    private static MessageRouterPublisher messageRouterPublisher;
+
+    private Supplier<MessageRouterPublishRequest> configSupplier;
+
+
+    @Captor
+    private ArgumentCaptor<Flux<JsonPrimitive>> fluxCaptor;
 
     @BeforeEach
-    public void beforeEach() throws SSLException {
-        dmaapPublisherConfiguration = createDefaultDmaapPublisherConfiguration();
-        consumerDmaapModel = mock(ConsumerDmaapModel.class);
-        cbsConfiguration = mock(CbsConfiguration.class);
-        requestDiagnosticContextOptionalMock = Optional.empty();
-        dmaapModel = mock(DmaapModel.class);
-        dMaaPPublisherReactiveHttpClient = mock(DMaaPPublisherReactiveHttpClient.class);
-        publisherReactiveHttpClientFactory = mock(PublisherReactiveHttpClientFactory.class);
-        when(cbsConfiguration.getDmaapPublisherConfiguration()).thenReturn(dmaapPublisherConfiguration);
-        when(publisherReactiveHttpClientFactory.create(dmaapPublisherConfiguration))
-            .thenReturn(dMaaPPublisherReactiveHttpClient);
-        configSupplier = () -> cbsConfiguration.getDmaapPublisherConfiguration();
+    void beforeEach() {
+        when(messageRouterPublisherClientResolver.resolveClient()).thenReturn(messageRouterPublisher);
+        MessageRouterPublishRequest mrRequest = createMRRequest();
+        configSupplier = () -> mrRequest;
     }
 
     @Test
     void execute_whenPassedObjectDoesntFit_ThrowsPrhTaskException() {
         //given
-        dmaapPublisherTask = new DmaapPublisherTaskImpl(configSupplier);
+        dmaapPublisherTask = new DmaapPublisherTaskImpl(configSupplier, messageRouterPublisherClientResolver);
         //when
         Executable executableFunction = () -> dmaapPublisherTask.execute(null);
         //then
         assertThrows(PrhTaskException.class, executableFunction, "The specified parameter is incorrect");
     }
 
-
     @Test
-    void execute_whenPassedObjectFits_ReturnsCorrectStatus() throws PrhTaskException, SSLException {
+    void execute_whenPassedObjectFits_ReturnsCorrectStatus() throws DmaapNotFoundException {
         //given
-        HttpResponseStatus httpResponseStatus = HttpResponseStatus.OK;
-        HttpResponse httpClientReponse = prepareMocksForTests(httpResponseStatus);
-        dmaapPublisherTask = new DmaapPublisherTaskImpl(configSupplier, publisherReactiveHttpClientFactory);
-
+        dmaapPublisherTask = new DmaapPublisherTaskImpl(configSupplier, messageRouterPublisherClientResolver);
         //when
-        StepVerifier.create(dmaapPublisherTask.execute(consumerDmaapModel)).expectSubscription()
-            .expectNext(httpClientReponse);
-
+        dmaapPublisherTask.execute(createConsumerDmaapModel());
         //then
-        verify(dMaaPPublisherReactiveHttpClient, times(1))
-            .getDMaaPProducerResponse(consumerDmaapModel, requestDiagnosticContextOptionalMock);
-
-        verifyNoMoreInteractions(dMaaPPublisherReactiveHttpClient);
+        verify(messageRouterPublisher).put(eq(configSupplier.get()), fluxCaptor.capture());
+        assertEquals(new JsonPrimitive("{\"correlationId\":\"NOKQTFCOC540002E\"}"), fluxCaptor.getValue().blockFirst());
     }
 
-    @Test
-    void execute_whenPassedObjectFits_butIncorrectResponseReturns() throws DmaapNotFoundException, SSLException {
-        //given
-        HttpResponseStatus httpResponseStatus = HttpResponseStatus.UNAUTHORIZED;
-        HttpResponse httpClientReponse = prepareMocksForTests(httpResponseStatus);
-        dmaapPublisherTask = new DmaapPublisherTaskImpl(configSupplier, publisherReactiveHttpClientFactory);
 
-        //when
-        StepVerifier.create(dmaapPublisherTask.execute(consumerDmaapModel)).expectSubscription()
-            .expectNext(httpClientReponse);
-
-        //then
-        verify(dMaaPPublisherReactiveHttpClient, times(1))
-            .getDMaaPProducerResponse(consumerDmaapModel, requestDiagnosticContextOptionalMock);
-        verifyNoMoreInteractions(dMaaPPublisherReactiveHttpClient);
+    private ImmutableConsumerDmaapModel createConsumerDmaapModel() {
+        return ImmutableConsumerDmaapModel.builder()
+                .ipv4("10.16.123.234")
+                .ipv6("0:0:0:0:0:FFFF:0A10:7BEA")
+                .correlationId("NOKQTFCOC540002E")
+                .serialNumber("QTFCOC540002E")
+                .equipVendor("nokia")
+                .equipModel("3310")
+                .equipType("type")
+                .nfRole("gNB")
+                .swVersion("v4.5.0.1")
+                .additionalFields(null)
+                .build();
     }
 
-    @Test()
-    void execute_whenConsumerDmaapModelIsNull() {
-        //given
-        HttpResponseStatus httpResponseStatus = HttpResponseStatus.UNAUTHORIZED;
-        HttpResponse httpClientReponse = prepareMocksForTests(httpResponseStatus);
-        dmaapPublisherTask = new DmaapPublisherTaskImpl(configSupplier, publisherReactiveHttpClientFactory);
-        assertThrows(DmaapNotFoundException.class, () -> {
-            dmaapPublisherTask.execute(null);
-        });
-    }
+    private MessageRouterPublishRequest createMRRequest() {
+        final MessageRouterSink sinkDefinition = ImmutableMessageRouterSink.builder()
+                .name("the topic")
+                .topicUrl("http://dmaap-mr:2222/events/unauthenticated.PNF_READY")
+                .build();
 
-    @Test
-    public void resolveClient() throws SSLException {
-        //given
-        dmaapPublisherTask = new DmaapPublisherTaskImpl(configSupplier, publisherReactiveHttpClientFactory);
-        //when
-        DMaaPPublisherReactiveHttpClient dMaaPPublisherReactiveHttpClientResolved = dmaapPublisherTask.resolveClient();
-        //then
-        assertSame(dMaaPPublisherReactiveHttpClientResolved, dMaaPPublisherReactiveHttpClient);
+        return ImmutableMessageRouterPublishRequest.builder()
+                .sinkDefinition(sinkDefinition)
+                .contentType("application/json")
+                .build();
     }
-
-    private HttpResponse prepareMocksForTests(HttpResponseStatus httpResponseStatus) {
-        HttpResponse httpClientResponse = mock(HttpResponse.class);
-        when(httpClientResponse.statusCode()).thenReturn(httpResponseStatus.code());
-        when(
-            dMaaPPublisherReactiveHttpClient.getDMaaPProducerResponse(dmaapModel, requestDiagnosticContextOptionalMock))
-            .thenReturn(Mono.just(httpClientResponse));
-        return httpClientResponse;
-    }
-
 }
\ No newline at end of file