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.cbs.client.impl;
  22
  23 import static org.assertj.core.api.Assertions.assertThat;
  24 import static org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer.sendResource;
  25 import static org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer.sendString;
  26 import static org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamPredicates.streamOfType;
  27 import static org.onap.dcaegen2.services.sdk.model.streams.StreamType.KAFKA;
  28 import static org.onap.dcaegen2.services.sdk.model.streams.StreamType.MESSAGE_ROUTER;
  29
  30 import com.google.gson.JsonObject;
  31 import io.vavr.collection.Stream;
  32 import java.time.Duration;
  33 import org.junit.jupiter.api.AfterAll;
  34 import org.junit.jupiter.api.BeforeAll;
  35 import org.junit.jupiter.api.Test;
  36 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer;
  37 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient;
  38 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClientFactory;
  39 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsRequests;
  40 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.exceptions.StreamParsingException;
  41 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.DataStreams;
  42 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParser;
  43 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParsers;
  44 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsRequest;
  45 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.EnvProperties;
  46 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.ImmutableEnvProperties;
  47 import org.onap.dcaegen2.services.sdk.model.streams.RawDataStream;
  48 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.KafkaSink;
  49 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.KafkaSource;
  50 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSink;
  51 import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
  52 import reactor.core.publisher.Flux;
  53 import reactor.core.publisher.Mono;
  54 import reactor.test.StepVerifier;
  55
  56 /**
  57  * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
  58  * @since February 2019
  59  */
  60 class CbsClientImplIT {
  61
  62     private static final String CONSUL_RESPONSE = "[\n"
  63             + "    {\n"
  64             + "        \"ServiceAddress\": \"HOST\",\n"
  65             + "        \"ServiceName\": \"the_cbs\",\n"
  66             + "        \"ServicePort\": PORT\n"
  67             + "    }\n"
  68             + "]\n";
  69     private static final String SAMPLE_CONFIG = "/sample_service_config.json";
  70     private static final String SAMPLE_ALL = "/sample_all.json";
  71     private static final String SAMPLE_KEY = "/sample_key.json";
  72     private static final String SAMPLE_CONFIG_KEY = "keystore.path";
  73     private static final String EXPECTED_CONFIG_VALUE = "/var/run/security/keystore.p12";
  74     private static EnvProperties sampleEnvironment;
  75     private static DummyHttpServer server;
  76
  77     @BeforeAll
  78     static void setUp() {
  79         server = DummyHttpServer.start(routes ->
  80                 routes.get("/v1/catalog/service/the_cbs", (req, resp) -> sendString(resp, lazyConsulResponse()))
  81                         .get("/service_component/dcae-component", (req, resp) -> sendResource(resp, SAMPLE_CONFIG))
  82                         .get("/service_component_all/dcae-component", (req, resp) -> sendResource(resp, SAMPLE_ALL))
  83                         .get("/sampleKey/dcae-component", (req, resp) -> sendResource(resp, SAMPLE_KEY))
  84         );
  85         sampleEnvironment = ImmutableEnvProperties.builder()
  86                 .appName("dcae-component")
  87                 .cbsName("the_cbs")
  88                 .consulHost(server.host())
  89                 .consulPort(server.port())
  90                 .build();
  91     }
  92
  93     @AfterAll
  94     static void tearDown() {
  95         server.close();
  96     }
  97
  98     @Test
  99     void testCbsClientWithSingleCall() {
 100         // given
 101         final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleEnvironment);
 102         final CbsRequest request = CbsRequests.getConfiguration(RequestDiagnosticContext.create());
 103
 104         // when
 105         final Mono<JsonObject> result = sut.flatMap(cbsClient -> cbsClient.get(request));
 106
 107         // then
 108         StepVerifier.create(result.map(this::sampleConfigValue))
 109                 .expectNext(EXPECTED_CONFIG_VALUE)
 110                 .expectComplete()
 111                 .verify(Duration.ofSeconds(5));
 112     }
 113
 114     @Test
 115     void testCbsClientWithPeriodicCall() {
 116         // given
 117         final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleEnvironment);
 118         final CbsRequest request = CbsRequests.getConfiguration(RequestDiagnosticContext.create());
 119
 120         // when
 121         final Flux<JsonObject> result = sut
 122                 .flatMapMany(cbsClient -> cbsClient.get(request, Duration.ZERO, Duration.ofMillis(10)));
 123
 124         // then
 125         final int itemsToTake = 5;
 126         StepVerifier.create(result.take(itemsToTake).map(this::sampleConfigValue))
 127                 .expectNextSequence(Stream.of(EXPECTED_CONFIG_VALUE).cycle(itemsToTake))
 128                 .expectComplete()
 129                 .verify(Duration.ofSeconds(5));
 130     }
 131
 132     @Test
 133     void testCbsClientWithUpdatesCall() {
 134         // given
 135         final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleEnvironment);
 136         final CbsRequest request = CbsRequests.getConfiguration(RequestDiagnosticContext.create());
 137         final Duration period = Duration.ofMillis(10);
 138
 139         // when
 140         final Flux<JsonObject> result = sut
 141                 .flatMapMany(cbsClient -> cbsClient.updates(request, Duration.ZERO, period));
 142
 143         // then
 144         final Duration timeToCollectItemsFor = period.multipliedBy(50);
 145         StepVerifier.create(result.take(timeToCollectItemsFor).map(this::sampleConfigValue))
 146                 .expectNext(EXPECTED_CONFIG_VALUE)
 147                 .expectComplete()
 148                 .verify(Duration.ofSeconds(5));
 149     }
 150
 151     @Test
 152     void testCbsClientWithStreamsParsing() {
 153         // given
 154         final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleEnvironment);
 155         final StreamFromGsonParser<KafkaSink> kafkaSinkParser = StreamFromGsonParsers.kafkaSinkParser();
 156         final CbsRequest request = CbsRequests.getConfiguration(RequestDiagnosticContext.create());
 157
 158         // when
 159         final Mono<KafkaSink> result = sut.flatMap(cbsClient -> cbsClient.get(request))
 160                 .map(json ->
 161                         DataStreams.namedSinks(json).map(kafkaSinkParser::unsafeParse).head()
 162                 );
 163
 164         // then
 165         StepVerifier.create(result)
 166                 .consumeNextWith(kafkaSink -> {
 167                     assertThat(kafkaSink.name()).isEqualTo("perf3gpp");
 168                     assertThat(kafkaSink.bootstrapServers()).isEqualTo("dmaap-mr-kafka:6060");
 169                     assertThat(kafkaSink.topicName()).isEqualTo("HVVES_PERF3GPP");
 170                 })
 171                 .expectComplete()
 172                 .verify(Duration.ofSeconds(5));
 173     }
 174
 175     @Test
 176     void testCbsClientWithStreamsParsingUsingSwitch() {
 177         // given
 178         final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleEnvironment);
 179         final CbsRequest request = CbsRequests.getConfiguration(RequestDiagnosticContext.create());
 180         // TODO: Use these parsers below
 181         final StreamFromGsonParser<KafkaSink> kafkaSinkParser = StreamFromGsonParsers.kafkaSinkParser();
 182         final StreamFromGsonParser<MessageRouterSink> mrSinkParser = StreamFromGsonParsers.messageRouterSinkParser();
 183
 184         // when
 185         final Mono<Void> result = sut.flatMap(cbsClient -> cbsClient.get(request))
 186                 .map(json -> {
 187                     final Stream<RawDataStream<JsonObject>> sinks = DataStreams.namedSinks(json);
 188
 189                     final Stream<KafkaSink> allKafkaSinks = sinks.filter(streamOfType(KAFKA))
 190                             .map(kafkaSinkParser::unsafeParse);
 191                     final Stream<MessageRouterSink> allMrSinks = sinks.filter(streamOfType(MESSAGE_ROUTER))
 192                             .map(mrSinkParser::unsafeParse);
 193
 194                     assertThat(allKafkaSinks.size())
 195                             .describedAs("Number of kafka sinks")
 196                             .isEqualTo(2);
 197                     assertThat(allMrSinks.size())
 198                             .describedAs("Number of DMAAP-MR sinks")
 199                             .isEqualTo(1);
 200
 201                     return true;
 202                 })
 203                 .then();
 204
 205         // then
 206         StepVerifier.create(result)
 207                 .expectComplete()
 208                 .verify(Duration.ofSeconds(5));
 209     }
 210
 211     @Test
 212     void testCbsClientWithStreamsParsingWhenUsingInvalidParser() {
 213         // given
 214         final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleEnvironment);
 215         final StreamFromGsonParser<KafkaSource> kafkaSourceParser = StreamFromGsonParsers.kafkaSourceParser();
 216         final CbsRequest request = CbsRequests.getConfiguration(RequestDiagnosticContext.create());
 217
 218         // when
 219         final Mono<KafkaSource> result = sut.flatMap(cbsClient -> cbsClient.get(request))
 220                 .map(json ->
 221                         DataStreams.namedSources(json).map(kafkaSourceParser::unsafeParse).head()
 222                 );
 223
 224         // then
 225         StepVerifier.create(result)
 226                 .expectErrorSatisfies(ex -> {
 227                     assertThat(ex).isInstanceOf(StreamParsingException.class);
 228                     assertThat(ex).hasMessageContaining("Invalid stream type");
 229                     assertThat(ex).hasMessageContaining(MESSAGE_ROUTER.toString());
 230                     assertThat(ex).hasMessageContaining(KAFKA.toString());
 231                 })
 232                 .verify(Duration.ofSeconds(5));
 233     }
 234
 235     @Test
 236     void testCbsClientWithSingleAllRequest() {
 237         // given
 238         final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleEnvironment);
 239         final CbsRequest request = CbsRequests.getAll(RequestDiagnosticContext.create());
 240
 241         // when
 242         final Mono<JsonObject> result = sut.flatMap(cbsClient -> cbsClient.get(request));
 243
 244         // then
 245         StepVerifier.create(result)
 246                 .assertNext(json -> {
 247                     assertThat(json.get("config")).isNotNull();
 248                     assertThat(json.get("policies")).isNotNull();
 249                     assertThat(json.get("sampleKey")).isNotNull();
 250                 })
 251                 .expectComplete()
 252                 .verify(Duration.ofSeconds(5));
 253     }
 254
 255
 256     @Test
 257     void testCbsClientWithSingleKeyRequest() {
 258         // given
 259         final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleEnvironment);
 260         final CbsRequest request = CbsRequests.getByKey(RequestDiagnosticContext.create(), "sampleKey");
 261
 262         // when
 263         final Mono<JsonObject> result = sut.flatMap(cbsClient -> cbsClient.get(request));
 264
 265         // then
 266         StepVerifier.create(result)
 267                 .assertNext(json -> {
 268                     assertThat(json.get("key")).isNotNull();
 269                     assertThat(json.get("key").getAsString()).isEqualTo("value");
 270                 })
 271                 .expectComplete()
 272                 .verify(Duration.ofSeconds(5));
 273     }
 274
 275     private String sampleConfigValue(JsonObject obj) {
 276         return obj.get(SAMPLE_CONFIG_KEY).getAsString();
 277     }
 278
 279     private static Mono<String> lazyConsulResponse() {
 280         return Mono.just(CONSUL_RESPONSE)
 281                 .map(CbsClientImplIT::processConsulResponseTemplate);
 282     }
 283
 284     private static String processConsulResponseTemplate(String resp) {
 285         return resp.replaceAll("HOST", server.host())
 286                 .replaceAll("PORT", Integer.toString(server.port()));
 287     }
 288 }