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
44 * @author Jakub Dudycz <jakub.dudycz@nokia.com>
47 internal object ConsulConfigurationProviderTest : Spek({
49 describe("Consul configuration provider") {
51 val httpAdapterMock: HttpAdapter = mock()
52 val healthStateProvider = HealthState.INSTANCE
54 given("valid resource url") {
55 val validUrl = "http://valid-url/"
56 val consulConfigProvider = constructConsulConfigProvider(validUrl, httpAdapterMock, healthStateProvider)
58 on("call to consul") {
59 whenever(httpAdapterMock.get(eq(validUrl), any(), Mockito.anyMap()))
60 .thenReturn(Mono.just(constructConsulResponse()))
62 it("should use received configuration") {
64 StepVerifier.create(consulConfigProvider().take(1))
67 val route1 = it.routing.routes[0]
68 assertThat(FAULT.domainName)
69 .describedAs("routed domain 1")
70 .isEqualTo(route1.domain)
71 assertThat("test-topic-1")
72 .describedAs("target topic 1")
73 .isEqualTo(route1.targetTopic)
75 val route2 = it.routing.routes[1]
76 assertThat(HEARTBEAT.domainName)
77 .describedAs("routed domain 2")
78 .isEqualTo(route2.domain)
79 assertThat("test-topic-2")
80 .describedAs("target topic 2")
81 .isEqualTo(route2.targetTopic)
88 given("invalid resource url") {
89 val invalidUrl = "http://invalid-url/"
91 val iterationCount = 3L
92 val consulConfigProvider = constructConsulConfigProvider(
93 invalidUrl, httpAdapterMock, healthStateProvider, iterationCount
96 on("call to consul") {
97 whenever(httpAdapterMock.get(eq(invalidUrl), any(), Mockito.anyMap()))
98 .thenReturn(Mono.error(RuntimeException("Test exception")))
100 it("should interrupt the flux") {
102 StepVerifier.create(consulConfigProvider())
103 .verifyErrorMessage("Test exception")
106 it("should update the health state") {
107 StepVerifier.create(healthStateProvider().take(iterationCount))
108 .expectNextCount(iterationCount - 1)
109 .expectNext(HealthDescription.RETRYING_FOR_DYNAMIC_CONFIGURATION)
118 private fun constructConsulConfigProvider(url: String,
119 httpAdapter: HttpAdapter,
120 healthState: HealthState,
121 iterationCount: Long = 1
122 ): ConsulConfigurationProvider {
124 val firstRequestDelay = Duration.ofMillis(1)
125 val requestInterval = Duration.ofMillis(1)
126 val retry = Retry.onlyIf<Any> { it.iteration() <= iterationCount }.fixedBackoff(Duration.ofNanos(1))
128 return ConsulConfigurationProvider(
138 fun constructConsulResponse(): String =
140 "whatever": "garbage",
141 "collector.routing": [
143 "fromDomain": "fault",
144 "toTopic": "test-topic-1"
147 "fromDomain": "heartbeat",
148 "toTopic": "test-topic-2"