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 org.assertj.core.api.Assertions.assertThat;
  24 import static org.junit.Assert.assertFalse;
  25 import static org.junit.jupiter.api.Assertions.assertTrue;
  26 import static org.mockito.ArgumentMatchers.any;
  27 import static org.mockito.ArgumentMatchers.anyString;
  28 import static org.mockito.Mockito.doReturn;
  29 import static org.mockito.Mockito.mock;
  30 import static org.mockito.Mockito.spy;
  31 import static org.mockito.Mockito.verify;
  32 import static org.mockito.Mockito.verifyNoMoreInteractions;
  33
  34 import ch.qos.logback.classic.Level;
  35 import ch.qos.logback.classic.spi.ILoggingEvent;
  36 import ch.qos.logback.core.read.ListAppender;
  37
  38 import com.google.gson.Gson;
  39 import com.google.gson.GsonBuilder;
  40 import com.google.gson.JsonObject;
  41
  42 import java.io.IOException;
  43 import java.util.Optional;
  44
  45 import org.junit.jupiter.api.BeforeEach;
  46 import org.junit.jupiter.api.Test;
  47 import org.mockito.ArgumentCaptor;
  48 import org.onap.dmaap.mr.client.MRBatchingPublisher;
  49 import org.onap.dmaap.mr.client.response.MRPublisherResponse;
  50 import org.oransc.policyagent.clients.AsyncRestClient;
  51 import org.oransc.policyagent.dmaap.DmaapRequestMessage.Operation;
  52 import org.oransc.policyagent.repository.ImmutablePolicyType;
  53 import org.oransc.policyagent.repository.PolicyType;
  54 import org.oransc.policyagent.utils.LoggingUtils;
  55 import org.springframework.http.HttpStatus;
  56
  57 import reactor.core.publisher.Mono;
  58 import reactor.test.StepVerifier;
  59
  60 public class DmaapMessageHandlerTest {
  61
  62     private static final String URL = "url";
  63
  64     private final MRBatchingPublisher dmaapClient = mock(MRBatchingPublisher.class);
  65     private final AsyncRestClient agentClient = mock(AsyncRestClient.class);
  66     private DmaapMessageHandler testedObject;
  67     private static Gson gson = new GsonBuilder() //
  68         .create(); //
  69
  70     @BeforeEach
  71     private void setUp() throws Exception {
  72         testedObject = spy(new DmaapMessageHandler(dmaapClient, agentClient));
  73     }
  74
  75     static JsonObject payloadAsJson() {
  76         return gson.fromJson(payloadAsString(), JsonObject.class);
  77     }
  78
  79     static String payloadAsString() {
  80         PolicyType pt = ImmutablePolicyType.builder().name("name").schema("schema").build();
  81         return gson.toJson(pt);
  82     }
  83
  84     DmaapRequestMessage dmaapRequestMessage(Operation operation) {
  85         Optional<JsonObject> payload =
  86             ((operation == Operation.PUT || operation == Operation.POST) ? Optional.of(payloadAsJson())
  87                 : Optional.empty());
  88         return ImmutableDmaapRequestMessage.builder().apiVersion("apiVersion") //
  89             .correlationId("correlationId") //
  90             .operation(operation) //
  91             .originatorId("originatorId") //
  92             .payload(payload) //
  93             .requestId("requestId") //
  94             .target("target") //
  95             .timestamp("timestamp") //
  96             .type("type") //
  97             .url(URL) //
  98             .build();
  99     }
 100
 101     private String dmaapInputMessage(Operation operation) {
 102         return gson.toJson(dmaapRequestMessage(operation));
 103     }
 104
 105     @Test
 106     public void testMessageParsing() {
 107         String message = dmaapInputMessage(Operation.DELETE);
 108         System.out.println(message);
 109         DmaapRequestMessage parsedMessage = gson.fromJson(message, ImmutableDmaapRequestMessage.class);
 110         assertTrue(parsedMessage != null);
 111         assertFalse(parsedMessage.payload().isPresent());
 112
 113         message = dmaapInputMessage(Operation.PUT);
 114         System.out.println(message);
 115         parsedMessage = gson.fromJson(message, ImmutableDmaapRequestMessage.class);
 116         assertTrue(parsedMessage != null);
 117         assertTrue(parsedMessage.payload().isPresent());
 118     }
 119
 120     @Test
 121     public void unparseableMessage_thenWarning() {
 122         final ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(DmaapMessageHandler.class);
 123
 124         testedObject.handleDmaapMsg("bad message");
 125
 126         assertThat(logAppender.list.get(0).getLevel()).isEqualTo(Level.WARN);
 127         assertThat(logAppender.list.toString().contains("handleDmaapMsg failure ")).isTrue();
 128     }
 129
 130     @Test
 131     public void successfulDelete() throws IOException {
 132         doReturn(Mono.just("OK")).when(agentClient).delete(anyString());
 133         doReturn(1).when(dmaapClient).send(anyString());
 134         doReturn(new MRPublisherResponse()).when(dmaapClient).sendBatchWithResponse();
 135
 136         String message = dmaapInputMessage(Operation.DELETE);
 137
 138         StepVerifier //
 139             .create(testedObject.createTask(message)) //
 140             .expectSubscription() //
 141             .expectNext("OK") //
 142             .verifyComplete(); //
 143
 144         verify(agentClient).delete(URL);
 145         verifyNoMoreInteractions(agentClient);
 146
 147         verify(dmaapClient).send(anyString());
 148         verify(dmaapClient).sendBatchWithResponse();
 149         verifyNoMoreInteractions(dmaapClient);
 150     }
 151
 152     @Test
 153     public void successfulGet() throws IOException {
 154         doReturn(Mono.just("OK")).when(agentClient).get(anyString());
 155         doReturn(1).when(dmaapClient).send(anyString());
 156         doReturn(new MRPublisherResponse()).when(dmaapClient).sendBatchWithResponse();
 157
 158         StepVerifier //
 159             .create(testedObject.createTask(dmaapInputMessage(Operation.GET))) //
 160             .expectSubscription() //
 161             .expectNext("OK") //
 162             .verifyComplete(); //
 163
 164         verify(agentClient).get(URL);
 165         verifyNoMoreInteractions(agentClient);
 166
 167         verify(dmaapClient).send(anyString());
 168         verify(dmaapClient).sendBatchWithResponse();
 169         verifyNoMoreInteractions(dmaapClient);
 170     }
 171
 172     @Test
 173     public void successfulPut() throws IOException {
 174         doReturn(Mono.just("OK")).when(agentClient).put(anyString(), anyString());
 175         doReturn(1).when(dmaapClient).send(anyString());
 176         doReturn(new MRPublisherResponse()).when(dmaapClient).sendBatchWithResponse();
 177
 178         StepVerifier //
 179             .create(testedObject.createTask(dmaapInputMessage(Operation.PUT))) //
 180             .expectSubscription() //
 181             .expectNext("OK") //
 182             .verifyComplete(); //
 183
 184         verify(agentClient).put(URL, payloadAsString());
 185         verifyNoMoreInteractions(agentClient);
 186
 187         verify(dmaapClient).send(anyString());
 188         verify(dmaapClient).sendBatchWithResponse();
 189         verifyNoMoreInteractions(dmaapClient);
 190     }
 191
 192     @Test
 193     public void successfulPost() throws IOException {
 194         doReturn(Mono.just("OK")).when(agentClient).post(anyString(), anyString());
 195         doReturn(1).when(dmaapClient).send(anyString());
 196         doReturn(new MRPublisherResponse()).when(dmaapClient).sendBatchWithResponse();
 197
 198         StepVerifier //
 199             .create(testedObject.createTask(dmaapInputMessage(Operation.POST))) //
 200             .expectSubscription() //
 201             .expectNext("OK") //
 202             .verifyComplete(); //
 203
 204         verify(agentClient).post(URL, payloadAsString());
 205         verifyNoMoreInteractions(agentClient);
 206
 207         verify(dmaapClient).send(anyString());
 208         verify(dmaapClient).sendBatchWithResponse();
 209         verifyNoMoreInteractions(dmaapClient);
 210     }
 211
 212     @Test
 213     public void exceptionWhenCallingPolicyAgent_thenNotFoundResponse() throws IOException {
 214         String errorCause = "Refused";
 215         doReturn(Mono.error(new Exception(errorCause))).when(agentClient).put(anyString(), any());
 216         doReturn(1).when(dmaapClient).send(anyString());
 217         doReturn(new MRPublisherResponse()).when(dmaapClient).sendBatchWithResponse();
 218
 219         StepVerifier //
 220             .create(testedObject.createTask(dmaapInputMessage(Operation.PUT))) //
 221             .expectSubscription() //
 222             .verifyComplete(); //
 223
 224         verify(agentClient).put(anyString(), anyString());
 225         verifyNoMoreInteractions(agentClient);
 226
 227         ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
 228         verify(dmaapClient).send(captor.capture());
 229         String actualMessage = captor.getValue();
 230         assertThat(actualMessage.contains(HttpStatus.NOT_FOUND + "\",\"message\":\"java.lang.Exception: " + errorCause))
 231             .isTrue();
 232
 233         verify(dmaapClient).sendBatchWithResponse();
 234         verifyNoMoreInteractions(dmaapClient);
 235     }
 236
 237     @Test
 238     public void unsupportedOperationInMessage_thenNotFoundResponseWithNotImplementedOperation() throws Exception {
 239         String message = dmaapInputMessage(Operation.PUT).toString();
 240         String badOperation = "BAD";
 241         message = message.replace(Operation.PUT.toString(), badOperation);
 242
 243         testedObject.handleDmaapMsg(message);
 244
 245         ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
 246         verify(dmaapClient).send(captor.capture());
 247         String actualMessage = captor.getValue();
 248         assertThat(actualMessage
 249             .contains(HttpStatus.NOT_FOUND + "\",\"message\":\"Not implemented operation: " + badOperation)).isTrue();
 250
 251         verify(dmaapClient).sendBatchWithResponse();
 252         verifyNoMoreInteractions(dmaapClient);
 253     }
 254
 255     @Test
 256     public void putWithoutPayload_thenNotFoundResponseWithWarning() throws Exception {
 257         String message = dmaapInputMessage(Operation.PUT).toString();
 258         message = message.replace(",\"payload\":{\"name\":\"name\",\"schema\":\"schema\"}", "");
 259
 260         final ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(DmaapMessageHandler.class);
 261
 262         testedObject.handleDmaapMsg(message);
 263
 264         assertThat(logAppender.list.get(0).getLevel()).isEqualTo(Level.WARN);
 265         assertThat(logAppender.list.toString().contains("Expected payload in message from DMAAP: ")).isTrue();
 266     }
 267 }