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.ImmutableMessageRouterSink;
  30 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSink;
  31 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer;
  32 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.ContentType;
  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.ImmutableMessageRouterPublishRequest;
  35 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishResponse;
  36 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest;
  37 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse;
  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.MessageRouterPublisherConfig;
  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.sendString;
  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 MessageRouterPublisherTest {
  56
  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 CONNECTION_ERROR_MESSAGE = "503 Service unavailable";
  60     private static final String SUCCESS_RESP_TOPIC_PATH = "/events/TOPIC";
  61     private static final String DELAY_RESP_TOPIC_PATH = "/events/DELAY";
  62     private static final String FAILING_WITH_400_RESP_PATH = "/events/TOPIC400";
  63     private static final String FAILING_WITH_401_RESP_PATH = "/events/TOPIC401";
  64     private static final String FAILING_WITH_403_RESP_PATH = "/events/TOPIC403";
  65     private static final String FAILING_WITH_404_RESP_PATH = "/events/TOPIC404";
  66     private static final String FAILING_WITH_500_RESP_PATH = "/events/TOPIC500";
  67     private static final Duration TIMEOUT = Duration.ofSeconds(10);
  68     private static final Flux<JsonPrimitive> messageBatch = Flux.just("ala", "ma", "kota")
  69             .map(JsonPrimitive::new);
  70     private static final List<String> messageBatchItems = List.of("ala", "ma", "kota");
  71     private static final DummyHttpServer DISPOSED_HTTP_SERVER = initialize().closeAndGet();
  72     private static final DummyHttpServer SERVER = initialize();
  73     private MessageRouterPublisher sut = DmaapClientFactory
  74             .createMessageRouterPublisher(MessageRouterPublisherConfig.createDefault());
  75
  76     private static DummyHttpServer initialize() {
  77         return DummyHttpServer.start(routes -> routes
  78                 .post(SUCCESS_RESP_TOPIC_PATH, (req, resp) -> sendString(resp, Mono.just("OK")))
  79                 .post(DELAY_RESP_TOPIC_PATH, (req, resp) -> sendWithDelay(resp, 200, TIMEOUT))
  80                 .post(FAILING_WITH_400_RESP_PATH, (req, resp) -> sendError(resp, 400, ERROR_MESSAGE))
  81                 .post(FAILING_WITH_401_RESP_PATH, (req, resp) -> sendError(resp, 401, ERROR_MESSAGE))
  82                 .post(FAILING_WITH_403_RESP_PATH, (req, resp) -> sendError(resp, 403, ERROR_MESSAGE))
  83                 .post(FAILING_WITH_404_RESP_PATH, (req, resp) -> sendError(resp, 404, ERROR_MESSAGE))
  84                 .post(FAILING_WITH_500_RESP_PATH, (req, resp) -> sendError(resp, 500, ERROR_MESSAGE))
  85         );
  86     }
  87
  88     @Test
  89     void test_put_givenMessageBatch_shouldMakeSuccessfulPostRequestReturningBatch() {
  90         //given
  91         final MessageRouterPublishRequest mrRequest = createTextPlainMRRequest(SUCCESS_RESP_TOPIC_PATH, SERVER);
  92         final List<JsonElement> expectedItems = messageBatchItems.map(JsonPrimitive::new);
  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     @ParameterizedTest
 105     @CsvSource({
 106             FAILING_WITH_400_RESP_PATH + "," + "400 Bad Request",
 107             FAILING_WITH_401_RESP_PATH + "," + "401 Unauthorized",
 108             FAILING_WITH_403_RESP_PATH + "," + "403 Forbidden",
 109             FAILING_WITH_404_RESP_PATH + "," + "404 Not Found",
 110             FAILING_WITH_500_RESP_PATH + "," + "500 Internal Server Error"
 111     })
 112     void publisher_shouldHandleError(String failingPath, String failReason) {
 113         //given
 114         final MessageRouterPublishRequest mrRequest = createTextPlainMRRequest(failingPath, SERVER);
 115         final MessageRouterPublishResponse expectedResponse = createErrorResponse(failReason);
 116
 117         //when
 118         final Flux<MessageRouterPublishResponse> result = sut.put(mrRequest, messageBatch);
 119
 120         //then
 121         StepVerifier.create(result)
 122                 .expectNext(expectedResponse)
 123                 .expectComplete()
 124                 .verify(TIMEOUT);
 125     }
 126
 127     @Test
 128     void publisher_shouldHandleClientTimeoutError() {
 129         //given
 130         final Duration requestTimeout = Duration.ofMillis(1);
 131         final MessageRouterPublishRequest mrRequest = createTextPlainMRRequest(DELAY_RESP_TOPIC_PATH, requestTimeout);
 132
 133         //when
 134         final Flux<MessageRouterPublishResponse> result = sut.put(mrRequest, messageBatch);
 135
 136         //then
 137         StepVerifier.create(result)
 138                 .consumeNextWith(this::assertTimeoutError)
 139                 .expectComplete()
 140                 .verify(TIMEOUT);
 141     }
 142
 143     @Test
 144     void publisher_shouldHandleConnectionError() {
 145         //given
 146         final MessageRouterPublishRequest mrRequest = createTextPlainMRRequest(
 147                 SUCCESS_RESP_TOPIC_PATH, DISPOSED_HTTP_SERVER);
 148
 149         //when
 150         final Flux<MessageRouterPublishResponse> result = sut.put(mrRequest, messageBatch);
 151
 152         //then
 153         StepVerifier.create(result)
 154                 .consumeNextWith(this::assertConnectionError)
 155                 .expectComplete()
 156                 .verify(TIMEOUT);
 157     }
 158
 159     private static MessageRouterPublishRequest createTextPlainMRRequest(String topicPath, DummyHttpServer dummyHttpServer) {
 160         final MessageRouterSink sinkDefinition = createMRSink(topicPath, dummyHttpServer);
 161         return ImmutableMessageRouterPublishRequest.builder()
 162                 .sinkDefinition(sinkDefinition)
 163                 .contentType(ContentType.TEXT_PLAIN)
 164                 .build();
 165     }
 166
 167     private static MessageRouterPublishRequest createTextPlainMRRequest(String topicPath, Duration timeout) {
 168         final MessageRouterSink sinkDefinition = createMRSink(topicPath, SERVER);
 169         return ImmutableMessageRouterPublishRequest.builder()
 170                 .sinkDefinition(sinkDefinition)
 171                 .contentType(ContentType.TEXT_PLAIN)
 172                 .timeoutConfig(ImmutableDmaapTimeoutConfig.builder().timeout(timeout).build())
 173                 .build();
 174     }
 175
 176     private static MessageRouterSink createMRSink(String topicPath, DummyHttpServer dummyHttpServer) {
 177         return ImmutableMessageRouterSink.builder()
 178                 .name("the topic")
 179                 .topicUrl(String.format("http://%s:%d%s",
 180                         dummyHttpServer.host(),
 181                         dummyHttpServer.port(),
 182                         topicPath)
 183                 )
 184                 .build();
 185     }
 186
 187     private static MessageRouterPublishResponse createErrorResponse(String failReason) {
 188         String failReasonFormat = failReason + "\n%s";
 189         return ImmutableMessageRouterPublishResponse
 190                 .builder()
 191                 .failReason(String.format(failReasonFormat, ERROR_MESSAGE))
 192                 .build();
 193     }
 194
 195     private void assertTimeoutError(DmaapResponse response) {
 196         assertThat(response.failed()).isTrue();
 197         assertThat(response.failReason()).startsWith(TIMEOUT_ERROR_MESSAGE);
 198     }
 199
 200     private void assertConnectionError(DmaapResponse response) {
 201         assertThat(response.failed()).isTrue();
 202         assertThat(response.failReason()).startsWith(CONNECTION_ERROR_MESSAGE);
 203     }
 204 }
 205