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 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;
36 import ch.qos.logback.classic.spi.ILoggingEvent;
37 import ch.qos.logback.core.read.ListAppender;
39 import com.google.gson.Gson;
40 import com.google.gson.GsonBuilder;
41 import com.google.gson.JsonObject;
43 import java.io.IOException;
44 import java.util.Optional;
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;
62 import reactor.core.publisher.Mono;
63 import reactor.test.StepVerifier;
65 class DmaapMessageHandlerTest {
66 private static final Logger logger = LoggerFactory.getLogger(DmaapMessageHandlerTest.class);
67 private static final String URL = "url";
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() //
76 private void setUp() throws Exception {
77 testedObject = spy(new DmaapMessageHandler(dmaapClient, agentClient));
80 static JsonObject payloadAsJson() {
81 return gson.fromJson(payloadAsString(), JsonObject.class);
84 static String payloadAsString() {
85 PolicyType pt = ImmutablePolicyType.builder().name("name").schema("schema").build();
86 return gson.toJson(pt);
89 DmaapRequestMessage dmaapRequestMessage(Operation operation) {
90 Optional<JsonObject> payload =
91 ((operation == Operation.PUT || operation == Operation.POST) ? Optional.of(payloadAsJson())
93 return ImmutableDmaapRequestMessage.builder() //
94 .apiVersion("apiVersion") //
95 .correlationId("correlationId") //
96 .operation(operation) //
97 .originatorId("originatorId") //
99 .requestId("requestId") //
101 .timestamp("timestamp") //
106 private String dmaapInputMessage(Operation operation) {
107 return gson.toJson(dmaapRequestMessage(operation));
110 private Mono<ResponseEntity<String>> okResponse() {
111 ResponseEntity<String> entity = new ResponseEntity<>("OK", HttpStatus.OK);
112 return Mono.just(entity);
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());
123 message = dmaapInputMessage(Operation.PUT);
124 logger.info(message);
125 parsedMessage = gson.fromJson(message, ImmutableDmaapRequestMessage.class);
126 assertNotNull(parsedMessage);
127 assertTrue(parsedMessage.payload().isPresent());
131 void unparseableMessage_thenWarning() {
132 final ListAppender<ILoggingEvent> logAppender =
133 LoggingUtils.getLogListAppender(DmaapMessageHandler.class, WARN);
135 String msg = "bad message";
136 testedObject.handleDmaapMsg(msg);
138 assertThat(logAppender.list.get(0).getFormattedMessage()).startsWith(
139 "handleDmaapMsg failure org.oransc.policyagent.exceptions.ServiceException: Received unparsable "
140 + "message from DMAAP: \"" + msg + "\", reason: ");
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();
149 String message = dmaapInputMessage(Operation.DELETE);
152 .create(testedObject.createTask(message)) //
153 .expectSubscription() //
155 .verifyComplete(); //
157 verify(agentClient).deleteForEntity(URL);
158 verifyNoMoreInteractions(agentClient);
160 verify(dmaapClient).send(anyString());
161 verify(dmaapClient).sendBatchWithResponse();
162 verifyNoMoreInteractions(dmaapClient);
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();
172 .create(testedObject.createTask(dmaapInputMessage(Operation.GET))) //
173 .expectSubscription() //
175 .verifyComplete(); //
177 verify(agentClient).getForEntity(URL);
178 verifyNoMoreInteractions(agentClient);
180 verify(dmaapClient).send(anyString());
181 verify(dmaapClient).sendBatchWithResponse();
182 verifyNoMoreInteractions(dmaapClient);
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();
192 .create(testedObject.createTask(dmaapInputMessage(Operation.PUT))) //
193 .expectSubscription() //
195 .verifyComplete(); //
197 verify(agentClient).putForEntity(URL, payloadAsString());
198 verifyNoMoreInteractions(agentClient);
200 verify(dmaapClient).send(anyString());
201 verify(dmaapClient).sendBatchWithResponse();
202 verifyNoMoreInteractions(dmaapClient);
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();
212 .create(testedObject.createTask(dmaapInputMessage(Operation.POST))) //
213 .expectSubscription() //
215 .verifyComplete(); //
217 verify(agentClient).postForEntity(URL, payloadAsString());
218 verifyNoMoreInteractions(agentClient);
220 verify(dmaapClient).send(anyString());
221 verify(dmaapClient).sendBatchWithResponse();
222 verifyNoMoreInteractions(dmaapClient);
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();
233 .create(testedObject.createTask(dmaapInputMessage(Operation.PUT))) //
234 .expectSubscription() //
235 .verifyComplete(); //
237 verify(agentClient).putForEntity(anyString(), anyString());
238 verifyNoMoreInteractions(agentClient);
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) //
247 verify(dmaapClient).sendBatchWithResponse();
248 verifyNoMoreInteractions(dmaapClient);
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);
257 testedObject.handleDmaapMsg(message);
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) //
267 verify(dmaapClient).sendBatchWithResponse();
268 verifyNoMoreInteractions(dmaapClient);
272 void putWithoutPayload_thenNotFoundResponseWithWarning() throws Exception {
273 String message = dmaapInputMessage(Operation.PUT).toString();
274 message = message.replace(",\"payload\":{\"name\":\"name\",\"schema\":\"schema\"}", "");
276 final ListAppender<ILoggingEvent> logAppender =
277 LoggingUtils.getLogListAppender(DmaapMessageHandler.class, WARN);
279 testedObject.handleDmaapMsg(message);
281 assertThat(logAppender.list.get(0).getFormattedMessage())
282 .startsWith("Expected payload in message from DMAAP: ");