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