gerrit.onap Code Review - dcaegen2/services/sdk.git/blob

   1 /*
   2  * ============LICENSE_START====================================
   3  * DCAEGEN2-SERVICES-SDK
   4  * =========================================================
   5  * Copyright (C) 2019 Nokia. 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.dcaegen2.services.sdk.rest.services.dmaap.client.impl;
  22
  23 import static org.assertj.core.api.Assertions.assertThat;
  24 import static org.mockito.ArgumentMatchers.any;
  25 import static org.mockito.BDDMockito.given;
  26 import static org.mockito.Mockito.mock;
  27 import static org.mockito.Mockito.times;
  28 import static org.mockito.Mockito.verify;
  29
  30 import com.google.gson.JsonArray;
  31 import com.google.gson.JsonElement;
  32 import com.google.gson.JsonObject;
  33 import com.google.gson.JsonParser;
  34 import com.google.gson.Gson;
  35 import io.netty.buffer.ByteBufAllocator;
  36 import io.netty.handler.codec.http.HttpHeaderValues;
  37 import java.nio.charset.StandardCharsets;
  38 import java.time.Duration;
  39 import java.util.Arrays;
  40 import java.util.List;
  41 import java.util.stream.Collectors;
  42 import java.util.stream.Stream;
  43 import org.junit.jupiter.api.Test;
  44 import org.mockito.ArgumentCaptor;
  45 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.ImmutableMessageRouterSink;
  46 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSink;
  47 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpHeaders;
  48 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpMethod;
  49 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpRequest;
  50 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
  51 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.ImmutableHttpResponse;
  52 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient;
  53 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterPublisher;
  54 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishRequest;
  55 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest;
  56 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse;
  57 import reactor.core.publisher.Flux;
  58 import reactor.core.publisher.Mono;
  59 import reactor.test.StepVerifier;
  60
  61 /**
  62  * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
  63  * @since April 2019
  64  */
  65 class MessageRouterPublisherImplTest {
  66     private static final Duration TIMEOUT = Duration.ofSeconds(5);
  67     private static final JsonParser parser = new JsonParser();
  68     private static final MessageRouterSink sinkDefinition = ImmutableMessageRouterSink.builder()
  69             .name("the topic")
  70             .topicUrl("https://dmaap-mr/TOPIC")
  71             .build();
  72     private final RxHttpClient httpClient = mock(RxHttpClient.class);
  73     private final MessageRouterPublisher cut = new MessageRouterPublisherImpl(httpClient, 3, Duration.ofMinutes(1));
  74     private final ArgumentCaptor<HttpRequest> httpRequestArgumentCaptor = ArgumentCaptor.forClass(HttpRequest.class);
  75     private final MessageRouterPublishRequest mrRequestTextPlain = createMRRPublishRequest();
  76     private final MessageRouterPublishRequest mrRequestJson = createMRRPublishRequest();
  77     private final HttpResponse successHttpResponse = createHttpResponse("OK", 200);
  78
  79     @Test
  80     void puttingElementsShouldYieldNonChunkedHttpRequest() {
  81         // given
  82         final List<String> threeJsonMessages = getAsMRJsonMessages(Arrays.asList("I", "like", "cookies"));
  83         final Flux<JsonObject> singleJsonMessageBatch = jsonBatch(threeJsonMessages);
  84         given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse));
  85
  86         // when
  87         final Flux<MessageRouterPublishResponse> responses = cut
  88                 .put(mrRequestTextPlain, singleJsonMessageBatch);
  89         responses.then().block();
  90
  91         // then
  92         verify(httpClient).call(httpRequestArgumentCaptor.capture());
  93         final HttpRequest httpRequest = httpRequestArgumentCaptor.getValue();
  94         assertThat(httpRequest.method()).isEqualTo(HttpMethod.POST);
  95         assertThat(httpRequest.url()).isEqualTo(sinkDefinition.topicUrl());
  96         assertThat(httpRequest.body()).isNotNull();
  97         assertThat(httpRequest.body().length()).isGreaterThan(0);
  98     }
  99
 100     @Test
 101     void puttingLowNumberOfElementsShouldYieldSingleHttpRequest() {
 102         // given
 103         final List<String> threeJsonMessages = getAsMRJsonMessages(Arrays.asList("I", "like", "cookies"));
 104         final Flux<JsonObject> singleJsonMessageBatch = jsonBatch(threeJsonMessages);
 105         given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse));
 106
 107         // when
 108         final Flux<MessageRouterPublishResponse> responses = cut
 109                 .put(mrRequestJson, singleJsonMessageBatch);
 110         responses.then().block();
 111
 112         // then
 113         verify(httpClient).call(httpRequestArgumentCaptor.capture());
 114         final HttpRequest httpRequest = httpRequestArgumentCaptor.getValue();
 115         final JsonArray elementsInRequest = extractNonEmptyJsonRequestBody(httpRequest);
 116         assertThat(elementsInRequest.size()).isEqualTo(3);
 117         assertThat(elementsInRequest.get(0).toString()).isEqualTo(threeJsonMessages.get(0));
 118         assertThat(elementsInRequest.get(1).toString()).isEqualTo(threeJsonMessages.get(1));
 119         assertThat(elementsInRequest.get(2).toString()).isEqualTo(threeJsonMessages.get(2));
 120     }
 121
 122     @Test
 123     void puttingElementsWithoutContentTypeSetShouldUseApplicationJson(){
 124         // given
 125         final List<String> threeJsonMessages = getAsMRJsonMessages(Arrays.asList("I", "like", "cookies"));
 126         final Flux<JsonObject> singleJsonMessageBatch = jsonBatch(threeJsonMessages);
 127         given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse));
 128
 129         // when
 130         final Flux<MessageRouterPublishResponse> responses = cut
 131                 .put(mrRequestJson, singleJsonMessageBatch);
 132         responses.then().block();
 133
 134         // then
 135         verify(httpClient).call(httpRequestArgumentCaptor.capture());
 136         final HttpRequest httpRequest = httpRequestArgumentCaptor.getValue();
 137         assertThat(httpRequest.headers().getOrElse(HttpHeaders.CONTENT_TYPE, ""))
 138                 .isEqualTo(HttpHeaderValues.APPLICATION_JSON.toString());
 139     }
 140
 141     @Test
 142     void puttingLowNumberOfElementsShouldReturnSingleResponse() {
 143         // given
 144         final List<String> threeJsonMessages = getAsMRJsonMessages(Arrays.asList("I", "like", "cookies"));
 145         final Flux<JsonObject> singleJsonMessageBatch = jsonBatch(threeJsonMessages);
 146         given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse));
 147
 148         // when
 149         final Flux<MessageRouterPublishResponse> responses = cut
 150                 .put(mrRequestJson, singleJsonMessageBatch);
 151
 152         // then
 153         StepVerifier.create(responses)
 154                 .consumeNextWith(response -> {
 155                     assertThat(response.successful()).describedAs("successful").isTrue();
 156                     assertThat(response.items()).containsExactly(
 157                             getAsJsonObject(threeJsonMessages.get(0)),
 158                             getAsJsonObject(threeJsonMessages.get(1)),
 159                             getAsJsonObject(threeJsonMessages.get(2)));
 160                 })
 161                 .expectComplete()
 162                 .verify(TIMEOUT);
 163     }
 164
 165     @Test
 166     void puttingHighNumberOfElementsShouldYieldMultipleHttpRequests() {
 167         // given
 168         final List<String> threeJsonMessages = getAsMRJsonMessages(Arrays.asList("I", "like", "cookies"));
 169         final List<String> twoJsonMessages = getAsMRJsonMessages(Arrays.asList("and", "pierogi"));
 170         final Flux<JsonObject> doubleJsonMessageBatch = jsonBatch(concat(
 171                 threeJsonMessages, twoJsonMessages));
 172
 173         given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse));
 174
 175         // when
 176         final Flux<MessageRouterPublishResponse> responses = cut
 177                 .put(mrRequestJson, doubleJsonMessageBatch);
 178         // then
 179         responses.then().block();
 180
 181         verify(httpClient, times(2)).call(httpRequestArgumentCaptor.capture());
 182         final List<HttpRequest> httpRequests = httpRequestArgumentCaptor.getAllValues();
 183         assertThat(httpRequests.size()).describedAs("number of requests").isEqualTo(2);
 184
 185         final JsonArray firstRequest = extractNonEmptyJsonRequestBody(httpRequests.get(0));
 186         assertThat(firstRequest.size()).isEqualTo(3);
 187         assertThat(firstRequest.get(0).toString()).isEqualTo(threeJsonMessages.get(0));
 188         assertThat(firstRequest.get(1).toString()).isEqualTo(threeJsonMessages.get(1));
 189         assertThat(firstRequest.get(2).toString()).isEqualTo(threeJsonMessages.get(2));
 190
 191         final JsonArray secondRequest = extractNonEmptyJsonRequestBody(httpRequests.get(1));
 192         assertThat(secondRequest.size()).isEqualTo(2);
 193         assertThat(secondRequest.get(0).toString()).isEqualTo(twoJsonMessages.get(0));
 194         assertThat(secondRequest.get(1).toString()).isEqualTo(twoJsonMessages.get(1));
 195     }
 196
 197     @Test
 198     void puttingHighNumberOfElementsShouldReturnMoreResponses() {
 199         // given
 200         final List<String> threeJsonMessages = getAsMRJsonMessages(Arrays.asList("I", "like", "cookies"));
 201         final List<String> twoJsonMessages = getAsMRJsonMessages(Arrays.asList("and", "pierogi"));
 202         final Flux<JsonObject> doubleJsonMessageBatch = jsonBatch(concat(
 203                 threeJsonMessages, twoJsonMessages));
 204         given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(successHttpResponse));
 205
 206         // when
 207         final Flux<MessageRouterPublishResponse> responses = cut
 208                 .put(mrRequestJson, doubleJsonMessageBatch);
 209
 210         // then
 211         StepVerifier.create(responses)
 212                 .consumeNextWith(response -> {
 213                     assertThat(response.successful()).describedAs("successful").isTrue();
 214                     assertThat(response.items()).containsExactly(
 215                             getAsJsonObject(threeJsonMessages.get(0)),
 216                             getAsJsonObject(threeJsonMessages.get(1)),
 217                             getAsJsonObject(threeJsonMessages.get(2)));
 218                 })
 219                 .consumeNextWith(response -> {
 220                     assertThat(response.successful()).describedAs("successful").isTrue();
 221                     assertThat(response.items()).containsExactly(
 222                             getAsJsonObject(twoJsonMessages.get(0)),
 223                             getAsJsonObject(twoJsonMessages.get(1)));
 224                 })
 225                 .expectComplete()
 226                 .verify(TIMEOUT);
 227     }
 228
 229     private static List<String> getAsMRJsonMessages(List<String> plainTextMessages){
 230         return plainTextMessages.stream()
 231                 .map(message -> String.format("{\"message\":\"%s\"}", message))
 232                 .collect(Collectors.toList());
 233     }
 234
 235
 236     private static Flux<JsonObject> jsonBatch(List<String> messages){
 237         return Flux.fromIterable(messages).map(parser::parse).map(JsonElement::getAsJsonObject);
 238     }
 239
 240     private static List<String> concat(List<String> firstList, List<String> secondList){
 241         return Stream.concat(firstList.stream(), secondList.stream()).collect(Collectors.toList());
 242     }
 243
 244     private static HttpResponse createHttpResponse(String statusReason, int statusCode){
 245         return ImmutableHttpResponse.builder()
 246                 .statusCode(statusCode)
 247                 .url(sinkDefinition.topicUrl())
 248                 .statusReason(statusReason)
 249                 .rawBody("[]".getBytes())
 250                 .build();
 251     }
 252
 253     private static MessageRouterPublishRequest createMRRPublishRequest(){
 254         return ImmutableMessageRouterPublishRequest
 255                 .builder()
 256                 .sinkDefinition(sinkDefinition)
 257                 .build();
 258     }
 259
 260     private String collectNonEmptyRequestBody(HttpRequest httpRequest){
 261         final String body = Flux.from(httpRequest.body().contents())
 262                 .collect(ByteBufAllocator.DEFAULT::compositeBuffer,
 263                         (byteBufs, buffer) -> byteBufs.addComponent(true, buffer))
 264                 .map(byteBufs -> byteBufs.toString(StandardCharsets.UTF_8))
 265                 .block();
 266         assertThat(body).describedAs("request body").isNotBlank();
 267
 268         return body;
 269     }
 270
 271     private JsonArray extractNonEmptyJsonRequestBody(HttpRequest httpRequest){
 272         return new Gson().fromJson(collectNonEmptyRequestBody(httpRequest), JsonArray.class);
 273     }
 274
 275     private JsonObject getAsJsonObject(String item){
 276         return new Gson().fromJson(item, JsonObject.class);
 277     }
 278 }