feat:Enhance sliceanalysis MS to use DCAE SDK dmaap-client lib

Issue-ID: DCAEGEN2-3120
Signed-off-by: zhaoyh6 <zhaoyh6@asiainfo.com>
Change-Id: I946c7a4b49906cb402062235a97452fb7856c8f0

8 files changed, 358 insertions, 177 deletions

diff --git a/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/AaiEventNotificationCallbackTest.java b/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/AaiEventNotificationCallbackTest.java
index eb88d537..3b0c32fa 100644
--- a/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/AaiEventNotificationCallbackTest.java
+++ b/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/AaiEventNotificationCallbackTest.java
@@ -3,6 +3,7 @@
  *  slice-analysis-ms
  *  ================================================================================
  *  Copyright (C) 2022 Huawei Canada Limited.
+ *  Copyright (C) 2022 CTC, Inc.
  *  ==============================================================================
  *     Licensed under the Apache License, Version 2.0 (the "License");
  *     you may not use this file except in compliance with the License.
@@ -21,11 +22,18 @@
 
 package org.onap.slice.analysis.ms.dmaap;
 
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonPrimitive;
+import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.InjectMocks;
+import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.Spy;
+import org.onap.slice.analysis.ms.models.Configuration;
+import org.onap.slice.analysis.ms.service.ccvpn.BandwidthEvaluator;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.junit4.SpringRunner;
 
@@ -33,6 +41,9 @@ import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Paths;
 
+import static org.mockito.ArgumentMatchers.any;
+import static org.powermock.api.mockito.PowerMockito.doNothing;
+
 
 @RunWith(SpringRunner.class)
 @SpringBootTest(classes = VesNotificationCallbackTest.class)
@@ -42,6 +53,45 @@ public class AaiEventNotificationCallbackTest {
     @InjectMocks
     AaiEventNotificationCallback aaiEventNotificationCallback;
 
+    @Mock
+    BandwidthEvaluator bandwidthEvaluator;
+
+    @Before
+    public void initConfiguration() {
+        JsonObject jsonObject = new JsonObject();
+        jsonObject.addProperty("sliceanalysisms.aaiNotif.targetAction", "UPDATE");
+        jsonObject.addProperty("sliceanalysisms.aaiNotif.targetEntity", "service-instance");
+        jsonObject.addProperty("sliceanalysisms.aaiNotif.targetSource", "UUI");
+        jsonObject.addProperty("postgres.port", "1");
+        jsonObject.addProperty("sliceanalysisms.pollingInterval", "1");
+        jsonObject.addProperty("postgres.password", "1");
+        jsonObject.addProperty("postgres.username", "1");
+        jsonObject.addProperty("postgres.host", "1");
+        jsonObject.addProperty("sliceanalysisms.cg", "1");
+        jsonObject.addProperty("sliceanalysisms.cid", "1");
+        jsonObject.addProperty("sliceanalysisms.configDb.service", "1");
+        jsonObject.addProperty("sliceanalysisms.configDbEnabled", "1");
+        jsonObject.addProperty("sliceanalysisms.pollingTimeout", "1");
+        jsonObject.addProperty("sliceanalysisms.samples", "1");
+        jsonObject.addProperty("sliceanalysisms.minPercentageChange", "1");
+        jsonObject.addProperty("sliceanalysisms.initialDelaySeconds", "1");
+        jsonObject.addProperty("sliceanalysisms.rannfnssiDetailsTemplateId", "1");
+        jsonObject.addProperty("sliceanalysisms.desUrl", "1");
+        jsonObject.addProperty("sliceanalysisms.pmDataDurationInWeeks", "1");
+        jsonObject.addProperty("sliceanalysisms.pollingInterval", "1");
+        jsonObject.addProperty("sliceanalysisms.vesNotifChangeIdentifier", "1");
+        jsonObject.addProperty("sliceanalysisms.vesNotifChangeType", "1");
+        jsonObject.addProperty("sliceanalysisms.vesNotifPollingInterval", "1");
+        jsonObject.addProperty("sliceanalysisms.ccvpnEvalInterval", "1");
+        jsonObject.addProperty("sliceanalysisms.ccvpnEvalThreshold", "1");
+        jsonObject.addProperty("sliceanalysisms.ccvpnEvalPrecision", "1");
+        jsonObject.addProperty("sliceanalysisms.ccvpnEvalPeriodicCheckOn", "1");
+        jsonObject.addProperty("sliceanalysisms.ccvpnEvalOnDemandCheckOn", "1");
+        Configuration configuration = Configuration.getInstance();
+        configuration.updateConfigurationFromJsonObject(jsonObject);
+        doNothing().when(bandwidthEvaluator).post(any());
+    }
+
     @Test
     public void initTest() {
         aaiEventNotificationCallback.init();
@@ -49,14 +99,21 @@ public class AaiEventNotificationCallbackTest {
     }
 
     @Test
-    public void activateCallBackTest() {
-        String input = null;
-        try {
-            input = new String(Files.readAllBytes(Paths.get("src/test/resources/aaiEventDmaapMsg.json")));
-        } catch (IOException e) {
-            e.printStackTrace();
-        }
+    public void activateCallBackTest() throws IOException {
+        aaiEventNotificationCallback.init();
+        String input = new String(Files.readAllBytes(Paths.get("src/test/resources/aaiEventDmaapMsg.json")));
         aaiEventNotificationCallback.activateCallBack(input);
         Mockito.verify(aaiEventNotificationCallback, Mockito.atLeastOnce()).activateCallBack(Mockito.anyString());
     }
+    @Test
+    public void activateCallBackArrayTest() throws IOException {
+        aaiEventNotificationCallback.init();
+        String input = new String(Files.readAllBytes(Paths.get("src/test/resources/aaiEventDmaapMsg.json")));
+        JsonArray jsonArray = new JsonArray();
+
+        JsonPrimitive jsonPrimitive = new JsonPrimitive(input);
+        jsonArray.add(jsonPrimitive);
+        aaiEventNotificationCallback.activateCallBack(jsonArray.toString());
+        Mockito.verify(aaiEventNotificationCallback, Mockito.atLeastOnce()).activateCallBack(Mockito.anyString());
+    }
 }
diff --git a/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/DmaapClientTest.java b/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/DmaapClientTest.java
index b74056db..7da9f95f 100644
--- a/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/DmaapClientTest.java
+++ b/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/DmaapClientTest.java
@@ -3,6 +3,7 @@
  *  slice-analysis-ms
  *  ================================================================================
  *   Copyright (C) 2020 Wipro Limited.
+ *   Copyright (C) 2022 CTC, Inc.
  *   ==============================================================================
  *     Licensed under the Apache License, Version 2.0 (the "License");
  *     you may not use this file except in compliance with the License.
@@ -22,28 +23,23 @@
 
 package org.onap.slice.analysis.ms.dmaap;
 
-import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
 
-import com.att.nsa.cambria.client.CambriaTopicManager;
 import com.google.gson.Gson;
 import com.google.gson.JsonObject;
 
 import java.io.BufferedReader;
 import java.io.FileReader;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
 
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.InjectMocks;
-import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
 import org.onap.slice.analysis.ms.models.Configuration;
+import org.powermock.api.mockito.PowerMockito;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.junit4.SpringRunner;
 
@@ -51,78 +47,43 @@ import org.springframework.test.context.junit4.SpringRunner;
 @SpringBootTest(classes = DmaapClientTest.class)
 public class DmaapClientTest {
 
-	@Mock
-	private CambriaTopicManager topicManager;
-
-	@InjectMocks
-	DmaapClient client;
-
-	@Before
-	public void setup() {
-		MockitoAnnotations.initMocks(this);
-	}
-
-	@Test
-	public void getAllTopicsTest() {
-		Set<String> topics = new HashSet<String>();
-		topics.add("topic1");
-		topics.add("topic2");
-		Configuration configuration = Configuration.getInstance();
-		List<String> list = new ArrayList<String>();
-		list.add("server");
-		configuration.setDmaapServers(list);
-		configuration.setCg("cg");
-		configuration.setCid("cid");
-		configuration.setPollingInterval(30);
-		configuration.setPollingTimeout(100);
-		configuration.setConfigDbService("sdnrService");
-
-		try {
-			when(topicManager.getTopics()).thenReturn(topics);
-
-			client=Mockito.mock(DmaapClient.class);
-			client.initClient();
-			Mockito.verify(client).initClient();      
-			// Mockito.verifycreateAndConfigureTopics();
-
-		} catch (IOException e) {
-			e.printStackTrace();
-		}
-	}
-
-	@Test
-	public void startClientTest() {		
-		try {
-			Configuration configuration = Configuration.getInstance();
-			String configAllJson = readFromFile("src/test/resources/config_all.json");
-
-			JsonObject configAll = new Gson().fromJson(configAllJson, JsonObject.class);
-
-			JsonObject config = configAll.getAsJsonObject("config");
-			System.out.println(configuration);
-			configuration.updateConfigurationFromJsonObject(config);
-			DmaapClient client= new DmaapClient();
-			client.initClient();
-			//Mockito.verify(client).startClient();      
-			// Mockito.verifycreateAndConfigureTopics();
-
-		} catch ( Exception e) {
-			e.printStackTrace();
-		}
-	}
-
-	private static String readFromFile(String file) {
-		String content = "";
-		try (BufferedReader bufferedReader = new BufferedReader(new FileReader(file))) {
-			content = bufferedReader.readLine();
-			String temp;
-			while ((temp = bufferedReader.readLine()) != null) {
-				content = content.concat(temp);
-			}
-			content = content.trim();
-		} catch (Exception e) {
-			content = null;
-		}
-		return content;
-	}
+    @InjectMocks
+    DmaapClient client;
+
+    @Before
+    public void setup() {
+        MockitoAnnotations.initMocks(this);
+    }
+
+    @Test
+    public void startClientTest() {
+        Configuration configuration = Configuration.getInstance();
+        String configAllJson = readFromFile("src/test/resources/config_all.json");
+
+        JsonObject configAll = new Gson().fromJson(configAllJson, JsonObject.class);
+
+        JsonObject config = configAll.getAsJsonObject("config");
+        System.out.println(configuration);
+        configuration.updateConfigurationFromJsonObject(config);
+        MRTopicMonitor mrTopicMonitor = Mockito.mock(MRTopicMonitor.class);
+        DmaapClient spy = PowerMockito.spy(client);
+        doReturn(mrTopicMonitor).when(spy).getMRTopicMonitor();
+        doNothing().when(mrTopicMonitor).start();
+        spy.initClient();
+    }
+
+    private static String readFromFile(String file) {
+        String content = "";
+        try (BufferedReader bufferedReader = new BufferedReader(new FileReader(file))) {
+            content = bufferedReader.readLine();
+            String temp;
+            while ((temp = bufferedReader.readLine()) != null) {
+                content = content.concat(temp);
+            }
+            content = content.trim();
+        } catch (Exception e) {
+            content = null;
+        }
+        return content;
+    }
 }
diff --git a/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/IntelligentSlicingCallbackTest.java b/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/IntelligentSlicingCallbackTest.java
new file mode 100644
index 00000000..59daae09
--- /dev/null
+++ b/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/IntelligentSlicingCallbackTest.java
@@ -0,0 +1,56 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  slice-analysis-ms
+ *  ================================================================================
+ *  Copyright (C) 2022 CTC, Inc.
+ *  ==============================================================================
+ *     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.slice.analysis.ms.dmaap;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.Spy;
+import org.onap.slice.analysis.ms.service.MLMessageProcessor;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = IntelligentSlicingCallbackTest.class)
+public class IntelligentSlicingCallbackTest {
+
+
+    @Spy
+    @InjectMocks
+    IntelligentSlicingCallback intelligentSlicingCallback;
+
+    @Mock
+    private MLMessageProcessor mlMsMessageProcessor;
+
+    @Test
+    public void activateCallBackTest() {
+        Mockito.doNothing().when(mlMsMessageProcessor).processMLMsg(Mockito.any());
+        intelligentSlicingCallback.activateCallBack("{}");
+        verify(mlMsMessageProcessor, times(1)).processMLMsg(any());
+    }
+}
diff --git a/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/MRTopicMonitorTest.java b/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/MRTopicMonitorTest.java
index ce920a0e..813bd463 100644
--- a/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/MRTopicMonitorTest.java
+++ b/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/MRTopicMonitorTest.java
@@ -3,6 +3,7 @@
  *  slice-analysis-ms
  *  ================================================================================
  *   Copyright (C) 2022 Huawei Canada Limited.
+ *   Copyright (C) 2022 CTC, Inc.
  *   ==============================================================================
  *     Licensed under the Apache License, Version 2.0 (the "License");
  *     you may not use this file except in compliance with the License.
@@ -31,19 +32,22 @@ import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
 import org.mockito.Spy;
-import org.mockito.stubbing.Answer;
 import org.onap.slice.analysis.ms.models.Configuration;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.modules.junit4.PowerMockRunnerDelegate;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.junit4.SpringRunner;
 
 import java.io.BufferedReader;
 import java.io.FileReader;
-import java.util.Map;
+import java.util.concurrent.Executors;
 
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-@RunWith(SpringRunner.class)
+@RunWith(PowerMockRunner.class)
+@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"})
+@PowerMockRunnerDelegate(SpringRunner.class)
+@PrepareForTest({ Executors.class})
 @SpringBootTest(classes = MRTopicMonitorTest.class)
 public class MRTopicMonitorTest {
 
@@ -67,12 +71,6 @@ public class MRTopicMonitorTest {
     }
 
     @Test
-    public void start() {
-        mrTopicMonitor.start();
-        Mockito.verify(mrTopicMonitor, Mockito.times(1)).start();
-    }
-
-    @Test
     public void run() {
         mrTopicMonitor.run();
         Mockito.verify(mrTopicMonitor, Mockito.times(1)).run();
diff --git a/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/MRTopicParamsTest.java b/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/MRTopicParamsTest.java
index a5066b17..2498306c 100644
--- a/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/MRTopicParamsTest.java
+++ b/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/MRTopicParamsTest.java
@@ -3,6 +3,7 @@
  *  slice-analysis-ms
  *  ================================================================================
  *   Copyright (C) 2022 Huawei Canada Limited.
+ *   Copyright (C) 2022 CTC, Inc.
  *   ==============================================================================
  *     Licensed under the Apache License, Version 2.0 (the "License");
  *     you may not use this file except in compliance with the License.
@@ -30,7 +31,16 @@ import com.openpojo.validation.rule.impl.SetterMustExistRule;
 import com.openpojo.validation.test.impl.GetterTester;
 import com.openpojo.validation.test.impl.SetterTester;
 import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
 
 public class MRTopicParamsTest {
     private static final String TEST_TOPIC = "test-topic";
@@ -93,4 +103,84 @@ public class MRTopicParamsTest {
                 .build();
         validator.validate(pojoclass);
     }
+
+    @Test
+    public void initProperty() {
+        List<String> strings = new ArrayList<>();
+        Map<String, String> map = new HashMap<>();
+        MRTopicParams params = MRTopicParams.builder()
+                .topic(TEST_TOPIC)
+                .hostname(TEST_HOST)
+                .clientName(MY_CLIENT)
+                .consumerGroup(MY_CG)
+                .consumerInstance(MY_CI)
+                .apiSecret(MY_API_SEC)
+                .apiKey(MY_API_KEY)
+                .fetchLimit(MY_FETCH_LIMIT)
+                .fetchTimeout(MY_FETCH_TIMEOUT)
+                .password(MY_PASS)
+                .userName(MY_USERNAME)
+                .port(MY_PORT)
+                .servers(strings)
+                .additionalProps(map)
+                .effectiveTopic("")
+                .useHttps(true)
+                .allowSelfSignedCerts(true)
+                .managed(true)
+                .environment("")
+                .aftEnvironment("")
+                .partner("")
+                .latitude("")
+                .longitude("")
+                .partitionId("")
+                .basePath("")
+                .serializationProvider("")
+                .build();
+        assertEquals(TEST_TOPIC, params.getTopic());
+        assertEquals(TEST_HOST, params.getHostname());
+        assertEquals(MY_CLIENT, params.getClientName());
+        assertEquals(MY_CG, params.getConsumerGroup());
+        assertEquals(MY_CI, params.getConsumerInstance());
+        assertEquals(MY_API_SEC, params.getApiSecret());
+        assertEquals(MY_API_KEY, params.getApiKey());
+        assertEquals(MY_FETCH_LIMIT, params.getFetchLimit());
+        assertEquals(MY_FETCH_TIMEOUT, params.getFetchTimeout());
+        assertEquals(MY_PASS, params.getPassword());
+        assertEquals(MY_USERNAME, params.getUserName());
+        assertEquals(MY_PORT, params.getPort());
+        assertEquals(strings, params.getServers());
+        assertEquals(map, params.getAdditionalProps());
+        assertEquals("", params.getEffectiveTopic());
+        assertTrue(params.isUseHttps());
+        assertTrue(params.isAllowSelfSignedCerts());
+        assertTrue(params.isManaged());
+        assertEquals("", params.getEnvironment());
+        assertEquals("", params.getAftEnvironment());
+        assertEquals("", params.getPartner());
+        assertEquals("", params.getLongitude());
+        assertEquals("", params.getPartitionId());
+        assertEquals("", params.getBasePath());
+        assertEquals("", params.getSerializationProvider());
+
+        assertTrue(params.isEnvironmentInvalid());
+        assertTrue(params.isAftEnvironmentInvalid());
+        assertTrue(params.isLatitudeInvalid());
+        assertTrue(params.isLongitudeInvalid());
+        assertTrue(params.isPartnerInvalid());
+        assertTrue(params.isServersInvalid());
+        assertTrue(params.isPartitionIdInvalid());
+        assertTrue(params.isApiKeyValid());
+        assertTrue(params.isApiSecretValid());
+        assertTrue(params.isUserNameValid());
+        assertTrue(params.isPasswordValid());
+        assertTrue(params.isAdditionalPropsValid());
+        assertFalse(params.isConsumerInstanceInvalid());
+        assertFalse(params.isConsumerGroupInvalid());
+        assertFalse(params.isClientNameInvalid());
+        assertFalse(params.isTopicInvalid());
+        assertFalse(params.isHostnameInvalid());
+        assertFalse(params.isPortInvalid());
+    }
+
+
 }
diff --git a/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/NotificationConsumerTest.java b/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/NotificationConsumerTest.java
index 9aab22c2..5b8e5f25 100644
--- a/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/NotificationConsumerTest.java
+++ b/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/NotificationConsumerTest.java
@@ -3,6 +3,7 @@
  *  slice-analysis-ms
  *  ================================================================================
  *   Copyright (C) 2020 Wipro Limited.
+ *   Copyright (C) 2022 CTC, Inc.
  *   ==============================================================================
  *     Licensed under the Apache License, Version 2.0 (the "License");
  *     you may not use this file except in compliance with the License.
@@ -23,44 +24,48 @@ package org.onap.slice.analysis.ms.dmaap;
 
 import static org.mockito.Mockito.when;
 
-import java.util.ArrayList;
-import java.util.List;
-
+import com.google.gson.JsonPrimitive;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.Mockito;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterSubscriber;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeResponse;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeResponse;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.junit4.SpringRunner;
 
-import com.att.nsa.cambria.client.CambriaConsumer;
+import reactor.core.publisher.Mono;
 
 @RunWith(SpringRunner.class)
 @SpringBootTest(classes = NotificationConsumerTest.class)
 public class NotificationConsumerTest {
-	
-	@Mock
-	CambriaConsumer cambriaConsumer;
-	
-	@Mock
-	NotificationCallback notificationCallback;
+    
+    @Mock
+    NotificationCallback notificationCallback;
+
+    @Mock
+    MessageRouterSubscriber subscriber;
+
+    @Mock
+    MessageRouterSubscribeRequest request;
 
-	@InjectMocks
-	NotificationConsumer notificationConsumer;
+    @InjectMocks
+    NotificationConsumer notificationConsumer;
 
-	@Test
-	public void testNotificationConsumer() {
-		try {
-			List<String> notifications = new ArrayList<>();
-			notifications.add("notification1");
-			when(cambriaConsumer.fetch()).thenReturn(notifications);
-			Mockito.doNothing().when(notificationCallback).activateCallBack(Mockito.anyString());
-			notificationConsumer.run();
-			
-		}catch(Exception e) {
-			e.printStackTrace();
-		}
-	}
+    @Test
+    public void testNotificationConsumer() {
+        io.vavr.collection.List<String> expectedItems = io.vavr.collection.List.of("I", "like", "pizza");
+        MessageRouterSubscribeResponse expectedResponse = ImmutableMessageRouterSubscribeResponse
+                .builder()
+                .items(expectedItems.map(JsonPrimitive::new))
+                .build();
+        Mono<MessageRouterSubscribeResponse> responses = Mono.just(expectedResponse);
+        when(subscriber.get(request)).thenReturn(responses);
+        Mockito.doNothing().when(notificationCallback).activateCallBack(Mockito.anyString());
+        notificationConsumer.run();
+    }
 
 }
diff --git a/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/NotificationProducerTest.java b/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/NotificationProducerTest.java
index 9dc51412..d39c0af6 100644
--- a/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/NotificationProducerTest.java
+++ b/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/NotificationProducerTest.java
@@ -3,6 +3,7 @@
  *  slice-analysis-ms
  *  ================================================================================
  *   Copyright (C) 2020 Wipro Limited.
+ *   Copyright (C) 2022 CTC, Inc.
  *   ==============================================================================
  *     Licensed under the Apache License, Version 2.0 (the "License");
  *     you may not use this file except in compliance with the License.
@@ -22,42 +23,47 @@
 
 package org.onap.slice.analysis.ms.dmaap;
 
-import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.when;
 
-import com.att.nsa.cambria.client.CambriaBatchingPublisher;
-
-import java.io.IOException;
-
+import com.google.gson.JsonPrimitive;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
-import org.mockito.Mockito;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterPublisher;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishResponse;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.junit4.SpringRunner;
+import reactor.core.publisher.Flux;
+
+import java.io.IOException;
 
 @RunWith(SpringRunner.class)
 @SpringBootTest(classes = NotificationProducerTest.class)
 public class NotificationProducerTest {
 
-	@Mock
-	CambriaBatchingPublisher cambriaBatchingPublisher;
-
-	@InjectMocks
-	NotificationProducer notificationProducer;
+    @Mock
+    MessageRouterPublisher publisher;
 
-	@Test
-	public void notificationProducerTest() {
+    @Mock
+    MessageRouterPublishRequest request;
 
-		try {
+    @InjectMocks
+    NotificationProducer notificationProducer;
 
-			when(cambriaBatchingPublisher.send(Mockito.anyString(), Mockito.anyString())).thenReturn(0);
-			int result = notificationProducer.sendNotification("msg");
-			assertEquals(0, result);
-		} catch (IOException e) {
-			e.printStackTrace();
-		}
+    @Test
+    public void notificationProducerTest() throws IOException {
+            io.vavr.collection.List<String> expectedItems = io.vavr.collection.List.of("I", "like", "pizza");
+            MessageRouterPublishResponse expectedResponse = ImmutableMessageRouterPublishResponse
+                    .builder().items(expectedItems.map(JsonPrimitive::new))
+                    .build();
+            Flux<MessageRouterPublishResponse> responses = Flux.just(expectedResponse);
+//            Flux<JsonPrimitive> singleMessage = Flux.just("msg").map(JsonPrimitive::new);
+            when(publisher.put(any(), any())).thenReturn(responses);
 
-	}
+            notificationProducer.sendNotification("msg");
+    }
 }
diff --git a/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/PolicyDmaapClientTest.java b/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/PolicyDmaapClientTest.java
index 3ff56ab1..26f691e3 100644
--- a/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/PolicyDmaapClientTest.java
+++ b/components/slice-analysis-ms/src/test/java/org/onap/slice/analysis/ms/dmaap/PolicyDmaapClientTest.java
@@ -3,6 +3,7 @@
  *  slice-analysis-ms
  *  ================================================================================
  *   Copyright (C) 2020 Wipro Limited.
+ *   Copyright (C) 2022 CTC, Inc.
  *   ==============================================================================
  *     Licensed under the Apache License, Version 2.0 (the "License");
  *     you may not use this file except in compliance with the License.
@@ -22,51 +23,50 @@
 
 package org.onap.slice.analysis.ms.dmaap;
 
-import static org.junit.Assert.assertTrue;
-
-import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 
+import com.google.gson.JsonPrimitive;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.Mockito;
-import org.mockito.runners.MockitoJUnitRunner;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterPublisher;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishResponse;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse;
 import org.onap.slice.analysis.ms.models.Configuration;
-import org.onap.slice.analysis.ms.utils.DmaapUtils;
+import org.onap.slice.analysis.ms.utils.DcaeDmaapUtil;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.modules.junit4.PowerMockRunnerDelegate;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+import reactor.core.publisher.Flux;
 
-import com.att.nsa.cambria.client.CambriaBatchingPublisher;
-import com.att.nsa.cambria.client.CambriaConsumer;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
 
-@RunWith(MockitoJUnitRunner.class)
-@SpringBootTest(classes = PolicyDmaapClient.class)
+@RunWith(PowerMockRunner.class)
+@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"})
+@PowerMockRunnerDelegate(SpringRunner.class)
+@PrepareForTest({ DcaeDmaapUtil.class})
+@SpringBootTest(classes = PolicyDmaapClientTest.class)
 public class PolicyDmaapClientTest {
     
     @Mock
     Configuration configurationMock;
     
-    @Mock
-    DmaapUtils dmaapUtilsMock;
-    
     @InjectMocks
     PolicyDmaapClient policyDmaapClient;
     
-    @Mock
-    CambriaConsumer policyResponseCambriaConsumerMock;
-    
-    @Mock
-    CambriaBatchingPublisher cambriaBatchingPublisherMock;
-    
-    @Mock
-    NotificationProducer notificationProducerMock;
-    
     @Before
     public void setup() {
-        policyDmaapClient = new PolicyDmaapClient(dmaapUtilsMock, configurationMock);
+        policyDmaapClient = new PolicyDmaapClient(configurationMock);
     }
     
     @Test
@@ -78,13 +78,21 @@ public class PolicyDmaapClientTest {
         dmaapInfo.put("dmaap_info", topics);
         streamsPublishes.put("CL_topic", dmaapInfo);
         Mockito.when(configurationMock.getStreamsPublishes()).thenReturn(streamsPublishes);
-        Mockito.when(dmaapUtilsMock.buildPublisher(configurationMock, "DCAE_CL_OUTPUT")).thenReturn(cambriaBatchingPublisherMock);
-        try {
-            Mockito.when(cambriaBatchingPublisherMock.send("", "hello")).thenReturn(0);
-        } catch (IOException e) {
-            e.printStackTrace();
-        }
-        assertTrue(policyDmaapClient.sendNotificationToPolicy("hello"));
-        
+
+        PowerMockito.mockStatic(DcaeDmaapUtil.class);
+        MessageRouterPublisher publisher = PowerMockito.mock(MessageRouterPublisher.class);
+        PowerMockito.when(DcaeDmaapUtil.buildPublisher()).thenReturn(publisher);
+
+        MessageRouterPublishRequest request = PowerMockito.mock(MessageRouterPublishRequest.class);
+        PowerMockito.when(DcaeDmaapUtil.buildPublisherRequest(any(),any())).thenReturn(request);
+
+        io.vavr.collection.List<String> expectedItems = io.vavr.collection.List.of("I", "like", "pizza");
+        MessageRouterPublishResponse expectedResponse = ImmutableMessageRouterPublishResponse
+                .builder().items(expectedItems.map(JsonPrimitive::new))
+                .build();
+        Flux<MessageRouterPublishResponse> responses = Flux.just(expectedResponse);
+        when(publisher.put(any(), any())).thenReturn(responses);
+
+        policyDmaapClient.sendNotificationToPolicy("msg");
     }
 }