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

   1 /*
   2  * ============LICENSE_START====================================
   3  * DCAEGEN2-SERVICES-SDK
   4  * =========================================================
   5  * Copyright (C) 2019-2021 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.api;
  22
  23 import com.google.gson.JsonElement;
  24 import com.google.gson.JsonPrimitive;
  25 import io.vavr.collection.List;
  26 import org.junit.jupiter.api.Test;
  27 import org.junit.jupiter.params.ParameterizedTest;
  28 import org.junit.jupiter.params.provider.CsvSource;
  29 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.ImmutableMessageRouterSource;
  30 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSource;
  31 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer;
  32 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.DmaapResponse;
  33 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeRequest;
  34 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeResponse;
  35 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeRequest;
  36 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeResponse;
  37 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.ImmutableDmaapTimeoutConfig;
  38 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.MessageRouterSubscriberConfig;
  39 import reactor.core.publisher.Flux;
  40 import reactor.core.publisher.Mono;
  41 import reactor.test.StepVerifier;
  42
  43 import java.time.Duration;
  44
  45 import static org.assertj.core.api.Assertions.assertThat;
  46 import static org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer.sendError;
  47 import static org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer.sendResource;
  48 import static org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer.sendWithDelay;
  49
  50 /**
  51  * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
  52  * @since May 2019
  53  */
  54 class MessageRouterSubscriberTest {
  55     private static final Duration TIMEOUT = Duration.ofSeconds(10);
  56     private static final String ERROR_MESSAGE = "Something went wrong";
  57     private static final String CONNECTION_ERROR_MESSAGE = "503 Service unavailable";
  58     private static final String TIMEOUT_ERROR_MESSAGE = "408 Request Timeout";
  59     private static final String CONSUMER_GROUP = "group1";
  60     private static final String SUCCESS_CONSUMER_ID = "consumer200";
  61     private static final String DELAY_CONSUMER_ID = "delay200";
  62     private static final String FAILING_WITH_401_CONSUMER_ID = "consumer401";
  63     private static final String FAILING_WITH_403_CONSUMER_ID = "consumer403";
  64     private static final String FAILING_WITH_409_CONSUMER_ID = "consumer409";
  65     private static final String FAILING_WITH_429_CONSUMER_ID = "consumer429";
  66     private static final String FAILING_WITH_500_CONSUMER_ID = "consumer500";
  67
  68     private static final String CONSUMER_PATH = String.format("/events/TOPIC/%s", CONSUMER_GROUP);
  69
  70     private static final String SUCCESS_RESP_PATH = String
  71             .format("%s/%s", CONSUMER_PATH, SUCCESS_CONSUMER_ID);
  72     private static final String DELAY_RESP_PATH = String
  73             .format("%s/%s", CONSUMER_PATH, DELAY_CONSUMER_ID);
  74     private static final String FAILING_WITH_401_RESP_PATH = String
  75             .format("%s/%s", CONSUMER_PATH, FAILING_WITH_401_CONSUMER_ID);
  76     private static final String FAILING_WITH_403_RESP_PATH = String
  77             .format("%s/%s", CONSUMER_PATH, FAILING_WITH_403_CONSUMER_ID);
  78     private static final String FAILING_WITH_409_RESP_PATH = String
  79             .format("%s/%s", CONSUMER_PATH, FAILING_WITH_409_CONSUMER_ID);
  80     private static final String FAILING_WITH_429_RESP_PATH = String
  81             .format("%s/%s", CONSUMER_PATH, FAILING_WITH_429_CONSUMER_ID);
  82     private static final String FAILING_WITH_500_RESP_PATH = String
  83             .format("%s/%s", CONSUMER_PATH, FAILING_WITH_500_CONSUMER_ID);
  84
  85     private static final DummyHttpServer DISPOSED_HTTP_SERVER = initialize().closeAndGet();
  86     private static final DummyHttpServer SERVER = initialize();
  87
  88     private MessageRouterSubscriber sut = DmaapClientFactory
  89             .createMessageRouterSubscriber(MessageRouterSubscriberConfig.createDefault());
  90     private static MessageRouterSource sourceDefinition = createMessageRouterSource(SERVER);
  91     private static MessageRouterSource failingSourceDefinition = createMessageRouterSource(DISPOSED_HTTP_SERVER);
  92     private static MessageRouterSubscribeRequest mrSuccessRequest = createSuccessRequest(sourceDefinition);
  93     private static MessageRouterSubscribeRequest mrFailingRequest = createFailingRequest(FAILING_WITH_500_CONSUMER_ID);
  94
  95     private static DummyHttpServer initialize() {
  96         return DummyHttpServer.start(routes -> routes
  97                 .get(SUCCESS_RESP_PATH, (req, resp) ->
  98                         sendResource(resp, "/sample-mr-subscribe-response.json"))
  99                 .get(DELAY_RESP_PATH, (req, resp) -> sendWithDelay(resp, 200, TIMEOUT))
 100                 .get(FAILING_WITH_401_RESP_PATH, (req, resp) -> sendError(resp, 401, ERROR_MESSAGE))
 101                 .get(FAILING_WITH_403_RESP_PATH, (req, resp) -> sendError(resp, 403, ERROR_MESSAGE))
 102                 .get(FAILING_WITH_409_RESP_PATH, (req, resp) -> sendError(resp, 409, ERROR_MESSAGE))
 103                 .get(FAILING_WITH_429_RESP_PATH, (req, resp) -> sendError(resp, 429, ERROR_MESSAGE))
 104                 .get(FAILING_WITH_500_RESP_PATH, (req, resp) -> sendError(resp, 500, ERROR_MESSAGE)));
 105     }
 106
 107     @Test
 108     void subscriber_shouldGetCorrectResponse() {
 109         Mono<MessageRouterSubscribeResponse> response = sut
 110                 .get(mrSuccessRequest);
 111
 112         List<String> expectedItems = List.of("I", "like", "pizza");
 113
 114         MessageRouterSubscribeResponse expectedResponse = ImmutableMessageRouterSubscribeResponse
 115                 .builder()
 116                 .items(expectedItems.map(JsonPrimitive::new))
 117                 .build();
 118
 119         StepVerifier.create(response)
 120                 .expectNext(expectedResponse)
 121                 .expectComplete()
 122                 .verify(TIMEOUT);
 123     }
 124
 125     @ParameterizedTest
 126     @CsvSource({
 127             FAILING_WITH_401_CONSUMER_ID + "," + "401 Unauthorized",
 128             FAILING_WITH_403_CONSUMER_ID + "," + "403 Forbidden",
 129             FAILING_WITH_409_CONSUMER_ID + "," + "409 Conflict",
 130             FAILING_WITH_429_CONSUMER_ID + "," + "429 Too Many Requests",
 131             FAILING_WITH_500_CONSUMER_ID + "," + "500 Internal Server Error"
 132     })
 133     void subscriber_shouldHandleError(String consumerId, String failReason) {
 134         MessageRouterSubscribeRequest request = createFailingRequest(consumerId);
 135         Mono<MessageRouterSubscribeResponse> response = sut.get(request);
 136
 137         MessageRouterSubscribeResponse expectedResponse = createErrorResponse(failReason);
 138
 139         StepVerifier.create(response)
 140                 .expectNext(expectedResponse)
 141                 .expectComplete()
 142                 .verify(TIMEOUT);
 143     }
 144
 145     @Test
 146     void subscriber_shouldParseCorrectResponse() {
 147         final Flux<String> result = sut
 148                 .getElements(mrSuccessRequest)
 149                 .map(JsonElement::getAsString);
 150
 151         StepVerifier.create(result)
 152                 .expectNext("I", "like", "pizza")
 153                 .expectComplete()
 154                 .verify(TIMEOUT);
 155     }
 156
 157     @Test
 158     void subscriber_shouldParseErrorResponse() {
 159         Flux<String> result = sut
 160                 .getElements(mrFailingRequest)
 161                 .map(JsonElement::getAsString);
 162
 163         StepVerifier.create(result)
 164                 .expectError(IllegalStateException.class)
 165                 .verify(TIMEOUT);
 166     }
 167
 168     @Test
 169     void subscriber_shouldSubscribeCorrectly() {
 170         Flux<String> subscriptionForElements = sut
 171                 .subscribeForElements(mrSuccessRequest, Duration.ofSeconds(1))
 172                 .map(JsonElement::getAsString);
 173
 174         StepVerifier.create(subscriptionForElements.take(2))
 175                 .expectNext("I", "like")
 176                 .expectComplete()
 177                 .verify(TIMEOUT);
 178     }
 179
 180     @Test
 181     void subscriber_shouldParseErrorWhenSubscribed() {
 182         Flux<String> subscriptionForElements = sut
 183                 .subscribeForElements(mrFailingRequest, Duration.ofSeconds(1))
 184                 .map(JsonElement::getAsString);
 185
 186         StepVerifier.create(subscriptionForElements.take(2))
 187                 .expectError(IllegalStateException.class)
 188                 .verify(TIMEOUT);
 189     }
 190
 191     @Test
 192     void subscriber_shouldHandleClientTimeoutError() {
 193         Duration requestTimeout = Duration.ofMillis(1);
 194         MessageRouterSubscribeRequest request = createDelayRequest(DELAY_CONSUMER_ID, requestTimeout);
 195         Mono<MessageRouterSubscribeResponse> response = sut.get(request);
 196
 197         StepVerifier.create(response)
 198                 .consumeNextWith(this::assertTimeoutError)
 199                 .expectComplete()
 200                 .verify(TIMEOUT);
 201     }
 202
 203     @Test
 204     void subscriber_shouldHandleConnectionError() {
 205         MessageRouterSubscribeRequest request = createSuccessRequest(failingSourceDefinition);
 206         Mono<MessageRouterSubscribeResponse> response = sut.get(request);
 207
 208         StepVerifier.create(response)
 209                 .consumeNextWith(this::assertConnectionError)
 210                 .expectComplete()
 211                 .verify(TIMEOUT);
 212     }
 213
 214     private static MessageRouterSource createMessageRouterSource(DummyHttpServer server) {
 215         return ImmutableMessageRouterSource.builder()
 216                 .name("the topic")
 217                 .topicUrl(String.format("http://%s:%d/events/TOPIC", server.host(), server.port()))
 218                 .build();
 219     }
 220
 221     private static MessageRouterSubscribeRequest createSuccessRequest(MessageRouterSource source) {
 222         return ImmutableMessageRouterSubscribeRequest.builder()
 223                 .sourceDefinition(source)
 224                 .consumerGroup(CONSUMER_GROUP)
 225                 .consumerId(SUCCESS_CONSUMER_ID)
 226                 .build();
 227     }
 228
 229     private static MessageRouterSubscribeRequest createDelayRequest(String consumerId, Duration timeout) {
 230         return ImmutableMessageRouterSubscribeRequest.builder()
 231                 .sourceDefinition(sourceDefinition)
 232                 .consumerGroup(CONSUMER_GROUP)
 233                 .consumerId(consumerId)
 234                 .timeoutConfig(ImmutableDmaapTimeoutConfig.builder().timeout(timeout).build())
 235                 .build();
 236     }
 237
 238     private static MessageRouterSubscribeRequest createFailingRequest(String consumerId) {
 239         return ImmutableMessageRouterSubscribeRequest
 240                 .builder()
 241                 .sourceDefinition(sourceDefinition)
 242                 .consumerGroup(CONSUMER_GROUP)
 243                 .consumerId(consumerId)
 244                 .build();
 245     }
 246
 247     private MessageRouterSubscribeResponse createErrorResponse(String failReason) {
 248         String failReasonFormat = failReason + "\n%s";
 249         return ImmutableMessageRouterSubscribeResponse
 250                 .builder()
 251                 .failReason(String.format(failReasonFormat, ERROR_MESSAGE))
 252                 .build();
 253     }
 254
 255     private void assertTimeoutError(DmaapResponse response) {
 256         assertThat(response.failed()).isTrue();
 257         assertThat(response.failReason()).startsWith(TIMEOUT_ERROR_MESSAGE);
 258     }
 259
 260     private void assertConnectionError(DmaapResponse response) {
 261         assertThat(response.failed()).isTrue();
 262         assertThat(response.failReason()).startsWith(CONNECTION_ERROR_MESSAGE);
 263     }
 264 }
 265