From b5fce77b85153783813d3c12ce2eb507b6b53daf Mon Sep 17 00:00:00 2001 From: Jim Hahn Date: Thu, 27 Aug 2020 07:59:19 -0400 Subject: Fix random failures in dmaap-sim Modified the code to use a separate topic for the format tests, as well as a separate consumer name for each test. Issue-ID: POLICY-2749 Change-Id: Ia93f17109b761e98a4f676066024b51dd9024dc0 Signed-off-by: Jim Hahn --- .../onap/policy/sim/dmaap/e2e/EndToEndTest.java | 58 ++++++++++++++++++++-- 1 file changed, 53 insertions(+), 5 deletions(-) (limited to 'models-sim/models-sim-dmaap') diff --git a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/e2e/EndToEndTest.java b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/e2e/EndToEndTest.java index 066c38bb6..5a83b4693 100644 --- a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/e2e/EndToEndTest.java +++ b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/e2e/EndToEndTest.java @@ -19,14 +19,22 @@ package org.onap.policy.sim.dmaap.e2e; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import com.att.nsa.cambria.client.CambriaClientBuilders; +import com.att.nsa.cambria.client.CambriaClientBuilders.ConsumerBuilder; +import com.att.nsa.cambria.client.CambriaConsumer; import java.io.File; import java.io.PrintWriter; import java.net.HttpURLConnection; +import java.net.MalformedURLException; import java.net.URL; import java.nio.charset.StandardCharsets; import java.nio.file.Files; +import java.security.GeneralSecurityException; import java.util.Arrays; +import java.util.Iterator; import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; @@ -39,7 +47,6 @@ import org.junit.Test; import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager; import org.onap.policy.common.endpoints.event.comm.TopicSink; import org.onap.policy.common.endpoints.parameters.TopicParameterGroup; -import org.onap.policy.common.endpoints.parameters.TopicParameters; import org.onap.policy.common.utils.coder.CoderException; import org.onap.policy.common.utils.coder.StandardCoder; import org.onap.policy.common.utils.network.NetworkUtil; @@ -56,6 +63,7 @@ public class EndToEndTest extends CommonRestServer { private static final int MAX_WAIT_SEC = 5; private static final String TOPIC = "MY-TOPIC"; private static final String TOPIC2 = "MY-TOPIC-B"; + private static final String TOPIC3 = "MY-TOPIC-C"; private static final int MAX_MSG = 200; private static Main main; @@ -75,6 +83,16 @@ public class EndToEndTest extends CommonRestServer { */ private static TopicParameterGroup topicConfig; + /** + * The "host:port", extracted from httpPrefix. + */ + private static String hostPort; + + /** + * Unique consumer name used by a single test case. + */ + private int consumerName; + /** * Starts the rest server. * @@ -105,6 +123,8 @@ public class EndToEndTest extends CommonRestServer { .register((infra, topic, event) -> queue.add(event)); TopicEndpointManager.getManager().getDmaapTopicSource(TOPIC2) .register((infra, topic, event) -> queue2.add(event)); + + hostPort = httpPrefix.substring(httpPrefix.indexOf("http://"), httpPrefix.length() - 1); } /** @@ -127,6 +147,8 @@ public class EndToEndTest extends CommonRestServer { */ @Before public void setUp() { + ++consumerName; + queue.clear(); queue2.clear(); } @@ -177,11 +199,16 @@ public class EndToEndTest extends CommonRestServer { */ private void test(String testName, String mediaType, BiConsumer> writeMessages) throws Exception { + + /* + * Force consumer name to be registered with the server by attempting to fetch a message. + */ + buildConsumer(0).fetch(); + String msg1 = "{'abc':10.0}".replace('\'', '"'); String msg2 = "{'def':20.0}".replace('\'', '"'); - TopicParameters sinkcfg = topicConfig.getTopicSinks().get(0); - URL url = new URL(httpPrefix + "events/" + sinkcfg.getTopic()); + URL url = new URL(httpPrefix + "events/" + TOPIC3); HttpURLConnection conn = (HttpURLConnection) url.openConnection(); conn.setRequestMethod("POST"); @@ -195,8 +222,29 @@ public class EndToEndTest extends CommonRestServer { assertEquals(testName + " response code", HttpURLConnection.HTTP_OK, conn.getResponseCode()); - assertEquals(testName + " message 1", msg1, queue.poll(MAX_WAIT_SEC, TimeUnit.SECONDS)); - assertEquals(testName + " message 2", msg2, queue.poll(MAX_WAIT_SEC, TimeUnit.SECONDS)); + // fetch the messages + Iterator iter = buildConsumer(1000).fetch().iterator(); + + assertTrue(testName + " have message 1", iter.hasNext()); + assertEquals(testName + " message 1", msg1, iter.next()); + + assertTrue(testName + " have message 2", iter.hasNext()); + assertEquals(testName + " message 2", msg2, iter.next()); + + // no more messages + assertFalse(testName + " extra message", iter.hasNext()); + } + + private CambriaConsumer buildConsumer(int timeoutMs) throws MalformedURLException, GeneralSecurityException { + ConsumerBuilder builder = new CambriaClientBuilders.ConsumerBuilder(); + + builder.knownAs(String.valueOf(consumerName), "my-consumer-id") + .usingHosts(hostPort).onTopic(TOPIC3) + .waitAtServer(timeoutMs).receivingAtMost(5); + + builder.withSocketTimeout(timeoutMs + 2000); + + return builder.build(); } /** -- cgit 1.2.3-korg