policy-agent/src/test/java/org/oransc/policyagent/dmaap/DmaapMessageHandlerTest.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2019 Nordix Foundation.
   4  * ================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  * SPDX-License-Identifier: Apache-2.0
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.oransc.policyagent.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.Assert.assertFalse;
  26 import static org.junit.Assert.assertNotNull;
  27 import static org.junit.jupiter.api.Assertions.assertTrue;
  28 import static org.mockito.ArgumentMatchers.any;
  29 import static org.mockito.ArgumentMatchers.anyString;
  30 import static org.mockito.Mockito.doReturn;
  31 import static org.mockito.Mockito.mock;
  32 import static org.mockito.Mockito.spy;
  33 import static org.mockito.Mockito.verify;
  34 import static org.mockito.Mockito.verifyNoMoreInteractions;
  35
  36 import ch.qos.logback.classic.spi.ILoggingEvent;
  37 import ch.qos.logback.core.read.ListAppender;
  38
  39 import com.google.gson.Gson;
  40 import com.google.gson.GsonBuilder;
  41 import com.google.gson.JsonObject;
  42
  43 import java.io.IOException;
  44 import java.util.Optional;
  45
  46 import org.junit.jupiter.api.BeforeEach;
  47 import org.junit.jupiter.api.Test;
  48 import org.mockito.ArgumentCaptor;
  49 import org.onap.dmaap.mr.client.MRBatchingPublisher;
  50 import org.onap.dmaap.mr.client.response.MRPublisherResponse;
  51 import org.oransc.policyagent.clients.AsyncRestClient;
  52 import org.oransc.policyagent.dmaap.DmaapRequestMessage.Operation;
  53 import org.oransc.policyagent.repository.ImmutablePolicyType;
  54 import org.oransc.policyagent.repository.PolicyType;
  55 import org.oransc.policyagent.utils.LoggingUtils;
  56 import org.slf4j.Logger;
  57 import org.slf4j.LoggerFactory;
  58 import org.springframework.http.HttpStatus;
  59 import org.springframework.http.ResponseEntity;
  60 import org.springframework.web.reactive.function.client.WebClientResponseException;
  61
  62 import reactor.core.publisher.Mono;
  63 import reactor.test.StepVerifier;
  64
  65 class DmaapMessageHandlerTest {
  66     private static final Logger logger = LoggerFactory.getLogger(DmaapMessageHandlerTest.class);
  67     private static final String URL = "url";
  68
  69     private final MRBatchingPublisher dmaapClient = mock(MRBatchingPublisher.class);
  70     private final AsyncRestClient agentClient = mock(AsyncRestClient.class);
  71     private DmaapMessageHandler testedObject;
  72     private static Gson gson = new GsonBuilder() //
  73         .create(); //
  74
  75     @BeforeEach
  76     private void setUp() throws Exception {
  77         testedObject = spy(new DmaapMessageHandler(dmaapClient, agentClient));
  78     }
  79
  80     static JsonObject payloadAsJson() {
  81         return gson.fromJson(payloadAsString(), JsonObject.class);
  82     }
  83
  84     static String payloadAsString() {
  85         PolicyType pt = ImmutablePolicyType.builder().name("name").schema("schema").build();
  86         return gson.toJson(pt);
  87     }
  88
  89     DmaapRequestMessage dmaapRequestMessage(Operation operation) {
  90         Optional<JsonObject> payload =
  91             ((operation == Operation.PUT || operation == Operation.POST) ? Optional.of(payloadAsJson())
  92                 : Optional.empty());
  93         return ImmutableDmaapRequestMessage.builder() //
  94             .apiVersion("apiVersion") //
  95             .correlationId("correlationId") //
  96             .operation(operation) //
  97             .originatorId("originatorId") //
  98             .payload(payload) //
  99             .requestId("requestId") //
 100             .target("target") //
 101             .timestamp("timestamp") //
 102             .url(URL) //
 103             .build();
 104     }
 105
 106     private String dmaapInputMessage(Operation operation) {
 107         return gson.toJson(dmaapRequestMessage(operation));
 108     }
 109
 110     private Mono<ResponseEntity<String>> okResponse() {
 111         ResponseEntity<String> entity = new ResponseEntity<>("OK", HttpStatus.OK);
 112         return Mono.just(entity);
 113     }
 114
 115     @Test
 116     void testMessageParsing() {
 117         String message = dmaapInputMessage(Operation.DELETE);
 118         logger.info(message);
 119         DmaapRequestMessage parsedMessage = gson.fromJson(message, ImmutableDmaapRequestMessage.class);
 120         assertNotNull(parsedMessage);
 121         assertFalse(parsedMessage.payload().isPresent());
 122
 123         message = dmaapInputMessage(Operation.PUT);
 124         logger.info(message);
 125         parsedMessage = gson.fromJson(message, ImmutableDmaapRequestMessage.class);
 126         assertNotNull(parsedMessage);
 127         assertTrue(parsedMessage.payload().isPresent());
 128     }
 129
 130     @Test
 131     void unparseableMessage_thenWarning() {
 132         final ListAppender<ILoggingEvent> logAppender =
 133             LoggingUtils.getLogListAppender(DmaapMessageHandler.class, WARN);
 134
 135         String msg = "bad message";
 136         testedObject.handleDmaapMsg(msg);
 137
 138         assertThat(logAppender.list.get(0).getFormattedMessage()).startsWith(
 139             "handleDmaapMsg failure org.oransc.policyagent.exceptions.ServiceException: Received unparsable "
 140                 + "message from DMAAP: \"" + msg + "\", reason: ");
 141     }
 142
 143     @Test
 144     void successfulDelete() throws IOException {
 145         doReturn(okResponse()).when(agentClient).deleteForEntity(anyString());
 146         doReturn(1).when(dmaapClient).send(anyString());
 147         doReturn(new MRPublisherResponse()).when(dmaapClient).sendBatchWithResponse();
 148
 149         String message = dmaapInputMessage(Operation.DELETE);
 150
 151         StepVerifier //
 152             .create(testedObject.createTask(message)) //
 153             .expectSubscription() //
 154             .expectNext("OK") //
 155             .verifyComplete(); //
 156
 157         verify(agentClient).deleteForEntity(URL);
 158         verifyNoMoreInteractions(agentClient);
 159
 160         verify(dmaapClient).send(anyString());
 161         verify(dmaapClient).sendBatchWithResponse();
 162         verifyNoMoreInteractions(dmaapClient);
 163     }
 164
 165     @Test
 166     void successfulGet() throws IOException {
 167         doReturn(okResponse()).when(agentClient).getForEntity(anyString());
 168         doReturn(1).when(dmaapClient).send(anyString());
 169         doReturn(new MRPublisherResponse()).when(dmaapClient).sendBatchWithResponse();
 170
 171         StepVerifier //
 172             .create(testedObject.createTask(dmaapInputMessage(Operation.GET))) //
 173             .expectSubscription() //
 174             .expectNext("OK") //
 175             .verifyComplete(); //
 176
 177         verify(agentClient).getForEntity(URL);
 178         verifyNoMoreInteractions(agentClient);
 179
 180         verify(dmaapClient).send(anyString());
 181         verify(dmaapClient).sendBatchWithResponse();
 182         verifyNoMoreInteractions(dmaapClient);
 183     }
 184
 185     @Test
 186     void successfulPut() throws IOException {
 187         doReturn(okResponse()).when(agentClient).putForEntity(anyString(), anyString());
 188         doReturn(1).when(dmaapClient).send(anyString());
 189         doReturn(new MRPublisherResponse()).when(dmaapClient).sendBatchWithResponse();
 190
 191         StepVerifier //
 192             .create(testedObject.createTask(dmaapInputMessage(Operation.PUT))) //
 193             .expectSubscription() //
 194             .expectNext("OK") //
 195             .verifyComplete(); //
 196
 197         verify(agentClient).putForEntity(URL, payloadAsString());
 198         verifyNoMoreInteractions(agentClient);
 199
 200         verify(dmaapClient).send(anyString());
 201         verify(dmaapClient).sendBatchWithResponse();
 202         verifyNoMoreInteractions(dmaapClient);
 203     }
 204
 205     @Test
 206     void successfulPost() throws IOException {
 207         doReturn(okResponse()).when(agentClient).postForEntity(anyString(), anyString());
 208         doReturn(1).when(dmaapClient).send(anyString());
 209         doReturn(new MRPublisherResponse()).when(dmaapClient).sendBatchWithResponse();
 210
 211         StepVerifier //
 212             .create(testedObject.createTask(dmaapInputMessage(Operation.POST))) //
 213             .expectSubscription() //
 214             .expectNext("OK") //
 215             .verifyComplete(); //
 216
 217         verify(agentClient).postForEntity(URL, payloadAsString());
 218         verifyNoMoreInteractions(agentClient);
 219
 220         verify(dmaapClient).send(anyString());
 221         verify(dmaapClient).sendBatchWithResponse();
 222         verifyNoMoreInteractions(dmaapClient);
 223     }
 224
 225     @Test
 226     void exceptionWhenCallingPolicyAgent_thenNotFoundResponse() throws IOException {
 227         WebClientResponseException except = new WebClientResponseException(400, "Refused", null, null, null, null);
 228         doReturn(Mono.error(except)).when(agentClient).putForEntity(anyString(), any());
 229         doReturn(1).when(dmaapClient).send(anyString());
 230         doReturn(new MRPublisherResponse()).when(dmaapClient).sendBatchWithResponse();
 231
 232         StepVerifier //
 233             .create(testedObject.createTask(dmaapInputMessage(Operation.PUT))) //
 234             .expectSubscription() //
 235             .verifyComplete(); //
 236
 237         verify(agentClient).putForEntity(anyString(), anyString());
 238         verifyNoMoreInteractions(agentClient);
 239
 240         ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
 241         verify(dmaapClient).send(captor.capture());
 242         String actualMessage = captor.getValue();
 243         assertThat(actualMessage.contains(HttpStatus.BAD_REQUEST.toString()))
 244             .as("Message \"%s\" sent to DMaaP contains %s", actualMessage, HttpStatus.BAD_REQUEST) //
 245             .isTrue();
 246
 247         verify(dmaapClient).sendBatchWithResponse();
 248         verifyNoMoreInteractions(dmaapClient);
 249     }
 250
 251     @Test
 252     void unsupportedOperationInMessage_thenNotFoundResponseWithNotImplementedOperation() throws Exception {
 253         String message = dmaapInputMessage(Operation.PUT).toString();
 254         String badOperation = "BAD";
 255         message = message.replace(Operation.PUT.toString(), badOperation);
 256
 257         testedObject.handleDmaapMsg(message);
 258
 259         ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
 260         verify(dmaapClient).send(captor.capture());
 261         String actualMessage = captor.getValue();
 262         assertThat(actualMessage
 263             .contains(HttpStatus.BAD_REQUEST + "\",\"message\":\"Not implemented operation: " + badOperation)) //
 264                 .as("Message \"%s\" sent to DMaaP contains %s", actualMessage, HttpStatus.BAD_REQUEST) //
 265                 .isTrue();
 266
 267         verify(dmaapClient).sendBatchWithResponse();
 268         verifyNoMoreInteractions(dmaapClient);
 269     }
 270
 271     @Test
 272     void putWithoutPayload_thenNotFoundResponseWithWarning() throws Exception {
 273         String message = dmaapInputMessage(Operation.PUT).toString();
 274         message = message.replace(",\"payload\":{\"name\":\"name\",\"schema\":\"schema\"}", "");
 275
 276         final ListAppender<ILoggingEvent> logAppender =
 277             LoggingUtils.getLogListAppender(DmaapMessageHandler.class, WARN);
 278
 279         testedObject.handleDmaapMsg(message);
 280
 281         assertThat(logAppender.list.get(0).getFormattedMessage())
 282             .startsWith("Expected payload in message from DMAAP: ");
 283     }
 284 }