models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/topic/ForwarderTest.java

   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.actorserviceprovider.topic;
  22
  23 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
  24 import static org.mockito.ArgumentMatchers.any;
  25 import static org.mockito.Mockito.doThrow;
  26 import static org.mockito.Mockito.never;
  27 import static org.mockito.Mockito.times;
  28 import static org.mockito.Mockito.verify;
  29
  30 import java.util.Arrays;
  31 import java.util.Map;
  32 import org.junit.Before;
  33 import org.junit.Test;
  34 import org.mockito.Mock;
  35 import org.mockito.MockitoAnnotations;
  36 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
  37 import org.onap.policy.common.endpoints.utils.PropertyUtils.TriConsumer;
  38 import org.onap.policy.common.utils.coder.StandardCoderObject;
  39 import org.onap.policy.controlloop.actorserviceprovider.Util;
  40
  41 public class ForwarderTest {
  42     private static final CommInfrastructure INFRA = CommInfrastructure.NOOP;
  43     private static final String TEXT = "some text";
  44
  45     private static final String KEY1 = "requestId";
  46     private static final String KEY2 = "container";
  47     private static final String SUBKEY = "subRequestId";
  48
  49     private static final String VALUEA_REQID = "hello";
  50     private static final String VALUEA_SUBREQID = "world";
  51
  52     // request id is shared with value A
  53     private static final String VALUEB_REQID = "hello";
  54     private static final String VALUEB_SUBREQID = "another world";
  55
  56     // unique values
  57     private static final String VALUEC_REQID = "bye";
  58     private static final String VALUEC_SUBREQID = "bye-bye";
  59
  60     @Mock
  61     private TriConsumer<CommInfrastructure, String, StandardCoderObject> listener1;
  62
  63     @Mock
  64     private TriConsumer<CommInfrastructure, String, StandardCoderObject> listener1b;
  65
  66     @Mock
  67     private TriConsumer<CommInfrastructure, String, StandardCoderObject> listener2;
  68
  69     @Mock
  70     private TriConsumer<CommInfrastructure, String, StandardCoderObject> listener3;
  71
  72     private Forwarder forwarder;
  73
  74
  75     /**
  76      * Sets up.
  77      */
  78     @Before
  79     public void setUp() {
  80         MockitoAnnotations.initMocks(this);
  81         forwarder = new Forwarder(Arrays.asList(new SelectorKey(KEY1), new SelectorKey(KEY2, SUBKEY)));
  82
  83         forwarder.register(Arrays.asList(VALUEA_REQID, VALUEA_SUBREQID), listener1);
  84         forwarder.register(Arrays.asList(VALUEA_REQID, VALUEA_SUBREQID), listener1b);
  85         forwarder.register(Arrays.asList(VALUEB_REQID, VALUEB_SUBREQID), listener2);
  86         forwarder.register(Arrays.asList(VALUEC_REQID, VALUEC_SUBREQID), listener3);
  87     }
  88
  89     @Test
  90     public void testRegister() {
  91         // key size mismatches
  92         assertThatIllegalArgumentException().isThrownBy(() -> forwarder.register(Arrays.asList(), listener1))
  93                         .withMessage("key/value mismatch");
  94         assertThatIllegalArgumentException()
  95                         .isThrownBy(() -> forwarder.register(Arrays.asList(VALUEA_REQID), listener1))
  96                         .withMessage("key/value mismatch");
  97     }
  98
  99     @Test
 100     public void testUnregister() {
 101         // remove listener1b
 102         forwarder.unregister(Arrays.asList(VALUEA_REQID, VALUEA_SUBREQID), listener1b);
 103
 104         StandardCoderObject sco = makeMessage(Map.of(KEY1, VALUEA_REQID, KEY2, Map.of(SUBKEY, VALUEA_SUBREQID)));
 105         forwarder.onMessage(INFRA, TEXT, sco);
 106
 107         verify(listener1).accept(INFRA, TEXT, sco);
 108         verify(listener1b, never()).accept(any(), any(), any());
 109
 110         // remove listener1
 111         forwarder.unregister(Arrays.asList(VALUEA_REQID, VALUEA_SUBREQID), listener1);
 112         forwarder.onMessage(INFRA, TEXT, sco);
 113
 114         // route a message to listener2
 115         sco = makeMessage(Map.of(KEY1, VALUEB_REQID, KEY2, Map.of(SUBKEY, VALUEB_SUBREQID)));
 116         forwarder.onMessage(INFRA, TEXT, sco);
 117         verify(listener2).accept(INFRA, TEXT, sco);
 118
 119         // no more messages to listener1 or 1b
 120         verify(listener1).accept(any(), any(), any());
 121         verify(listener1b, never()).accept(any(), any(), any());
 122     }
 123
 124     @Test
 125     public void testOnMessage() {
 126         StandardCoderObject sco = makeMessage(Map.of(KEY1, VALUEA_REQID, KEY2, Map.of(SUBKEY, VALUEA_SUBREQID)));
 127         forwarder.onMessage(INFRA, TEXT, sco);
 128
 129         verify(listener1).accept(INFRA, TEXT, sco);
 130         verify(listener1b).accept(INFRA, TEXT, sco);
 131
 132         // repeat - counts should increment
 133         forwarder.onMessage(INFRA, TEXT, sco);
 134
 135         verify(listener1, times(2)).accept(INFRA, TEXT, sco);
 136         verify(listener1b, times(2)).accept(INFRA, TEXT, sco);
 137
 138         // should not have been invoked
 139         verify(listener2, never()).accept(any(), any(), any());
 140         verify(listener3, never()).accept(any(), any(), any());
 141
 142         // try other listeners now
 143         sco = makeMessage(Map.of(KEY1, VALUEB_REQID, KEY2, Map.of(SUBKEY, VALUEB_SUBREQID)));
 144         forwarder.onMessage(INFRA, TEXT, sco);
 145         verify(listener2).accept(INFRA, TEXT, sco);
 146
 147         sco = makeMessage(Map.of(KEY1, VALUEC_REQID, KEY2, Map.of(SUBKEY, VALUEC_SUBREQID)));
 148         forwarder.onMessage(INFRA, TEXT, sco);
 149         verify(listener3).accept(INFRA, TEXT, sco);
 150
 151         // message has no listeners
 152         sco = makeMessage(Map.of(KEY1, "xyzzy", KEY2, Map.of(SUBKEY, VALUEB_SUBREQID)));
 153         forwarder.onMessage(INFRA, TEXT, sco);
 154
 155         // message doesn't have both keys
 156         sco = makeMessage(Map.of(KEY1, VALUEA_REQID));
 157         forwarder.onMessage(INFRA, TEXT, sco);
 158
 159         // counts should not have incremented
 160         verify(listener1, times(2)).accept(any(), any(), any());
 161         verify(listener1b, times(2)).accept(any(), any(), any());
 162         verify(listener2).accept(any(), any(), any());
 163         verify(listener3).accept(any(), any(), any());
 164
 165         // listener throws an exception
 166         doThrow(new IllegalStateException("expected exception")).when(listener1).accept(any(), any(), any());
 167     }
 168
 169     /*
 170      * Tests onMessage() when listener1 throws an exception.
 171      */
 172     @Test
 173     public void testOnMessageListenerException1() {
 174         doThrow(new IllegalStateException("expected exception")).when(listener1).accept(any(), any(), any());
 175
 176         StandardCoderObject sco = makeMessage(Map.of(KEY1, VALUEA_REQID, KEY2, Map.of(SUBKEY, VALUEA_SUBREQID)));
 177         forwarder.onMessage(INFRA, TEXT, sco);
 178
 179         verify(listener1b).accept(INFRA, TEXT, sco);
 180     }
 181
 182     /*
 183      * Tests onMessage() when listener1b throws an exception.
 184      */
 185     @Test
 186     public void testOnMessageListenerException1b() {
 187         doThrow(new IllegalStateException("expected exception")).when(listener1b).accept(any(), any(), any());
 188
 189         StandardCoderObject sco = makeMessage(Map.of(KEY1, VALUEA_REQID, KEY2, Map.of(SUBKEY, VALUEA_SUBREQID)));
 190         forwarder.onMessage(INFRA, TEXT, sco);
 191
 192         verify(listener1).accept(INFRA, TEXT, sco);
 193     }
 194
 195     /**
 196      * Makes a message from a map.
 197      */
 198     private StandardCoderObject makeMessage(Map<String, Object> map) {
 199         return Util.translate("", map, StandardCoderObject.class);
 200     }
 201 }