controlloop/common/rules-test/src/test/java/org/onap/policy/controlloop/common/rules/test/ListenerTest.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP
   4  * ================================================================================
   5  * Copyright (C) 2020-2021 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.junit.runner.RunWith;
  41 import org.mockito.Mock;
  42 import org.mockito.junit.MockitoJUnitRunner;
  43 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
  44 import org.onap.policy.common.endpoints.event.comm.TopicEndpoint;
  45 import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager;
  46 import org.onap.policy.common.endpoints.event.comm.bus.NoopTopicSink;
  47 import org.onap.policy.common.endpoints.parameters.TopicParameters;
  48
  49 @RunWith(MockitoJUnitRunner.class)
  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         when(mgr.getNoopTopicSink(MY_TOPIC)).thenReturn(sink);
  91
  92         listener = new Listener<>(MY_TOPIC, msg -> msg + MSG_SUFFIX) {
  93             @Override
  94             protected TopicEndpoint getTopicManager() {
  95                 return mgr;
  96             }
  97         };
  98     }
  99
 100     @Test
 101     public void testListener() {
 102         verify(sink).register(listener);
 103     }
 104
 105     @Test
 106     public void testAwait_testAwaitLongTimeUnit_testIsEmpty() {
 107         assertTrue(listener.isEmpty());
 108
 109         listener.onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MESSAGE);
 110         assertFalse(listener.isEmpty());
 111
 112         assertEquals(DECODED_MESSAGE, listener.await());
 113
 114         assertTrue(listener.isEmpty());
 115     }
 116
 117     @Test
 118     public void testAwaitPredicateOfT() {
 119         listener.onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MESSAGE);
 120         listener.onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MESSAGE2);
 121         assertEquals(MESSAGE2 + MSG_SUFFIX, listener.await(msg -> msg.startsWith("other-")));
 122     }
 123
 124     /**
 125      * Tests await() when the remaining time is negative.
 126      */
 127     @Test
 128     public void testAwaitLongTimeUnitPredicateNoTime() {
 129         assertThatThrownBy(() -> listener.await(-1, TimeUnit.SECONDS)).isInstanceOf(TopicException.class);
 130     }
 131
 132     /**
 133      * Tests await() when the poll() returns {@code null}.
 134      */
 135     @Test
 136     public void testAwaitLongTimeUnitPredicateNoMessage() {
 137         assertThatThrownBy(() -> listener.await(1, TimeUnit.MILLISECONDS)).isInstanceOf(TopicException.class);
 138     }
 139
 140     /**
 141      * Tests await() when the poll() is interrupted.
 142      */
 143     @Test
 144     public void testAwaitLongTimeUnitPredicateInterrupted() throws InterruptedException {
 145         listener = new Listener<String>(MY_TOPIC, msg -> msg) {
 146             @Override
 147             protected String pollMessage(long remainingMs) throws InterruptedException {
 148                 throw new InterruptedException(EXPECTED_EXCEPTION);
 149             }
 150         };
 151
 152         AtomicReference<TopicException> exref = new AtomicReference<>();
 153         CountDownLatch interrupted = new CountDownLatch(1);
 154
 155         Thread thread = new Thread() {
 156             @Override
 157             public void run() {
 158                 try {
 159                     listener.await();
 160                 } catch (TopicException e) {
 161                     exref.set(e);
 162                 }
 163
 164                 if (Thread.currentThread().isInterrupted()) {
 165                     interrupted.countDown();
 166                 }
 167             }
 168         };
 169
 170         thread.start();
 171         assertTrue(interrupted.await(5, TimeUnit.SECONDS));
 172         assertNotNull(exref.get());
 173     }
 174
 175     @Test
 176     public void testUnregister() {
 177         listener.unregister();
 178         verify(sink).unregister(listener);
 179     }
 180
 181     @Test
 182     public void testOnTopicEvent() {
 183         listener = new Listener<>(MY_TOPIC, msg -> {
 184             throw new IllegalArgumentException(EXPECTED_EXCEPTION);
 185         });
 186
 187         // onTopicEvent() should not throw an exception
 188         assertThatCode(() -> listener.onTopicEvent(CommInfrastructure.NOOP, MY_TOPIC, MESSAGE))
 189                         .doesNotThrowAnyException();
 190
 191         // should not have queued a message
 192         assertTrue(listener.isEmpty());
 193     }
 194
 195     @Test
 196     public void testGetTopicManager() {
 197         // use a listener with a real manager
 198         assertNotNull(new Listener<>(MY_TOPIC, msg -> msg).getTopicManager());
 199     }
 200 }