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