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
11 * http://www.apache.org/licenses/LICENSE-2.0
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=========================================================
20 package org.onap.dcae.collectors.veshv.impl.adapters
22 import com.nhaarman.mockitokotlin2.any
23 import com.nhaarman.mockitokotlin2.eq
24 import com.nhaarman.mockitokotlin2.mock
25 import com.nhaarman.mockitokotlin2.whenever
26 import org.assertj.core.api.Assertions.assertThat
27 import org.jetbrains.spek.api.Spek
28 import org.jetbrains.spek.api.dsl.describe
29 import org.jetbrains.spek.api.dsl.given
30 import org.jetbrains.spek.api.dsl.it
31 import org.jetbrains.spek.api.dsl.on
32 import org.mockito.Mockito
33 import org.onap.dcae.collectors.veshv.domain.VesEventDomain.HEARTBEAT
34 import org.onap.dcae.collectors.veshv.domain.VesEventDomain.FAULT
35 import org.onap.dcae.collectors.veshv.healthcheck.api.HealthDescription
36 import org.onap.dcae.collectors.veshv.healthcheck.api.HealthState
38 import reactor.core.publisher.Mono
39 import reactor.retry.Retry
40 import reactor.test.StepVerifier
41 import java.time.Duration
43 import kotlin.test.assertEquals
46 * @author Jakub Dudycz <jakub.dudycz@nokia.com>
49 internal object ConsulConfigurationProviderTest : Spek({
51 describe("Consul configuration provider") {
53 val httpAdapterMock: HttpAdapter = mock()
54 val healthStateProvider = HealthState.INSTANCE
56 given("valid resource url") {
57 val validUrl = "http://valid-url/"
58 val consulConfigProvider = constructConsulConfigProvider(validUrl, httpAdapterMock, healthStateProvider)
60 on("call to consul") {
61 whenever(httpAdapterMock.get(eq(validUrl), any(), Mockito.anyMap()))
62 .thenReturn(Mono.just(constructConsulResponse()))
64 it("should use received configuration") {
66 StepVerifier.create(consulConfigProvider().take(1))
69 val route1 = it.routing.routes[0]
70 assertThat(FAULT.domainName)
71 .describedAs("routed domain 1")
72 .isEqualTo(route1.domain)
73 assertThat("test-topic-1")
74 .describedAs("target topic 1")
75 .isEqualTo(route1.targetTopic)
77 val route2 = it.routing.routes[1]
78 assertThat(HEARTBEAT.domainName)
79 .describedAs("routed domain 2")
80 .isEqualTo(route2.domain)
81 assertThat("test-topic-2")
82 .describedAs("target topic 2")
83 .isEqualTo(route2.targetTopic)
90 given("invalid resource url") {
91 val invalidUrl = "http://invalid-url/"
93 val iterationCount = 3L
94 val consulConfigProvider = constructConsulConfigProvider(
95 invalidUrl, httpAdapterMock, healthStateProvider, iterationCount
98 on("call to consul") {
99 whenever(httpAdapterMock.get(eq(invalidUrl), any(), Mockito.anyMap()))
100 .thenReturn(Mono.error(RuntimeException("Test exception")))
102 it("should interrupt the flux") {
104 StepVerifier.create(consulConfigProvider())
105 .verifyErrorMessage("Test exception")
108 it("should update the health state") {
109 StepVerifier.create(healthStateProvider().take(iterationCount))
110 .expectNextCount(iterationCount - 1)
111 .expectNext(HealthDescription.RETRYING_FOR_CONSUL_CONFIGURATION)
120 private fun constructConsulConfigProvider(url: String,
121 httpAdapter: HttpAdapter,
122 healthState: HealthState,
123 iterationCount: Long = 1
124 ): ConsulConfigurationProvider {
126 val firstRequestDelay = Duration.ofMillis(1)
127 val requestInterval = Duration.ofMillis(1)
128 val retry = Retry.onlyIf<Any> { it.iteration() <= iterationCount }.fixedBackoff(Duration.ofNanos(1))
130 return ConsulConfigurationProvider(
140 fun constructConsulResponse(): String =
142 "whatever": "garbage",
143 "collector.routing": [
145 "fromDomain": "fault",
146 "toTopic": "test-topic-1"
149 "fromDomain": "heartbeat",
150 "toTopic": "test-topic-2"