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

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * dcaegen2-collectors-veshv
   4  * ================================================================================
   5  * Copyright (C) 2018 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.impl.adapters
  21
  22 import com.nhaarman.mockito_kotlin.eq
  23 import com.nhaarman.mockito_kotlin.mock
  24 import com.nhaarman.mockito_kotlin.whenever
  25 import org.jetbrains.spek.api.Spek
  26 import org.jetbrains.spek.api.dsl.describe
  27 import org.jetbrains.spek.api.dsl.given
  28 import org.jetbrains.spek.api.dsl.it
  29 import org.jetbrains.spek.api.dsl.on
  30 import org.mockito.Mockito
  31 import org.onap.dcae.collectors.veshv.healthcheck.api.HealthDescription
  32 import org.onap.dcae.collectors.veshv.healthcheck.api.HealthState
  33 import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain
  34 import reactor.core.publisher.Mono
  35 import reactor.retry.Retry
  36 import reactor.test.StepVerifier
  37 import java.time.Duration
  38 import java.util.*
  39 import kotlin.test.assertEquals
  40
  41 /**
  42  * @author Jakub Dudycz <jakub.dudycz@nokia.com>
  43  * @since May 2018
  44  */
  45 internal object ConsulConfigurationProviderTest : Spek({
  46
  47     describe("Consul configuration provider") {
  48
  49         val httpAdapterMock: HttpAdapter = mock()
  50         val healthStateProvider = HealthState.INSTANCE
  51
  52         given("valid resource url") {
  53             val validUrl = "http://valid-url/"
  54             val consulConfigProvider = constructConsulConfigProvider(validUrl, httpAdapterMock, healthStateProvider)
  55
  56             on("call to consul") {
  57                 whenever(httpAdapterMock.get(eq(validUrl), Mockito.anyMap()))
  58                         .thenReturn(Mono.just(constructConsulResponse()))
  59
  60                 it("should use received configuration") {
  61
  62                     StepVerifier.create(consulConfigProvider().take(1))
  63                             .consumeNextWith {
  64
  65                                 assertEquals("kafka:9093", it.kafkaBootstrapServers)
  66
  67                                 val route1 = it.routing.routes[0]
  68                                 assertEquals(Domain.FAULT, route1.domain)
  69                                 assertEquals("test-topic-1", route1.targetTopic)
  70
  71                                 val route2 = it.routing.routes[1]
  72                                 assertEquals(Domain.HEARTBEAT, route2.domain)
  73                                 assertEquals("test-topic-2", route2.targetTopic)
  74
  75                             }.verifyComplete()
  76                 }
  77             }
  78
  79         }
  80         given("invalid resource url") {
  81             val invalidUrl = "http://invalid-url/"
  82
  83             val iterationCount = 3L
  84             val consulConfigProvider = constructConsulConfigProvider(
  85                     invalidUrl, httpAdapterMock, healthStateProvider, iterationCount
  86             )
  87
  88             on("call to consul") {
  89                 whenever(httpAdapterMock.get(eq(invalidUrl), Mockito.anyMap()))
  90                         .thenReturn(Mono.error(RuntimeException("Test exception")))
  91
  92                 it("should interrupt the flux") {
  93
  94                     StepVerifier.create(consulConfigProvider())
  95                             .verifyErrorMessage("Test exception")
  96                 }
  97
  98                 it("should update the health state"){
  99                     StepVerifier.create(healthStateProvider().take(iterationCount))
 100                             .expectNextCount(iterationCount - 1)
 101                             .expectNext(HealthDescription.RETRYING_FOR_CONSUL_CONFIGURATION)
 102                             .verifyComplete()
 103                 }
 104             }
 105         }
 106     }
 107
 108 })
 109
 110 private fun constructConsulConfigProvider(url: String,
 111                                           httpAdapter: HttpAdapter,
 112                                           healthState: HealthState,
 113                                           iterationCount: Long = 1
 114 ): ConsulConfigurationProvider {
 115
 116     val firstRequestDelay = Duration.ofMillis(1)
 117     val requestInterval = Duration.ofMillis(1)
 118     val retry = Retry.onlyIf<Any> { it.iteration() <= iterationCount }.fixedBackoff(Duration.ofNanos(1))
 119
 120     return ConsulConfigurationProvider(
 121             httpAdapter,
 122             url,
 123             firstRequestDelay,
 124             requestInterval,
 125             healthState,
 126             retry
 127     )
 128 }
 129
 130
 131 fun constructConsulResponse(): String {
 132
 133     val config = """{
 134     "kafkaBootstrapServers": "kafka:9093",
 135     "routing": [
 136             {
 137                 "fromDomain": 1,
 138                 "toTopic": "test-topic-1"
 139             },
 140             {
 141                 "fromDomain": 2,
 142                 "toTopic": "test-topic-2"
 143             }
 144     ]
 145 }"""
 146
 147     val encodedValue = String(Base64.getEncoder().encode(config.toByteArray()))
 148
 149     return """[
 150         {
 151             "CreateIndex": 100,
 152             "ModifyIndex": 200,
 153             "LockIndex": 200,
 154             "Key": "zip",
 155             "Flags": 0,
 156             "Value": "$encodedValue",
 157             "Session": "adf4238a-882b-9ddc-4a9d-5b6758e4159e"
 158         }
 159     ]"""
 160 }