Implement simple health check mechanism

Change-Id: Ic4b8b59ced9dc19c9ebf26131036a9e1a752164f
Issue-ID: DCAEGEN2-659
Signed-off-by: Jakub Dudycz <jakub.dudycz@nokia.com>

1 files changed, 72 insertions, 41 deletions

diff --git a/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ConsulConfigurationProviderTest.kt b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ConsulConfigurationProviderTest.kt
index 1626c028..f9a9ba60 100644
--- a/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ConsulConfigurationProviderTest.kt
+++ b/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ConsulConfigurationProviderTest.kt
@@ -23,9 +23,13 @@ import com.nhaarman.mockito_kotlin.eq
 import com.nhaarman.mockito_kotlin.mock
 import com.nhaarman.mockito_kotlin.whenever
 import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.describe
 import org.jetbrains.spek.api.dsl.given
 import org.jetbrains.spek.api.dsl.it
+import org.jetbrains.spek.api.dsl.on
 import org.mockito.Mockito
+import org.onap.dcae.collectors.veshv.healthcheck.api.HealthState
+import org.onap.dcae.collectors.veshv.healthcheck.api.HealthStateProvider
 import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain
 import reactor.core.publisher.Mono
 import reactor.retry.Retry
@@ -40,62 +44,89 @@ import kotlin.test.assertEquals
  */
 internal object ConsulConfigurationProviderTest : Spek({
 
-    val httpAdapterMock: HttpAdapter = mock()
-    val firstRequestDelay = Duration.ofMillis(1)
-    val requestInterval = Duration.ofMillis(1)
-    val retry = Retry.onlyIf<Any> { it.iteration() < 2 }.fixedBackoff(Duration.ofNanos(1))
+    describe("Consul configuration provider") {
 
-    given("valid resource url") {
+        val httpAdapterMock: HttpAdapter = mock()
+        val healthStateProvider = HealthStateProvider.INSTANCE
 
-        val validUrl = "http://valid-url/"
-        val consulConfigProvider = ConsulConfigurationProvider(
-                httpAdapterMock,
-                validUrl,
-                firstRequestDelay,
-                requestInterval,
-                retry)
+        given("valid resource url") {
+            val validUrl = "http://valid-url/"
+            val consulConfigProvider = constructConsulConfigProvider(validUrl, httpAdapterMock, healthStateProvider)
 
-        whenever(httpAdapterMock.get(eq(validUrl), Mockito.anyMap()))
-                .thenReturn(Mono.just(constructConsulResponse()))
+            on("call to consul") {
+                whenever(httpAdapterMock.get(eq(validUrl), Mockito.anyMap()))
+                        .thenReturn(Mono.just(constructConsulResponse()))
 
-        it("should use received configuration") {
+                it("should use received configuration") {
 
-            StepVerifier.create(consulConfigProvider().take(1))
-                    .consumeNextWith {
+                    StepVerifier.create(consulConfigProvider().take(1))
+                            .consumeNextWith {
 
-                        assertEquals("kafka:9093", it.kafkaBootstrapServers)
+                                assertEquals("kafka:9093", it.kafkaBootstrapServers)
 
-                        val route1 = it.routing.routes[0]
-                        assertEquals(Domain.FAULT, route1.domain)
-                        assertEquals("test-topic-1", route1.targetTopic)
+                                val route1 = it.routing.routes[0]
+                                assertEquals(Domain.FAULT, route1.domain)
+                                assertEquals("test-topic-1", route1.targetTopic)
 
-                        val route2 = it.routing.routes[1]
-                        assertEquals(Domain.HEARTBEAT, route2.domain)
-                        assertEquals("test-topic-2", route2.targetTopic)
+                                val route2 = it.routing.routes[1]
+                                assertEquals(Domain.HEARTBEAT, route2.domain)
+                                assertEquals("test-topic-2", route2.targetTopic)
+
+                            }.verifyComplete()
+                }
+            }
 
-                    }.verifyComplete()
+        }
+        given("invalid resource url") {
+            val invalidUrl = "http://invalid-url/"
+
+            val iterationCount = 3L
+            val consulConfigProvider = constructConsulConfigProvider(
+                    invalidUrl, httpAdapterMock, healthStateProvider, iterationCount
+            )
+
+            on("call to consul") {
+                whenever(httpAdapterMock.get(eq(invalidUrl), Mockito.anyMap()))
+                        .thenReturn(Mono.error(RuntimeException("Test exception")))
+
+                it("should interrupt the flux") {
+
+                    StepVerifier.create(consulConfigProvider())
+                            .verifyErrorMessage("Test exception")
+                }
+
+                it("should update the health state"){
+                    StepVerifier.create(healthStateProvider().take(iterationCount))
+                            .expectNextCount(iterationCount - 1)
+                            .expectNext(HealthState.WAITING_FOR_CONSUL_CONFIGURATION)
+                            .verifyComplete()
+                }
+            }
         }
     }
-    given("invalid resource url") {
 
-        val invalidUrl = "http://invalid-url/"
-        val consulConfigProvider = ConsulConfigurationProvider(
-                httpAdapterMock,
-                invalidUrl,
-                firstRequestDelay,
-                requestInterval,
-                retry)
+})
 
-        whenever(httpAdapterMock.get(eq(invalidUrl), Mockito.anyMap()))
-                .thenReturn(Mono.error(RuntimeException("Test exception")))
+private fun constructConsulConfigProvider(url: String,
+                                          httpAdapter: HttpAdapter,
+                                          healthStateProvider: HealthStateProvider,
+                                          iterationCount: Long = 1
+): ConsulConfigurationProvider {
 
-        it("should interrupt the flux") {
+    val firstRequestDelay = Duration.ofMillis(1)
+    val requestInterval = Duration.ofMillis(1)
+    val retry = Retry.onlyIf<Any> { it.iteration() <= iterationCount }.fixedBackoff(Duration.ofNanos(1))
+
+    return ConsulConfigurationProvider(
+            httpAdapter,
+            url,
+            firstRequestDelay,
+            requestInterval,
+            healthStateProvider,
+            retry
+    )
+}
 
-            StepVerifier.create(consulConfigProvider())
-                    .verifyErrorMessage("Test exception")
-        }
-    }
-})
 
 fun constructConsulResponse(): String {