gerrit.onap Code Review - dcaegen2/collectors/hv-ves.git/blob

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * dcaegen2-collectors-veshv
   4  * ================================================================================
   5  * Copyright (C) 2018-2019 NOKIA
   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 package org.onap.dcae.collectors.veshv.config.impl
  21
  22 import arrow.core.Some
  23 import com.google.gson.JsonParser
  24 import com.nhaarman.mockitokotlin2.any
  25 import com.nhaarman.mockitokotlin2.eq
  26 import com.nhaarman.mockitokotlin2.mock
  27 import com.nhaarman.mockitokotlin2.times
  28 import com.nhaarman.mockitokotlin2.verify
  29 import com.nhaarman.mockitokotlin2.whenever
  30 import org.assertj.core.api.Assertions.assertThat
  31 import org.jetbrains.spek.api.Spek
  32 import org.jetbrains.spek.api.dsl.describe
  33 import org.jetbrains.spek.api.dsl.given
  34 import org.jetbrains.spek.api.dsl.it
  35 import org.jetbrains.spek.api.dsl.on
  36 import org.onap.dcae.collectors.veshv.config.api.ConfigurationStateListener
  37 import org.onap.dcae.collectors.veshv.config.api.model.CbsConfiguration
  38 import org.onap.dcaegen2.services.sdk.model.streams.ImmutableAafCredentials
  39 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient
  40 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParsers
  41 import reactor.core.publisher.Flux
  42 import reactor.core.publisher.Mono
  43 import reactor.retry.Retry
  44 import reactor.test.StepVerifier
  45 import java.time.Duration
  46
  47 /**
  48  * @author Jakub Dudycz <jakub.dudycz@nokia.com>
  49  * @since May 2018
  50  */
  51 internal object CbsConfigurationProviderTest : Spek({
  52
  53     describe("Configuration provider") {
  54
  55         val cbsClient = mock<CbsClient>()
  56         val cbsClientMock = Mono.just(cbsClient)
  57         val configStateListener = mock<ConfigurationStateListener>()
  58
  59         given("configuration is never in cbs") {
  60             val configProvider = constructConfigurationProvider(cbsClientMock, configStateListener)
  61
  62             on("waiting for configuration") {
  63                 val waitTime = Duration.ofMillis(100)
  64
  65                 it("should not get it") {
  66                     StepVerifier.create(configProvider().take(1))
  67                             .expectNoEvent(waitTime)
  68                 }
  69             }
  70         }
  71
  72         given("valid configuration from cbs") {
  73             val configProvider = constructConfigurationProvider(cbsClientMock, configStateListener)
  74
  75             on("new configuration") {
  76                 whenever(cbsClient.updates(any(), eq(firstRequestDelay), eq(requestInterval)))
  77                         .thenReturn(Flux.just(validConfiguration))
  78                 it("should use received configuration") {
  79
  80                     StepVerifier.create(configProvider().take(1))
  81                             .consumeNextWith {
  82
  83                                 assertThat(it.listenPort).isEqualTo(Some(6061))
  84                                 assertThat(it.idleTimeoutSec).isEqualTo(Some(60L))
  85
  86                                 val sinks = it.streamPublishers.orNull()!!
  87                                 val sink1 = sinks[0]
  88                                 val sink2 = sinks[1]
  89
  90                                 assertThat(sink1.name()).isEqualTo(PERF3GPP_REGIONAL)
  91                                 assertThat(sink1.aafCredentials()).isEqualTo(aafCredentials1)
  92                                 assertThat(sink1.bootstrapServers())
  93                                         .isEqualTo("dmaap-mr-kafka-0.regional:6060,dmaap-mr-kafka-1.regional:6060")
  94                                 assertThat(sink1.topicName()).isEqualTo("REG_HVVES_PERF3GPP")
  95
  96                                 assertThat(sink2.name()).isEqualTo(PERF3GPP_CENTRAL)
  97                                 assertThat(sink2.aafCredentials()).isEqualTo(aafCredentials2)
  98                                 assertThat(sink2.bootstrapServers())
  99                                         .isEqualTo("dmaap-mr-kafka-0.central:6060,dmaap-mr-kafka-1.central:6060")
 100                                 assertThat(sink2.topicName()).isEqualTo("CEN_HVVES_PERF3GPP")
 101                             }.verifyComplete()
 102                 }
 103             }
 104         }
 105
 106         given("invalid configuration from cbs") {
 107             val iterationCount = 3L
 108             val configProvider = constructConfigurationProvider(
 109                     cbsClientMock, configStateListener, iterationCount
 110             )
 111
 112             on("new configuration") {
 113                 whenever(cbsClient.updates(any(), eq(firstRequestDelay), eq(requestInterval)))
 114                         .thenReturn(Flux.just(invalidConfiguration))
 115
 116                 it("should interrupt the flux") {
 117                     StepVerifier
 118                             .create(configProvider())
 119                             .verifyError()
 120                 }
 121
 122                 it("should call state listener when retrying") {
 123                     verify(configStateListener, times(iterationCount.toInt())).retrying()
 124                 }
 125             }
 126         }
 127     }
 128
 129 })
 130
 131
 132 private const val PERF3GPP_REGIONAL = "perf3gpp_regional"
 133 private const val PERF3GPP_CENTRAL = "perf3gpp_central"
 134
 135 private val aafCredentials1 = ImmutableAafCredentials.builder()
 136         .username("client")
 137         .password("very secure password")
 138         .build()
 139
 140 private val aafCredentials2 = ImmutableAafCredentials.builder()
 141         .username("other_client")
 142         .password("another very secure password")
 143         .build()
 144
 145 private val validConfiguration = JsonParser().parse("""
 146 {
 147     "server.listenPort": 6061,
 148     "server.idleTimeoutSec": 60,
 149     "streams_publishes": {
 150         "$PERF3GPP_REGIONAL": {
 151             "type": "kafka",
 152             "aaf_credentials": {
 153                 "username": "client",
 154                 "password": "very secure password"
 155             },
 156             "kafka_info": {
 157                 "bootstrap_servers": "dmaap-mr-kafka-0.regional:6060,dmaap-mr-kafka-1.regional:6060",
 158                 "topic_name": "REG_HVVES_PERF3GPP"
 159             }
 160         },
 161         "$PERF3GPP_CENTRAL": {
 162             "type": "kafka",
 163             "aaf_credentials": {
 164                 "username": "other_client",
 165                 "password": "another very secure password"
 166             },
 167             "kafka_info": {
 168                 "bootstrap_servers": "dmaap-mr-kafka-0.central:6060,dmaap-mr-kafka-1.central:6060",
 169                 "topic_name": "CEN_HVVES_PERF3GPP"
 170             }
 171         }
 172     }
 173 }""").asJsonObject
 174
 175 private val invalidConfiguration = JsonParser().parse("""
 176 {
 177     "streams_publishes": {
 178         "$PERF3GPP_REGIONAL": {
 179             "type": "kafka",
 180             "aaf_credentials": {
 181                 "user": "client",
 182                 "password": "very secure password"
 183             },
 184             "kafka_info": {
 185                 "servers": "dmaap-mr-kafka-0.regional:6060,dmaap-mr-kafka-1.regional:6060",
 186                 "name": "REG_HVVES_PERF3GPP"
 187             }
 188         }
 189     }
 190 }""").asJsonObject
 191
 192 private val firstRequestDelay = Duration.ofMillis(1)
 193 private val requestInterval = Duration.ofMillis(1)
 194 private val streamParser = StreamFromGsonParsers.kafkaSinkParser()
 195 private val configParser = JsonConfigurationParser()
 196
 197 private fun constructConfigurationProvider(cbsClientMono: Mono<CbsClient>,
 198                                            configurationStateListener: ConfigurationStateListener,
 199                                            iterationCount: Long = 1
 200 ): CbsConfigurationProvider {
 201
 202     val retry = Retry
 203             .onlyIf<Any> { it.iteration() <= iterationCount }
 204             .fixedBackoff(Duration.ofNanos(1))
 205
 206     return CbsConfigurationProvider(
 207             cbsClientMono,
 208             CbsConfiguration(firstRequestDelay, requestInterval),
 209             configParser,
 210             streamParser,
 211             configurationStateListener,
 212             { mapOf("k" to "v") },
 213             retry
 214     )
 215 }