a1-policy-management/src/test/java/org/onap/ccsdk/oran/a1policymanagementservice/dmaap/DmaapMessageConsumerTest.java

   1 /*-
   2  * ========================LICENSE_START=================================
   3  * ONAP : ccsdk oran
   4  * ======================================================================
   5  * Copyright (C) 2020 Nordix Foundation. 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.ccsdk.oran.a1policymanagementservice.dmaap;
  22
  23 import static ch.qos.logback.classic.Level.WARN;
  24 import static org.assertj.core.api.Assertions.assertThat;
  25 import static org.junit.jupiter.api.Assertions.assertNotNull;
  26 import static org.junit.jupiter.api.Assertions.assertThrows;
  27 import static org.junit.jupiter.api.Assertions.assertTrue;
  28 import static org.mockito.ArgumentMatchers.any;
  29 import static org.mockito.Mockito.doNothing;
  30 import static org.mockito.Mockito.doReturn;
  31 import static org.mockito.Mockito.inOrder;
  32 import static org.mockito.Mockito.spy;
  33 import static org.mockito.Mockito.verify;
  34 import static org.mockito.Mockito.verifyNoMoreInteractions;
  35 import static org.mockito.Mockito.when;
  36
  37 import ch.qos.logback.classic.spi.ILoggingEvent;
  38 import ch.qos.logback.core.read.ListAppender;
  39
  40 import com.google.gson.Gson;
  41 import com.google.gson.GsonBuilder;
  42 import com.google.gson.JsonObject;
  43
  44 import java.time.Duration;
  45 import java.util.LinkedList;
  46 import java.util.List;
  47
  48 import org.junit.jupiter.api.AfterEach;
  49 import org.junit.jupiter.api.Test;
  50 import org.junit.jupiter.api.extension.ExtendWith;
  51 import org.mockito.ArgumentCaptor;
  52 import org.mockito.InOrder;
  53 import org.mockito.Mock;
  54 import org.mockito.junit.jupiter.MockitoExtension;
  55 import org.onap.ccsdk.oran.a1policymanagementservice.clients.AsyncRestClient;
  56 import org.onap.ccsdk.oran.a1policymanagementservice.configuration.ApplicationConfig;
  57 import org.onap.ccsdk.oran.a1policymanagementservice.dmaap.DmaapRequestMessage.Operation;
  58 import org.onap.ccsdk.oran.a1policymanagementservice.exceptions.ServiceException;
  59 import org.onap.ccsdk.oran.a1policymanagementservice.utils.LoggingUtils;
  60 import org.springframework.http.HttpStatus;
  61 import org.springframework.http.ResponseEntity;
  62 import reactor.core.publisher.Mono;
  63
  64 @ExtendWith(MockitoExtension.class)
  65 class DmaapMessageConsumerTest {
  66     @Mock
  67     private ApplicationConfig applicationConfigMock;
  68     @Mock
  69     private AsyncRestClient messageRouterConsumerMock;
  70     @Mock
  71     private DmaapMessageHandler messageHandlerMock;
  72
  73     private DmaapMessageConsumer messageConsumerUnderTest;
  74
  75     private Gson gson = new GsonBuilder().create();
  76
  77     @AfterEach
  78     void resetLogging() {
  79         LoggingUtils.getLogListAppender(DmaapMessageConsumer.class);
  80     }
  81
  82     @Test
  83     void dmaapNotConfigured_thenSleepAndRetryUntilConfig() throws Exception {
  84         messageConsumerUnderTest = spy(new DmaapMessageConsumer(applicationConfigMock));
  85
  86         doNothing().when(messageConsumerUnderTest).sleep(any(Duration.class));
  87         doReturn(false, false, false, true).when(messageConsumerUnderTest).isStopped();
  88         doReturn(false, true, true).when(messageConsumerUnderTest).isDmaapConfigured();
  89         doReturn(new LinkedList<>()).when(messageConsumerUnderTest).fetchAllMessages();
  90
  91         messageConsumerUnderTest.start().join();
  92
  93         InOrder orderVerifier = inOrder(messageConsumerUnderTest);
  94         orderVerifier.verify(messageConsumerUnderTest).sleep(DmaapMessageConsumer.TIME_BETWEEN_DMAAP_RETRIES);
  95         orderVerifier.verify(messageConsumerUnderTest).fetchAllMessages();
  96     }
  97
  98     @Test
  99     void dmaapConfigurationRemoved_thenStopPollingDmaapSleepAndRetry() throws Exception {
 100         messageConsumerUnderTest = spy(new DmaapMessageConsumer(applicationConfigMock));
 101
 102         doNothing().when(messageConsumerUnderTest).sleep(any(Duration.class));
 103         doReturn(false, false, false, false, true).when(messageConsumerUnderTest).isStopped();
 104         doReturn(true, true, false).when(messageConsumerUnderTest).isDmaapConfigured();
 105         doReturn(new LinkedList<>()).when(messageConsumerUnderTest).fetchAllMessages();
 106
 107         messageConsumerUnderTest.start().join();
 108
 109         InOrder orderVerifier = inOrder(messageConsumerUnderTest);
 110         orderVerifier.verify(messageConsumerUnderTest).fetchAllMessages();
 111         orderVerifier.verify(messageConsumerUnderTest).sleep(DmaapMessageConsumer.TIME_BETWEEN_DMAAP_RETRIES);
 112     }
 113
 114     @Test
 115     void dmaapConfiguredAndNoMessages_thenPollOnce() throws Exception {
 116         setUpMrConfig();
 117
 118         messageConsumerUnderTest = spy(new DmaapMessageConsumer(applicationConfigMock));
 119
 120         Mono<ResponseEntity<String>> response = Mono.empty();
 121
 122         doReturn(false, true).when(messageConsumerUnderTest).isStopped();
 123         doReturn(messageRouterConsumerMock).when(messageConsumerUnderTest).getMessageRouterConsumer();
 124         doReturn(response).when(messageRouterConsumerMock).getForEntity(any());
 125
 126         messageConsumerUnderTest.start().join();
 127
 128         verify(messageRouterConsumerMock).getForEntity(any());
 129         verifyNoMoreInteractions(messageRouterConsumerMock);
 130     }
 131
 132     @Test
 133     void dmaapConfiguredAndErrorGettingMessages_thenLogWarningAndSleep() throws Exception {
 134         setUpMrConfig();
 135
 136         messageConsumerUnderTest = spy(new DmaapMessageConsumer(applicationConfigMock));
 137
 138         doNothing().when(messageConsumerUnderTest).sleep(any(Duration.class));
 139         doReturn(false, true).when(messageConsumerUnderTest).isStopped();
 140         doReturn(messageRouterConsumerMock).when(messageConsumerUnderTest).getMessageRouterConsumer();
 141
 142         Mono<ResponseEntity<String>> response = Mono.just(new ResponseEntity<>("Error", HttpStatus.BAD_REQUEST));
 143         when(messageRouterConsumerMock.getForEntity(any())).thenReturn(response);
 144
 145         final ListAppender<ILoggingEvent> logAppender =
 146                 LoggingUtils.getLogListAppender(DmaapMessageConsumer.class, WARN);
 147
 148         messageConsumerUnderTest.start().join();
 149
 150         assertThat(logAppender.list.get(0).getFormattedMessage())
 151                 .isEqualTo("Cannot fetch because of Error respons: 400 BAD_REQUEST Error");
 152
 153         verify(messageConsumerUnderTest).sleep(DmaapMessageConsumer.TIME_BETWEEN_DMAAP_RETRIES);
 154     }
 155
 156     @Test
 157     void dmaapConfiguredAndOneMessage_thenPollOnceAndProcessMessage() throws Exception {
 158         // The message from MR is here an array of Json objects
 159         setUpMrConfig();
 160         messageConsumerUnderTest = spy(new DmaapMessageConsumer(applicationConfigMock));
 161
 162         String messages = jsonArray(gson.toJson(dmaapRequestMessage(Operation.PUT)));
 163
 164         doReturn(false, true).when(messageConsumerUnderTest).isStopped();
 165         doReturn(messageRouterConsumerMock).when(messageConsumerUnderTest).getMessageRouterConsumer();
 166
 167         Mono<ResponseEntity<String>> response = Mono.just(new ResponseEntity<>(messages, HttpStatus.OK));
 168         when(messageRouterConsumerMock.getForEntity(any())).thenReturn(response);
 169
 170         doReturn(messageHandlerMock).when(messageConsumerUnderTest).getDmaapMessageHandler();
 171
 172         messageConsumerUnderTest.start().join();
 173
 174         ArgumentCaptor<DmaapRequestMessage> captor = ArgumentCaptor.forClass(DmaapRequestMessage.class);
 175         verify(messageHandlerMock).handleDmaapMsg(captor.capture());
 176         DmaapRequestMessage messageAfterJsonParsing = captor.getValue();
 177         assertThat(messageAfterJsonParsing.apiVersion()).isNotEmpty();
 178
 179         verifyNoMoreInteractions(messageHandlerMock);
 180     }
 181
 182     @Test
 183     void testMessageParsing() throws ServiceException {
 184         messageConsumerUnderTest = new DmaapMessageConsumer(applicationConfigMock);
 185         String json = gson.toJson(dmaapRequestMessage(Operation.PUT));
 186         {
 187             String jsonArrayOfObject = jsonArray(json);
 188             List<DmaapRequestMessage> parsedMessage = messageConsumerUnderTest.parseMessages(jsonArrayOfObject);
 189             assertNotNull(parsedMessage);
 190             assertTrue(parsedMessage.get(0).payload().isPresent());
 191         }
 192         {
 193             String jsonArrayOfString = jsonArray(quote(json));
 194             List<DmaapRequestMessage> parsedMessage = messageConsumerUnderTest.parseMessages(jsonArrayOfString);
 195             assertNotNull(parsedMessage);
 196             assertTrue(parsedMessage.get(0).payload().isPresent());
 197         }
 198
 199     }
 200
 201     @Test
 202     void incomingUnparsableRequest_thenSendResponse() throws Exception {
 203         messageConsumerUnderTest = spy(new DmaapMessageConsumer(applicationConfigMock));
 204         doReturn(messageHandlerMock).when(messageConsumerUnderTest).getDmaapMessageHandler();
 205         doReturn(Mono.just("OK")).when(messageHandlerMock).sendDmaapResponse(any(), any(), any());
 206         Exception actualException =
 207                 assertThrows(ServiceException.class, () -> messageConsumerUnderTest.parseMessages("[\"abc:\"def\"]"));
 208         assertThat(actualException.getMessage())
 209                 .contains("Could not parse incomming request. Reason :com.google.gson.stream.MalformedJsonException");
 210
 211         verify(messageHandlerMock).sendDmaapResponse(any(), any(), any());
 212     }
 213
 214     @Test
 215     void incomingUnparsableRequest_thenSendingResponseFailed() throws Exception {
 216         messageConsumerUnderTest = spy(new DmaapMessageConsumer(applicationConfigMock));
 217         doReturn(messageHandlerMock).when(messageConsumerUnderTest).getDmaapMessageHandler();
 218         doReturn(Mono.error(new Exception("Sending response failed"))).when(messageHandlerMock).sendDmaapResponse(any(),
 219                 any(), any());
 220         Exception actualException =
 221                 assertThrows(Exception.class, () -> messageConsumerUnderTest.parseMessages("[\"abc:\"def\"]"));
 222         assertThat(actualException.getMessage()).contains("Sending response failed");
 223
 224         verify(messageHandlerMock).sendDmaapResponse(any(), any(), any());
 225     }
 226
 227     private void setUpMrConfig() {
 228         when(applicationConfigMock.getDmaapConsumerTopicUrl()).thenReturn("url");
 229         when(applicationConfigMock.getDmaapProducerTopicUrl()).thenReturn("url");
 230     }
 231
 232     private String jsonArray(String s) {
 233         return "[" + s + "]";
 234     }
 235
 236     private String quote(String s) {
 237         return "\"" + s.replace("\"", "\\\"") + "\"";
 238     }
 239
 240     private DmaapRequestMessage dmaapRequestMessage(Operation operation) {
 241         return ImmutableDmaapRequestMessage.builder() //
 242                 .apiVersion("apiVersion") //
 243                 .correlationId("correlationId") //
 244                 .operation(operation) //
 245                 .originatorId("originatorId") //
 246                 .payload(new JsonObject()) //
 247                 .requestId("requestId") //
 248                 .target("target") //
 249                 .timestamp("timestamp") //
 250                 .url("URL") //
 251                 .build();
 252     }
 253
 254 }