Adding NOOP sources support

In addition, Noop* classes have been refactored to
increase code reuse and clean some checkstyle issues.
Additional Junits have been added for existing functionality.

Change-Id: I072f9ff2f415630ac82eca949a8360249f73da86
Issue-ID: POLICY-1397
Signed-off-by: Jorge Hernandez <jorge.hernandez-herrero@att.com>

8 files changed, 803 insertions, 271 deletions

diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxyTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxyTest.java
new file mode 100644
index 00000000..19dde432
--- /dev/null
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxyTest.java
@@ -0,0 +1,358 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2019 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.policy.common.endpoints.event.comm;
+
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import java.util.List;
+import java.util.Properties;
+import org.junit.Test;
+import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
+import org.onap.policy.common.endpoints.event.comm.bus.DmaapTopicPropertyBuilder;
+import org.onap.policy.common.endpoints.event.comm.bus.NoopTopicPropertyBuilder;
+import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
+
+public class TopicEndpointProxyTest {
+
+    private static final String NOOP_SOURCE_TOPIC = "noop-source";
+    private static final String NOOP_SINK_TOPIC = "noop-sink";
+
+    private static final String UEB_SOURCE_TOPIC = "ueb-source";
+    private static final String UEB_SINK_TOPIC = "ueb-sink";
+
+    private static final String DMAAP_SOURCE_TOPIC = "dmaap-source";
+    private static final String DMAAP_SINK_TOPIC = "dmaap-sink";
+
+    private Properties configuration = new Properties();
+
+    /**
+     * Constructor.
+     */
+    public TopicEndpointProxyTest() {
+        Properties noopSourceProperties =
+            new NoopTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_NOOP_SOURCE_TOPICS)
+                .makeTopic(NOOP_SOURCE_TOPIC).build();
+
+        Properties noopSinkProperties =
+            new NoopTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_NOOP_SINK_TOPICS)
+                .makeTopic(NOOP_SINK_TOPIC).build();
+
+        Properties uebSourceProperties =
+            new NoopTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS)
+                .makeTopic(UEB_SOURCE_TOPIC).build();
+
+        Properties uebSinkProperties =
+            new NoopTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS)
+                .makeTopic(UEB_SINK_TOPIC).build();
+
+        Properties dmaapSourceProperties =
+            new DmaapTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS)
+                .makeTopic(DMAAP_SOURCE_TOPIC).build();
+
+        Properties dmaapSinkProperties =
+            new DmaapTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS)
+                .makeTopic(DMAAP_SINK_TOPIC).build();
+
+        configuration.putAll(noopSourceProperties);
+        configuration.putAll(noopSinkProperties);
+        configuration.putAll(uebSourceProperties);
+        configuration.putAll(uebSinkProperties);
+        configuration.putAll(dmaapSourceProperties);
+        configuration.putAll(dmaapSinkProperties);
+    }
+
+    private <T extends Topic> boolean exists(List<T> topics, String topicName) {
+        return topics.stream().map(Topic::getTopic).anyMatch(topicName::equals);
+    }
+
+    private <T extends Topic> boolean allSources(List<T> topics) {
+        return exists(topics, NOOP_SOURCE_TOPIC)
+            && exists(topics, UEB_SOURCE_TOPIC)
+            && exists(topics, DMAAP_SOURCE_TOPIC);
+    }
+
+    private <T extends Topic> boolean allSinks(List<T> topics) {
+        return exists(topics, NOOP_SINK_TOPIC)
+            && exists(topics, UEB_SINK_TOPIC)
+            && exists(topics, DMAAP_SINK_TOPIC);
+    }
+
+    private <T extends Topic> boolean anySource(List<T> topics) {
+        return exists(topics, NOOP_SOURCE_TOPIC)
+            || exists(topics, UEB_SOURCE_TOPIC)
+            || exists(topics, DMAAP_SOURCE_TOPIC);
+    }
+
+    private <T extends Topic> boolean anySink(List<T> topics) {
+        return exists(topics, NOOP_SINK_TOPIC)
+            || exists(topics, UEB_SINK_TOPIC)
+            || exists(topics, DMAAP_SINK_TOPIC);
+    }
+
+    @Test
+    public void addTopicSources() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+
+        List<TopicSource>  sources = manager.addTopicSources(configuration);
+        assertSame(3, sources.size());
+
+        assertTrue(allSources(sources));
+        assertFalse(anySink(sources));
+    }
+
+    @Test
+    public void addTopicSinks() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+
+        List<TopicSink>  sinks = manager.addTopicSinks(configuration);
+        assertSame(3, sinks.size());
+
+        assertFalse(anySource(sinks));
+        assertTrue(allSinks(sinks));
+    }
+
+    @Test
+    public void getTopicSources() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+
+        manager.addTopicSources(configuration);
+        manager.addTopicSinks(configuration);
+
+        List<TopicSource>  sources = manager.getTopicSources();
+        assertSame(3, sources.size());
+
+        assertTrue(allSources(sources));
+        assertFalse(anySink(sources));
+    }
+
+    @Test
+    public void getTopicSinks() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+
+        manager.addTopicSources(configuration);
+        manager.addTopicSinks(configuration);
+
+        List<TopicSink>  sinks = manager.getTopicSinks();
+        assertSame(3, sinks.size());
+
+        assertFalse(anySource(sinks));
+        assertTrue(allSinks(sinks));
+    }
+
+    @Test
+    public void getUebTopicSources() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+
+        manager.addTopicSources(configuration);
+        assertSame(1, manager.getUebTopicSources().size());
+    }
+
+    @Test
+    public void getDmaapTopicSources() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+
+        manager.addTopicSources(configuration);
+        assertSame(1, manager.getDmaapTopicSources().size());
+    }
+
+    @Test
+    public void getNoopTopicSources() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+
+        manager.addTopicSources(configuration);
+        assertSame(1, manager.getNoopTopicSources().size());
+    }
+
+    @Test
+    public void getUebTopicSinks() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+
+        manager.addTopicSinks(configuration);
+        assertSame(1, manager.getUebTopicSinks().size());
+    }
+
+    @Test
+    public void getDmaapTopicSinks() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+
+        manager.addTopicSinks(configuration);
+        assertSame(1, manager.getDmaapTopicSinks().size());
+    }
+
+    @Test
+    public void getNoopTopicSinks() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+
+        manager.addTopicSinks(configuration);
+        assertSame(1, manager.getNoopTopicSinks().size());
+    }
+
+    @Test
+    public void lifecycle() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+
+        assertTrue(manager.start());
+        assertTrue(manager.isAlive());
+
+        assertTrue(manager.stop());
+        assertFalse(manager.isAlive());
+
+        assertTrue(manager.start());
+        assertTrue(manager.isAlive());
+
+        manager.shutdown();
+        assertFalse(manager.isAlive());
+    }
+
+    @Test
+    public void lock() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+
+        manager.lock();
+        assertTrue(manager.isLocked());
+
+        manager.unlock();
+        assertFalse(manager.isLocked());
+    }
+
+    @Test
+    public void getTopicSource() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+        manager.addTopicSources(configuration);
+
+        assertSame(NOOP_SOURCE_TOPIC, manager.getTopicSource(CommInfrastructure.NOOP, NOOP_SOURCE_TOPIC).getTopic());
+        assertSame(UEB_SOURCE_TOPIC, manager.getTopicSource(CommInfrastructure.UEB, UEB_SOURCE_TOPIC).getTopic());
+        assertSame(DMAAP_SOURCE_TOPIC, manager.getTopicSource(CommInfrastructure.DMAAP, DMAAP_SOURCE_TOPIC).getTopic());
+
+        assertThatIllegalStateException()
+            .isThrownBy(() -> manager.getTopicSource(CommInfrastructure.NOOP, NOOP_SINK_TOPIC));
+        assertThatIllegalStateException()
+            .isThrownBy(() -> manager.getTopicSource(CommInfrastructure.UEB, UEB_SINK_TOPIC));
+        assertThatIllegalStateException()
+            .isThrownBy(() -> manager.getTopicSource(CommInfrastructure.DMAAP, DMAAP_SINK_TOPIC));
+    }
+
+    @Test
+    public void getTopicSink() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+        manager.addTopicSinks(configuration);
+
+        assertSame(NOOP_SINK_TOPIC, manager.getTopicSink(CommInfrastructure.NOOP, NOOP_SINK_TOPIC).getTopic());
+        assertSame(UEB_SINK_TOPIC, manager.getTopicSink(CommInfrastructure.UEB, UEB_SINK_TOPIC).getTopic());
+        assertSame(DMAAP_SINK_TOPIC, manager.getTopicSink(CommInfrastructure.DMAAP, DMAAP_SINK_TOPIC).getTopic());
+
+        assertThatIllegalStateException()
+            .isThrownBy(() -> manager.getTopicSink(CommInfrastructure.NOOP, NOOP_SOURCE_TOPIC));
+        assertThatIllegalStateException()
+            .isThrownBy(() -> manager.getTopicSink(CommInfrastructure.UEB, UEB_SOURCE_TOPIC));
+        assertThatIllegalStateException()
+            .isThrownBy(() -> manager.getTopicSink(CommInfrastructure.DMAAP, DMAAP_SOURCE_TOPIC));
+    }
+
+    @Test
+    public void getUebTopicSource() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+        manager.addTopicSources(configuration);
+
+        assertSame(UEB_SOURCE_TOPIC, manager.getUebTopicSource(UEB_SOURCE_TOPIC).getTopic());
+
+        assertThatIllegalStateException().isThrownBy(() -> manager.getUebTopicSource(NOOP_SOURCE_TOPIC));
+        assertThatIllegalStateException().isThrownBy(() -> manager.getUebTopicSource(DMAAP_SOURCE_TOPIC));
+
+        assertThatIllegalArgumentException().isThrownBy(() -> manager.getUebTopicSource(null));
+        assertThatIllegalArgumentException().isThrownBy(() -> manager.getUebTopicSource(""));
+    }
+
+    @Test
+    public void getUebTopicSink() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+        manager.addTopicSinks(configuration);
+
+        assertSame(UEB_SINK_TOPIC, manager.getUebTopicSink(UEB_SINK_TOPIC).getTopic());
+
+        assertThatIllegalStateException().isThrownBy(() -> manager.getUebTopicSink(NOOP_SINK_TOPIC));
+        assertThatIllegalStateException().isThrownBy(() -> manager.getUebTopicSink(DMAAP_SINK_TOPIC));
+
+        assertThatIllegalArgumentException().isThrownBy(() -> manager.getUebTopicSink(null));
+        assertThatIllegalArgumentException().isThrownBy(() -> manager.getUebTopicSink(""));
+    }
+
+    @Test
+    public void getDmaapTopicSource() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+        manager.addTopicSources(configuration);
+
+        assertSame(DMAAP_SOURCE_TOPIC, manager.getDmaapTopicSource(DMAAP_SOURCE_TOPIC).getTopic());
+
+        assertThatIllegalStateException().isThrownBy(() -> manager.getDmaapTopicSource(NOOP_SOURCE_TOPIC));
+        assertThatIllegalStateException().isThrownBy(() -> manager.getDmaapTopicSource(UEB_SOURCE_TOPIC));
+
+        assertThatIllegalArgumentException().isThrownBy(() -> manager.getDmaapTopicSource(null));
+        assertThatIllegalArgumentException().isThrownBy(() -> manager.getDmaapTopicSource(""));
+    }
+
+    @Test
+    public void getDmaapTopicSink() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+        manager.addTopicSinks(configuration);
+
+        assertSame(DMAAP_SINK_TOPIC, manager.getDmaapTopicSink(DMAAP_SINK_TOPIC).getTopic());
+
+        assertThatIllegalStateException().isThrownBy(() -> manager.getDmaapTopicSink(NOOP_SINK_TOPIC));
+        assertThatIllegalStateException().isThrownBy(() -> manager.getDmaapTopicSink(UEB_SINK_TOPIC));
+
+        assertThatIllegalArgumentException().isThrownBy(() -> manager.getDmaapTopicSink(null));
+        assertThatIllegalArgumentException().isThrownBy(() -> manager.getDmaapTopicSink(""));
+    }
+
+
+    @Test
+    public void getNoopTopicSource() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+        manager.addTopicSources(configuration);
+
+        assertSame(NOOP_SOURCE_TOPIC, manager.getNoopTopicSource(NOOP_SOURCE_TOPIC).getTopic());
+
+        assertThatIllegalStateException().isThrownBy(() -> manager.getNoopTopicSource(DMAAP_SOURCE_TOPIC));
+        assertThatIllegalStateException().isThrownBy(() -> manager.getNoopTopicSource(UEB_SOURCE_TOPIC));
+
+        assertThatIllegalArgumentException().isThrownBy(() -> manager.getNoopTopicSource(null));
+        assertThatIllegalArgumentException().isThrownBy(() -> manager.getNoopTopicSource(""));
+    }
+
+    @Test
+    public void getNoopTopicSink() {
+        TopicEndpoint manager = new TopicEndpointProxy();
+        manager.addTopicSinks(configuration);
+
+        assertSame(NOOP_SINK_TOPIC, manager.getNoopTopicSink(NOOP_SINK_TOPIC).getTopic());
+
+        assertThatIllegalStateException().isThrownBy(() -> manager.getNoopTopicSink(DMAAP_SINK_TOPIC));
+        assertThatIllegalStateException().isThrownBy(() -> manager.getNoopTopicSink(UEB_SINK_TOPIC));
+
+        assertThatIllegalArgumentException().isThrownBy(() -> manager.getNoopTopicSink(null));
+        assertThatIllegalArgumentException().isThrownBy(() -> manager.getNoopTopicSink(""));
+    }
+}
\ No newline at end of file
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicFactoryTestBase.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicFactoryTestBase.java
index 877246e2..67304f91 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicFactoryTestBase.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicFactoryTestBase.java
@@ -21,6 +21,7 @@
 package org.onap.policy.common.endpoints.event.comm.bus;
 
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
 import static org.junit.Assert.assertEquals;
 import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_CONN_TIMEOUT_SUFFIX;
 import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_READ_TIMEOUT_MS_SUFFIX;
@@ -126,6 +127,6 @@ public abstract class DmaapTopicFactoryTestBase<T extends Topic> extends BusTopi
         initFactory();
         buildTopics(makePropBuilder().makeTopic(MY_TOPIC).build());
 
-        assertThatIllegalArgumentException().as("unknown topic").isThrownBy(() -> getTopic(TOPIC2));
+        assertThatIllegalStateException().as("unknown topic").isThrownBy(() -> getTopic(TOPIC2));
     }
 }
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicEndpointTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicEndpointTest.java
new file mode 100644
index 00000000..5bbc80cd
--- /dev/null
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicEndpointTest.java
@@ -0,0 +1,117 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2019 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.policy.common.endpoints.event.comm.bus;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+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 java.util.Arrays;
+import java.util.Collections;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
+import org.onap.policy.common.endpoints.event.comm.TopicListener;
+
+public abstract class NoopTopicEndpointTest<F extends NoopTopicFactory<T>, T extends NoopTopicEndpoint>
+    extends TopicTestBase {
+
+    protected final F factory;
+    protected T endpoint;
+
+    protected abstract boolean io(String message);
+
+    public NoopTopicEndpointTest(F factory) {
+        this.factory = factory;
+    }
+
+    @Before
+    public void setUp() {
+        super.setUp();
+        this.endpoint = this.factory.build(servers, MY_TOPIC);
+    }
+
+    @Test
+    public void tesIo() {
+        TopicListener listener = mock(TopicListener.class);
+        this.endpoint.register(listener);
+        this.endpoint.start();
+
+        assertTrue(io(MY_MESSAGE));
+        assertSame(MY_MESSAGE, this.endpoint.getRecentEvents()[0]);
+        assertEquals(Collections.singletonList(MY_MESSAGE), Arrays.asList(this.endpoint.getRecentEvents()));
+        verify(listener).onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MY_MESSAGE);
+
+        this.endpoint.unregister(listener);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testIoNullMessage() {
+        io(null);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testIoEmptyMessage() {
+        io("");
+    }
+
+    @Test(expected = IllegalStateException.class)
+    public void testOfferNotStarted() {
+        io(MY_MESSAGE);
+    }
+
+    @Test
+    public void testGetTopicCommInfrastructure() {
+        assertEquals(CommInfrastructure.NOOP, this.endpoint.getTopicCommInfrastructure());
+    }
+
+    @Test
+    public void testStart_testStop_testShutdown() {
+        this.endpoint.start();
+        assertTrue(this.endpoint.isAlive());
+
+        // start again
+        this.endpoint.start();
+        assertTrue(this.endpoint.isAlive());
+
+        // stop
+        this.endpoint.stop();
+        assertFalse(this.endpoint.isAlive());
+
+        // re-start again
+        this.endpoint.start();
+        assertTrue(this.endpoint.isAlive());
+
+        // shutdown
+        this.endpoint.shutdown();
+        assertFalse(this.endpoint.isAlive());
+    }
+
+    @Test(expected = IllegalStateException.class)
+    public void testStart_Locked() {
+        this.endpoint.lock();
+        this.endpoint.start();
+    }
+
+}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicFactoryTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicFactoryTest.java
new file mode 100644
index 00000000..16d9e539
--- /dev/null
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicFactoryTest.java
@@ -0,0 +1,224 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2019 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.policy.common.endpoints.event.comm.bus;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import org.apache.commons.lang3.RandomStringUtils;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
+
+public abstract class NoopTopicFactoryTest<F extends NoopTopicFactory<T>, T extends NoopTopicEndpoint>
+    extends TopicFactoryTestBase<T> {
+
+    private static final List<String> NOOP_SERVERS = Arrays.asList(CommInfrastructure.NOOP.toString());
+    private F factory = null;
+
+    protected abstract F buildFactory();
+
+    /**
+     * Creates the object to be tested.
+     */
+    @Before
+    public void setUp() {
+        super.setUp();
+        initFactory();
+    }
+
+    @After
+    public void tearDown() {
+        factory.destroy();
+    }
+
+    @Test
+    public void testBuildListOfStringStringBoolean() {
+        initFactory();
+
+        T item1 = buildTopic(servers, MY_TOPIC, true);
+        assertNotNull(item1);
+
+        assertEquals(servers, item1.getServers());
+        assertEquals(MY_TOPIC, item1.getTopic());
+
+        // managed topic - should not build a new one
+        assertEquals(item1, buildTopic(servers, MY_TOPIC, true));
+
+        T item2 = buildTopic(servers, TOPIC2, true);
+        assertNotNull(item2);
+        assertNotSame(item1, item2);
+
+        // duplicate - should be the same, as these topics are managed
+        List<String> randomServers = new ArrayList<>();
+        randomServers.add(RandomStringUtils.randomAlphanumeric(8));
+        T item3 = buildTopic(randomServers, TOPIC2, true);
+        assertSame(item2, item3);
+
+        T item4 = buildTopic(Collections.emptyList(), TOPIC2, true);
+        assertSame(item3, item4);
+
+        // null server list
+        initFactory();
+        assertEquals(NOOP_SERVERS, buildTopic(null, MY_TOPIC, true).getServers());
+
+        // empty server list
+        initFactory();
+        assertEquals(NOOP_SERVERS, buildTopic(Collections.emptyList(), MY_TOPIC, true).getServers());
+
+        // unmanaged topic
+        initFactory();
+        item1 = buildTopic(servers, MY_TOPIC, false);
+        assertNotSame(item1, buildTopic(servers, MY_TOPIC, false));
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testBuildListOfStringStringBoolean_NullTopic() {
+        buildTopic(servers, null, true);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testBuildListOfStringStringBoolean_EmptyTopic() {
+        buildTopic(servers, "", true);
+    }
+
+    @Test
+    public void testBuildProperties() {
+        // managed topic
+        initFactory();
+        assertEquals(1, buildTopics(makePropBuilder().makeTopic(MY_TOPIC).build()).size());
+        assertNotNull(factory.get(MY_TOPIC));
+
+        // unmanaged topic - get() will throw an exception
+        initFactory();
+        assertEquals(1, buildTopics(makePropBuilder().makeTopic(MY_TOPIC)
+                        .setTopicProperty(PROPERTY_MANAGED_SUFFIX, "false").build()).size());
+        assertThatThrownBy(() -> factory.get(MY_TOPIC));
+
+        // managed undefined - default to true
+        initFactory();
+        assertEquals(1, buildTopics(
+                        makePropBuilder().makeTopic(MY_TOPIC).removeTopicProperty(PROPERTY_MANAGED_SUFFIX).build())
+                                        .size());
+        assertNotNull(factory.get(MY_TOPIC));
+
+        // managed empty - default to true
+        initFactory();
+        assertEquals(1, buildTopics(
+                        makePropBuilder().makeTopic(MY_TOPIC).setTopicProperty(PROPERTY_MANAGED_SUFFIX, "").build())
+                                        .size());
+        assertNotNull(factory.get(MY_TOPIC));
+
+        initFactory();
+
+        // null topic list
+        assertTrue(buildTopics(makePropBuilder().build()).isEmpty());
+
+        // empty topic list
+        assertTrue(buildTopics(makePropBuilder().addTopic("").build()).isEmpty());
+
+        // null server list
+        initFactory();
+        T endpoint = buildTopics(makePropBuilder().makeTopic(MY_TOPIC)
+                        .removeTopicProperty(PROPERTY_TOPIC_SERVERS_SUFFIX).build()).get(0);
+        assertEquals(NOOP_SERVERS, endpoint.getServers());
+
+        // empty server list
+        initFactory();
+        endpoint = buildTopics(makePropBuilder().makeTopic(MY_TOPIC).setTopicProperty(PROPERTY_TOPIC_SERVERS_SUFFIX, "")
+                        .build()).get(0);
+        assertEquals(NOOP_SERVERS, endpoint.getServers());
+
+        // test other options
+        super.testBuildProperties_Multiple();
+    }
+
+    @Test
+    public void testDestroyString_testGet_testInventory() {
+        super.testDestroyString_testGet_testInventory();
+        super.testDestroyString_Ex();
+    }
+
+    @Test
+    public void testDestroy() {
+        super.testDestroy();
+    }
+
+    @Test
+    public void testGet() {
+        super.testGet_Ex();
+    }
+
+    @Override
+    protected void initFactory() {
+        if (factory != null) {
+            factory.destroy();
+        }
+
+        factory = buildFactory();
+    }
+
+    @Override
+    protected List<T> buildTopics(Properties properties) {
+        return factory.build(properties);
+    }
+
+    protected T buildTopic(List<String> servers, String topic, boolean managed) {
+        return factory.build(servers, topic, managed);
+    }
+
+    @Override
+    protected void destroyFactory() {
+        factory.destroy();
+    }
+
+    @Override
+    protected void destroyTopic(String topic) {
+        factory.destroy(topic);
+    }
+
+    @Override
+    protected List<T> getInventory() {
+        return factory.inventory();
+    }
+
+    @Override
+    protected T getTopic(String topic) {
+        return factory.get(topic);
+    }
+
+    @Override
+    protected TopicPropertyBuilder makePropBuilder() {
+        return new NoopTopicPropertyBuilder(factory.getTopicsPropertyName());
+    }
+}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkFactoryTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkFactoryTest.java
index 82a9df4c..cc44716f 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkFactoryTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkFactoryTest.java
@@ -1,6 +1,6 @@
 /*
  * ============LICENSE_START=======================================================
- * policy-endpoints
+ * ONAP
  * ================================================================================
  * Copyright (C) 2018-2019 AT&T Intellectual Property. All rights reserved.
  * ================================================================================
@@ -20,195 +20,10 @@
 
 package org.onap.policy.common.endpoints.event.comm.bus;
 
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_NOOP_SINK_TOPICS;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX;
-
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.Properties;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
-public class NoopTopicSinkFactoryTest extends TopicFactoryTestBase<NoopTopicSink> {
-
-    private static final List<String> NOOP_SERVERS = Arrays.asList("noop");
-
-    private IndexedNoopTopicSinkFactory factory;
-
-    /**
-     * Creates the object to be tested.
-     */
-    @Before
-    public void setUp() {
-        super.setUp();
-
-        factory = new IndexedNoopTopicSinkFactory();
-    }
-
-    @After
-    public void tearDown() {
-        factory.destroy();
-    }
-
-    @Test
-    public void testBuildListOfStringStringBoolean() {
-        initFactory();
-
-        NoopTopicSink item1 = buildTopic(servers, MY_TOPIC, true);
-        assertNotNull(item1);
-
-        assertEquals(servers, item1.getServers());
-        assertEquals(MY_TOPIC, item1.getTopic());
-
-        // managed topic - should not build a new one
-        assertEquals(item1, buildTopic(servers, MY_TOPIC, true));
-
-        NoopTopicSink item2 = buildTopic(servers, TOPIC2, true);
-        assertNotNull(item2);
-        assertTrue(item1 != item2);
-
-        // duplicate - should be the same, as these topics are managed
-        NoopTopicSink item3 = buildTopic(Collections.emptyList(), TOPIC2, true);
-        assertTrue(item2 == item3);
-
-        // null server list
-        initFactory();
-        assertEquals(NOOP_SERVERS, buildTopic(null, MY_TOPIC, true).getServers());
-
-        // empty server list
-        initFactory();
-        assertEquals(NOOP_SERVERS, buildTopic(Collections.emptyList(), MY_TOPIC, true).getServers());
-
-        // unmanaged topic
-        initFactory();
-        item1 = buildTopic(servers, MY_TOPIC, false);
-        assertTrue(item1 != buildTopic(servers, MY_TOPIC, false));
-    }
-
-    @Test(expected = IllegalArgumentException.class)
-    public void testBuildListOfStringStringBoolean_NullTopic() {
-        buildTopic(servers, null, true);
-    }
-
-    @Test(expected = IllegalArgumentException.class)
-    public void testBuildListOfStringStringBoolean_EmptyTopic() {
-        buildTopic(servers, "", true);
-    }
-
-    @Test
-    public void testBuildProperties() {
-        // managed topic
-        initFactory();
-        assertEquals(1, buildTopics(makePropBuilder().makeTopic(MY_TOPIC).build()).size());
-        assertNotNull(factory.get(MY_TOPIC));
-
-        // unmanaged topic - get() will throw an exception
-        initFactory();
-        assertEquals(1, buildTopics(makePropBuilder().makeTopic(MY_TOPIC)
-                        .setTopicProperty(PROPERTY_MANAGED_SUFFIX, "false").build()).size());
-        assertThatThrownBy(() -> factory.get(MY_TOPIC));
-
-        // managed undefined - default to true
-        initFactory();
-        assertEquals(1, buildTopics(
-                        makePropBuilder().makeTopic(MY_TOPIC).removeTopicProperty(PROPERTY_MANAGED_SUFFIX).build())
-                                        .size());
-        assertNotNull(factory.get(MY_TOPIC));
-
-        // managed empty - default to true
-        initFactory();
-        assertEquals(1, buildTopics(
-                        makePropBuilder().makeTopic(MY_TOPIC).setTopicProperty(PROPERTY_MANAGED_SUFFIX, "").build())
-                                        .size());
-        assertNotNull(factory.get(MY_TOPIC));
-
-        initFactory();
-
-        // null topic list
-        assertTrue(buildTopics(makePropBuilder().build()).isEmpty());
-
-        // empty topic list
-        assertTrue(buildTopics(makePropBuilder().addTopic("").build()).isEmpty());
-
-        // null server list
-        initFactory();
-        NoopTopicSink sink = buildTopics(makePropBuilder().makeTopic(MY_TOPIC)
-                        .removeTopicProperty(PROPERTY_TOPIC_SERVERS_SUFFIX).build()).get(0);
-        assertEquals(NOOP_SERVERS, sink.getServers());
-
-        // empty server list
-        initFactory();
-        sink = buildTopics(makePropBuilder().makeTopic(MY_TOPIC).setTopicProperty(PROPERTY_TOPIC_SERVERS_SUFFIX, "")
-                        .build()).get(0);
-        assertEquals(NOOP_SERVERS, sink.getServers());
-
-        // test other options
-        super.testBuildProperties_Multiple();
-    }
-
-    @Test
-    public void testDestroyString_testGet_testInventory() {
-        super.testDestroyString_testGet_testInventory();
-        super.testDestroyString_Ex();
-    }
-
-    @Test
-    public void testDestroy() {
-        super.testDestroy();
-    }
-
-    @Test
-    public void testGet() {
-        super.testGet_Ex();
-    }
-
-    @Override
-    protected void initFactory() {
-        if (factory != null) {
-            factory.destroy();
-        }
-
-        factory = new IndexedNoopTopicSinkFactory();
-    }
-
-    @Override
-    protected List<NoopTopicSink> buildTopics(Properties properties) {
-        return factory.build(properties);
-    }
-
-    protected NoopTopicSink buildTopic(List<String> servers, String topic, boolean managed) {
-        return factory.build(servers, topic, managed);
-    }
-
-    @Override
-    protected void destroyFactory() {
-        factory.destroy();
-    }
-
-    @Override
-    protected void destroyTopic(String topic) {
-        factory.destroy(topic);
-    }
-
-    @Override
-    protected List<NoopTopicSink> getInventory() {
-        return factory.inventory();
-    }
-
-    @Override
-    protected NoopTopicSink getTopic(String topic) {
-        return factory.get(topic);
-    }
+public class NoopTopicSinkFactoryTest extends NoopTopicFactoryTest<NoopTopicSinkFactory, NoopTopicSink> {
 
     @Override
-    protected TopicPropertyBuilder makePropBuilder() {
-        return new NoopTopicPropertyBuilder(PROPERTY_NOOP_SINK_TOPICS);
+    protected NoopTopicSinkFactory buildFactory() {
+        return new NoopTopicSinkFactory();
     }
 }
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkTest.java
index 8a5b7b2e..02478367 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkTest.java
@@ -1,8 +1,8 @@
 /*
  * ============LICENSE_START=======================================================
- * policy-endpoints
+ * ONAP
  * ================================================================================
- * Copyright (C) 2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2019 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.
@@ -20,50 +20,30 @@
 
 package org.onap.policy.common.endpoints.event.comm.bus;
 
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
 
-import java.util.Arrays;
-import org.junit.Before;
 import org.junit.Test;
-import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
-import org.onap.policy.common.endpoints.event.comm.TopicListener;
 
-public class NoopTopicSinkTest extends TopicTestBase {
+public class NoopTopicSinkTest extends NoopTopicEndpointTest<NoopTopicSinkFactory, NoopTopicSink> {
 
-    private NoopTopicSink sink;
-
-    /**
-     * Creates the object to be tested.
-     */
-    @Before
-    public void setUp() {
-        super.setUp();
+    public NoopTopicSinkTest() {
+        super(new NoopTopicSinkFactory());
+    }
 
-        sink = new NoopTopicSink(servers, MY_TOPIC);
+    @Override
+    protected boolean io(String message) {
+        return endpoint.send(message);
     }
 
     @Test
     public void testToString() {
-        assertTrue(sink.toString().startsWith("NoopTopicSink ["));
+        assertThat(endpoint.toString()).startsWith("NoopTopicSink");
     }
 
     @Test
     public void testSend() {
-        TopicListener listener = mock(TopicListener.class);
-        sink.register(listener);
-        sink.start();
-
-        assertTrue(sink.send(MY_MESSAGE));
-
-        assertEquals(Arrays.asList(MY_MESSAGE), Arrays.asList(sink.getRecentEvents()));
-        verify(listener).onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MY_MESSAGE);
-
-        // generate exception during broadcast
-        sink = new NoopTopicSink(servers, MY_TOPIC) {
+        NoopTopicSink sink = new NoopTopicSink(servers, MY_TOPIC) {
             @Override
             protected boolean broadcast(String message) {
                 throw new RuntimeException(EXPECTED);
@@ -74,53 +54,4 @@ public class NoopTopicSinkTest extends TopicTestBase {
         sink.start();
         assertFalse(sink.send(MY_MESSAGE));
     }
-
-    @Test(expected = IllegalArgumentException.class)
-    public void testSend_NullMessage() {
-        sink.send(null);
-    }
-
-    @Test(expected = IllegalArgumentException.class)
-    public void testSend_EmptyMessage() {
-        sink.send("");
-    }
-
-    @Test(expected = IllegalStateException.class)
-    public void testSend_NotStarted() {
-        sink.send(MY_MESSAGE);
-    }
-
-    @Test
-    public void testGetTopicCommInfrastructure() {
-        assertEquals(CommInfrastructure.NOOP, sink.getTopicCommInfrastructure());
-    }
-
-    @Test
-    public void testStart_testStop_testShutdown() {
-        sink.start();
-        assertTrue(sink.isAlive());
-
-        // start again
-        sink.start();
-        assertTrue(sink.isAlive());
-
-        // stop
-        sink.stop();
-        assertFalse(sink.isAlive());
-
-        // re-start again
-        sink.start();
-        assertTrue(sink.isAlive());
-
-        // shutdown
-        sink.shutdown();
-        assertFalse(sink.isAlive());
-    }
-
-    @Test(expected = IllegalStateException.class)
-    public void testStart_Locked() {
-        sink.lock();
-        sink.start();
-    }
-
 }
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSourceFactoryTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSourceFactoryTest.java
new file mode 100644
index 00000000..c8a44292
--- /dev/null
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSourceFactoryTest.java
@@ -0,0 +1,29 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2019 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.policy.common.endpoints.event.comm.bus;
+
+public class NoopTopicSourceFactoryTest extends NoopTopicFactoryTest<NoopTopicSourceFactory, NoopTopicSource>   {
+
+    @Override
+    protected NoopTopicSourceFactory buildFactory() {
+        return new NoopTopicSourceFactory();
+    }
+}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSourceTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSourceTest.java
new file mode 100644
index 00000000..22ccb9a8
--- /dev/null
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSourceTest.java
@@ -0,0 +1,57 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2019 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.policy.common.endpoints.event.comm.bus;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Test;
+
+public class NoopTopicSourceTest extends NoopTopicEndpointTest<NoopTopicSourceFactory, NoopTopicSource> {
+
+    public NoopTopicSourceTest() {
+        super(new NoopTopicSourceFactory());
+    }
+
+    @Override
+    protected boolean io(String message) {
+        return this.endpoint.offer(message);
+    }
+
+    @Test
+    public void testToString() {
+        assertTrue(this.endpoint.toString().startsWith("NoopTopicSource"));
+    }
+
+    @Test
+    public void testOffer() {
+        NoopTopicSource source = new NoopTopicSource(servers, MY_TOPIC) {
+            @Override
+            protected boolean broadcast(String message) {
+                throw new RuntimeException(EXPECTED);
+            }
+
+        };
+
+        source.start();
+        assertFalse(source.offer(MY_MESSAGE));
+    }
+}