rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisherTest.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.ImmutableMessageRouterSink;
  31 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSink;
  32 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer;
  33 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.ContentType;
  34 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.DmaapResponse;
  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.ImmutableDmaapTimeoutConfig;
  40 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.MessageRouterPublisherConfig;
  41 import reactor.core.publisher.Flux;
  42 import reactor.core.publisher.Mono;
  43 import reactor.test.StepVerifier;
  44
  45 import java.time.Duration;
  46
  47 import static org.assertj.core.api.Assertions.assertThat;
  48 import static org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer.sendError;
  49 import static org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer.sendString;
  50 import static org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer.sendWithDelay;
  51
  52 /**
  53  * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
  54  * @since May 2019
  55  */
  56 class MessageRouterPublisherTest {
  57
  58     private static final String ERROR_MESSAGE = "Something went wrong";
  59     private static final String TIMEOUT_ERROR_MESSAGE = "408 Request Timeout";
  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
  72     private static DummyHttpServer server;
  73     private MessageRouterPublisher sut = DmaapClientFactory
  74             .createMessageRouterPublisher(MessageRouterPublisherConfig.createDefault());
  75
  76
  77     @BeforeAll
  78     static void setUp() {
  79         server = DummyHttpServer.start(routes -> routes
  80                 .post(SUCCESS_RESP_TOPIC_PATH, (req, resp) -> sendString(resp, Mono.just("OK")))
  81                 .post(DELAY_RESP_TOPIC_PATH, (req, resp) -> sendWithDelay(resp, 200, TIMEOUT))
  82                 .post(FAILING_WITH_400_RESP_PATH, (req, resp) -> sendError(resp, 400, ERROR_MESSAGE))
  83                 .post(FAILING_WITH_401_RESP_PATH, (req, resp) -> sendError(resp, 401, ERROR_MESSAGE))
  84                 .post(FAILING_WITH_403_RESP_PATH, (req, resp) -> sendError(resp, 403, ERROR_MESSAGE))
  85                 .post(FAILING_WITH_404_RESP_PATH, (req, resp) -> sendError(resp, 404, ERROR_MESSAGE))
  86                 .post(FAILING_WITH_500_RESP_PATH, (req, resp) -> sendError(resp, 500, ERROR_MESSAGE))
  87         );
  88     }
  89
  90     @Test
  91     void test_put_givenMessageBatch_shouldMakeSuccessfulPostRequestReturningBatch() {
  92         //given
  93         final MessageRouterPublishRequest mrRequest = createTextPlainMRRequest(SUCCESS_RESP_TOPIC_PATH);
  94         final List<JsonElement> expectedItems = messageBatchItems.map(JsonPrimitive::new);
  95
  96         //when
  97         final Flux<MessageRouterPublishResponse> result = sut.put(mrRequest, messageBatch);
  98
  99         //then
 100         StepVerifier.create(result)
 101                 .expectNext(ImmutableMessageRouterPublishResponse.builder().items(expectedItems).build())
 102                 .expectComplete()
 103                 .verify(TIMEOUT);
 104     }
 105
 106     @ParameterizedTest
 107     @CsvSource({
 108             FAILING_WITH_400_RESP_PATH + "," + "400 Bad Request",
 109             FAILING_WITH_401_RESP_PATH + "," + "401 Unauthorized",
 110             FAILING_WITH_403_RESP_PATH + "," + "403 Forbidden",
 111             FAILING_WITH_404_RESP_PATH + "," + "404 Not Found",
 112             FAILING_WITH_500_RESP_PATH + "," + "500 Internal Server Error"
 113     })
 114     void publisher_shouldHandleError(String failingPath, String failReason) {
 115         //given
 116         final MessageRouterPublishRequest mrRequest = createTextPlainMRRequest(failingPath);
 117         final MessageRouterPublishResponse expectedResponse = createErrorResponse(failReason);
 118
 119         //when
 120         final Flux<MessageRouterPublishResponse> result = sut.put(mrRequest, messageBatch);
 121
 122         //then
 123         StepVerifier.create(result)
 124                 .expectNext(expectedResponse)
 125                 .expectComplete()
 126                 .verify(TIMEOUT);
 127     }
 128
 129     @Test
 130     void publisher_shouldHandleClientTimeoutError() {
 131         //given
 132         final Duration requestTimeout = Duration.ofMillis(1);
 133         final MessageRouterPublishRequest mrRequest = createTextPlainMRRequest(DELAY_RESP_TOPIC_PATH, requestTimeout);
 134
 135         //when
 136         final Flux<MessageRouterPublishResponse> result = sut.put(mrRequest, messageBatch);
 137
 138         //then
 139         StepVerifier.create(result)
 140                 .consumeNextWith(this::assertTimeoutError)
 141                 .expectComplete()
 142                 .verify(TIMEOUT);
 143     }
 144
 145     private static MessageRouterPublishRequest createTextPlainMRRequest(String topicPath) {
 146         final MessageRouterSink sinkDefinition = createMRSink(topicPath);
 147         return ImmutableMessageRouterPublishRequest.builder()
 148                 .sinkDefinition(sinkDefinition)
 149                 .contentType(ContentType.TEXT_PLAIN)
 150                 .build();
 151     }
 152
 153     private static MessageRouterPublishRequest createTextPlainMRRequest(String topicPath, Duration timeout) {
 154         final MessageRouterSink sinkDefinition = createMRSink(topicPath);
 155         return ImmutableMessageRouterPublishRequest.builder()
 156                 .sinkDefinition(sinkDefinition)
 157                 .contentType(ContentType.TEXT_PLAIN)
 158                 .timeoutConfig(ImmutableDmaapTimeoutConfig.builder().timeout(timeout).build())
 159                 .build();
 160     }
 161
 162     private static MessageRouterSink createMRSink(String topicPath) {
 163         return ImmutableMessageRouterSink.builder()
 164                 .name("the topic")
 165                 .topicUrl(String.format("http://%s:%d%s",
 166                         server.host(),
 167                         server.port(),
 168                         topicPath)
 169                 )
 170                 .build();
 171     }
 172
 173     private static MessageRouterPublishResponse createErrorResponse(String failReason) {
 174         String failReasonFormat = failReason + "\n%s";
 175         return ImmutableMessageRouterPublishResponse
 176                 .builder()
 177                 .failReason(String.format(failReasonFormat, ERROR_MESSAGE))
 178                 .build();
 179     }
 180
 181     private void assertTimeoutError(DmaapResponse response) {
 182         assertThat(response.failed()).isTrue();
 183         assertThat(response.failReason()).startsWith(TIMEOUT_ERROR_MESSAGE);
 184     }
 185 }
 186