add testcases

Issue-ID: DMAAP-1140
Change-Id: I6dce89a14f874e7788acd7bea1fd48ebaafed71e
Signed-off-by: sunil.unnava <sunil.unnava@att.com>

11 files changed, 388 insertions, 125 deletions

diff --git a/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/KafkaPublisher.java b/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/KafkaPublisher.java
index 2a9e0ab..4bdd9f3 100644
--- a/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/KafkaPublisher.java
+++ b/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/KafkaPublisher.java
@@ -32,7 +32,7 @@ import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.json.JSONException;
 import org.springframework.beans.factory.annotation.Qualifier;
-
+import org.springframework.util.StringUtils;
 import org.onap.dmaap.dmf.mr.backends.Publisher;
 import org.onap.dmaap.dmf.mr.constants.CambriaConstants;
 import org.onap.dmaap.dmf.mr.utils.Utils;
@@ -61,15 +61,10 @@ public class KafkaPublisher implements Publisher {
 	 * @throws rrNvReadable.missingReqdSetting
 	 */
 	public KafkaPublisher(@Qualifier("propertyReader") rrNvReadable settings) throws rrNvReadable.missingReqdSetting {
-		//fSettings = settings;
 
 		final Properties props = new Properties();
-		/*transferSetting(fSettings, props, "metadata.broker.list", "localhost:9092");
-		transferSetting(fSettings, props, "request.required.acks", "1");
-		transferSetting(fSettings, props, "message.send.max.retries", "5");
-		transferSetting(fSettings, props, "retry.backoff.ms", "150"); */
 		String kafkaConnUrl= com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,"kafka.metadata.broker.list");
-		if(null==kafkaConnUrl){
+		if(StringUtils.isEmpty(kafkaConnUrl)){
 			
 			kafkaConnUrl="localhost:9092";
 		}
@@ -209,7 +204,7 @@ try{
    */
 	private void transferSetting(Properties props, String key, String defVal) {
 		String kafka_prop= com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,"kafka." + key);
-		if (null==kafka_prop) kafka_prop=defVal;
+		if (StringUtils.isEmpty(kafka_prop)) kafka_prop=defVal;
 		//props.put(key, settings.getString("kafka." + key, defVal));
 		props.put(key, kafka_prop);
 	}
diff --git a/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/JUnitTestSuite.java b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/JUnitTestSuite.java
index 5904de5..de77603 100644
--- a/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/JUnitTestSuite.java
+++ b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/JUnitTestSuite.java
@@ -28,7 +28,7 @@ import org.junit.runners.Suite.SuiteClasses;
 import org.apache.log4j.Logger;
 
 @RunWith(Suite.class)
-@SuiteClasses({ KafkaConsumerCacheTest.class, KafkaPublisherTest.class, })
+@SuiteClasses({ KafkaConsumerCacheTest.class, KafkaPublisherTest.class,Kafka011ConsumerTest.class,KafkaLiveLockAvoider2Test.class })
 public class JUnitTestSuite {
 	private static final Logger LOGGER = Logger.getLogger(JUnitTestSuite.class);
 
diff --git a/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/Kafka011ConsumerTest.java b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/Kafka011ConsumerTest.java
new file mode 100644
index 0000000..88d703e
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/Kafka011ConsumerTest.java
@@ -0,0 +1,91 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 2017 AT&T 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=========================================================
+ *  
+ *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ *  
+ *******************************************************************************/
+package org.onap.dmaap.mr.cambria.backends.kafka;
+
+import static org.junit.Assert.assertNotNull;
+
+
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.onap.dmaap.dmf.mr.backends.kafka.Kafka011Consumer;
+import org.onap.dmaap.dmf.mr.backends.kafka.KafkaLiveLockAvoider2;
+import org.onap.dmaap.dmf.mr.constants.CambriaConstants;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import com.att.ajsc.filemonitor.AJSCPropertiesMap;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ AJSCPropertiesMap.class })
+public class Kafka011ConsumerTest {
+	
+		
+	@Mock
+	private KafkaConsumer<String, String> cc;
+	@Mock
+	private KafkaLiveLockAvoider2 klla;
+
+	@Before
+	public void setUp() throws Exception {
+		MockitoAnnotations.initMocks(this);
+		
+	}
+
+	@After
+	public void tearDown() throws Exception {
+	}
+	
+	@Test
+	public void testKafka011Consumer() {
+		PowerMockito.mockStatic(AJSCPropertiesMap.class);
+		PowerMockito.when(AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "consumer.timeout")).thenReturn("10");
+		Kafka011Consumer consumer=null;
+		try {
+			 consumer= new Kafka011Consumer("topic", "group", "id", cc, klla)	;
+			 consumer.commitOffsets();
+			 consumer.touch();
+			 consumer.setOffset(10);
+		} catch (Exception e) {
+			
+		}
+		assertNotNull(consumer);
+		assertNotNull(consumer.getConsumer());
+		assertNotNull(consumer.getConsumerGroup());
+		assertNotNull(consumer.getConsumerId());
+		assertNotNull(consumer.getConsumerId());
+		assertNotNull(consumer.getCreateTimeMs());
+		assertNotNull(consumer.getLastAccessMs());
+		assertNotNull(consumer.getName());
+		assertNotNull(consumer.getOffset());
+		assertNotNull(consumer.getLastTouch());
+		
+		
+	}
+	
+
+}
diff --git a/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/KafkaConsumerCacheTest.java b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/KafkaConsumerCacheTest.java
index 83866cf..ced6fc4 100644
--- a/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/KafkaConsumerCacheTest.java
+++ b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/KafkaConsumerCacheTest.java
@@ -207,7 +207,7 @@ public class KafkaConsumerCacheTest {
 			}
 		}
 
-	}
+	} 
 
 	@Test
 	public void testSignalOwnership() {
@@ -216,22 +216,14 @@ public class KafkaConsumerCacheTest {
 
 		try {
 			kafka = new KafkaConsumerCache();
-			// kafka.signalOwnership("testTopic", "CG1", "23");
-		} catch (NoClassDefFoundError e) {
-			try {
-				kafka.signalOwnership("testTopic", "CG1", "23");
-			} catch (KafkaConsumerCacheException e1) {
-				// TODO Auto-generated catch block
-				e.printStackTrace();
-			} catch (NullPointerException e1) {
-				// TODO Auto-generated catch block
-				// assertTrue(true);
-				e1.printStackTrace();
-			}
-
+		 try {
+			kafka.signalOwnership("testTopic", "CG1", "23");
+		} catch (KafkaConsumerCacheException e) {
+			assertTrue(true);
 		}
+		} catch (NoClassDefFoundError e) {}
 
-		// assertTrue(true);
+		// 
 	}
 
 	@Test
@@ -252,5 +244,6 @@ public class KafkaConsumerCacheTest {
 		}
 
 	}
+	
 
 }
diff --git a/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/KafkaLiveLockAvoider2Test.java b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/KafkaLiveLockAvoider2Test.java
new file mode 100644
index 0000000..71d50e8
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/KafkaLiveLockAvoider2Test.java
@@ -0,0 +1,108 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 2017 AT&T 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=========================================================
+ *  
+ *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ *  
+ *******************************************************************************/
+package org.onap.dmaap.mr.cambria.backends.kafka;
+
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.api.CreateBuilder;
+import org.apache.curator.framework.api.ExistsBuilder;
+import org.apache.curator.framework.api.GetChildrenBuilder;
+import org.apache.curator.framework.api.ProtectACLCreateModeStatPathAndBytesable;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.onap.dmaap.dmf.mr.backends.kafka.KafkaLiveLockAvoider2;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+
+@RunWith(PowerMockRunner.class)
+public class KafkaLiveLockAvoider2Test {
+	
+	@Mock
+	private CuratorFramework curatorFramework;
+	@Mock
+	private ExistsBuilder existsBuilder;
+	@Mock
+	private CreateBuilder createBuilder;
+	@Mock
+	private GetChildrenBuilder childrenBuilder;
+	@Mock
+	ProtectACLCreateModeStatPathAndBytesable<String> acl;
+	@InjectMocks
+	private KafkaLiveLockAvoider2 liveLockAvoider;
+	
+	public static final String ZNODE_ROOT = "/live-lock-avoid";
+	public static final String ZNODE_LOCKS = "/locks";
+	public static final String ZNODE_UNSTICK_TASKS ="/unstick-tasks";
+	
+	private static String locksPath = ZNODE_ROOT+ZNODE_LOCKS;
+	private static String tasksPath = ZNODE_ROOT+ZNODE_UNSTICK_TASKS;
+	
+
+	@Before
+	public void setUp() throws Exception {
+		List<String> taskNodes= new ArrayList<String>();
+		taskNodes.add("appId");
+		MockitoAnnotations.initMocks(this);
+		PowerMockito.when(acl.forPath(locksPath)).thenReturn(locksPath);
+		PowerMockito.when(acl.forPath(tasksPath)).thenReturn(tasksPath);
+		PowerMockito.when(createBuilder.creatingParentsIfNeeded()).thenReturn(acl);
+		PowerMockito.when(curatorFramework.create()).thenReturn(createBuilder);
+		PowerMockito.when(curatorFramework.checkExists()).thenReturn(existsBuilder);
+		PowerMockito.when(childrenBuilder.forPath(tasksPath)).thenReturn(taskNodes);
+		PowerMockito.when(curatorFramework.getChildren()).thenReturn(childrenBuilder);
+		
+	}
+
+	@After
+	public void tearDown() throws Exception {
+	}
+
+	@Test
+	public void testUnlock(){
+		liveLockAvoider.init();
+		try {
+			liveLockAvoider.unlockConsumerGroup("appId", "groupName");
+		} catch (Exception e) {
+			assertTrue(true);
+		}
+	}
+	
+	@Test
+	public void testWatcher(){
+		try {
+			liveLockAvoider.startNewWatcherForServer("appId", null);
+		} catch (Exception e) {
+			assertTrue(true);
+		}
+	}
+
+}
diff --git a/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/KafkaPublisherTest.java b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/KafkaPublisherTest.java
index 982fbf2..8292c2c 100644
--- a/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/KafkaPublisherTest.java
+++ b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/KafkaPublisherTest.java
@@ -18,35 +18,43 @@
  * ============LICENSE_END=========================================================
  */
 
- package org.onap.dmaap.mr.cambria.backends.kafka;
+package org.onap.dmaap.mr.cambria.backends.kafka;
 
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
-import java.io.File;
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
 
+import org.apache.kafka.clients.producer.Producer;
 import org.junit.After;
 import org.junit.Before;
-import org.junit.ClassRule;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.onap.dmaap.dmf.mr.backends.Publisher.message;
+import org.onap.dmaap.dmf.mr.backends.kafka.KafkaPublisher;
+import org.onap.dmaap.dmf.mr.beans.LogDetails;
+import org.onap.dmaap.dmf.mr.constants.CambriaConstants;
+import org.onap.dmaap.dmf.mr.utils.Utils;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
 
 import com.att.ajsc.filemonitor.AJSCPropertiesMap;
-import org.onap.dmaap.dmf.mr.backends.Publisher.message;
 import com.att.nsa.drumlin.till.nv.rrNvReadable.missingReqdSetting;
 
-import kafka.common.FailedToSendMessageException;
-import kafka.producer.KeyedMessage;
-
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ Utils.class })
 public class KafkaPublisherTest {
-	
-	
 
-	/*@Before
+	@Before
 	public void setUp() throws Exception {
-		ClassLoader classLoader = getClass().getClassLoader();		
-		AJSCPropertiesMap.refresh(new File(classLoader.getResource("MsgRtrApi.properties").getFile()));
+		MockitoAnnotations.initMocks(this);
+		PowerMockito.mockStatic(Utils.class);
+		PowerMockito.when(Utils.isCadiEnabled()).thenReturn(true);
+
 	}
 
 	@After
@@ -54,100 +62,22 @@ public class KafkaPublisherTest {
 	}
 
 	@Test
-	public void testSendMessages() {
-
-		String topic = "testTopic";
+	public void testPublisherInit() {
+		
+		
 
-		KafkaPublisher kafka = null;
 		try {
-			kafka = new KafkaPublisher(null);
-			
-		} catch (missingReqdSetting e) {
-			// TODO Auto-generated catch block
-			e.printStackTrace();
-		} catch (NoClassDefFoundError e) {
 			try {
-				kafka.sendMessage(topic, null);
-			} catch (NullPointerException e1) {
-				// TODO Auto-generated catch block
+				KafkaPublisher kafkaPublisher = new KafkaPublisher(null);
+			} catch (missingReqdSetting e) {
 				assertTrue(true);
-			} catch (FailedToSendMessageException e1) {
-				// TODO Auto-generated catch block
-				e1.printStackTrace();
-			} catch (IOException e1) {
-				// TODO Auto-generated catch block
-				e1.printStackTrace();
 			}
-		} catch (FailedToSendMessageException e) {
-			// TODO Auto-generated catch block
-			e.printStackTrace();
+		} catch (LinkageError e) {
+			assertTrue(true);
 		}
 
 	}
 
-	@Test
-	public void testSendBatchMessage() {
-
-		String topic = "testTopic";
-
-		KafkaPublisher kafka = null;
-		ArrayList<KeyedMessage<String, String>> kms = null;
-		try {
-			kafka = new KafkaPublisher(null);
-			
-		} catch (missingReqdSetting e) {
-			// TODO Auto-generated catch block
-			e.printStackTrace();
-		} catch (NoClassDefFoundError e) {
-			try {
-				kafka.sendBatchMessage(topic, kms);
-			} catch (NullPointerException e1) {
-				// TODO Auto-generated catch block
-				assertTrue(true);
-			} catch (IOException e1) {
-				// TODO Auto-generated catch block
-				e1.printStackTrace();
-			}
-		} catch (FailedToSendMessageException e) {
-			// TODO Auto-generated catch block
-			e.printStackTrace();
-		}
-
-	}
 	
-	@Test
-	public void sendMessages() {
-
-		String topic = "testTopic";
-		
-		List<message> msgs = null;
-
-		KafkaPublisher kafka = null;
-		//ArrayList<KeyedMessage<String, String>> kms = null;
-		try {
-			kafka = new KafkaPublisher(null);
-			
-		} catch (missingReqdSetting e) {
-			// TODO Auto-generated catch block
-			e.printStackTrace();
-		} catch (NoClassDefFoundError e) {
-			try {
-				kafka.sendMessages(topic, msgs);
-			} catch (NullPointerException e1) {
-				// TODO Auto-generated catch block
-				assertTrue(true);
-			} catch (FailedToSendMessageException e1) {
-				// TODO Auto-generated catch block
-				e1.printStackTrace();
-			} catch (IOException e1) {
-				// TODO Auto-generated catch block
-				e1.printStackTrace();
-			}
-		} catch (FailedToSendMessageException e) {
-			// TODO Auto-generated catch block
-			e.printStackTrace();
-		}
-
-	}*/
 
 }
diff --git a/src/test/java/org/onap/dmaap/mr/cambria/beans/DMaaPKafkaConsumerFactoryTest.java b/src/test/java/org/onap/dmaap/mr/cambria/beans/DMaaPKafkaConsumerFactoryTest.java
new file mode 100644
index 0000000..fd30359
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/mr/cambria/beans/DMaaPKafkaConsumerFactoryTest.java
@@ -0,0 +1,55 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 2017 AT&T 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=========================================================
+ *  
+ *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ *  
+ *******************************************************************************/
+package org.onap.dmaap.mr.cambria.beans;
+
+import static org.junit.Assert.assertTrue;
+
+import org.junit.After;
+import org.junit.Test;
+import org.onap.dmaap.dmf.mr.beans.DMaaPKafkaConsumerFactory;
+import org.onap.dmaap.dmf.mr.utils.ConfigurationReader;
+import org.onap.dmaap.mr.cambria.embed.EmbedConfigurationReader;
+
+
+public class DMaaPKafkaConsumerFactoryTest {
+	
+	EmbedConfigurationReader embedConfigurationReader = new EmbedConfigurationReader();
+	
+	@After
+	public void tearDown() throws Exception {
+		embedConfigurationReader.tearDown();
+	}
+
+@Test	
+public void testConsumerFactory(){
+	
+	try {
+		ConfigurationReader configurationReader = embedConfigurationReader.buildConfigurationReader();
+		DMaaPKafkaConsumerFactory consumerFactory=(DMaaPKafkaConsumerFactory) configurationReader.getfConsumerFactory();
+		consumerFactory.getConsumerFor("topic", "consumerGroupName", "consumerId", 10, "remotehost");
+	} catch (Exception e) {
+		assertTrue(false);
+	}
+	assertTrue(true);
+}
+
+}
diff --git a/src/test/java/org/onap/dmaap/mr/cambria/beans/TopicBeanTest.java b/src/test/java/org/onap/dmaap/mr/cambria/beans/TopicBeanTest.java
index 844fc08..bb3af25 100644
--- a/src/test/java/org/onap/dmaap/mr/cambria/beans/TopicBeanTest.java
+++ b/src/test/java/org/onap/dmaap/mr/cambria/beans/TopicBeanTest.java
@@ -1,6 +1,6 @@
 /*-
  * ============LICENSE_START=======================================================
- * ONAP Policy Engine
+ * 
  * ================================================================================
  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
  * ================================================================================
diff --git a/src/test/java/org/onap/dmaap/mr/cambria/utils/DMaaPResponseBuilderTest.java b/src/test/java/org/onap/dmaap/mr/cambria/utils/DMaaPResponseBuilderTest.java
index aaabb7d..002fc4b 100644
--- a/src/test/java/org/onap/dmaap/mr/cambria/utils/DMaaPResponseBuilderTest.java
+++ b/src/test/java/org/onap/dmaap/mr/cambria/utils/DMaaPResponseBuilderTest.java
@@ -84,19 +84,40 @@ public class DMaaPResponseBuilderTest {
 	}
 	
 	@Test
+	public void testrespondOkNoContentError(){
+		dMaapContext.setResponse(null);
+		DMaaPResponseBuilder.respondOkNoContent(dMaapContext);
+		assertNull(dMaapContext.getResponse());
+	}
+	
+	@Test
 	public void testrespondOkWithHtml(){
 		DMaaPResponseBuilder.respondOkWithHtml(dMaapContext, "<head></head>");
 		
 		assertEquals("text/html", response.getContentType());
+		DMaaPResponseBuilder.respondOkWithHtml(dMaapContext, "<head></head>");
 		assertEquals(200, response.getStatus());
 	}
 	
 	@Test
+	public void testrespondOkWithHtmlError(){
+		dMaapContext.setResponse(null);
+		DMaaPResponseBuilder.respondOkWithHtml(dMaapContext, "<head></head>");
+		assertNull(dMaapContext.getResponse());
+	}
+	
+	@Test
 	public void testrespondWithError(){
 		DMaaPResponseBuilder.respondWithError(dMaapContext, 500, "InternalServerError");
 		assertEquals(500, response.getStatus());
 	}
 	
+	@Test(expected=NullPointerException.class)
+	public void testInvalidrespondWithError(){
+		dMaapContext.setResponse(null);
+		DMaaPResponseBuilder.respondWithError(dMaapContext, 500, "InternalServerError");
+	}
+	
 	@Test
 	public void testrespondWithJsonError(){
 		JSONObject o = new JSONObject();
@@ -107,6 +128,16 @@ public class DMaaPResponseBuilderTest {
 	}
 	
 	@Test
+	public void testInvalidrespondWithJsonError(){
+		JSONObject o = new JSONObject();
+		o.put("status", 500);
+		o.put("message", "InternalServerError");
+		dMaapContext.setResponse(null);
+		DMaaPResponseBuilder.respondWithError(dMaapContext, 500, o);
+		assertNull(dMaapContext.getResponse());
+	}
+	
+	@Test
 	public void testrespondWithErrorInJson(){
 		DMaaPResponseBuilder.respondWithErrorInJson(dMaapContext, 500, "InternalServerError");
 		
@@ -137,5 +168,11 @@ public class DMaaPResponseBuilderTest {
 		assertEquals("application/octet-stream", response.getContentType());
 		assertEquals(200, response.getStatus());
 	}
+	
+	@Test(expected=NullPointerException.class)
+	public void testgetStreamForBinaryResponseError() throws IOException{
+		dMaapContext.setResponse(null);
+		DMaaPResponseBuilder.getStreamForBinaryResponse(dMaapContext);
+	}
 
 }
diff --git a/src/test/java/org/onap/dmaap/mr/cambria/utils/EMailerTest.java b/src/test/java/org/onap/dmaap/mr/cambria/utils/EMailerTest.java
new file mode 100644
index 0000000..ab81bab
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/mr/cambria/utils/EMailerTest.java
@@ -0,0 +1,43 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP Policy Engine
+ * ================================================================================
+ * Copyright (C) 2017 AT&T 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.dmaap.mr.cambria.utils;
+
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+
+import org.junit.Test;
+import org.onap.dmaap.dmf.mr.utils.Emailer;
+
+public class EMailerTest {
+	
+	@Test
+	public void testEmailer(){
+		
+		Emailer emailer= new Emailer();
+		try {
+			emailer.send("dummy@dummy.com", "subj", "body");
+		} catch (IOException e) {
+			assertTrue(true);
+		}
+		
+	}
+
+}
diff --git a/src/test/java/org/onap/dmaap/mr/cambria/utils/UtilsTest.java b/src/test/java/org/onap/dmaap/mr/cambria/utils/UtilsTest.java
index b4645a3..8a4009b 100644
--- a/src/test/java/org/onap/dmaap/mr/cambria/utils/UtilsTest.java
+++ b/src/test/java/org/onap/dmaap/mr/cambria/utils/UtilsTest.java
@@ -27,7 +27,6 @@ import java.security.Principal;
 import java.text.SimpleDateFormat;
 import java.util.Date;
 
-import javax.servlet.http.HttpServletRequest;
 
 import org.apache.http.auth.BasicUserPrincipal;
 import org.junit.After;
@@ -122,4 +121,16 @@ public class UtilsTest {
 		
 		
 	}
+	
+	@Test
+	public void testGetKey(){
+		assertNotNull(Utils.getKafkaproperty());
+		
+	}
+	
+	@Test
+	public void testCadiEnable(){
+		assertFalse(Utils.isCadiEnabled());
+		
+	}
 }