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
9 * http://www.apache.org/licenses/LICENSE-2.0
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.
17 * SPDX-License-Identifier: Apache-2.0
18 * ============LICENSE_END=========================================================
21 package org.oransc.policyagent.dmaap;
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;
34 import ch.qos.logback.classic.Level;
35 import ch.qos.logback.classic.spi.ILoggingEvent;
36 import ch.qos.logback.core.read.ListAppender;
38 import com.google.gson.Gson;
39 import com.google.gson.GsonBuilder;
40 import com.google.gson.JsonObject;
42 import java.io.IOException;
43 import java.util.Optional;
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;
57 import reactor.core.publisher.Mono;
58 import reactor.test.StepVerifier;
60 public class DmaapMessageHandlerTest {
62 private static final String URL = "url";
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() //
71 private void setUp() throws Exception {
72 testedObject = spy(new DmaapMessageHandler(dmaapClient, agentClient));
75 static JsonObject payloadAsJson() {
76 return gson.fromJson(payloadAsString(), JsonObject.class);
79 static String payloadAsString() {
80 PolicyType pt = ImmutablePolicyType.builder().name("name").schema("schema").build();
81 return gson.toJson(pt);
84 DmaapRequestMessage dmaapRequestMessage(Operation operation) {
85 Optional<JsonObject> payload =
86 ((operation == Operation.PUT || operation == Operation.POST) ? Optional.of(payloadAsJson())
88 return ImmutableDmaapRequestMessage.builder().apiVersion("apiVersion") //
89 .correlationId("correlationId") //
90 .operation(operation) //
91 .originatorId("originatorId") //
93 .requestId("requestId") //
95 .timestamp("timestamp") //
101 private String dmaapInputMessage(Operation operation) {
102 return gson.toJson(dmaapRequestMessage(operation));
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());
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());
121 public void unparseableMessage_thenWarning() {
122 final ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(DmaapMessageHandler.class);
124 testedObject.handleDmaapMsg("bad message");
126 assertThat(logAppender.list.get(0).getLevel()).isEqualTo(Level.WARN);
127 assertThat(logAppender.list.toString().contains("handleDmaapMsg failure ")).isTrue();
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();
136 String message = dmaapInputMessage(Operation.DELETE);
139 .create(testedObject.createTask(message)) //
140 .expectSubscription() //
142 .verifyComplete(); //
144 verify(agentClient).delete(URL);
145 verifyNoMoreInteractions(agentClient);
147 verify(dmaapClient).send(anyString());
148 verify(dmaapClient).sendBatchWithResponse();
149 verifyNoMoreInteractions(dmaapClient);
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();
159 .create(testedObject.createTask(dmaapInputMessage(Operation.GET))) //
160 .expectSubscription() //
162 .verifyComplete(); //
164 verify(agentClient).get(URL);
165 verifyNoMoreInteractions(agentClient);
167 verify(dmaapClient).send(anyString());
168 verify(dmaapClient).sendBatchWithResponse();
169 verifyNoMoreInteractions(dmaapClient);
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();
179 .create(testedObject.createTask(dmaapInputMessage(Operation.PUT))) //
180 .expectSubscription() //
182 .verifyComplete(); //
184 verify(agentClient).put(URL, payloadAsString());
185 verifyNoMoreInteractions(agentClient);
187 verify(dmaapClient).send(anyString());
188 verify(dmaapClient).sendBatchWithResponse();
189 verifyNoMoreInteractions(dmaapClient);
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();
199 .create(testedObject.createTask(dmaapInputMessage(Operation.POST))) //
200 .expectSubscription() //
202 .verifyComplete(); //
204 verify(agentClient).post(URL, payloadAsString());
205 verifyNoMoreInteractions(agentClient);
207 verify(dmaapClient).send(anyString());
208 verify(dmaapClient).sendBatchWithResponse();
209 verifyNoMoreInteractions(dmaapClient);
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();
220 .create(testedObject.createTask(dmaapInputMessage(Operation.PUT))) //
221 .expectSubscription() //
222 .verifyComplete(); //
224 verify(agentClient).put(anyString(), anyString());
225 verifyNoMoreInteractions(agentClient);
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))
233 verify(dmaapClient).sendBatchWithResponse();
234 verifyNoMoreInteractions(dmaapClient);
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);
243 testedObject.handleDmaapMsg(message);
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();
251 verify(dmaapClient).sendBatchWithResponse();
252 verifyNoMoreInteractions(dmaapClient);
256 public void putWithoutPayload_thenNotFoundResponseWithWarning() throws Exception {
257 String message = dmaapInputMessage(Operation.PUT).toString();
258 message = message.replace(",\"payload\":{\"name\":\"name\",\"schema\":\"schema\"}", "");
260 final ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(DmaapMessageHandler.class);
262 testedObject.handleDmaapMsg(message);
264 assertThat(logAppender.list.get(0).getLevel()).isEqualTo(Level.WARN);
265 assertThat(logAppender.list.toString().contains("Expected payload in message from DMAAP: ")).isTrue();