diff options
4 files changed, 252 insertions, 17 deletions
diff --git a/feature-simulators/pom.xml b/feature-simulators/pom.xml index d4ecf7a7..7e7a2064 100644 --- a/feature-simulators/pom.xml +++ b/feature-simulators/pom.xml @@ -44,6 +44,12 @@ <scope>test</scope> </dependency> <dependency> + <groupId>org.mockito</groupId> + <artifactId>mockito-core</artifactId> + <version>2.13.0</version> + <scope>test</scope> + </dependency> + <dependency> <groupId>org.onap.policy.common</groupId> <artifactId>policy-endpoints</artifactId> <version>${policy.common.version}</version> diff --git a/feature-simulators/src/main/java/org/onap/policy/drools/simulators/DMaaPSimulatorJaxRs.java b/feature-simulators/src/main/java/org/onap/policy/drools/simulators/DMaaPSimulatorJaxRs.java index 5e7861d2..8b653aae 100644 --- a/feature-simulators/src/main/java/org/onap/policy/drools/simulators/DMaaPSimulatorJaxRs.java +++ b/feature-simulators/src/main/java/org/onap/policy/drools/simulators/DMaaPSimulatorJaxRs.java @@ -1,8 +1,8 @@ -/* +/*- * ============LICENSE_START======================================================= * feature-simulators * ================================================================================ - * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved. + * Copyright (C) 2017-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. @@ -43,14 +43,16 @@ import org.slf4j.LoggerFactory; @Path("/events") public class DMaaPSimulatorJaxRs { - private static final String NO_DATA_MSG = "No Data"; + public static final String NO_TOPIC_MSG = "No topic"; + public static final String NO_DATA_MSG = "No Data"; + private static final Map<String, BlockingQueue<String>> queues = new ConcurrentHashMap<>(); private static final Logger logger = LoggerFactory.getLogger(DMaaPSimulatorJaxRs.class); private static int responseCode = 200; /** * Get consumer ID. - * + * * @param timeout timeout value * @param topicName the dmaap topic * @param httpResponse http response object @@ -58,7 +60,7 @@ public class DMaaPSimulatorJaxRs { */ @GET @Path("/{topicName}/{consumeGroup}/{consumerId}") - public String subscribe(@DefaultValue("0") @QueryParam("timeout") int timeout, @PathParam("topicName") + public String subscribe(@DefaultValue("0") @QueryParam("timeout") int timeout, @PathParam("topicName") String topicName, @Context final HttpServletResponse httpResponse) { int currentRespCode = responseCode; httpResponse.setStatus(currentRespCode); @@ -79,14 +81,14 @@ public class DMaaPSimulatorJaxRs { else if (timeout > 0) { return waitForNextMessageFromQueue(timeout, topicName); } - return "No topic"; + return NO_TOPIC_MSG; } private String getNextMessageFromQueue(final int timeout, final String topicName) { BlockingQueue<String> queue = queues.get(topicName); String response = NO_DATA_MSG; try { - response = queue.poll(timeout, TimeUnit.MILLISECONDS); + response = poll(queue, timeout); } catch (InterruptedException e) { logger.debug("error in DMaaP simulator", e); Thread.currentThread().interrupt(); @@ -99,7 +101,7 @@ public class DMaaPSimulatorJaxRs { private String waitForNextMessageFromQueue(int timeout, String topicName) { try { - Thread.sleep(timeout); + sleep(timeout); if (queues.containsKey(topicName)) { BlockingQueue<String> queue = queues.get(topicName); String response = queue.poll(); @@ -112,12 +114,12 @@ public class DMaaPSimulatorJaxRs { logger.debug("error in DMaaP simulator", e); Thread.currentThread().interrupt(); } - return "No topic"; + return NO_TOPIC_MSG; } /** * Post to a topic. - * + * * @param topicName name of the topic * @param body message * @return empty string @@ -142,12 +144,30 @@ public class DMaaPSimulatorJaxRs { return "Status code set"; } + // the following non-static methods may be overridden by junit tests + + protected String poll(BlockingQueue<String> queue, final int timeout) throws InterruptedException { + return queue.poll(timeout, TimeUnit.MILLISECONDS); + } + + protected void sleep(int timeout) throws InterruptedException { + Thread.sleep(timeout); + } + /** * Static method to set static response code, synchronized for multiple possible uses. - * + * * @param incomingResponseCode the response code to set */ private static synchronized void setResponseCode(final int incomingResponseCode) { - responseCode = incomingResponseCode; + responseCode = incomingResponseCode; + } + + /** + * Used only by junit tests to reset the simulator. + */ + protected static void reset() { + responseCode = 200; + queues.clear(); } } diff --git a/feature-simulators/src/test/java/org/onap/policy/drools/simulators/DMaaPSimulatorJaxRsTest.java b/feature-simulators/src/test/java/org/onap/policy/drools/simulators/DMaaPSimulatorJaxRsTest.java new file mode 100644 index 00000000..7200bdce --- /dev/null +++ b/feature-simulators/src/test/java/org/onap/policy/drools/simulators/DMaaPSimulatorJaxRsTest.java @@ -0,0 +1,207 @@ +/*- + * ============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.drools.simulators; + +import static org.junit.Assert.assertEquals; +import static org.mockito.Mockito.doThrow; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingQueue; +import javax.servlet.http.HttpServletResponse; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +public class DMaaPSimulatorJaxRsTest { + private static final String MESSAGE = "hello"; + private static final String MESSAGE2 = "world"; + private static final String EXPECTED_EXCEPTION = "expected exception"; + private static final String TOPIC = "my-topic"; + private static final int TIMEOUT_MS = 10; + private static final int LONG_TIMEOUT_MS = 250; + + @Mock + private HttpServletResponse resp; + + private DMaaPSimulatorJaxRs sim; + + /** + * Initializes objects and creates the simulator. + */ + @Before + public void setUp() { + MockitoAnnotations.initMocks(this); + + sim = new DMaaPSimulatorJaxRs(); + } + + @After + public void tearDown() { + DMaaPSimulatorJaxRs.reset(); + } + + @Test + public void testSubscribe() { + sim.publish(TOPIC, MESSAGE); + + assertEquals(MESSAGE, sim.subscribe(0, TOPIC, resp)); + } + + @Test + public void testSubscribe_FlushEx() throws IOException { + doThrow(new IOException(EXPECTED_EXCEPTION)).when(resp).flushBuffer(); + + assertEquals("Got an error", sim.subscribe(TIMEOUT_MS, TOPIC, resp)); + } + + @Test + public void testSubscribe_BadStatus_testSetResponseCode() { + sim.setStatus(199); + assertEquals("You got response code: 199", sim.subscribe(TIMEOUT_MS, TOPIC, resp)); + + sim.setStatus(300); + assertEquals("You got response code: 300", sim.subscribe(TIMEOUT_MS, TOPIC, resp)); + } + + @Test + public void testSubscribe_UnknownTopic_ZeroTimeout() { + assertEquals(DMaaPSimulatorJaxRs.NO_TOPIC_MSG, sim.subscribe(0, TOPIC, resp)); + } + + @Test + public void testSubscribe_UnknownTopic_NonZeroTimeout() { + assertEquals(DMaaPSimulatorJaxRs.NO_TOPIC_MSG, sim.subscribe(TIMEOUT_MS, TOPIC, resp)); + } + + @Test + public void testGetNextMessageFromQueue() { + sim.publish(TOPIC, MESSAGE); + sim.publish(TOPIC, MESSAGE2); + + assertEquals(MESSAGE, sim.subscribe(0, TOPIC, resp)); + assertEquals(MESSAGE2, sim.subscribe(0, TOPIC, resp)); + + // repeat - no message + assertEquals(DMaaPSimulatorJaxRs.NO_DATA_MSG, sim.subscribe(0, TOPIC, resp)); + } + + @Test + public void testGetNextMessageFromQueue_Interrupted() throws InterruptedException { + sim = new DMaaPSimulatorJaxRs() { + @Override + protected String poll(BlockingQueue<String> queue, int timeout) throws InterruptedException { + throw new InterruptedException(EXPECTED_EXCEPTION); + } + }; + + sim.publish(TOPIC, MESSAGE); + + // put it in the background so we don't interrupt the test thread + BlockingQueue<String> queue = new LinkedBlockingQueue<>(); + backgroundSubscribe(queue); + + assertEquals(DMaaPSimulatorJaxRs.NO_DATA_MSG, queue.take()); + } + + @Test + public void testWaitForNextMessageFromQueue() throws InterruptedException { + BlockingQueue<String> queue = new LinkedBlockingQueue<>(); + + CountDownLatch latch1 = backgroundSubscribe(queue); + CountDownLatch latch2 = backgroundSubscribe(queue); + + // wait for both threads to start + latch1.await(); + latch2.await(); + + /* + * Must pause to prevent the topic from being created before subscribe() is + * invoked. + */ + Thread.sleep(LONG_TIMEOUT_MS / 3); + + // only publish one message + sim.publish(TOPIC, MESSAGE); + + // wait for both subscribers to add their messages to the queue + List<String> messages = new ArrayList<>(); + messages.add(queue.take()); + messages.add(queue.take()); + + // sort them so the order is consistent + Collections.sort(messages); + + assertEquals("[No Data, hello]", messages.toString()); + } + + @Test + public void testWaitForNextMessageFromQueue_Interrupted() throws InterruptedException { + sim = new DMaaPSimulatorJaxRs() { + @Override + protected void sleep(int timeout) throws InterruptedException { + throw new InterruptedException(EXPECTED_EXCEPTION); + } + }; + + // put it in the background so we don't interrupt the test thread + BlockingQueue<String> queue = new LinkedBlockingQueue<>(); + backgroundSubscribe(queue); + + assertEquals(DMaaPSimulatorJaxRs.NO_TOPIC_MSG, queue.take()); + } + + @Test + public void testPublish() { + assertEquals("", sim.publish(TOPIC, MESSAGE)); + assertEquals(MESSAGE, sim.subscribe(0, TOPIC, resp)); + } + + @Test + public void testSetStatus() { + assertEquals("Status code set", sim.setStatus(500)); + assertEquals("You got response code: 500", sim.subscribe(TIMEOUT_MS, TOPIC, resp)); + } + + /** + * Invokes subscribe() in a background thread. + * + * @param queue where to place the returned result + * @return a latch that will be counted down just before the background thread invokes + * subscribe() + */ + private CountDownLatch backgroundSubscribe(BlockingQueue<String> queue) { + CountDownLatch latch = new CountDownLatch(1); + + new Thread(() -> { + latch.countDown(); + queue.add(sim.subscribe(LONG_TIMEOUT_MS, TOPIC, resp)); + }).start(); + + return latch; + } +} diff --git a/feature-simulators/src/test/java/org/onap/policy/drools/simulators/DMaaPSimulatorTest.java b/feature-simulators/src/test/java/org/onap/policy/drools/simulators/DMaaPSimulatorTest.java index 9740ac6d..bb1b1cb9 100644 --- a/feature-simulators/src/test/java/org/onap/policy/drools/simulators/DMaaPSimulatorTest.java +++ b/feature-simulators/src/test/java/org/onap/policy/drools/simulators/DMaaPSimulatorTest.java @@ -43,15 +43,17 @@ import org.onap.policy.drools.utils.logging.LoggerUtil; public class DMaaPSimulatorTest { - private static final int DMAAPSIM_SERVER_PORT = 6670; + private static int DMAAPSIM_SERVER_PORT; /** * Setup the simulator. + * @throws IOException if a server port cannot be allocated */ @BeforeClass - public static void setUpSimulator() { + public static void setUpSimulator() throws IOException { LoggerUtil.setLevel("ROOT", "INFO"); LoggerUtil.setLevel("org.eclipse.jetty", "WARN"); + DMAAPSIM_SERVER_PORT = NetworkUtil.allocPort(); try { final HttpServletServer testServer = HttpServletServerFactoryInstance.getServerFactory().build("dmaapSim", "localhost", DMAAPSIM_SERVER_PORT, "/", false, true); @@ -76,7 +78,7 @@ public class DMaaPSimulatorTest { Pair<Integer, String> response = dmaapGet("myTopicNoData", timeout); assertNotNull(response); assertNotNull(response.first); - assertEquals("No topic", response.second); + assertEquals(DMaaPSimulatorJaxRs.NO_TOPIC_MSG, response.second); } @Test @@ -148,7 +150,7 @@ public class DMaaPSimulatorTest { response = dmaapGet(topics[1], 1000); assertNotNull(response); assertNotNull(response.first); - assertEquals("No topic", response.second); + assertEquals(DMaaPSimulatorJaxRs.NO_TOPIC_MSG, response.second); response = dmaapPost(topics[1], data[1][0]); assertNotNull(response); @@ -183,7 +185,7 @@ public class DMaaPSimulatorTest { response = dmaapGet(topics[0], 1000); assertNotNull(response); assertNotNull(response.first); - assertEquals("No Data", response.second); + assertEquals(DMaaPSimulatorJaxRs.NO_DATA_MSG, response.second); } @Test |