summaryrefslogtreecommitdiffstats
path: root/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka
diff options
context:
space:
mode:
Diffstat (limited to 'src/test/java/org/onap/dmaap/mr/cambria/backends/kafka')
-rw-r--r--src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/CuratorFrameworkImpl.java278
-rw-r--r--src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/JUnitTestSuite.java43
-rw-r--r--src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/Kafka011ConsumerTest.java91
-rw-r--r--src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/KafkaConsumerCacheTest.java250
-rw-r--r--src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/KafkaLiveLockAvoider2Test.java109
-rw-r--r--src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/KafkaPublisherTest.java74
-rw-r--r--src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/MetricsSetImpl.java122
-rw-r--r--src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/TestRunner.java41
8 files changed, 1008 insertions, 0 deletions
diff --git a/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/CuratorFrameworkImpl.java b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/CuratorFrameworkImpl.java
new file mode 100644
index 0000000..2f57abf
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/CuratorFrameworkImpl.java
@@ -0,0 +1,278 @@
+/*-
+ * ============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.backends.kafka;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.curator.CuratorZookeeperClient;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.WatcherRemoveCuratorFramework;
+import org.apache.curator.framework.api.CreateBuilder;
+import org.apache.curator.framework.api.CuratorListener;
+import org.apache.curator.framework.api.DeleteBuilder;
+import org.apache.curator.framework.api.ExistsBuilder;
+import org.apache.curator.framework.api.GetACLBuilder;
+import org.apache.curator.framework.api.GetChildrenBuilder;
+import org.apache.curator.framework.api.GetConfigBuilder;
+import org.apache.curator.framework.api.GetDataBuilder;
+import org.apache.curator.framework.api.ReconfigBuilder;
+import org.apache.curator.framework.api.RemoveWatchesBuilder;
+import org.apache.curator.framework.api.SetACLBuilder;
+import org.apache.curator.framework.api.SetDataBuilder;
+import org.apache.curator.framework.api.SyncBuilder;
+import org.apache.curator.framework.api.UnhandledErrorListener;
+import org.apache.curator.framework.api.transaction.CuratorMultiTransaction;
+import org.apache.curator.framework.api.transaction.CuratorTransaction;
+import org.apache.curator.framework.api.transaction.TransactionOp;
+import org.apache.curator.framework.imps.CuratorFrameworkState;
+import org.apache.curator.framework.listen.Listenable;
+import org.apache.curator.framework.schema.SchemaSet;
+import org.apache.curator.framework.state.ConnectionStateErrorPolicy;
+import org.apache.curator.framework.state.ConnectionStateListener;
+import org.apache.curator.utils.EnsurePath;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
+
+public class CuratorFrameworkImpl implements CuratorFramework {
+
+ @Override
+ public void blockUntilConnected() throws InterruptedException {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public boolean blockUntilConnected(int arg0, TimeUnit arg1) throws InterruptedException {
+ // TODO Auto-generated method stub
+ return false;
+ }
+
+ @Override
+ public ExistsBuilder checkExists() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public void clearWatcherReferences(Watcher arg0) {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public void close() {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public CreateBuilder create() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public DeleteBuilder delete() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public GetACLBuilder getACL() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public GetChildrenBuilder getChildren() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public Listenable<ConnectionStateListener> getConnectionStateListenable() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public Listenable<CuratorListener> getCuratorListenable() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public GetDataBuilder getData() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public String getNamespace() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public CuratorFrameworkState getState() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public Listenable<UnhandledErrorListener> getUnhandledErrorListenable() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public CuratorZookeeperClient getZookeeperClient() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public CuratorTransaction inTransaction() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public boolean isStarted() {
+ // TODO Auto-generated method stub
+ return false;
+ }
+
+ @Override
+ public EnsurePath newNamespaceAwareEnsurePath(String arg0) {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public CuratorFramework nonNamespaceView() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public SetACLBuilder setACL() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public SetDataBuilder setData() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public void start() {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public SyncBuilder sync() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public void sync(String arg0, Object arg1) {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public CuratorFramework usingNamespace(String arg0) {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public ReconfigBuilder reconfig() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public GetConfigBuilder getConfig() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public CuratorMultiTransaction transaction() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public TransactionOp transactionOp() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public void createContainers(String path) throws Exception {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public RemoveWatchesBuilder watches() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public WatcherRemoveCuratorFramework newWatcherRemoveCuratorFramework() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public ConnectionStateErrorPolicy getConnectionStateErrorPolicy() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public QuorumVerifier getCurrentConfig() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public SchemaSet getSchemaSet() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public boolean isZk34CompatibilityMode() {
+ // TODO Auto-generated method stub
+ return false;
+ }
+
+}
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
new file mode 100644
index 0000000..d299fdd
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/JUnitTestSuite.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.backends.kafka;
+
+import junit.framework.TestSuite;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+import org.junit.runners.Suite.SuiteClasses;
+
+@RunWith(Suite.class)
+@SuiteClasses({KafkaConsumerCacheTest.class, KafkaPublisherTest.class, Kafka011ConsumerTest.class,
+ KafkaLiveLockAvoider2Test.class})
+public class JUnitTestSuite {
+
+ private static final Logger LOGGER = LogManager.getLogger(JUnitTestSuite.class);
+
+ public static void main(String[] args) {
+ LOGGER.info("Running the test suite");
+
+ TestSuite tstSuite = new TestSuite();
+ LOGGER.info("Total Test Counts " + tstSuite.countTestCases());
+ }
+}
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..10526c5
--- /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 com.att.ajsc.filemonitor.AJSCPropertiesMap;
+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.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"})
+@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
new file mode 100644
index 0000000..457fff4
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/KafkaConsumerCacheTest.java
@@ -0,0 +1,250 @@
+/*-
+ * ============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.backends.kafka;
+
+import static org.junit.Assert.*;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.curator.framework.CuratorFramework;
+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.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+
+import com.att.ajsc.filemonitor.AJSCPropertiesMap;
+
+import org.onap.dmaap.dmf.mr.backends.MetricsSet;
+import org.onap.dmaap.dmf.mr.backends.kafka.Kafka011Consumer;
+import org.onap.dmaap.dmf.mr.backends.kafka.KafkaConsumerCache;
+import org.onap.dmaap.dmf.mr.backends.kafka.KafkaConsumerCache.KafkaConsumerCacheException;
+import org.onap.dmaap.dmf.mr.constants.CambriaConstants;
+
+@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"})
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ AJSCPropertiesMap.class })
+public class KafkaConsumerCacheTest {
+ private KafkaConsumerCache kafkaConsumerCache =null;
+ @Mock
+ private ConcurrentHashMap<String, Kafka011Consumer> fConsumers;
+ @Mock
+ private MetricsSet fMetrics;
+
+ @Before
+ public void setUp() throws Exception {
+ MockitoAnnotations.initMocks(this);
+
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ }
+
+
+ @Test
+ public void testSweep() {
+ kafkaConsumerCache = new KafkaConsumerCache();
+ PowerMockito.mockStatic(AJSCPropertiesMap.class);
+ PowerMockito.when(AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "kSetting_TouchEveryMs")).thenReturn("100");
+ kafkaConsumerCache.sweep();
+
+ }
+
+
+ // DOES NOT WORK
+ @Test
+ public void testStartCache() {
+
+ /*
+ * KafkaConsumerCache kafka = null;
+ *
+ * try { kafka = new KafkaConsumerCache("123", null);
+ *
+ * } catch (NoClassDefFoundError e) { try { kafka.startCache("DMAAP",
+ * null); } catch (NullPointerException e1) { // TODO Auto-generated
+ * catch block assertTrue(true); } catch (KafkaConsumerCacheException
+ * e1) { // TODO Auto-generated catch block e1.printStackTrace(); } }
+ */
+
+
+ new CuratorFrameworkImpl();
+ new MetricsSetImpl();
+ KafkaConsumerCache kafka=null;
+ try {
+ kafka = new KafkaConsumerCache();
+ kafka.setfApiId("1");
+ kafka.startCache("DMAAP", null);
+ } catch (NoClassDefFoundError e) {
+
+ } catch (KafkaConsumerCacheException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+
+ }
+
+ @Test
+ public void testGetCuratorFramework() {
+
+ CuratorFramework curator = new CuratorFrameworkImpl();
+ new MetricsSetImpl();
+ try {
+
+ } catch (NoClassDefFoundError e) {
+
+ KafkaConsumerCache.getCuratorFramework(curator);
+ }
+
+ }
+
+ /*
+ * @Test public void testStopCache() {
+ *
+ * KafkaConsumerCache kafka = null; new CuratorFrameworkImpl(); new
+ * MetricsSetImpl(); try { kafka = new KafkaConsumerCache("123", null);
+ * kafka.stopCache(); } catch (NoClassDefFoundError e) {
+ *
+ * }
+ *
+ * }
+ */
+
+ @Test
+ public void testGetConsumerFor() {
+
+ KafkaConsumerCache kafka = null;
+
+ try {
+ kafka = new KafkaConsumerCache();
+ kafka.getConsumerFor("testTopic", "CG1", "23");
+ } catch (NoClassDefFoundError e) {
+
+ } catch (KafkaConsumerCacheException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+
+ }
+
+ @Test
+ public void testPutConsumerFor() {
+
+ Kafka011Consumer consumer = null;
+ KafkaConsumerCache kafka = null;
+
+ try {
+ kafka = new KafkaConsumerCache();
+
+ } catch (NoClassDefFoundError e) {
+ try {
+ kafka.putConsumerFor("testTopic", "CG1", "23", consumer);
+ } catch (NullPointerException e1) {
+ // TODO Auto-generated catch block
+ assertTrue(true);
+ } catch (KafkaConsumerCacheException e1) {
+ // TODO Auto-generated catch block
+ e1.printStackTrace();
+ }
+ }
+
+ }
+
+ @Test
+ public void testGetConsumers() {
+
+ KafkaConsumerCache kafka = null;
+
+ try {
+ kafka = new KafkaConsumerCache();
+
+ } catch (NoClassDefFoundError e) {
+ try {
+ kafka.getConsumers();
+ } catch (NullPointerException e1) {
+ // TODO Auto-generated catch block
+ assertTrue(true);
+ }
+ }
+
+ }
+
+ @Test
+ public void testDropAllConsumers() {
+
+ KafkaConsumerCache kafka = null;
+ try {
+ kafka = new KafkaConsumerCache();
+
+ } catch (NoClassDefFoundError e) {
+ try {
+ kafka.dropAllConsumers();
+ } catch (NullPointerException e1) {
+ // TODO Auto-generated catch block
+ assertTrue(true);
+ }
+ }
+
+ }
+
+ @Test
+ public void testSignalOwnership() {
+
+ KafkaConsumerCache kafka = null;
+
+ try {
+ kafka = new KafkaConsumerCache();
+ try {
+ kafka.signalOwnership("testTopic", "CG1", "23");
+ } catch (KafkaConsumerCacheException e) {
+ assertTrue(true);
+ }
+ } catch (NoClassDefFoundError e) {}
+
+ //
+ }
+
+ @Test
+ public void testDropConsumer() {
+
+ KafkaConsumerCache kafka = null;
+
+ try {
+ kafka = new KafkaConsumerCache();
+ // kafka.dropConsumer("testTopic", "CG1", "23");
+ } catch (NoClassDefFoundError e) {
+ try {
+ kafka.dropConsumer("testTopic", "CG1", "23");
+ } catch (NullPointerException e1) {
+ // TODO Auto-generated catch block
+ assertTrue(true);
+ }
+ }
+
+ }
+
+
+}
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..7f81641
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/KafkaLiveLockAvoider2Test.java
@@ -0,0 +1,109 @@
+/*******************************************************************************
+ * ============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.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"})
+@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
new file mode 100644
index 0000000..7a0fe78
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/KafkaPublisherTest.java
@@ -0,0 +1,74 @@
+/*-
+ * ============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.backends.kafka;
+
+import static org.junit.Assert.assertTrue;
+
+import com.att.nsa.drumlin.till.nv.rrNvReadable.missingReqdSetting;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.MockitoAnnotations;
+import org.onap.dmaap.dmf.mr.backends.kafka.KafkaPublisher;
+import org.onap.dmaap.dmf.mr.utils.Utils;
+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;
+
+@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"})
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ Utils.class })
+public class KafkaPublisherTest {
+
+ @Before
+ public void setUp() throws Exception {
+ MockitoAnnotations.initMocks(this);
+ PowerMockito.mockStatic(Utils.class);
+ PowerMockito.when(Utils.isCadiEnabled()).thenReturn(true);
+
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ }
+
+ @Test
+ public void testPublisherInit() {
+
+
+
+ try {
+ try {
+ KafkaPublisher kafkaPublisher = new KafkaPublisher(null);
+ } catch (missingReqdSetting e) {
+ assertTrue(true);
+ }
+ } catch (LinkageError e) {
+ assertTrue(true);
+ }
+
+ }
+
+
+
+}
diff --git a/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/MetricsSetImpl.java b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/MetricsSetImpl.java
new file mode 100644
index 0000000..ea36d86
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/MetricsSetImpl.java
@@ -0,0 +1,122 @@
+/*-
+ * ============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.backends.kafka;
+
+import com.att.nsa.metrics.CdmMeasuredItem;
+import org.json.JSONObject;
+import org.onap.dmaap.dmf.mr.backends.MetricsSet;
+
+import java.util.List;
+import java.util.Map;
+
+public class MetricsSetImpl implements MetricsSet {
+
+ @Override
+ public List<? extends CdmMetricEntry> getEntries() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public CdmMeasuredItem getItem(String arg0) {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public Map<String, CdmMeasuredItem> getItems() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public void putItem(String arg0, CdmMeasuredItem arg1) {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public void removeItem(String arg0) {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public int size() {
+ // TODO Auto-generated method stub
+ return 0;
+ }
+
+ @Override
+ public JSONObject toJson() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public void setupCambriaSender() {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public void onRouteComplete(String name, long durationMs) {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public void publishTick(int amount) {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public void consumeTick(int amount) {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public void onKafkaConsumerCacheMiss() {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public void onKafkaConsumerCacheHit() {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public void onKafkaConsumerClaimed() {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public void onKafkaConsumerTimeout() {
+ // TODO Auto-generated method stub
+
+ }
+
+}
diff --git a/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/TestRunner.java b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/TestRunner.java
new file mode 100644
index 0000000..38261e1
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/mr/cambria/backends/kafka/TestRunner.java
@@ -0,0 +1,41 @@
+/*-
+ * ============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.backends.kafka;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.junit.runner.JUnitCore;
+import org.junit.runner.Result;
+import org.junit.runner.notification.Failure;
+
+public class TestRunner {
+
+ private static final Logger LOGGER = LogManager.getLogger(TestRunner.class);
+
+ public static void main(String[] args) {
+ // TODO Auto-generated method stub
+ Result result = JUnitCore.runClasses(JUnitTestSuite.class);
+ for (Failure failure : result.getFailures()) {
+ LOGGER.info(failure.toString());
+ }
+ LOGGER.info(result.wasSuccessful());
+ }
+}