Add RestConf Collector

Issue-ID: DCAEGEN2-612

1. Instantiated to support CCVPN Close Loop Use Case
2. In general, this supports data collection from all PNF or devices that supports RestConf protocol

Change-Id: I6311ad618e8d68badc5423a63d7781a19dc62829
Signed-off-by: rama-huawei <rama.subba.reddy.s@huawei.com>

12 files changed, 658 insertions, 0 deletions

diff --git a/src/test/java/org/onap/dcae/collectors/restconf/common/event/publishing/DMaaPConfigurationParserTest.java b/src/test/java/org/onap/dcae/collectors/restconf/common/event/publishing/DMaaPConfigurationParserTest.java
new file mode 100755
index 0000000..1b709bd
--- /dev/null
+++ b/src/test/java/org/onap/dcae/collectors/restconf/common/event/publishing/DMaaPConfigurationParserTest.java
@@ -0,0 +1,111 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.restconf
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. 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.dcae.collectors.restconf.common.event.publishing;
+
+import io.vavr.collection.Map;
+import io.vavr.control.Try;
+import org.junit.Test;
+
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+import static io.vavr.API.List;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.onap.dcae.collectors.restconf.common.event.publishing.DMaaPConfigurationParser.parseToDomainMapping;
+
+public class DMaaPConfigurationParserTest {
+    @Test
+    public void testParseCredentialsForGen2() {
+        Path path = Paths.get("src/test/resources/testParseDMaaPCredentialsGen2.json");
+        Try<Map<String, PublisherConfig>> publisherConfigs = parseToDomainMapping(path);
+
+        PublisherConfig authCredentialsNulls = publisherConfigs.get().get("auth-credentials-null").getOrNull();
+        assertThat(authCredentialsNulls.userName().isEmpty()).isTrue();
+        assertThat(authCredentialsNulls.password().isEmpty()).isTrue();
+        assertThat(authCredentialsNulls.isSecured()).isFalse();
+
+        PublisherConfig authCredentialsPresent = publisherConfigs.get().get("auth-credentials-present").getOrNull();
+        assertThat(authCredentialsPresent.userName().getOrNull()).isEqualTo("sampleUser");
+        assertThat(authCredentialsPresent.password().getOrNull()).isEqualTo("samplePassword");
+        assertThat(authCredentialsPresent.isSecured()).isTrue();
+
+        PublisherConfig authCredentialsKeysMissing = publisherConfigs.get().get("auth-credentials-missing").getOrNull();
+        assertThat(authCredentialsKeysMissing.userName().isEmpty()).isTrue();
+        assertThat(authCredentialsKeysMissing.password().isEmpty()).isTrue();
+        assertThat(authCredentialsKeysMissing.isSecured()).isFalse();
+    }
+
+
+    @Test
+    public void testParseCredentialsForLegacy() {
+        Path path = Paths.get("src/test/resources/testParseDMaaPCredentialsLegacy.json");
+        Try<Map<String, PublisherConfig>> publisherConfigs = parseToDomainMapping(path);
+
+        PublisherConfig authCredentialsNull = publisherConfigs.get().get("auth-credentials-null").getOrNull();
+        assertThat(authCredentialsNull.userName().isEmpty()).isTrue();
+        assertThat(authCredentialsNull.password().isEmpty()).isTrue();
+        assertThat(authCredentialsNull.isSecured()).isFalse();
+
+        PublisherConfig authCredentialsPresent = publisherConfigs.get().get("auth-credentials-present").getOrNull();
+        assertThat(authCredentialsPresent.userName().getOrNull()).isEqualTo("sampleUser");
+        assertThat(authCredentialsPresent.password().getOrNull()).isEqualTo("samplePassword");
+        assertThat(authCredentialsPresent.isSecured()).isTrue();
+
+        PublisherConfig authCredentialsMissing = publisherConfigs.get().get("auth-credentials-missing").getOrNull();
+        assertThat(authCredentialsMissing.userName().isEmpty()).isTrue();
+        assertThat(authCredentialsMissing.password().isEmpty()).isTrue();
+        assertThat(authCredentialsMissing.isSecured()).isFalse();
+    }
+
+
+    @Test
+    public void testParseGen2() {
+        Path path = Paths.get("src/test/resources/testParseDMaaPGen2.json");
+        Try<Map<String, PublisherConfig>> publisherConfigs = parseToDomainMapping(path);
+
+        PublisherConfig withEventsSegment = publisherConfigs.get().get("event-segments-with-port").getOrNull();
+        assertThat(withEventsSegment.destinations()).isEqualTo(List("UEBHOST:3904"));
+        assertThat(withEventsSegment.topic()).isEqualTo("DCAE-RESTCONF-COLLECTOR-EVENTS-DEV");
+
+        PublisherConfig withOtherSegment = publisherConfigs.get().get("other-segments-without-ports").getOrNull();
+        assertThat(withOtherSegment.destinations()).isEqualTo(List("UEBHOST"));
+        assertThat(withOtherSegment.topic()).isEqualTo("DCAE-RESTCONF-COLLECTOR-EVENTS-DEV");
+    }
+
+    @Test
+    public void testParseLegacy() {
+        Path exemplaryConfig = Paths.get("src/test/resources/testParseDMaaPLegacy.json");
+        Try<Map<String, PublisherConfig>> publisherConfigs =
+                parseToDomainMapping(exemplaryConfig);
+
+        PublisherConfig urlFirstThenHosts = publisherConfigs.get().get("url-precedes-hosts").getOrNull();
+        assertThat(urlFirstThenHosts.destinations()).isEqualTo(List("127.0.0.1:3904"));
+        assertThat(urlFirstThenHosts.topic()).isEqualTo("DCAE-RESTCONF-COLLECTOR-EVENTS-DEV");
+
+        PublisherConfig urlKeyMissing = publisherConfigs.get().get("url-key-missing").getOrNull();
+        assertThat(urlKeyMissing.destinations()).isEqualTo(List("h1.att.com", "h2.att.com"));
+        assertThat(urlKeyMissing.topic()).isEqualTo("DCAE-RESTCONF-COLLECTOR-EVENTS-DEV");
+
+        PublisherConfig urlIsMissing = publisherConfigs.get().get("url-is-null").getOrNull();
+        assertThat(urlIsMissing.destinations()).isEqualTo(List("h1.att.com", "h2.att.com"));
+        assertThat(urlIsMissing.topic()).isEqualTo("DCAE-RESTCONF-COLLECTOR-EVENTS-DEV");
+    }
+}
diff --git a/src/test/java/org/onap/dcae/collectors/restconf/common/event/publishing/DMaaPEventPublisherTest.java b/src/test/java/org/onap/dcae/collectors/restconf/common/event/publishing/DMaaPEventPublisherTest.java
new file mode 100755
index 0000000..f66748c
--- /dev/null
+++ b/src/test/java/org/onap/dcae/collectors/restconf/common/event/publishing/DMaaPEventPublisherTest.java
@@ -0,0 +1,77 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.restconf
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. 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.dcae.collectors.restconf.common.event.publishing;
+
+import com.att.nsa.cambria.client.CambriaBatchingPublisher;
+import org.json.JSONObject;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+
+import java.io.IOException;
+
+import static io.vavr.API.Option;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class DMaaPEventPublisherTest {
+
+    private static final String STREAM_ID = "sampleStreamId";
+
+    private DMaaPEventPublisher eventPublisher;
+    private CambriaBatchingPublisher cambriaPublisher;
+    private DMaaPPublishersCache DMaaPPublishersCache;
+
+    @Before
+    public void setUp() {
+        cambriaPublisher = mock(CambriaBatchingPublisher.class);
+        DMaaPPublishersCache = mock(DMaaPPublishersCache.class);
+        when(DMaaPPublishersCache.getPublisher(anyString())).thenReturn(Option(cambriaPublisher));
+        eventPublisher = new DMaaPEventPublisher(DMaaPPublishersCache, mock(Logger.class));
+    }
+
+    @Test
+    public void shouldSendEventToTopic() throws Exception {
+        // given
+        JSONObject event = new JSONObject("{}");
+
+        // when
+        eventPublisher.sendEvent(event, STREAM_ID);
+
+        // then
+        verify(cambriaPublisher).send("MyPartitionKey", event.toString());
+    }
+
+    @Test
+    public void shouldCloseConnectionWhenExceptionOccurred() throws Exception {
+        // given
+        JSONObject event = new JSONObject("{}");
+        given(cambriaPublisher.send(anyString(), anyString())).willThrow(new IOException("epic fail"));
+
+        // when
+        eventPublisher.sendEvent(event, STREAM_ID);
+
+        // then
+        verify(DMaaPPublishersCache).closePublisherFor(STREAM_ID);
+    }
+}
diff --git a/src/test/java/org/onap/dcae/collectors/restconf/common/event/publishing/DMaaPPublishersCacheTest.java b/src/test/java/org/onap/dcae/collectors/restconf/common/event/publishing/DMaaPPublishersCacheTest.java
new file mode 100755
index 0000000..49f37c3
--- /dev/null
+++ b/src/test/java/org/onap/dcae/collectors/restconf/common/event/publishing/DMaaPPublishersCacheTest.java
@@ -0,0 +1,92 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.restconf
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. 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.dcae.collectors.restconf.common.event.publishing;
+
+import com.att.nsa.cambria.client.CambriaBatchingPublisher;
+import io.vavr.collection.Map;
+import io.vavr.control.Option;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.dcae.collectors.restconf.common.event.publishing.DMaaPPublishersCache.OnPublisherRemovalListener;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+import static io.vavr.API.List;
+import static io.vavr.API.Map;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class DMaaPPublishersCacheTest {
+
+    private String streamId1;
+    private Map<String, PublisherConfig> dMaaPConfigs;
+
+    @Before
+    public void setUp() {
+        streamId1 = "sampleStream1";
+        dMaaPConfigs = Map("sampleStream1", new PublisherConfig(List("destination1"), "topic1"));
+    }
+
+    @Test
+    public void shouldReturnTheSameCachedInstanceOnConsecutiveRetrievals() {
+        // given
+        DMaaPPublishersCache dMaaPPublishersCache = new DMaaPPublishersCache(dMaaPConfigs);
+
+        // when
+        Option<CambriaBatchingPublisher> firstPublisher = dMaaPPublishersCache.getPublisher(streamId1);
+        Option<CambriaBatchingPublisher> secondPublisher = dMaaPPublishersCache.getPublisher(streamId1);
+
+        // then
+        assertSame("should return same instance", firstPublisher.get(), secondPublisher.get());
+    }
+
+    @Test
+    public void shouldCloseCambriaPublisherOnCacheInvalidate() throws IOException, InterruptedException {
+        // given
+        CambriaBatchingPublisher cambriaPublisherMock1 = mock(CambriaBatchingPublisher.class);
+        DMaaPPublishersCache.CambriaPublishersCacheLoader cacheLoaderMock = mock(DMaaPPublishersCache.CambriaPublishersCacheLoader.class);
+        DMaaPPublishersCache dMaaPPublishersCache = new DMaaPPublishersCache(cacheLoaderMock,
+                                                                             new OnPublisherRemovalListener(),
+                                                                             dMaaPConfigs);
+        when(cacheLoaderMock.load(streamId1)).thenReturn(cambriaPublisherMock1);
+
+        // when
+        dMaaPPublishersCache.getPublisher(streamId1);
+        dMaaPPublishersCache.closePublisherFor(streamId1);
+
+        // then
+        verify(cambriaPublisherMock1).close(20, TimeUnit.SECONDS);
+
+    }
+
+    @Test
+    public void shouldReturnNoneIfThereIsNoDMaaPConfigurationForGivenStreamID() {
+        // given
+        DMaaPPublishersCache dMaaPPublishersCache = new DMaaPPublishersCache(dMaaPConfigs);
+
+        // then
+        assertTrue("should not exist", dMaaPPublishersCache.getPublisher("non-existing").isEmpty());
+    }
+
+}
\ No newline at end of file
diff --git a/src/test/java/org/onap/dcae/collectors/restconf/restconftest/AnyNodeTest.java b/src/test/java/org/onap/dcae/collectors/restconf/restconftest/AnyNodeTest.java
new file mode 100755
index 0000000..746023d
--- /dev/null
+++ b/src/test/java/org/onap/dcae/collectors/restconf/restconftest/AnyNodeTest.java
@@ -0,0 +1,62 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.restconf
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. 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.dcae.collectors.restconf.restconftest;
+
+import com.google.common.collect.Sets;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.onap.dcae.collectors.restconf.common.AnyNode;
+
+import java.util.Set;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class AnyNodeTest {
+
+    private static final String SAMPLE_JSON_FILEPATH = "{\n"
+            + "  \"channels\": [{\n"
+            + "    \"one\": \"number1\", \"two\": \"number2\", \"three\": \"number3\"}],\n"
+            + "  \"sampleStrList\": [\"1\", \"2\", \"3\", \"4\", \"5\"],\n"
+            + "  \"sampleNestedObject\": {\"a\": 1, \"b\": 2},\n"
+            + "  \"sampleInt\": 1,\n"
+            + "  \"sampleString\": \"str\",\n"
+            + "  \"sampleNull\": null\n"
+            + "}\n";
+    private static final Set<String> EXPECTED_JSON_KEYS = Sets
+            .newHashSet("channels", "sampleStrList", "sampleNestedObject", "sampleInt", "sampleString", "sampleNull");
+    private static AnyNode node;
+
+
+    @BeforeClass
+    public static void setUpClass() {
+        node = AnyNode.fromString(SAMPLE_JSON_FILEPATH);
+    }
+
+    @Test
+    public void testShouldReturnJsonObjectKeySet() {
+        assertThat(node.keys()).containsOnlyElementsOf(EXPECTED_JSON_KEYS);
+    }
+
+    @Test(expected = ClassCastException.class)
+    public void whenInvokedOnJsonObjInsteadOfJsonArrShouldRaiseRuntimeEx() {
+        node.toList();
+    }
+}
\ No newline at end of file
diff --git a/src/test/java/org/onap/dcae/collectors/restconf/restconftest/RestConfProcTest.java b/src/test/java/org/onap/dcae/collectors/restconf/restconftest/RestConfProcTest.java
new file mode 100755
index 0000000..0084f40
--- /dev/null
+++ b/src/test/java/org/onap/dcae/collectors/restconf/restconftest/RestConfProcTest.java
@@ -0,0 +1,75 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.restconf
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. 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.dcae.collectors.restconf.restconftest;
+
+import com.att.nsa.cmdLine.NsaCommandLineUtil;
+import com.att.nsa.drumlin.service.framework.DrumlinServlet;
+import com.att.nsa.drumlin.till.nv.impl.nvPropertiesFile;
+import com.att.nsa.drumlin.till.nv.impl.nvReadableStack;
+import com.att.nsa.drumlin.till.nv.impl.nvReadableTable;
+import org.glassfish.jersey.grizzly2.httpserver.GrizzlyHttpServerFactory;
+import org.glassfish.jersey.media.sse.SseFeature;
+import org.glassfish.jersey.server.ResourceConfig;
+import org.junit.Test;
+import org.onap.dcae.collectors.restconf.common.Constants;
+import org.onap.dcae.collectors.restconf.common.RestConfCollector;
+import org.onap.dcae.collectors.restconf.common.RestConfContext;
+import org.onap.dcae.collectors.restconf.common.RestConfProc;
+
+import java.net.URI;
+import java.net.URL;
+import java.util.HashMap;
+import java.util.Map;
+
+public class RestConfProcTest {
+
+    private static final URI CONTEXT = URI.create("http://localhost:8080/");
+
+    @Test
+    public void testEstablishPersistentConnection() throws Exception {
+
+        final Map<String, String> argMap = new HashMap<>();
+        final String config = NsaCommandLineUtil.getSetting(argMap, Constants.KCONFIG, "collector.properties");
+        final URL settingStream = DrumlinServlet.findStream(config, RestConfCollector.class);
+
+        final nvReadableStack settings = new nvReadableStack();
+
+        settings.push(new nvPropertiesFile(settingStream));
+        settings.push(new nvReadableTable(argMap));
+
+        RestConfProc restConfProc = new RestConfProc(settings);
+
+        final ResourceConfig resourceConfig = new ResourceConfig(SseResource.class, SseFeature.class);
+        GrizzlyHttpServerFactory.createHttpServer(CONTEXT, resourceConfig);
+        RestConfContext ctx = new RestConfContext();
+        ctx.setAttribute("prop.encoding-json", "encoding-json");
+        ctx.setAttribute("restapi-result.response-code", "200");
+        ctx.setAttribute("restapi-result.ietf-subscribed-notifications:output.identifier", "100");
+
+        Map<String, String> p = new HashMap<>();
+        p.put("sseConnectURL", "http://localhost:8080/ssevents");
+        p.put("subscriberId", "networkId");
+        p.put("responsePrefix", "restapi-result");
+
+        restConfProc.establishPersistentConnection(p, ctx);
+        Thread.sleep(1000);
+    }
+}
diff --git a/src/test/java/org/onap/dcae/collectors/restconf/restconftest/SseResource.java b/src/test/java/org/onap/dcae/collectors/restconf/restconftest/SseResource.java
new file mode 100755
index 0000000..db81886
--- /dev/null
+++ b/src/test/java/org/onap/dcae/collectors/restconf/restconftest/SseResource.java
@@ -0,0 +1,69 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.restconf
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. 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.dcae.collectors.restconf.restconftest;
+
+import org.glassfish.jersey.media.sse.EventOutput;
+import org.glassfish.jersey.media.sse.OutboundEvent;
+import org.glassfish.jersey.media.sse.SseFeature;
+
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import java.io.IOException;
+
+@Path("ssevents")
+public class SseResource {
+
+    @GET
+    @Produces(SseFeature.SERVER_SENT_EVENTS)
+    public EventOutput getServerSentEvents() throws IOException {
+        String data = "{"
+                + "\"ietf-notification:notification\" : {"
+                + "  \"eventTime\" : \"2017-10-25T08:22:33.44Z\","
+                + "    \"ietf-yang-push:push-change-update\": {"
+                + "\"subscription-id\":\"89\","
+                + "\"datastore-changes\": {"
+                + "\"ietf-yang-patch:yang-patch\":{"
+                + "\"patch-id\":\"1\","
+                + "\"edit\":[{"
+                + "\"edit-id\":\"edit1\","
+                + "\"operation\":\"merge\","
+                + "\"target\":\"/ietf-interfaces:interfaces-state\","
+                + "\"value\": {"
+                + "\"ietf-interfaces:interfaces-state\":{"
+                + "\"interface\": {"
+                + "\"name\":\"eth0\","
+                + "\"oper-status\":\"down\","
+                + "}"
+                + "}"
+                + "}"
+                + "}]"
+                + "}"
+                + "}"
+                + "}"
+                + "}"
+                + "}";
+        final EventOutput result = new EventOutput();
+        result.write(new OutboundEvent.Builder().data(String.class, data).build());
+        result.close();
+        return result;
+    }
+}
diff --git a/src/test/java/org/onap/dcae/collectors/restconf/restconftest/TestRestConfCollector.java b/src/test/java/org/onap/dcae/collectors/restconf/restconftest/TestRestConfCollector.java
new file mode 100755
index 0000000..6bedc8e
--- /dev/null
+++ b/src/test/java/org/onap/dcae/collectors/restconf/restconftest/TestRestConfCollector.java
@@ -0,0 +1,66 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.restconf
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. 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.dcae.collectors.restconf.restconftest;
+
+import com.att.nsa.cmdLine.NsaCommandLineUtil;
+import com.att.nsa.drumlin.till.nv.impl.nvReadableStack;
+import com.att.nsa.drumlin.till.nv.impl.nvReadableTable;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonParser;
+import org.json.JSONArray;
+import org.json.JSONObject;
+import org.junit.Test;
+import org.onap.dcae.collectors.restconf.common.Constants;
+import org.onap.dcae.collectors.restconf.common.RestConfProc;
+
+import java.io.FileReader;
+import java.util.Map;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.junit.Assert.assertEquals;
+
+public class TestRestConfCollector {
+    @Test
+    public void testParseCLIArguments() {
+        // given
+        String args[] = {"-a", "aa"};
+        Map<String, String> argMap = NsaCommandLineUtil.processCmdLine(args, true);
+        // when
+        nvReadableStack settings = new nvReadableStack();
+        settings.push(new nvReadableTable(argMap));
+
+        // then
+        assertEquals(settings.getString("a", "default"), "aa");
+    }
+
+    @Test
+    public void shouldPutValidRestConfEventOnProcessingQueueWithoutExceptions() throws Exception {
+        // given
+        RestConfProc.fProcessingInputQueue = new LinkedBlockingQueue<>(
+                Constants.KDEFAULT_MAXQUEUEDEVENTS);
+        JsonElement restConfEvent = new JsonParser().parse(new FileReader("src/test/resources/RestConfEvent.json"));
+        JSONObject validRestConfEvent = new JSONObject(restConfEvent.toString());
+        JSONArray jsonArrayMod = new JSONArray().put(validRestConfEvent);
+
+        // then
+        RestConfProc.handleEvents(jsonArrayMod);
+    }
+}
diff --git a/src/test/resources/RestConfEvent.json b/src/test/resources/RestConfEvent.json
new file mode 100755
index 0000000..93d0b64
--- /dev/null
+++ b/src/test/resources/RestConfEvent.json
@@ -0,0 +1,26 @@
+{
+    "ietf-notification:notification" : {
+        "eventTime" : "eventtime",
+        "ietf-yang-push:push-change-update": {
+            "subscription-id":"100",
+            "datastore-changes": {
+                "ietf-yang-patch:yang-patch":{
+                    "patch-id":"patch-id",
+                    "edit":[{
+                        "edit-id":"edit-id",
+                        "operation":"create",
+                        "target":"target",
+                        "value": {
+                            "ietf-interfaces:interfaces-state":{
+                                "interface": {
+                                    "name":"eth0",
+                                    "oper-status":"up"
+                                }
+                            }
+                        }
+                    }]
+                }
+            }
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/test/resources/testParseDMaaPCredentialsGen2.json b/src/test/resources/testParseDMaaPCredentialsGen2.json
new file mode 100755
index 0000000..23230c1
--- /dev/null
+++ b/src/test/resources/testParseDMaaPCredentialsGen2.json
@@ -0,0 +1,21 @@
+{
+  "auth-credentials-null": {
+    "aaf_username": null,
+    "dmaap_info": {
+      "topic_url": "http://UEBHOST:3904/events/DCAE-RESTCONF-COLLECTOR-EVENTS-DEV"
+    },
+    "aaf_password": null
+  },
+  "auth-credentials-present": {
+    "aaf_username": "sampleUser",
+    "dmaap_info": {
+      "topic_url": "http://UEBHOST:3904/events/DCAE-RESTCONF-COLLECTOR-EVENTS-DEV"
+    },
+    "aaf_password": "samplePassword"
+  },
+  "auth-credentials-missing": {
+    "dmaap_info": {
+      "topic_url": "http://UEBHOST:3904/events/DCAE-RESTCONF-COLLECTOR-EVENTS-DEV"
+    }
+  }
+}
\ No newline at end of file
diff --git a/src/test/resources/testParseDMaaPCredentialsLegacy.json b/src/test/resources/testParseDMaaPCredentialsLegacy.json
new file mode 100755
index 0000000..ee215e2
--- /dev/null
+++ b/src/test/resources/testParseDMaaPCredentialsLegacy.json
@@ -0,0 +1,26 @@
+{
+  "channels": [
+    {
+      "name": "auth-credentials-null",
+      "cambria.url": "127.0.0.1:3904",
+      "cambria.hosts": "uebsb91kcdc.it.att.com,uebsb92kcdc.it.att.com,uebsb93kcdc.it.att.com",
+      "cambria.topic": "DCAE-RESTCONF-COLLECTOR-EVENTS-DEV",
+      "basicAuthPassword": null,
+      "basicAuthUsername": null
+    },
+    {
+      "name": "auth-credentials-present",
+      "cambria.url": "127.0.0.1:3904",
+      "cambria.hosts": "uebsb91kcdc.it.att.com,uebsb92kcdc.it.att.com,uebsb93kcdc.it.att.com",
+      "cambria.topic": "DCAE-RESTCONF-COLLECTOR-EVENTS-DEV",
+      "basicAuthPassword": "samplePassword",
+      "basicAuthUsername": "sampleUser"
+    },
+    {
+      "name": "auth-credentials-missing",
+      "cambria.url": "127.0.0.1:3904",
+      "cambria.hosts": "uebsb91kcdc.it.att.com,uebsb92kcdc.it.att.com,uebsb93kcdc.it.att.com",
+      "cambria.topic": "DCAE-RESTCONF-COLLECTOR-EVENTS-DEV"
+    }
+  ]
+}
\ No newline at end of file
diff --git a/src/test/resources/testParseDMaaPGen2.json b/src/test/resources/testParseDMaaPGen2.json
new file mode 100755
index 0000000..4ef04a2
--- /dev/null
+++ b/src/test/resources/testParseDMaaPGen2.json
@@ -0,0 +1,12 @@
+{
+  "event-segments-with-port": {
+    "dmaap_info": {
+      "topic_url": "http://UEBHOST:3904/events/DCAE-RESTCONF-COLLECTOR-EVENTS-DEV"
+    }
+  },
+  "other-segments-without-ports": {
+    "dmaap_info": {
+      "topic_url": "http://UEBHOST:3904/somethingHere/DCAE-RESTCONF-COLLECTOR-EVENTS-DEV"
+    }
+  }
+}
\ No newline at end of file
diff --git a/src/test/resources/testParseDMaaPLegacy.json b/src/test/resources/testParseDMaaPLegacy.json
new file mode 100755
index 0000000..fda9c60
--- /dev/null
+++ b/src/test/resources/testParseDMaaPLegacy.json
@@ -0,0 +1,21 @@
+{
+  "channels": [
+    {
+      "name": "url-precedes-hosts",
+      "cambria.url": "127.0.0.1:3904",
+      "cambria.hosts": "h1.att.com,h2.att.com",
+      "cambria.topic": "DCAE-RESTCONF-COLLECTOR-EVENTS-DEV"
+    },
+    {
+      "name": "url-key-missing",
+      "cambria.hosts": "h1.att.com,h2.att.com",
+      "cambria.topic": "DCAE-RESTCONF-COLLECTOR-EVENTS-DEV"
+    },
+    {
+      "name": "url-is-null",
+      "cambria.url": null,
+      "cambria.hosts": "h1.att.com,h2.att.com",
+      "cambria.topic": "DCAE-RESTCONF-COLLECTOR-EVENTS-DEV"
+    }
+  ]
+}
\ No newline at end of file