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.sendString;
  25
  26 import com.google.gson.JsonElement;
  27 import com.google.gson.JsonPrimitive;
  28 import io.vavr.collection.List;
  29 import java.time.Duration;
  30 import org.junit.jupiter.api.BeforeAll;
  31 import org.junit.jupiter.api.Test;
  32 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.ImmutableMessageRouterSink;
  33 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSink;
  34 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer;
  35 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishRequest;
  36 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishResponse;
  37 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest;
  38 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse;
  39 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.MessageRouterPublisherConfig;
  40 import reactor.core.publisher.Flux;
  41 import reactor.core.publisher.Mono;
  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 MessageRouterPublisherIT {
  49
  50     private static final String ERROR_MESSAGE = "Something went wrong";
  51     private static final String TEXT_PLAIN_CONTENT_TYPE = "text/plain";
  52     private static final String JSON_CONTENT_TYPE = "application/json";
  53     private static final String SUCCESS_RESP_TOPIC_PATH = "/events/TOPIC";
  54     private static final String FAILING_WITH_400_RESP_PATH = "/events/TOPIC400";
  55     private static final String FAILING_WITH_401_RESP_PATH = "/events/TOPIC401";
  56     private static final String FAILING_WITH_403_RESP_PATH = "/events/TOPIC403";
  57     private static final String FAILING_WITH_404_RESP_PATH = "/events/TOPIC404";
  58     private static final String FAILING_WITH_500_TOPIC_PATH = "/events/TOPIC500";
  59     private static final Duration TIMEOUT = Duration.ofSeconds(10);
  60     private static final Flux<JsonPrimitive> messageBatch = Flux.just("ala", "ma", "kota")
  61             .map(JsonPrimitive::new);
  62     private static final List<String> messageBatchItems = List.of("ala", "ma", "kota");
  63
  64     private static DummyHttpServer server;
  65     private MessageRouterPublisher sut = DmaapClientFactory
  66             .createMessageRouterPublisher(MessageRouterPublisherConfig.createDefault());
  67
  68
  69     @BeforeAll
  70     static void setUp() {
  71         server = DummyHttpServer.start(routes ->
  72                 routes.post(SUCCESS_RESP_TOPIC_PATH, (req, resp) -> sendString(resp, Mono.just("OK")))
  73                         .post(FAILING_WITH_400_RESP_PATH, (req, resp) ->
  74                                 sendError(resp, 400, ERROR_MESSAGE))
  75                         .post(FAILING_WITH_401_RESP_PATH, (req, resp) ->
  76                                 sendError(resp, 401, ERROR_MESSAGE))
  77                         .post(FAILING_WITH_403_RESP_PATH, (req, resp) ->
  78                                 sendError(resp, 403, ERROR_MESSAGE))
  79                         .post(FAILING_WITH_404_RESP_PATH, (req, resp) ->
  80                                 sendError(resp, 404, ERROR_MESSAGE))
  81                         .post(FAILING_WITH_500_TOPIC_PATH, (req, resp) ->
  82                                 sendError(resp, 500, ERROR_MESSAGE))
  83         );
  84     }
  85
  86     @Test
  87     void test_put_givenMessageBatch_shouldMakeSuccessfulPostRequestReturningBatch(){
  88         //given
  89         final MessageRouterPublishRequest mrRequest = createMRRequest(SUCCESS_RESP_TOPIC_PATH,
  90                 TEXT_PLAIN_CONTENT_TYPE);
  91         final List<JsonElement> expectedItems = messageBatchItems.map(JsonPrimitive::new);
  92
  93
  94         //when
  95         final Flux<MessageRouterPublishResponse> result = sut.put(mrRequest, messageBatch);
  96
  97         //then
  98         StepVerifier.create(result)
  99                 .expectNext(ImmutableMessageRouterPublishResponse.builder().items(expectedItems).build())
 100                 .expectComplete()
 101                 .verify(TIMEOUT);
 102     }
 103
 104     @Test
 105     void publisher_shouldHandleBadRequestError(){
 106         //given
 107         final MessageRouterPublishRequest mrRequest = createMRRequest(FAILING_WITH_400_RESP_PATH,
 108                 JSON_CONTENT_TYPE);
 109         final MessageRouterPublishResponse expectedResponse = createErrorResponse(
 110                 "400 Bad Request\n%s", ERROR_MESSAGE);
 111
 112         //when
 113         final Flux<MessageRouterPublishResponse> result = sut.put(mrRequest, messageBatch);
 114
 115         //then
 116         StepVerifier.create(result)
 117                 .expectNext(expectedResponse)
 118                 .expectComplete()
 119                 .verify(TIMEOUT);
 120     }
 121
 122     @Test
 123     void publisher_shouldHandleUnauthorizedError(){
 124         //given
 125         final MessageRouterPublishRequest mrRequest = createMRRequest(FAILING_WITH_401_RESP_PATH,
 126                 TEXT_PLAIN_CONTENT_TYPE);
 127         final MessageRouterPublishResponse expectedResponse = createErrorResponse(
 128                 "401 Unauthorized\n%s", ERROR_MESSAGE);
 129
 130         //when
 131         final Flux<MessageRouterPublishResponse> result = sut.put(mrRequest, messageBatch);
 132
 133         //then
 134         StepVerifier.create(result)
 135                 .expectNext(expectedResponse)
 136                 .expectComplete()
 137                 .verify(TIMEOUT);
 138     }
 139
 140     @Test
 141     void publisher_shouldHandleForbiddenError(){
 142         //given
 143         final MessageRouterPublishRequest mrRequest = createMRRequest(FAILING_WITH_403_RESP_PATH,
 144                 TEXT_PLAIN_CONTENT_TYPE);
 145         final MessageRouterPublishResponse expectedResponse = createErrorResponse(
 146                 "403 Forbidden\n%s", ERROR_MESSAGE);
 147
 148         //when
 149         final Flux<MessageRouterPublishResponse> result = sut
 150                 .put(mrRequest, messageBatch);
 151
 152         //then
 153         StepVerifier.create(result)
 154                 .expectNext(expectedResponse)
 155                 .expectComplete()
 156                 .verify(TIMEOUT);
 157     }
 158
 159     @Test
 160     void publisher_shouldHandleNotFoundError(){
 161         //given
 162         final MessageRouterPublishRequest mrRequest = createMRRequest(FAILING_WITH_404_RESP_PATH,
 163                 TEXT_PLAIN_CONTENT_TYPE);
 164         final MessageRouterPublishResponse expectedResponse = createErrorResponse(
 165                 "404 Not Found\n%s", ERROR_MESSAGE);
 166
 167         //when
 168         final Flux<MessageRouterPublishResponse> result = sut
 169                 .put(mrRequest, messageBatch);
 170
 171         //then
 172         StepVerifier.create(result)
 173                 .expectNext(expectedResponse)
 174                 .expectComplete()
 175                 .verify(TIMEOUT);
 176     }
 177
 178     @Test
 179     void publisher_shouldHandleInternalServerError(){
 180         //given
 181         final MessageRouterPublishRequest mrRequest = createMRRequest(FAILING_WITH_500_TOPIC_PATH,
 182                 TEXT_PLAIN_CONTENT_TYPE);
 183         final MessageRouterPublishResponse expectedResponse = createErrorResponse(
 184                 "500 Internal Server Error\n%s", ERROR_MESSAGE);
 185
 186         //when
 187         final Flux<MessageRouterPublishResponse> result = sut
 188                 .put(mrRequest, messageBatch);
 189
 190         //then
 191         StepVerifier.create(result)
 192                 .expectNext(expectedResponse)
 193                 .expectComplete()
 194                 .verify(TIMEOUT);
 195     }
 196
 197
 198     private MessageRouterPublishRequest createMRRequest(String topicPath, String contentType){
 199         final MessageRouterSink sinkDefinition = ImmutableMessageRouterSink.builder()
 200                 .name("the topic")
 201                 .topicUrl(String.format("http://%s:%d%s",
 202                         server.host(),
 203                         server.port(),
 204                         topicPath)
 205                 )
 206                 .build();
 207
 208         return ImmutableMessageRouterPublishRequest.builder()
 209                 .sinkDefinition(sinkDefinition)
 210                 .contentType(contentType)
 211                 .build();
 212     }
 213
 214     private MessageRouterPublishResponse createErrorResponse(String failReasonFormat, Object... formatArgs){
 215         return ImmutableMessageRouterPublishResponse
 216                 .builder()
 217                 .failReason(String.format(failReasonFormat, formatArgs))
 218                 .build();
 219     }
 220 }