Merge "More actor clean-up"
[policy/models.git] / models-interactions / model-actors / actorServiceProvider / src / test / java / org / onap / policy / controlloop / actorserviceprovider / impl / BidirectionalTopicOperationTest.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.impl;
22
23 import static org.assertj.core.api.Assertions.assertThatCode;
24 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
25 import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
26 import static org.junit.Assert.assertEquals;
27 import static org.junit.Assert.assertFalse;
28 import static org.junit.Assert.assertNotNull;
29 import static org.junit.Assert.assertSame;
30 import static org.junit.Assert.assertTrue;
31 import static org.mockito.ArgumentMatchers.any;
32 import static org.mockito.ArgumentMatchers.eq;
33 import static org.mockito.Mockito.never;
34 import static org.mockito.Mockito.verify;
35 import static org.mockito.Mockito.when;
36
37 import java.util.Arrays;
38 import java.util.List;
39 import java.util.concurrent.CompletableFuture;
40 import java.util.function.BiConsumer;
41 import lombok.Getter;
42 import lombok.Setter;
43 import org.junit.Before;
44 import org.junit.Test;
45 import org.mockito.ArgumentCaptor;
46 import org.mockito.Captor;
47 import org.mockito.Mock;
48 import org.mockito.MockitoAnnotations;
49 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
50 import org.onap.policy.common.utils.coder.Coder;
51 import org.onap.policy.common.utils.coder.CoderException;
52 import org.onap.policy.common.utils.coder.StandardCoder;
53 import org.onap.policy.common.utils.coder.StandardCoderObject;
54 import org.onap.policy.common.utils.time.PseudoExecutor;
55 import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
56 import org.onap.policy.controlloop.actorserviceprovider.parameters.BidirectionalTopicConfig;
57 import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
58 import org.onap.policy.controlloop.actorserviceprovider.topic.BidirectionalTopicHandler;
59 import org.onap.policy.controlloop.actorserviceprovider.topic.Forwarder;
60 import org.onap.policy.controlloop.policy.PolicyResult;
61
62 public class BidirectionalTopicOperationTest {
63     private static final CommInfrastructure SINK_INFRA = CommInfrastructure.NOOP;
64     private static final IllegalStateException EXPECTED_EXCEPTION = new IllegalStateException("expected exception");
65     private static final String ACTOR = "my-actor";
66     private static final String OPERATION = "my-operation";
67     private static final String REQ_ID = "my-request-id";
68     private static final String TEXT = "some text";
69     private static final int TIMEOUT_SEC = 10;
70     private static final long TIMEOUT_MS = 1000 * TIMEOUT_SEC;
71     private static final int MAX_REQUESTS = 100;
72
73     private static final StandardCoder coder = new StandardCoder();
74
75     @Mock
76     private BidirectionalTopicConfig config;
77     @Mock
78     private BidirectionalTopicHandler handler;
79     @Mock
80     private Forwarder forwarder;
81
82     @Captor
83     private ArgumentCaptor<BiConsumer<String, StandardCoderObject>> listenerCaptor;
84
85     private ControlLoopOperationParams params;
86     private OperationOutcome outcome;
87     private StandardCoderObject stdResponse;
88     private String responseText;
89     private PseudoExecutor executor;
90     private int ntimes;
91     private BidirectionalTopicOperation<MyRequest, MyResponse> oper;
92
93     /**
94      * Sets up.
95      */
96     @Before
97     public void setUp() throws CoderException {
98         MockitoAnnotations.initMocks(this);
99
100         when(config.getTopicHandler()).thenReturn(handler);
101         when(config.getForwarder()).thenReturn(forwarder);
102         when(config.getTimeoutMs()).thenReturn(TIMEOUT_MS);
103
104         when(handler.send(any())).thenReturn(true);
105         when(handler.getSinkTopicCommInfrastructure()).thenReturn(SINK_INFRA);
106
107         executor = new PseudoExecutor();
108
109         params = ControlLoopOperationParams.builder().actor(ACTOR).operation(OPERATION).executor(executor).build();
110         outcome = params.makeOutcome();
111
112         responseText = coder.encode(new MyResponse());
113         stdResponse = coder.decode(responseText, StandardCoderObject.class);
114
115         ntimes = 1;
116
117         oper = new MyOperation();
118     }
119
120     @Test
121     public void testConstructor_testGetTopicHandler_testGetForwarder_testGetTopicParams() {
122         assertEquals(ACTOR, oper.getActorName());
123         assertEquals(OPERATION, oper.getName());
124         assertSame(handler, oper.getTopicHandler());
125         assertSame(forwarder, oper.getForwarder());
126         assertEquals(TIMEOUT_MS, oper.getTimeoutMs());
127         assertSame(MyResponse.class, oper.getResponseClass());
128     }
129
130     @Test
131     public void testStartOperationAsync() throws Exception {
132
133         // tell it to expect three responses
134         ntimes = 3;
135
136         CompletableFuture<OperationOutcome> future = oper.startOperationAsync(1, outcome);
137         assertFalse(future.isDone());
138
139         verify(forwarder).register(eq(Arrays.asList(REQ_ID)), listenerCaptor.capture());
140
141         verify(forwarder, never()).unregister(any(), any());
142
143         verify(handler).send(any());
144
145         // provide first response
146         listenerCaptor.getValue().accept(responseText, stdResponse);
147         assertTrue(executor.runAll(MAX_REQUESTS));
148         assertFalse(future.isDone());
149
150         // provide second response
151         listenerCaptor.getValue().accept(responseText, stdResponse);
152         assertTrue(executor.runAll(MAX_REQUESTS));
153         assertFalse(future.isDone());
154
155         // provide final response
156         listenerCaptor.getValue().accept(responseText, stdResponse);
157         assertTrue(executor.runAll(MAX_REQUESTS));
158         assertTrue(future.isDone());
159
160         assertSame(outcome, future.get());
161         assertEquals(PolicyResult.SUCCESS, outcome.getResult());
162
163         verify(forwarder).unregister(eq(Arrays.asList(REQ_ID)), eq(listenerCaptor.getValue()));
164     }
165
166     /**
167      * Tests startOperationAsync() when the publisher throws an exception.
168      */
169     @Test
170     public void testStartOperationAsyncException() throws Exception {
171         // indicate that nothing was published
172         when(handler.send(any())).thenReturn(false);
173
174         assertThatIllegalStateException().isThrownBy(() -> oper.startOperationAsync(1, outcome));
175
176         verify(forwarder).register(eq(Arrays.asList(REQ_ID)), listenerCaptor.capture());
177
178         // must still unregister
179         verify(forwarder).unregister(eq(Arrays.asList(REQ_ID)), eq(listenerCaptor.getValue()));
180     }
181
182     @Test
183     public void testGetTimeoutMsInteger() {
184         // use default
185         assertEquals(TIMEOUT_MS, oper.getTimeoutMs(null));
186         assertEquals(TIMEOUT_MS, oper.getTimeoutMs(0));
187
188         // use provided value
189         assertEquals(5000, oper.getTimeoutMs(5));
190     }
191
192     @Test
193     public void testPublishRequest() {
194         assertThatCode(() -> oper.publishRequest(new MyRequest())).doesNotThrowAnyException();
195     }
196
197     /**
198      * Tests publishRequest() when nothing is published.
199      */
200     @Test
201     public void testPublishRequestUnpublished() {
202         when(handler.send(any())).thenReturn(false);
203         assertThatIllegalStateException().isThrownBy(() -> oper.publishRequest(new MyRequest()));
204     }
205
206     /**
207      * Tests publishRequest() when the request type is a String.
208      */
209     @Test
210     public void testPublishRequestString() {
211         MyStringOperation oper2 = new MyStringOperation();
212         assertThatCode(() -> oper2.publishRequest(TEXT)).doesNotThrowAnyException();
213     }
214
215     /**
216      * Tests publishRequest() when the coder throws an exception.
217      */
218     @Test
219     public void testPublishRequestException() {
220         setOperCoderException();
221         assertThatIllegalArgumentException().isThrownBy(() -> oper.publishRequest(new MyRequest()));
222     }
223
224     /**
225      * Tests processResponse() when it's a success and the response type is a String.
226      */
227     @Test
228     public void testProcessResponseSuccessString() {
229         MyStringOperation oper2 = new MyStringOperation();
230
231         assertSame(outcome, oper2.processResponse(outcome, TEXT, null));
232         assertEquals(PolicyResult.SUCCESS, outcome.getResult());
233     }
234
235     /**
236      * Tests processResponse() when it's a success and the response type is a
237      * StandardCoderObject.
238      */
239     @Test
240     public void testProcessResponseSuccessSco() {
241         MyScoOperation oper2 = new MyScoOperation();
242
243         assertSame(outcome, oper2.processResponse(outcome, responseText, stdResponse));
244         assertEquals(PolicyResult.SUCCESS, outcome.getResult());
245     }
246
247     /**
248      * Tests processResponse() when it's a failure.
249      */
250     @Test
251     public void testProcessResponseFailure() throws CoderException {
252         // indicate error in the response
253         MyResponse resp = new MyResponse();
254         resp.setOutput("error");
255
256         responseText = coder.encode(resp);
257         stdResponse = coder.decode(responseText, StandardCoderObject.class);
258
259         assertSame(outcome, oper.processResponse(outcome, responseText, stdResponse));
260         assertEquals(PolicyResult.FAILURE, outcome.getResult());
261     }
262
263     /**
264      * Tests processResponse() when the decoder succeeds.
265      */
266     @Test
267     public void testProcessResponseDecodeOk() throws CoderException {
268         assertSame(outcome, oper.processResponse(outcome, responseText, stdResponse));
269         assertEquals(PolicyResult.SUCCESS, outcome.getResult());
270     }
271
272     /**
273      * Tests processResponse() when the decoder throws an exception.
274      */
275     @Test
276     public void testProcessResponseDecodeExcept() throws CoderException {
277         // @formatter:off
278         assertThatIllegalArgumentException().isThrownBy(
279             () -> oper.processResponse(outcome, "{invalid json", stdResponse));
280         // @formatter:on
281     }
282
283     @Test
284     public void testPostProcessResponse() {
285         assertThatCode(() -> oper.postProcessResponse(outcome, null, null)).doesNotThrowAnyException();
286     }
287
288     @Test
289     public void testMakeCoder() {
290         assertNotNull(oper.makeCoder());
291     }
292
293     /**
294      * Creates a new {@link #oper} whose coder will throw an exception.
295      */
296     private void setOperCoderException() {
297         oper = new MyOperation() {
298             @Override
299             protected Coder makeCoder() {
300                 return new StandardCoder() {
301                     @Override
302                     public String encode(Object object, boolean pretty) throws CoderException {
303                         throw new CoderException(EXPECTED_EXCEPTION);
304                     }
305                 };
306             }
307         };
308     }
309
310     @Getter
311     @Setter
312     public static class MyRequest {
313         private String theRequestId = REQ_ID;
314         private String input;
315     }
316
317     @Getter
318     @Setter
319     public static class MyResponse {
320         private String requestId = REQ_ID;
321         private String output;
322     }
323
324
325     private class MyStringOperation extends BidirectionalTopicOperation<String, String> {
326         public MyStringOperation() {
327             super(BidirectionalTopicOperationTest.this.params, config, String.class);
328         }
329
330         @Override
331         protected String makeRequest(int attempt) {
332             return TEXT;
333         }
334
335         @Override
336         protected List<String> getExpectedKeyValues(int attempt, String request) {
337             return Arrays.asList(REQ_ID);
338         }
339
340         @Override
341         protected Status detmStatus(String rawResponse, String response) {
342             return (response != null ? Status.SUCCESS : Status.FAILURE);
343         }
344     }
345
346
347     private class MyScoOperation extends BidirectionalTopicOperation<MyRequest, StandardCoderObject> {
348         public MyScoOperation() {
349             super(BidirectionalTopicOperationTest.this.params, config, StandardCoderObject.class);
350         }
351
352         @Override
353         protected MyRequest makeRequest(int attempt) {
354             return new MyRequest();
355         }
356
357         @Override
358         protected List<String> getExpectedKeyValues(int attempt, MyRequest request) {
359             return Arrays.asList(REQ_ID);
360         }
361
362         @Override
363         protected Status detmStatus(String rawResponse, StandardCoderObject response) {
364             return (response.getString("output") == null ? Status.SUCCESS : Status.FAILURE);
365         }
366     }
367
368
369     private class MyOperation extends BidirectionalTopicOperation<MyRequest, MyResponse> {
370         public MyOperation() {
371             super(BidirectionalTopicOperationTest.this.params, config, MyResponse.class);
372         }
373
374         @Override
375         protected MyRequest makeRequest(int attempt) {
376             return new MyRequest();
377         }
378
379         @Override
380         protected List<String> getExpectedKeyValues(int attempt, MyRequest request) {
381             return Arrays.asList(REQ_ID);
382         }
383
384         @Override
385         protected Status detmStatus(String rawResponse, MyResponse response) {
386             if (--ntimes <= 0) {
387                 return (response.getOutput() == null ? Status.SUCCESS : Status.FAILURE);
388             }
389
390             return Status.STILL_WAITING;
391         }
392     }
393 }