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 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.JsonElement;
  27 import com.google.gson.JsonPrimitive;
  28 import io.vavr.collection.List;
  29
  30 import java.time.Duration;
  31
  32 import org.junit.jupiter.api.BeforeAll;
  33 import org.junit.jupiter.api.Test;
  34 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.ImmutableMessageRouterSource;
  35 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSource;
  36 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer;
  37 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeRequest;
  38 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeResponse;
  39 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeRequest;
  40 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeResponse;
  41 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.MessageRouterSubscriberConfig;
  42 import reactor.core.publisher.Flux;
  43 import reactor.core.publisher.Mono;
  44 import reactor.netty.http.server.HttpServerRoutes;
  45 import reactor.test.StepVerifier;
  46
  47 /**
  48  * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
  49  * @since May 2019
  50  */
  51 class MessageRouterSubscriberTest {
  52     private static final Duration TIMEOUT = Duration.ofSeconds(10);
  53     private static final String ERROR_MESSAGE = "Something went wrong";
  54     private static final String CONSUMER_GROUP = "group1";
  55     private static final String SUCCESS_CONSUMER_ID = "consumer200";
  56     private static final String FAILING_WITH_401_CONSUMER_ID = "consumer401";
  57     private static final String FAILING_WITH_403_CONSUMER_ID = "consumer403";
  58     private static final String FAILING_WITH_409_CONSUMER_ID = "consumer409";
  59     private static final String FAILING_WITH_429_CONSUMER_ID = "consumer429";
  60     private static final String FAILING_WITH_500_CONSUMER_ID = "consumer500";
  61
  62     private static final String CONSUMER_PATH = String.format("/events/TOPIC/%s", CONSUMER_GROUP);
  63
  64     private static final String SUCCESS_RESP_PATH = String
  65             .format("%s/%s", CONSUMER_PATH, SUCCESS_CONSUMER_ID);
  66     private static final String FAILING_WITH_401_RESP_PATH = String
  67             .format("%s/%s", CONSUMER_PATH, FAILING_WITH_401_CONSUMER_ID);
  68     private static final String FAILING_WITH_403_RESP_PATH = String
  69             .format("%s/%s", CONSUMER_PATH, FAILING_WITH_403_CONSUMER_ID);
  70     private static final String FAILING_WITH_409_RESP_PATH = String
  71             .format("%s/%s", CONSUMER_PATH, FAILING_WITH_409_CONSUMER_ID);
  72     private static final String FAILING_WITH_429_RESP_PATH = String
  73             .format("%s/%s", CONSUMER_PATH, FAILING_WITH_429_CONSUMER_ID);
  74     private static final String FAILING_WITH_500_RESP_PATH = String
  75             .format("%s/%s", CONSUMER_PATH, FAILING_WITH_500_CONSUMER_ID);
  76
  77     private static MessageRouterSubscribeRequest mrSuccessRequest;
  78     private static MessageRouterSubscribeRequest mrFailingRequest;
  79     private MessageRouterSubscriber sut = DmaapClientFactory
  80             .createMessageRouterSubscriber(MessageRouterSubscriberConfig.createDefault());
  81     private static MessageRouterSource sourceDefinition;
  82
  83
  84     @BeforeAll
  85     static void setUp() {
  86         DummyHttpServer server = DummyHttpServer.start(MessageRouterSubscriberTest::setRoutes);
  87
  88         sourceDefinition = createMessageRouterSource(server);
  89
  90         mrSuccessRequest = createSuccessRequest();
  91
  92         mrFailingRequest = createFailingRequest(FAILING_WITH_500_CONSUMER_ID);
  93     }
  94
  95     @Test
  96     void subscriber_shouldGetCorrectResponse() {
  97         Mono<MessageRouterSubscribeResponse> response = sut
  98                 .get(mrSuccessRequest);
  99
 100         List<String> expectedItems = List.of("I", "like", "pizza");
 101
 102         MessageRouterSubscribeResponse expectedResponse = ImmutableMessageRouterSubscribeResponse
 103                 .builder()
 104                 .items(expectedItems.map(JsonPrimitive::new))
 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 }
 276