aboutsummaryrefslogtreecommitdiffstats
path: root/policy-endpoints/src/test/java
diff options
context:
space:
mode:
Diffstat (limited to 'policy-endpoints/src/test/java')
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxyTest.java100
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicFactoryTestBase.java2
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSinkFactoryTest.java14
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSinkTest.java2
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSourceFactoryTest.java12
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSourceTest.java2
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicFactoryTestBase.java48
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicPropertyBuilder.java107
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkFactoryTest.java177
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkTest.java34
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceFactoryTest.java210
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceTest.java34
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumerTest.java73
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusPublisherTest.java100
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicParamsTest.java25
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineBusTopicSinkTest.java13
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineUebTopicSinkTest.java74
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedUebTopicSourceTest.java70
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/client/BidirectionalTopicClientTest.java3
19 files changed, 60 insertions, 1040 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
index 92dd6483..b6777db7 100644
--- 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
@@ -37,8 +37,6 @@ import org.junit.Test;
import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
import org.onap.policy.common.endpoints.event.comm.bus.NoopTopicFactories;
import org.onap.policy.common.endpoints.event.comm.bus.NoopTopicPropertyBuilder;
-import org.onap.policy.common.endpoints.event.comm.bus.UebTopicFactories;
-import org.onap.policy.common.endpoints.event.comm.bus.UebTopicPropertyBuilder;
import org.onap.policy.common.endpoints.parameters.TopicParameterGroup;
import org.onap.policy.common.endpoints.parameters.TopicParameters;
import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
@@ -49,11 +47,8 @@ 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 Properties configuration = new Properties();
- private TopicParameterGroup group = new TopicParameterGroup();
+ private final Properties configuration = new Properties();
+ private final TopicParameterGroup group = new TopicParameterGroup();
/**
* Constructor.
@@ -74,18 +69,6 @@ public class TopicEndpointProxyTest {
configuration.putAll(noopSinkBuilder.build());
group.getTopicSinks().add(noopSinkBuilder.getParams());
- UebTopicPropertyBuilder uebSourceBuilder =
- new UebTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS)
- .makeTopic(UEB_SOURCE_TOPIC);
- configuration.putAll(uebSourceBuilder.build());
- group.getTopicSources().add(uebSourceBuilder.getParams());
-
- UebTopicPropertyBuilder uebSinkBuilder =
- new UebTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS)
- .makeTopic(UEB_SINK_TOPIC);
- configuration.putAll(uebSinkBuilder.build());
- group.getTopicSinks().add(uebSinkBuilder.getParams());
-
TopicParameters invalidCommInfraParams =
new NoopTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_NOOP_SOURCE_TOPICS)
.makeTopic(NOOP_SOURCE_TOPIC).getParams();
@@ -99,19 +82,19 @@ public class TopicEndpointProxyTest {
}
private <T extends Topic> boolean allSources(List<T> topics) {
- return exists(topics, NOOP_SOURCE_TOPIC) && exists(topics, UEB_SOURCE_TOPIC);
+ return exists(topics, NOOP_SOURCE_TOPIC);
}
private <T extends Topic> boolean allSinks(List<T> topics) {
- return exists(topics, NOOP_SINK_TOPIC) && exists(topics, UEB_SINK_TOPIC);
+ return exists(topics, NOOP_SINK_TOPIC);
}
private <T extends Topic> boolean anySource(List<T> topics) {
- return exists(topics, NOOP_SOURCE_TOPIC) || exists(topics, UEB_SOURCE_TOPIC);
+ return exists(topics, NOOP_SOURCE_TOPIC);
}
private <T extends Topic> boolean anySink(List<T> topics) {
- return exists(topics, NOOP_SINK_TOPIC) || exists(topics, UEB_SINK_TOPIC);
+ return exists(topics, NOOP_SINK_TOPIC);
}
/**
@@ -121,9 +104,6 @@ public class TopicEndpointProxyTest {
public void tearDown() {
NoopTopicFactories.getSinkFactory().destroy();
NoopTopicFactories.getSourceFactory().destroy();
-
- UebTopicFactories.getSinkFactory().destroy();
- UebTopicFactories.getSourceFactory().destroy();
}
@Test
@@ -142,7 +122,7 @@ public class TopicEndpointProxyTest {
TopicEndpoint manager = new TopicEndpointProxy();
List<TopicSource> sources = manager.addTopicSources(group.getTopicSources());
- assertSame(2, sources.size());
+ assertSame(1, sources.size());
assertTrue(allSources(sources));
assertFalse(anySink(sources));
@@ -153,7 +133,7 @@ public class TopicEndpointProxyTest {
TopicEndpoint manager = new TopicEndpointProxy();
List<TopicSource> sources = manager.addTopicSources(configuration);
- assertSame(2, sources.size());
+ assertSame(1, sources.size());
assertTrue(allSources(sources));
assertFalse(anySink(sources));
@@ -164,7 +144,7 @@ public class TopicEndpointProxyTest {
TopicEndpoint manager = new TopicEndpointProxy();
List<TopicSink> sinks = manager.addTopicSinks(group.getTopicSinks());
- assertSame(2, sinks.size());
+ assertSame(1, sinks.size());
assertFalse(anySource(sinks));
assertTrue(allSinks(sinks));
@@ -175,7 +155,7 @@ public class TopicEndpointProxyTest {
TopicEndpoint manager = new TopicEndpointProxy();
List<TopicSink> sinks = manager.addTopicSinks(configuration);
- assertSame(2, sinks.size());
+ assertSame(1, sinks.size());
assertFalse(anySource(sinks));
assertTrue(allSinks(sinks));
@@ -186,7 +166,7 @@ public class TopicEndpointProxyTest {
TopicEndpoint manager = new TopicEndpointProxy();
List<Topic> topics = manager.addTopics(configuration);
- assertSame(4, topics.size());
+ assertSame(2, topics.size());
assertTrue(allSources(topics));
assertTrue(allSinks(topics));
@@ -197,7 +177,7 @@ public class TopicEndpointProxyTest {
TopicEndpoint manager = new TopicEndpointProxy();
List<Topic> topics = manager.addTopics(group);
- assertSame(4, topics.size());
+ assertSame(2, topics.size());
assertTrue(allSources(topics));
assertTrue(allSinks(topics));
@@ -236,7 +216,7 @@ public class TopicEndpointProxyTest {
manager.addTopicSinks(configuration);
List<TopicSource> sources = manager.getTopicSources();
- assertSame(2, sources.size());
+ assertSame(1, sources.size());
assertTrue(allSources(sources));
assertFalse(anySink(sources));
@@ -250,21 +230,13 @@ public class TopicEndpointProxyTest {
manager.addTopicSinks(configuration);
List<TopicSink> sinks = manager.getTopicSinks();
- assertSame(2, sinks.size());
+ assertSame(1, sinks.size());
assertFalse(anySource(sinks));
assertTrue(allSinks(sinks));
}
@Test
- public void testGetUebTopicSources() {
- TopicEndpoint manager = new TopicEndpointProxy();
-
- manager.addTopicSources(configuration);
- assertSame(1, manager.getUebTopicSources().size());
- }
-
- @Test
public void testGetNoopTopicSources() {
TopicEndpoint manager = new TopicEndpointProxy();
@@ -273,14 +245,6 @@ public class TopicEndpointProxyTest {
}
@Test
- public void testGetUebTopicSinks() {
- TopicEndpoint manager = new TopicEndpointProxy();
-
- manager.addTopicSinks(configuration);
- assertSame(1, manager.getUebTopicSinks().size());
- }
-
- @Test
public void testGetNoopTopicSinks() {
TopicEndpoint manager = new TopicEndpointProxy();
@@ -322,12 +286,9 @@ public class TopicEndpointProxyTest {
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());
assertThatIllegalStateException()
.isThrownBy(() -> manager.getTopicSource(CommInfrastructure.NOOP, NOOP_SINK_TOPIC));
- assertThatIllegalStateException()
- .isThrownBy(() -> manager.getTopicSource(CommInfrastructure.UEB, UEB_SINK_TOPIC));
}
@Test
@@ -336,38 +297,9 @@ public class TopicEndpointProxyTest {
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());
assertThatIllegalStateException()
.isThrownBy(() -> manager.getTopicSink(CommInfrastructure.NOOP, NOOP_SOURCE_TOPIC));
- assertThatIllegalStateException()
- .isThrownBy(() -> manager.getTopicSink(CommInfrastructure.UEB, UEB_SOURCE_TOPIC));
- }
-
- @Test
- public void testGetUebTopicSource() {
- TopicEndpoint manager = new TopicEndpointProxy();
- manager.addTopicSources(configuration);
-
- assertSame(UEB_SOURCE_TOPIC, manager.getUebTopicSource(UEB_SOURCE_TOPIC).getTopic());
-
- assertThatIllegalStateException().isThrownBy(() -> manager.getUebTopicSource(NOOP_SOURCE_TOPIC));
-
- assertThatIllegalArgumentException().isThrownBy(() -> manager.getUebTopicSource(null));
- assertThatIllegalArgumentException().isThrownBy(() -> manager.getUebTopicSource(""));
- }
-
- @Test
- public void testGetUebTopicSink() {
- TopicEndpoint manager = new TopicEndpointProxy();
- manager.addTopicSinks(configuration);
-
- assertSame(UEB_SINK_TOPIC, manager.getUebTopicSink(UEB_SINK_TOPIC).getTopic());
-
- assertThatIllegalStateException().isThrownBy(() -> manager.getUebTopicSink(NOOP_SINK_TOPIC));
-
- assertThatIllegalArgumentException().isThrownBy(() -> manager.getUebTopicSink(null));
- assertThatIllegalArgumentException().isThrownBy(() -> manager.getUebTopicSink(""));
}
@Test
@@ -377,8 +309,6 @@ public class TopicEndpointProxyTest {
assertSame(NOOP_SOURCE_TOPIC, manager.getNoopTopicSource(NOOP_SOURCE_TOPIC).getTopic());
- assertThatIllegalStateException().isThrownBy(() -> manager.getNoopTopicSource(UEB_SOURCE_TOPIC));
-
assertThatIllegalArgumentException().isThrownBy(() -> manager.getNoopTopicSource(null));
assertThatIllegalArgumentException().isThrownBy(() -> manager.getNoopTopicSource(""));
}
@@ -390,8 +320,6 @@ public class TopicEndpointProxyTest {
assertSame(NOOP_SINK_TOPIC, manager.getNoopTopicSink(NOOP_SINK_TOPIC).getTopic());
- assertThatIllegalStateException().isThrownBy(() -> manager.getNoopTopicSink(UEB_SINK_TOPIC));
-
assertThatIllegalArgumentException().isThrownBy(() -> manager.getNoopTopicSink(null));
assertThatIllegalArgumentException().isThrownBy(() -> manager.getNoopTopicSink(""));
}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicFactoryTestBase.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicFactoryTestBase.java
index 3986549c..3dfd96dd 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicFactoryTestBase.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicFactoryTestBase.java
@@ -2,7 +2,7 @@
* ============LICENSE_START=======================================================
* policy-endpoints
* ================================================================================
- * Copyright (C) 2018-2019 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2022 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSinkFactoryTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSinkFactoryTest.java
index c109e70a..52868c44 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSinkFactoryTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSinkFactoryTest.java
@@ -2,7 +2,7 @@
* ============LICENSE_START=======================================================
* policy-endpoints
* ================================================================================
- * Copyright (C) 2022 Nordix Foundation.
+ * Copyright (C) 2022, 2024 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -21,11 +21,11 @@
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.assertTrue;
import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_KAFKA_SINK_TOPICS;
import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_EFFECTIVE_TOPIC_SUFFIX;
-import java.util.Arrays;
import java.util.Deque;
import java.util.LinkedList;
import java.util.List;
@@ -70,7 +70,7 @@ public class KafkaTopicSinkFactoryTest extends KafkaTopicFactoryTestBase<KafkaTo
// check parameters that were used
BusTopicParams params = getLastParams();
- assertEquals(false, params.isAllowSelfSignedCerts());
+ assertFalse(params.isAllowSelfSignedCerts());
}
@Test
@@ -82,9 +82,9 @@ public class KafkaTopicSinkFactoryTest extends KafkaTopicFactoryTestBase<KafkaTo
assertEquals(MY_EFFECTIVE_TOPIC, topics.get(0).getEffectiveTopic());
BusTopicParams params = getLastParams();
- assertEquals(true, params.isManaged());
- assertEquals(false, params.isUseHttps());
- assertEquals(Arrays.asList(KAFKA_SERVER), params.getServers());
+ assertTrue(params.isManaged());
+ assertFalse(params.isUseHttps());
+ assertEquals(List.of(KAFKA_SERVER), params.getServers());
assertEquals(MY_TOPIC, params.getTopic());
assertEquals(MY_EFFECTIVE_TOPIC, params.getEffectiveTopic());
assertEquals(MY_PARTITION, params.getPartitionId());
@@ -178,7 +178,7 @@ public class KafkaTopicSinkFactoryTest extends KafkaTopicFactoryTestBase<KafkaTo
}
/**
- * Factory that records the parameters of all of the sinks it creates.
+ * Factory that records the parameters of all the sinks it creates.
*/
private static class SinkFactory extends IndexedKafkaTopicSinkFactory {
private Deque<BusTopicParams> params = new LinkedList<>();
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSinkTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSinkTest.java
index 503e5131..483e4e99 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSinkTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSinkTest.java
@@ -2,7 +2,7 @@
* ============LICENSE_START=======================================================
* ONAP Policy Engine - Common Modules
* ================================================================================
- * Copyright (C) 2018-2019 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2022, 2024 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSourceFactoryTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSourceFactoryTest.java
index 1c6985ad..392cefe9 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSourceFactoryTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSourceFactoryTest.java
@@ -2,8 +2,7 @@
* ============LICENSE_START=======================================================
* ONAP Policy Engine - Common Modules
* ================================================================================
- * Copyright (C) 2018-2019 AT&T Intellectual Property. All rights reserved.
- * Modifications Copyright (C) 2023 Nordix Foundation.
+ * Copyright (C) 2022-2024 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,6 +21,7 @@
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.assertTrue;
import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_KAFKA_SOURCE_TOPICS;
@@ -69,9 +69,9 @@ public class KafkaTopicSourceFactoryTest extends KafkaTopicFactoryTestBase<Kafka
assertEquals(MY_EFFECTIVE_TOPIC, topics.get(0).getEffectiveTopic());
BusTopicParams params = getLastParams();
- assertEquals(true, params.isManaged());
- assertEquals(false, params.isUseHttps());
- assertEquals(Arrays.asList(KAFKA_SERVER), params.getServers());
+ assertTrue(params.isManaged());
+ assertFalse(params.isUseHttps());
+ assertEquals(List.of(KAFKA_SERVER), params.getServers());
assertEquals(MY_TOPIC, params.getTopic());
assertEquals(MY_EFFECTIVE_TOPIC, params.getEffectiveTopic());
}
@@ -154,7 +154,7 @@ public class KafkaTopicSourceFactoryTest extends KafkaTopicFactoryTestBase<Kafka
}
/**
- * Factory that records the parameters of all of the sources it creates.
+ * Factory that records the parameters of all the sources it creates.
*/
private static class SourceFactory extends IndexedKafkaTopicSourceFactory {
private Deque<BusTopicParams> params = new LinkedList<>();
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSourceTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSourceTest.java
index ee2d1d7b..5079e601 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSourceTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/KafkaTopicSourceTest.java
@@ -2,7 +2,7 @@
* ============LICENSE_START=======================================================
* ONAP Policy Engine - Common Modules
* ================================================================================
- * Copyright (C) 2018-2019 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2022 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicFactoryTestBase.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicFactoryTestBase.java
deleted file mode 100644
index 41dbac8c..00000000
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicFactoryTestBase.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * policy-endpoints
- * ================================================================================
- * Copyright (C) 2018-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.assertThatIllegalArgumentException;
-
-import java.util.Collections;
-import org.onap.policy.common.endpoints.event.comm.Topic;
-
-/**
- * Base class for UebTopicXxxFactory tests.
- *
- * @param <T> type of topic managed by the factory
- */
-public abstract class UebTopicFactoryTestBase<T extends Topic> extends BusTopicFactoryTestBase<T> {
-
- @Override
- public void testBuildBusTopicParams_Ex() {
-
- super.testBuildBusTopicParams_Ex();
-
- // null servers
- assertThatIllegalArgumentException().as("null servers")
- .isThrownBy(() -> buildTopic(makeBuilder().servers(null).build()));
-
- // empty servers
- assertThatIllegalArgumentException().as("empty servers")
- .isThrownBy(() -> buildTopic(makeBuilder().servers(Collections.emptyList()).build()));
- }
-}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicPropertyBuilder.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicPropertyBuilder.java
deleted file mode 100644
index 42ea6eba..00000000
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicPropertyBuilder.java
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * ONAP
- * ================================================================================
- * Copyright (C) 2018-2019 AT&T Intellectual Property. All rights reserved.
- * Modifications Copyright (C) 2024 Nordix Foundation.
- * ================================================================================
- * 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.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_AFT_ENV;
-import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_API_KEY;
-import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_API_SECRET;
-import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_CONS_GROUP;
-import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_CONS_INST;
-import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_EFFECTIVE_TOPIC;
-import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_FETCH_LIMIT;
-import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_FETCH_TIMEOUT;
-import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_PARTITION;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_API_KEY_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_API_SECRET_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_EFFECTIVE_TOPIC_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SINK_PARTITION_KEY_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_GROUP_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_INSTANCE_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX;
-
-import java.util.List;
-import lombok.Getter;
-import org.onap.policy.common.endpoints.parameters.TopicParameters;
-
-@Getter
-public class UebTopicPropertyBuilder extends TopicPropertyBuilder {
-
- public static final String SERVER = "my-server";
- public static final String TOPIC2 = "my-topic-2";
-
- private final TopicParameters params = new TopicParameters();
-
- /**
- * Constructs the object.
- *
- * @param prefix the prefix for the properties to be built
- */
- public UebTopicPropertyBuilder(String prefix) {
- super(prefix);
- }
-
- /**
- * Adds a topic and configures its properties with default values.
- *
- * @param topic the topic to be added
- * @return this builder
- */
- public UebTopicPropertyBuilder makeTopic(String topic) {
- addTopic(topic);
-
- setTopicProperty(PROPERTY_TOPIC_EFFECTIVE_TOPIC_SUFFIX, MY_EFFECTIVE_TOPIC);
- setTopicProperty(PROPERTY_TOPIC_SOURCE_CONSUMER_GROUP_SUFFIX, MY_CONS_GROUP);
- setTopicProperty(PROPERTY_TOPIC_SOURCE_CONSUMER_INSTANCE_SUFFIX, MY_CONS_INST);
- setTopicProperty(PROPERTY_MANAGED_SUFFIX, "true");
- setTopicProperty(PROPERTY_HTTP_HTTPS_SUFFIX, "true");
- setTopicProperty(PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX, "true");
- setTopicProperty(PROPERTY_TOPIC_API_KEY_SUFFIX, MY_API_KEY);
- setTopicProperty(PROPERTY_TOPIC_API_SECRET_SUFFIX, MY_API_SECRET);
- setTopicProperty(PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX, MY_FETCH_LIMIT);
- setTopicProperty(PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX, MY_FETCH_TIMEOUT);
- setTopicProperty(PROPERTY_TOPIC_SINK_PARTITION_KEY_SUFFIX, MY_PARTITION);
- setTopicProperty(PROPERTY_TOPIC_SERVERS_SUFFIX, SERVER);
-
- params.setTopicCommInfrastructure("ueb");
- params.setTopic(topic);
- params.setEffectiveTopic(MY_EFFECTIVE_TOPIC);
- params.setConsumerGroup(MY_CONS_GROUP);
- params.setConsumerInstance(MY_CONS_INST);
- params.setManaged(true);
- params.setUseHttps(true);
- params.setAftEnvironment(MY_AFT_ENV);
- params.setAllowSelfSignedCerts(true);
- params.setApiKey(MY_API_KEY);
- params.setApiSecret(MY_API_SECRET);
- params.setFetchLimit(MY_FETCH_LIMIT);
- params.setFetchTimeout(MY_FETCH_TIMEOUT);
- params.setPartitionId(MY_PARTITION);
- params.setServers(List.of(SERVER));
-
- return this;
- }
-}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkFactoryTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkFactoryTest.java
deleted file mode 100644
index 4896a9df..00000000
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkFactoryTest.java
+++ /dev/null
@@ -1,177 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * policy-endpoints
- * ================================================================================
- * Copyright (C) 2018-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.assertTrue;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS;
-
-import java.util.Deque;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Properties;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
-
-public class UebTopicSinkFactoryTest extends UebTopicFactoryTestBase<UebTopicSink> {
-
- private SinkFactory factory;
-
- /**
- * Creates the object to be tested.
- */
- @Before
- @Override
- public void setUp() {
- super.setUp();
-
- factory = new SinkFactory();
- }
-
- @After
- public void tearDown() {
- factory.destroy();
- }
-
- @Test
- @Override
- public void testBuildBusTopicParams() {
- super.testBuildBusTopicParams();
- super.testBuildBusTopicParams_Ex();
- }
-
- @Test
- @Override
- public void testBuildListOfStringString() {
- super.testBuildListOfStringString();
-
- // check parameters that were used
- BusTopicParams params = getLastParams();
- assertEquals(false, params.isAllowSelfSignedCerts());
- }
-
- @Test
- @Override
- public void testBuildProperties() {
- super.testBuildProperties();
- super.testBuildProperties_Variations();
- super.testBuildProperties_Multiple();
-
- initFactory();
-
- assertEquals(1, buildTopics(makePropBuilder().makeTopic(MY_TOPIC).build()).size());
-
- BusTopicParams params = getLastParams();
- assertEquals(MY_PARTITION, params.getPartitionId());
- }
-
- @Test
- @Override
- public void testDestroyString_testGet_testInventory() {
- super.testDestroyString_testGet_testInventory();
- super.testDestroyString_Ex();
- }
-
- @Test
- @Override
- public void testDestroy() {
- super.testDestroy();
- }
-
- @Test
- public void testGet() {
- super.testGet_Ex();
- }
-
- @Test
- public void testToString() {
- assertTrue(factory.toString().startsWith("IndexedUebTopicSinkFactory ["));
- }
-
- @Override
- protected void initFactory() {
- if (factory != null) {
- factory.destroy();
- }
-
- factory = new SinkFactory();
- }
-
- @Override
- protected List<UebTopicSink> buildTopics(Properties properties) {
- return factory.build(properties);
- }
-
- @Override
- protected UebTopicSink buildTopic(BusTopicParams params) {
- return factory.build(params);
- }
-
- @Override
- protected UebTopicSink buildTopic(List<String> servers, String topic) {
- return factory.build(servers, topic);
- }
-
- @Override
- protected void destroyFactory() {
- factory.destroy();
- }
-
- @Override
- protected void destroyTopic(String topic) {
- factory.destroy(topic);
- }
-
- @Override
- protected List<UebTopicSink> getInventory() {
- return factory.inventory();
- }
-
- @Override
- protected UebTopicSink getTopic(String topic) {
- return factory.get(topic);
- }
-
- @Override
- protected BusTopicParams getLastParams() {
- return factory.params.getLast();
- }
-
- @Override
- protected TopicPropertyBuilder makePropBuilder() {
- return new UebTopicPropertyBuilder(PROPERTY_UEB_SINK_TOPICS);
- }
-
- /**
- * Factory that records the parameters of all of the sinks it creates.
- */
- private static class SinkFactory extends IndexedUebTopicSinkFactory {
- private Deque<BusTopicParams> params = new LinkedList<>();
-
- @Override
- protected UebTopicSink makeSink(BusTopicParams busTopicParams) {
- params.add(busTopicParams);
- return super.makeSink(busTopicParams);
- }
- }
-}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkTest.java
deleted file mode 100644
index 77452604..00000000
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkTest.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * ONAP Policy Engine - Common Modules
- * ================================================================================
- * Copyright (C) 2018-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.assertNotNull;
-
-import org.junit.Test;
-
-public class UebTopicSinkTest {
-
- @Test
- public void test() {
- assertNotNull(UebTopicFactories.getSinkFactory());
- }
-
-}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceFactoryTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceFactoryTest.java
deleted file mode 100644
index 81e30756..00000000
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceFactoryTest.java
+++ /dev/null
@@ -1,210 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * ONAP Policy Engine - Common Modules
- * ================================================================================
- * Copyright (C) 2018-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.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS;
-
-import java.util.Deque;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Properties;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
-import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
-
-public class UebTopicSourceFactoryTest extends UebTopicFactoryTestBase<UebTopicSource> {
-
- private SourceFactory factory;
-
- /**
- * Creates the object to be tested.
- */
- @Before
- @Override
- public void setUp() {
- super.setUp();
-
- factory = new SourceFactory();
- }
-
- @After
- public void tearDown() {
- factory.destroy();
- }
-
- @Test
- @Override
- public void testBuildBusTopicParams() {
- super.testBuildBusTopicParams();
- super.testBuildBusTopicParams_Ex();
- }
-
- @Test
- @Override
- public void testBuildProperties() {
-
- super.testBuildProperties();
-
- // check source-specific parameters that were used
- BusTopicParams params = factory.params.getFirst();
- assertEquals(MY_CONS_GROUP, params.getConsumerGroup());
- assertEquals(MY_CONS_INST, params.getConsumerInstance());
- assertEquals(MY_FETCH_LIMIT, params.getFetchLimit());
- assertEquals(MY_FETCH_TIMEOUT, params.getFetchTimeout());
-
- super.testBuildProperties_Variations();
- super.testBuildProperties_Multiple();
-
- // check default values for source-specific parameters
- checkDefault(PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX,
- params2 -> params2.getFetchLimit() == PolicyEndPointProperties.DEFAULT_LIMIT_FETCH,
- null, "", "invalid-limit-number");
-
- checkDefault(PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX,
- params2 -> params2.getFetchTimeout() == PolicyEndPointProperties.DEFAULT_TIMEOUT_MS_FETCH,
- null, "", "invalid-timeout-number");
- }
-
- @Test
- public void testBuildListOfStringStringStringString() {
- UebTopicSource source1 = factory.build(servers, MY_TOPIC, MY_API_KEY, MY_API_SECRET);
- assertNotNull(source1);
-
- // check source-specific parameters that were used
- BusTopicParams params = factory.params.getFirst();
- assertEquals(MY_API_KEY, params.getApiKey());
- assertEquals(MY_API_SECRET, params.getApiSecret());
- assertEquals(PolicyEndPointProperties.DEFAULT_LIMIT_FETCH, params.getFetchLimit());
- assertEquals(PolicyEndPointProperties.DEFAULT_TIMEOUT_MS_FETCH, params.getFetchTimeout());
- }
-
- @Test
- @Override
- public void testBuildListOfStringString() {
- super.testBuildListOfStringString();
-
- // check source-specific parameters that were used
- BusTopicParams params = factory.params.getFirst();
- assertEquals(null, params.getApiKey());
- assertEquals(null, params.getApiSecret());
- assertEquals(PolicyEndPointProperties.DEFAULT_LIMIT_FETCH, params.getFetchLimit());
- assertEquals(PolicyEndPointProperties.DEFAULT_TIMEOUT_MS_FETCH, params.getFetchTimeout());
-
- assertEquals(true, params.isAllowSelfSignedCerts());
- }
-
- @Test
- @Override
- public void testDestroyString_testGet_testInventory() {
- super.testDestroyString_testGet_testInventory();
- super.testDestroyString_Ex();
- }
-
- @Test
- @Override
- public void testDestroy() {
- super.testDestroy();
- }
-
- @Test
- public void testGet() {
- super.testGet_Ex();
- }
-
- @Test
- public void testToString() {
- assertTrue(factory.toString().startsWith("IndexedUebTopicSourceFactory ["));
- }
-
- @Override
- protected void initFactory() {
- if (factory != null) {
- factory.destroy();
- }
-
- factory = new SourceFactory();
- }
-
- @Override
- protected List<UebTopicSource> buildTopics(Properties properties) {
- return factory.build(properties);
- }
-
- @Override
- protected UebTopicSource buildTopic(BusTopicParams params) {
- return factory.build(params);
- }
-
- @Override
- protected UebTopicSource buildTopic(List<String> servers, String topic) {
- return factory.build(servers, topic);
- }
-
- @Override
- protected void destroyFactory() {
- factory.destroy();
- }
-
- @Override
- protected void destroyTopic(String topic) {
- factory.destroy(topic);
- }
-
- @Override
- protected List<UebTopicSource> getInventory() {
- return factory.inventory();
- }
-
- @Override
- protected UebTopicSource getTopic(String topic) {
- return factory.get(topic);
- }
-
- @Override
- protected BusTopicParams getLastParams() {
- return factory.params.getLast();
- }
-
- @Override
- protected TopicPropertyBuilder makePropBuilder() {
- return new UebTopicPropertyBuilder(PROPERTY_UEB_SOURCE_TOPICS);
- }
-
- /**
- * Factory that records the parameters of all of the sources it creates.
- */
- private static class SourceFactory extends IndexedUebTopicSourceFactory {
- private Deque<BusTopicParams> params = new LinkedList<>();
-
- @Override
- protected UebTopicSource makeSource(BusTopicParams busTopicParams) {
- params.add(busTopicParams);
- return super.makeSource(busTopicParams);
- }
- }
-}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceTest.java
deleted file mode 100644
index 9ef8af84..00000000
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceTest.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * ONAP Policy Engine - Common Modules
- * ================================================================================
- * Copyright (C) 2018-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.assertNotNull;
-
-import org.junit.Test;
-
-public class UebTopicSourceTest {
-
- @Test
- public void test() {
- assertNotNull(UebTopicFactories.getSourceFactory());
- }
-
-}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumerTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumerTest.java
index 70fa83c6..2c33a257 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumerTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumerTest.java
@@ -23,42 +23,35 @@ package org.onap.policy.common.endpoints.event.comm.bus.internal;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
-import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThrows;
-import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-import com.att.nsa.cambria.client.CambriaConsumer;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
-import org.apache.commons.collections4.IteratorUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
+import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase;
-import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.CambriaConsumerWrapper;
import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.FetchingBusConsumer;
import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.KafkaConsumerWrapper;
import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
-import org.springframework.test.util.ReflectionTestUtils;
public class BusConsumerTest extends TopicTestBase {
@@ -68,11 +61,18 @@ public class BusConsumerTest extends TopicTestBase {
@Mock
KafkaConsumer<String, String> mockedKafkaConsumer;
+ AutoCloseable closeable;
+
@Before
@Override
public void setUp() {
super.setUp();
- MockitoAnnotations.initMocks(this);
+ closeable = MockitoAnnotations.openMocks(this);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ closeable.close();
}
@@ -139,61 +139,6 @@ public class BusConsumerTest extends TopicTestBase {
}
@Test
- public void testCambriaConsumerWrapper() {
- // verify that different wrappers can be built
- new CambriaConsumerWrapper(makeBuilder().build());
- new CambriaConsumerWrapper(makeBuilder().useHttps(false).build());
- new CambriaConsumerWrapper(makeBuilder().useHttps(true).build());
- new CambriaConsumerWrapper(makeBuilder().useHttps(true).allowSelfSignedCerts(false).build());
- new CambriaConsumerWrapper(makeBuilder().useHttps(true).allowSelfSignedCerts(true).build());
- new CambriaConsumerWrapper(makeBuilder().apiKey(null).build());
- new CambriaConsumerWrapper(makeBuilder().apiSecret(null).build());
- new CambriaConsumerWrapper(makeBuilder().apiKey(null).apiSecret(null).build());
- new CambriaConsumerWrapper(makeBuilder().userName(null).build());
- new CambriaConsumerWrapper(makeBuilder().password(null).build());
-
- assertThatCode(() -> new CambriaConsumerWrapper(makeBuilder().userName(null).password(null).build()))
- .doesNotThrowAnyException();
- }
-
- @Test
- public void testCambriaConsumerWrapperFetch() throws Exception {
- CambriaConsumer inner = mock(CambriaConsumer.class);
- List<String> lst = Arrays.asList(MY_MESSAGE, MY_MESSAGE2);
- when(inner.fetch()).thenReturn(lst);
-
- CambriaConsumerWrapper cons = new CambriaConsumerWrapper(builder.build());
- ReflectionTestUtils.setField(cons, "consumer", inner);
-
- assertEquals(lst, IteratorUtils.toList(cons.fetch().iterator()));
-
- // arrange to throw exception next time fetch is called
- IOException ex = new IOException(EXPECTED);
- when(inner.fetch()).thenThrow(ex);
-
- cons.fetchTimeout = 10;
-
- try {
- cons.fetch();
- fail("missing exception");
-
- } catch (IOException e) {
- assertEquals(ex, e);
- }
- }
-
- @Test
- public void testCambriaConsumerWrapperClose() {
- CambriaConsumerWrapper cons = new CambriaConsumerWrapper(builder.build());
- assertThatCode(cons::close).doesNotThrowAnyException();
- }
-
- @Test
- public void testCambriaConsumerWrapperToString() {
- assertNotNull(new CambriaConsumerWrapper(makeBuilder().build()).toString());
- }
-
- @Test
public void testKafkaConsumerWrapper() {
// verify that different wrappers can be built
assertThatCode(() -> new KafkaConsumerWrapper(makeKafkaBuilder().build())).doesNotThrowAnyException();
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusPublisherTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusPublisherTest.java
deleted file mode 100644
index 66029b61..00000000
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusPublisherTest.java
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * policy-endpoints
- * ================================================================================
- * Copyright (C) 2018-2020 AT&T Intellectual Property. All rights reserved.
- * Modifications Copyright (C) 2024 Nordix Foundation.
- * ================================================================================
- * 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.internal;
-
-import static org.assertj.core.api.Assertions.assertThatCode;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.Mockito.doThrow;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-import com.att.nsa.cambria.client.CambriaBatchingPublisher;
-import java.io.IOException;
-import org.junit.Before;
-import org.junit.Test;
-import org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase;
-import org.onap.policy.common.endpoints.event.comm.bus.internal.BusPublisher.CambriaPublisherWrapper;
-
-
-public class BusPublisherTest extends TopicTestBase {
-
- @Before
- @Override
- public void setUp() {
- super.setUp();
- }
-
- @Test
- public void testCambriaPublisherWrapper() {
- // verify that different wrappers can be built
- new CambriaPublisherWrapper(makeBuilder().build());
- new CambriaPublisherWrapper(makeBuilder().useHttps(false).build());
- new CambriaPublisherWrapper(makeBuilder().useHttps(true).build());
- new CambriaPublisherWrapper(makeBuilder().useHttps(true).allowSelfSignedCerts(false).build());
- new CambriaPublisherWrapper(makeBuilder().useHttps(true).allowSelfSignedCerts(true).build());
- new CambriaPublisherWrapper(makeBuilder().apiKey(null).build());
- new CambriaPublisherWrapper(makeBuilder().apiSecret(null).build());
- new CambriaPublisherWrapper(makeBuilder().apiKey(null).apiSecret(null).build());
- new CambriaPublisherWrapper(makeBuilder().userName(null).build());
- new CambriaPublisherWrapper(makeBuilder().password(null).build());
- assertThatCode(() -> new CambriaPublisherWrapper(makeBuilder().userName(null).password(null).build()))
- .doesNotThrowAnyException();
- }
-
- @Test
- public void testCambriaPublisherWrapperSend() throws Exception {
- CambriaBatchingPublisher pub = mock(CambriaBatchingPublisher.class);
- CambriaPublisherWrapper cambria = new CambriaPublisherWrapper(makeBuilder().build());
- cambria.publisher = pub;
-
- assertTrue(cambria.send(MY_PARTITION, MY_MESSAGE));
-
- // publisher exception
- when(pub.send(anyString(), anyString())).thenThrow(new IOException(EXPECTED));
- assertFalse(cambria.send(MY_PARTITION2, MY_MESSAGE2));
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testCambriaPublisherWrapperSend_InvalidMsg() {
- CambriaPublisherWrapper cambria = new CambriaPublisherWrapper(makeBuilder().build());
- cambria.publisher = mock(CambriaBatchingPublisher.class);
-
- cambria.send(MY_PARTITION, null);
- }
-
- @Test
- public void testCambriaPublisherWrapperClose() {
- CambriaBatchingPublisher pub = mock(CambriaBatchingPublisher.class);
- CambriaPublisherWrapper cambria = new CambriaPublisherWrapper(makeBuilder().build());
- cambria.publisher = pub;
-
- cambria.close();
- verify(pub).close();
-
- // try again, this time with an exception
- doThrow(new RuntimeException(EXPECTED)).when(pub).close();
- cambria.close();
- }
-}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicParamsTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicParamsTest.java
index 93e5067e..3abb8b10 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicParamsTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicParamsTest.java
@@ -3,6 +3,7 @@
* policy-endpoints
* ================================================================================
* Copyright (C) 2018-2020 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2024 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -24,8 +25,8 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
-import java.util.Arrays;
import java.util.LinkedList;
+import java.util.List;
import java.util.function.BiConsumer;
import org.junit.Before;
import org.junit.Test;
@@ -46,7 +47,7 @@ public class BusTopicParamsTest extends TopicTestBase {
assertEquals(addProps, params.getAdditionalProps());
assertEquals(MY_AFT_ENV, params.getAftEnvironment());
- assertEquals(true, params.isAllowSelfSignedCerts());
+ assertTrue(params.isAllowSelfSignedCerts());
assertEquals(MY_API_KEY, params.getApiKey());
assertEquals(MY_API_SECRET, params.getApiSecret());
assertEquals(MY_BASE_PATH, params.getBasePath());
@@ -59,7 +60,7 @@ public class BusTopicParamsTest extends TopicTestBase {
assertEquals(MY_HOST, params.getHostname());
assertEquals(MY_LAT, params.getLatitude());
assertEquals(MY_LONG, params.getLongitude());
- assertEquals(true, params.isManaged());
+ assertTrue(params.isManaged());
assertEquals(MY_PARTITION, params.getPartitionId());
assertEquals(MY_PARTNER, params.getPartner());
assertEquals(MY_PASS, params.getPassword());
@@ -67,16 +68,16 @@ public class BusTopicParamsTest extends TopicTestBase {
assertEquals(servers, params.getServers());
assertEquals(MY_TOPIC, params.getTopic());
assertEquals(MY_EFFECTIVE_TOPIC, params.getEffectiveTopic());
- assertEquals(true, params.isUseHttps());
+ assertTrue(params.isUseHttps());
assertEquals(MY_USERNAME, params.getUserName());
}
@Test
public void testBooleanGetters() {
// ensure that booleans are independent of each other
- testBoolean("true:false:false", (bldr, flag) -> bldr.allowSelfSignedCerts(flag));
- testBoolean("false:true:false", (bldr, flag) -> bldr.managed(flag));
- testBoolean("false:false:true", (bldr, flag) -> bldr.useHttps(flag));
+ testBoolean("true:false:false", TopicParamsBuilder::allowSelfSignedCerts);
+ testBoolean("false:true:false", TopicParamsBuilder::managed);
+ testBoolean("false:false:true", TopicParamsBuilder::useHttps);
}
@Test
@@ -124,15 +125,15 @@ public class BusTopicParamsTest extends TopicTestBase {
assertTrue(makeBuilder().port(65536).build().isPortInvalid());
assertTrue(makeBuilder().servers(null).build().isServersInvalid());
assertTrue(makeBuilder().servers(new LinkedList<>()).build().isServersInvalid());
- assertTrue(makeBuilder().servers(Arrays.asList("")).build().isServersInvalid());
- assertFalse(makeBuilder().servers(Arrays.asList("one-server")).build().isServersInvalid());
+ assertTrue(makeBuilder().servers(List.of("")).build().isServersInvalid());
+ assertFalse(makeBuilder().servers(List.of("one-server")).build().isServersInvalid());
assertTrue(makeBuilder().topic("").build().isTopicInvalid());
assertFalse(makeBuilder().userName("").build().isUserNameValid());
}
/**
* Tests the boolean methods by applying a function, once with {@code false} and once
- * with {@code true}. Verifies that all of the boolean methods return the correct
+ * with {@code true}. Verifies that all the boolean methods return the correct
* value by concatenating them.
*
* @param expectedTrue the string that is expected when {@code true} is passed to the
@@ -147,7 +148,7 @@ public class BusTopicParamsTest extends TopicTestBase {
BusTopicParams params = builder.build();
assertEquals("false:false:false",
- "" + params.isAllowSelfSignedCerts() + ":" + params.isManaged() + ":" + params.isUseHttps());
+ params.isAllowSelfSignedCerts() + ":" + params.isManaged() + ":" + params.isUseHttps());
// now try the "true" case
@@ -155,6 +156,6 @@ public class BusTopicParamsTest extends TopicTestBase {
params = builder.build();
assertEquals(expectedTrue,
- "" + params.isAllowSelfSignedCerts() + ":" + params.isManaged() + ":" + params.isUseHttps());
+ params.isAllowSelfSignedCerts() + ":" + params.isManaged() + ":" + params.isUseHttps());
}
}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineBusTopicSinkTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineBusTopicSinkTest.java
index e6eec799..7aa70b2a 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineBusTopicSinkTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineBusTopicSinkTest.java
@@ -3,6 +3,7 @@
* policy-endpoints
* ================================================================================
* Copyright (C) 2018-2020 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2024 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -32,6 +33,7 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.Arrays;
+import java.util.List;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -124,7 +126,7 @@ public class InlineBusTopicSinkTest extends TopicTestBase {
verify(pub).send(MY_PARTITION, MY_MESSAGE);
verify(listener).onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MY_MESSAGE);
- assertEquals(Arrays.asList(MY_MESSAGE), Arrays.asList(sink.getRecentEvents()));
+ assertEquals(List.of(MY_MESSAGE), Arrays.asList(sink.getRecentEvents()));
// arrange for send to throw an exception
when(pub.send(anyString(), anyString())).thenThrow(new RuntimeException(EXPECTED));
@@ -138,8 +140,7 @@ public class InlineBusTopicSinkTest extends TopicTestBase {
@Test(expected = IllegalArgumentException.class)
public void testSend_NullMessage() {
sink.start();
- BusPublisher pub = mock(BusPublisher.class);
- sink.publisher = pub;
+ sink.publisher = mock(BusPublisher.class);
sink.send(null);
}
@@ -147,16 +148,14 @@ public class InlineBusTopicSinkTest extends TopicTestBase {
@Test(expected = IllegalArgumentException.class)
public void testSend_EmptyMessage() {
sink.start();
- BusPublisher pub = mock(BusPublisher.class);
- sink.publisher = pub;
+ sink.publisher = mock(BusPublisher.class);
sink.send("");
}
@Test(expected = IllegalStateException.class)
public void testSend_NotStarted() {
- BusPublisher pub = mock(BusPublisher.class);
- sink.publisher = pub;
+ sink.publisher = mock(BusPublisher.class);
sink.send(MY_MESSAGE);
}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineUebTopicSinkTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineUebTopicSinkTest.java
deleted file mode 100644
index 674f379f..00000000
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineUebTopicSinkTest.java
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * policy-endpoints
- * ================================================================================
- * Copyright (C) 2018-2020 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.internal;
-
-import static org.assertj.core.api.Assertions.assertThatCode;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-import org.junit.After;
-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.bus.TopicTestBase;
-import org.onap.policy.common.utils.gson.GsonTestUtils;
-
-public class InlineUebTopicSinkTest extends TopicTestBase {
- private InlineUebTopicSink sink;
-
- /**
- * Creates the object to be tested.
- */
- @Before
- @Override
- public void setUp() {
- super.setUp();
-
- sink = new InlineUebTopicSink(makeBuilder().build());
- }
-
- @After
- public void tearDown() {
- sink.shutdown();
- }
-
- @Test
- public void testSerialize() {
- assertThatCode(() -> new GsonTestUtils().compareGson(sink, InlineUebTopicSinkTest.class))
- .doesNotThrowAnyException();
- }
-
- @Test
- public void testToString() {
- assertTrue(sink.toString().startsWith("InlineUebTopicSink ["));
- }
-
- @Test
- public void testInit() {
- assertThatCode(() -> sink.init()).doesNotThrowAnyException();
- }
-
- @Test
- public void testGetTopicCommInfrastructure() {
- assertEquals(CommInfrastructure.UEB, sink.getTopicCommInfrastructure());
- }
-
-}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedUebTopicSourceTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedUebTopicSourceTest.java
deleted file mode 100644
index 6536d0e8..00000000
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedUebTopicSourceTest.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * policy-endpoints
- * ================================================================================
- * Copyright (C) 2018-2020 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.internal;
-
-import static org.assertj.core.api.Assertions.assertThatCode;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-import org.junit.After;
-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.bus.TopicTestBase;
-import org.onap.policy.common.utils.gson.GsonTestUtils;
-
-public class SingleThreadedUebTopicSourceTest extends TopicTestBase {
- private SingleThreadedUebTopicSource source;
-
- /**
- * Creates the object to be tested.
- */
- @Before
- @Override
- public void setUp() {
- super.setUp();
-
- source = new SingleThreadedUebTopicSource(makeBuilder().build());
- }
-
- @After
- public void tearDown() {
- source.shutdown();
- }
-
- @Test
- public void testSerialize() {
- assertThatCode(() -> new GsonTestUtils().compareGson(source, SingleThreadedUebTopicSourceTest.class))
- .doesNotThrowAnyException();
- }
-
- @Test
- public void testToString() {
- assertTrue(source.toString().startsWith("SingleThreadedUebTopicSource ["));
- source.shutdown();
- }
-
- @Test
- public void testGetTopicCommInfrastructure() {
- assertEquals(CommInfrastructure.UEB, source.getTopicCommInfrastructure());
- }
-
-}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/client/BidirectionalTopicClientTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/client/BidirectionalTopicClientTest.java
index 2605c14b..704b2cb0 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/client/BidirectionalTopicClientTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/client/BidirectionalTopicClientTest.java
@@ -3,6 +3,7 @@
* ONAP
* ================================================================================
* Copyright (C) 2020-2021 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2024 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -68,7 +69,7 @@ public class BidirectionalTopicClientTest {
private static final String SOURCE_TOPIC = "my-source-topic";
private static final String MY_TEXT = "my-text";
- private static final CommInfrastructure SINK_INFRA = CommInfrastructure.UEB;
+ private static final CommInfrastructure SINK_INFRA = CommInfrastructure.NOOP;
private static final CommInfrastructure SOURCE_INFRA = CommInfrastructure.NOOP;
@Mock