gerrit.onap Code Review - policy/drools-applications.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP
   4  * ================================================================================
   5  * Copyright (C) 2020 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.controlloop.common.rules.test;
  22
  23 import static org.assertj.core.api.Assertions.assertThatCode;
  24 import static org.assertj.core.api.Assertions.assertThatThrownBy;
  25 import static org.junit.Assert.assertEquals;
  26 import static org.junit.Assert.assertFalse;
  27 import static org.junit.Assert.assertNotNull;
  28 import static org.junit.Assert.assertTrue;
  29 import static org.mockito.Mockito.verify;
  30 import static org.mockito.Mockito.when;
  31
  32 import java.util.List;
  33 import java.util.concurrent.CountDownLatch;
  34 import java.util.concurrent.TimeUnit;
  35 import java.util.concurrent.atomic.AtomicReference;
  36 import org.junit.AfterClass;
  37 import org.junit.Before;
  38 import org.junit.BeforeClass;
  39 import org.junit.Test;
  40 import org.mockito.Mock;
  41 import org.mockito.MockitoAnnotations;
  42 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
  43 import org.onap.policy.common.endpoints.event.comm.TopicEndpoint;
  44 import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager;
  45 import org.onap.policy.common.endpoints.event.comm.bus.NoopTopicSink;
  46 import org.onap.policy.common.endpoints.parameters.TopicParameters;
  47 import org.onap.policy.controlloop.common.rules.test.Listener;
  48 import org.onap.policy.controlloop.common.rules.test.TopicException;
  49
  50 public class ListenerTest {
  51     private static final String EXPECTED_EXCEPTION = "expected exception";
  52     private static final String MY_TOPIC = "my-topic";
  53     private static final String MESSAGE = "the-message";
  54     private static final String MESSAGE2 = "other-message";
  55     private static final String MSG_SUFFIX = "s";
  56     private static final String DECODED_MESSAGE = MESSAGE + MSG_SUFFIX;
  57
  58     @Mock
  59     private NoopTopicSink sink;
  60     @Mock
  61     private TopicEndpoint mgr;
  62
  63     private Listener<String> listener;
  64
  65     /**
  66      * Creates topics.
  67      */
  68     @BeforeClass
  69     public static void setUpBeforeClass() {
  70         TopicEndpointManager.getManager().shutdown();
  71
  72         TopicParameters params = new TopicParameters();
  73         params.setTopic(MY_TOPIC);
  74         params.setManaged(true);
  75         params.setTopicCommInfrastructure("NOOP");
  76
  77         TopicEndpointManager.getManager().addTopicSinks(List.of(params));
  78     }
  79
  80     @AfterClass
  81     public static void tearDownAfterClass() {
  82         TopicEndpointManager.getManager().shutdown();
  83     }
  84
  85     /**
  86      * Sets up.
  87      */
  88     @Before
  89     public void setUp() {
  90         MockitoAnnotations.initMocks(this);
  91
  92         when(mgr.getNoopTopicSink(MY_TOPIC)).thenReturn(sink);
  93
  94         listener = new Listener<>(MY_TOPIC, msg -> msg + MSG_SUFFIX) {
  95             @Override
  96             protected TopicEndpoint getTopicManager() {
  97                 return mgr;
  98             }
  99         };
 100     }
 101
 102     @Test
 103     public void testListener() {
 104         verify(sink).register(listener);
 105     }
 106
 107     @Test
 108     public void testAwait_testAwaitLongTimeUnit_testIsEmpty() {
 109         assertTrue(listener.isEmpty());
 110
 111         listener.onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MESSAGE);
 112         assertFalse(listener.isEmpty());
 113
 114         assertEquals(DECODED_MESSAGE, listener.await());
 115
 116         assertTrue(listener.isEmpty());
 117     }
 118
 119     @Test
 120     public void testAwaitPredicateOfT() {
 121         listener.onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MESSAGE);
 122         listener.onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MESSAGE2);
 123         assertEquals(MESSAGE2 + MSG_SUFFIX, listener.await(msg -> msg.startsWith("other-")));
 124     }
 125
 126     /**
 127      * Tests await() when the remaining time is negative.
 128      */
 129     @Test
 130     public void testAwaitLongTimeUnitPredicateNoTime() {
 131         assertThatThrownBy(() -> listener.await(-1, TimeUnit.SECONDS)).isInstanceOf(TopicException.class);
 132     }
 133
 134     /**
 135      * Tests await() when the poll() returns {@code null}.
 136      */
 137     @Test
 138     public void testAwaitLongTimeUnitPredicateNoMessage() {
 139         assertThatThrownBy(() -> listener.await(1, TimeUnit.MILLISECONDS)).isInstanceOf(TopicException.class);
 140     }
 141
 142     /**
 143      * Tests await() when the poll() is interrupted.
 144      */
 145     @Test
 146     public void testAwaitLongTimeUnitPredicateInterrupted() throws InterruptedException {
 147         listener = new Listener<String>(MY_TOPIC, msg -> msg) {
 148             @Override
 149             protected String pollMessage(long remainingMs) throws InterruptedException {
 150                 throw new InterruptedException(EXPECTED_EXCEPTION);
 151             }
 152         };
 153
 154         AtomicReference<TopicException> exref = new AtomicReference<>();
 155         CountDownLatch interrupted = new CountDownLatch(1);
 156
 157         Thread thread = new Thread() {
 158             @Override
 159             public void run() {
 160                 try {
 161                     listener.await();
 162                 } catch (TopicException e) {
 163                     exref.set(e);
 164                 }
 165
 166                 if (Thread.currentThread().isInterrupted()) {
 167                     interrupted.countDown();
 168                 }
 169             }
 170         };
 171
 172         thread.start();
 173         assertTrue(interrupted.await(5, TimeUnit.SECONDS));
 174         assertNotNull(exref.get());
 175     }
 176
 177     @Test
 178     public void testUnregister() {
 179         listener.unregister();
 180         verify(sink).unregister(listener);
 181     }
 182
 183     @Test
 184     public void testOnTopicEvent() {
 185         listener = new Listener<>(MY_TOPIC, msg -> {
 186             throw new IllegalArgumentException(EXPECTED_EXCEPTION);
 187         });
 188
 189         // onTopicEvent() should not throw an exception
 190         assertThatCode(() -> listener.onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MESSAGE))
 191                         .doesNotThrowAnyException();
 192
 193         // should not have queued a message
 194         assertTrue(listener.isEmpty());
 195     }
 196
 197     @Test
 198     public void testGetTopicManager() {
 199         // use a listener with a real manager
 200         assertNotNull(new Listener<>(MY_TOPIC, msg -> msg).getTopicManager());
 201     }
 202 }