rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberTest.java

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