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;
* @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";
}
*/
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);
}
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);
--- /dev/null
+/*******************************************************************************
+ * ============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());
+
+
+ }
+
+
+}
}
}
- }
+ }
@Test
public void testSignalOwnership() {
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
}
}
+
}
--- /dev/null
+/*******************************************************************************
+ * ============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);
+ }
+ }
+
+}
* ============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
}
@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();
- }
-
- }*/
}
--- /dev/null
+/*******************************************************************************
+ * ============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);
+}
+
+}
/*-
* ============LICENSE_START=======================================================
- * ONAP Policy Engine
+ *
* ================================================================================
* Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
* ================================================================================
assertEquals(204, response.getStatus());
}
+ @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();
assertEquals(500, response.getStatus());
}
+ @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");
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);
+ }
}
--- /dev/null
+/*-
+ * ============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);
+ }
+
+ }
+
+}
import java.text.SimpleDateFormat;
import java.util.Date;
-import javax.servlet.http.HttpServletRequest;
import org.apache.http.auth.BasicUserPrincipal;
import org.junit.After;
}
+
+ @Test
+ public void testGetKey(){
+ assertNotNull(Utils.getKafkaproperty());
+
+ }
+
+ @Test
+ public void testCadiEnable(){
+ assertFalse(Utils.isCadiEnabled());
+
+ }
}
Code Review / dmaap / messagerouter / msgrtr.git / commitdiff