Merge "Parse whole dynamic configuration"

4 files changed, 121 insertions, 95 deletions

diff --git a/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/CbsConfigurationProviderTest.kt b/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/CbsConfigurationProviderTest.kt
index d5fe588e..94eb519d 100644
--- a/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/CbsConfigurationProviderTest.kt
+++ b/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/CbsConfigurationProviderTest.kt
@@ -19,6 +19,7 @@
  */
 package org.onap.dcae.collectors.veshv.config.impl
 
+import arrow.core.Some
 import com.google.gson.JsonParser
 import com.nhaarman.mockitokotlin2.any
 import com.nhaarman.mockitokotlin2.eq
@@ -51,9 +52,9 @@ internal object CbsConfigurationProviderTest : Spek({
 
     describe("Configuration provider") {
 
-        val cbsClient: CbsClient = mock()
-        val cbsClientMock: Mono<CbsClient> = Mono.just(cbsClient)
-        val configStateListener: ConfigurationStateListener = mock()
+        val cbsClient = mock<CbsClient>()
+        val cbsClientMock = Mono.just(cbsClient)
+        val configStateListener = mock<ConfigurationStateListener>()
 
         given("configuration is never in cbs") {
             val configProvider = constructConfigurationProvider(cbsClientMock, configStateListener)
@@ -78,29 +79,32 @@ internal object CbsConfigurationProviderTest : Spek({
 
                     StepVerifier.create(configProvider().take(1))
                             .consumeNextWith {
-                                val routes = it.routing.orNull()!!
-                                val route1 = routes.elementAt(0)
-                                val route2 = routes.elementAt(1)
-                                val receivedSink1 = route1.sink
-                                val receivedSink2 = route2.sink
-
-                                assertThat(route1.domain).isEqualTo(PERF3GPP_REGIONAL)
-                                assertThat(receivedSink1.aafCredentials()).isEqualTo(aafCredentials1)
-                                assertThat(receivedSink1.bootstrapServers())
+
+                                assertThat(it.listenPort).isEqualTo(Some(6061))
+                                assertThat(it.idleTimeoutSec).isEqualTo(Some(60L))
+                                assertThat(it.maxPayloadSizeBytes).isEqualTo(Some(1048576))
+
+
+                                val sinks = it.streamPublishers.orNull()!!
+                                val sink1 = sinks[0]
+                                val sink2 = sinks[1]
+
+                                assertThat(sink1.name()).isEqualTo(PERF3GPP_REGIONAL)
+                                assertThat(sink1.aafCredentials()).isEqualTo(aafCredentials1)
+                                assertThat(sink1.bootstrapServers())
                                         .isEqualTo("dmaap-mr-kafka-0.regional:6060,dmaap-mr-kafka-1.regional:6060")
-                                assertThat(receivedSink1.topicName()).isEqualTo("REG_HVVES_PERF3GPP")
+                                assertThat(sink1.topicName()).isEqualTo("REG_HVVES_PERF3GPP")
 
-                                assertThat(route2.domain).isEqualTo(PERF3GPP_CENTRAL)
-                                assertThat(receivedSink2.aafCredentials()).isEqualTo(aafCredentials2)
-                                assertThat(receivedSink2.bootstrapServers())
+                                assertThat(sink2.name()).isEqualTo(PERF3GPP_CENTRAL)
+                                assertThat(sink2.aafCredentials()).isEqualTo(aafCredentials2)
+                                assertThat(sink2.bootstrapServers())
                                         .isEqualTo("dmaap-mr-kafka-0.central:6060,dmaap-mr-kafka-1.central:6060")
-                                assertThat(receivedSink2.topicName()).isEqualTo("CEN_HVVES_PERF3GPP")
-
+                                assertThat(sink2.topicName()).isEqualTo("CEN_HVVES_PERF3GPP")
                             }.verifyComplete()
                 }
             }
-
         }
+
         given("invalid configuration from cbs") {
             val iterationCount = 3L
             val configProvider = constructConfigurationProvider(
@@ -112,7 +116,8 @@ internal object CbsConfigurationProviderTest : Spek({
                         .thenReturn(Flux.just(invalidConfiguration))
 
                 it("should interrupt the flux") {
-                    StepVerifier.create(configProvider())
+                    StepVerifier
+                            .create(configProvider())
                             .verifyError()
                 }
 
@@ -126,8 +131,8 @@ internal object CbsConfigurationProviderTest : Spek({
 })
 
 
-val PERF3GPP_REGIONAL = "perf3gpp_regional"
-val PERF3GPP_CENTRAL = "perf3gpp_central"
+private const val PERF3GPP_REGIONAL = "perf3gpp_regional"
+private const val PERF3GPP_CENTRAL = "perf3gpp_central"
 
 private val aafCredentials1 = ImmutableAafCredentials.builder()
         .username("client")
@@ -141,6 +146,9 @@ private val aafCredentials2 = ImmutableAafCredentials.builder()
 
 private val validConfiguration = JsonParser().parse("""
 {
+    "server.listenPort": 6061,
+    "server.idleTimeoutSec": 60,
+    "server.maxPayloadSizeBytes": 1048576,
     "streams_publishes": {
         "$PERF3GPP_REGIONAL": {
             "type": "kafka",
@@ -173,12 +181,12 @@ private val invalidConfiguration = JsonParser().parse("""
         "$PERF3GPP_REGIONAL": {
             "type": "kafka",
             "aaf_credentials": {
-                "username": "client",
+                "user": "client",
                 "password": "very secure password"
             },
             "kafka_info": {
-                "bootstrap_servers": "dmaap-mr-kafka-0.regional:6060,dmaap-mr-kafka-1.regional:6060",
-                "popic_name": "REG_HVVES_PERF3GPP"
+                "servers": "dmaap-mr-kafka-0.regional:6060,dmaap-mr-kafka-1.regional:6060",
+                "name": "REG_HVVES_PERF3GPP"
             }
         }
     }
@@ -187,20 +195,24 @@ private val invalidConfiguration = JsonParser().parse("""
 private val firstRequestDelay = Duration.ofMillis(1)
 private val requestInterval = Duration.ofMillis(1)
 private val streamParser = StreamFromGsonParsers.kafkaSinkParser()
+private val configParser = JsonConfigurationParser()
 
 private fun constructConfigurationProvider(cbsClientMono: Mono<CbsClient>,
                                            configurationStateListener: ConfigurationStateListener,
                                            iterationCount: Long = 1
 ): CbsConfigurationProvider {
 
-    val retry = Retry.onlyIf<Any> { it.iteration() <= iterationCount }.fixedBackoff(Duration.ofNanos(1))
+    val retry = Retry
+            .onlyIf<Any> { it.iteration() <= iterationCount }
+            .fixedBackoff(Duration.ofNanos(1))
 
     return CbsConfigurationProvider(
             cbsClientMono,
             CbsConfiguration(firstRequestDelay, requestInterval),
+            configParser,
             streamParser,
             configurationStateListener,
-            retry,
-            { mapOf("k" to "v") }
+            { mapOf("k" to "v") },
+            retry
     )
 }
diff --git a/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationMergerTest.kt b/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationMergerTest.kt
index bc61b57d..4cd2ba97 100644
--- a/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationMergerTest.kt
+++ b/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationMergerTest.kt
@@ -55,9 +55,9 @@ internal object ConfigurationMergerTest : Spek({
         }
 
         it("merges single parameter into full config") {
-            val actual = FileConfigurationReader().loadConfig(
+            val actual = JsonConfigurationParser().parse(
                     InputStreamReader(
-                            FileConfigurationReaderTest.javaClass.getResourceAsStream("/sampleConfig.json")) as Reader)
+                            JsonConfigurationParserTest.javaClass.getResourceAsStream("/sampleConfig.json")) as Reader)
             val diff = PartialConfiguration(logLevel = Some(LogLevel.INFO))
 
             val result = ConfigurationMerger().merge(actual, diff)
@@ -66,31 +66,31 @@ internal object ConfigurationMergerTest : Spek({
         }
 
         it("merges single embedded parameter into full config") {
-            val actual = FileConfigurationReader().loadConfig(
+            val actual = JsonConfigurationParser().parse(
                     InputStreamReader(
-                            FileConfigurationReaderTest.javaClass.getResourceAsStream("/sampleConfig.json")) as Reader)
+                            JsonConfigurationParserTest.javaClass.getResourceAsStream("/sampleConfig.json")) as Reader)
             val diff = PartialConfiguration(listenPort = someListenPort)
 
             val result = ConfigurationMerger().merge(actual, diff)
 
             assertThat(result.listenPort).isEqualTo(someListenPort)
             assertThat(result.idleTimeoutSec.isEmpty()).isFalse()
-            assertThat(result.idleTimeoutSec).isEqualTo(Some(Duration.ofSeconds(1200)))
+            assertThat(result.idleTimeoutSec).isEqualTo(Some(1200L))
             assertThat(result.maxPayloadSizeBytes.isEmpty()).isFalse()
             assertThat(result.maxPayloadSizeBytes).isEqualTo(Some(1048576))
         }
 
         it("merges full config into single parameter") {
             val actual = PartialConfiguration(logLevel = Some(LogLevel.INFO))
-            val diff = FileConfigurationReader().loadConfig(
+            val diff = JsonConfigurationParser().parse(
                     InputStreamReader(
-                            FileConfigurationReaderTest.javaClass.getResourceAsStream("/sampleConfig.json")) as Reader)
+                            JsonConfigurationParserTest.javaClass.getResourceAsStream("/sampleConfig.json")) as Reader)
 
             val result = ConfigurationMerger().merge(actual, diff)
 
             assertThat(result.logLevel).isEqualTo(Some(LogLevel.ERROR))
             assertThat(result.maxPayloadSizeBytes).isEqualTo(Some(1048576))
-            assertThat(result.idleTimeoutSec).isEqualTo(Some(Duration.ofSeconds(1200)))
+            assertThat(result.idleTimeoutSec).isEqualTo(Some(1200L))
 
             assertThat(result.keyStoreFile.isEmpty()).isFalse()
             assertThat(result.firstRequestDelaySec.isEmpty()).isFalse()
diff --git a/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationValidatorTest.kt b/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationValidatorTest.kt
index e43acfa3..5fa1fd62 100644
--- a/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationValidatorTest.kt
+++ b/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationValidatorTest.kt
@@ -23,14 +23,17 @@ import arrow.core.None
 import arrow.core.Option
 import arrow.core.Some
 import arrow.core.getOrElse
+import com.nhaarman.mockitokotlin2.mock
+import com.nhaarman.mockitokotlin2.whenever
 import org.assertj.core.api.Assertions.assertThat
 import org.assertj.core.api.Assertions.fail
 import org.jetbrains.spek.api.Spek
 import org.jetbrains.spek.api.dsl.describe
 import org.jetbrains.spek.api.dsl.it
-import org.onap.dcae.collectors.veshv.config.api.model.Routing
+import org.onap.dcae.collectors.veshv.config.api.model.Route
 import org.onap.dcae.collectors.veshv.config.impl.ConfigurationValidator.Companion.DEFAULT_LOG_LEVEL
 import org.onap.dcae.collectors.veshv.utils.logging.LogLevel
+import org.onap.dcaegen2.services.sdk.model.streams.dmaap.KafkaSink
 import org.onap.dcaegen2.services.sdk.security.ssl.SecurityKeys
 import java.io.File
 import java.time.Duration
@@ -81,7 +84,7 @@ internal object ConfigurationValidatorTest : Spek({
                     keyStorePassword = Some(KEYSTORE_PASSWORD),
                     trustStoreFile = Some(TRUSTSTORE),
                     trustStorePassword = Some(TRUSTSTORE_PASSWORD),
-                    routing = Some(emptyRouting),
+                    streamPublishers = Some(sampleStreamsDefinition),
                     logLevel = Some(LogLevel.TRACE)
             )
 
@@ -92,9 +95,12 @@ internal object ConfigurationValidatorTest : Spek({
                             fail("Configuration should have been created successfully")
                         },
                         {
-                            assertThat(it.server.listenPort).isEqualTo(defaultListenPort)
-                            assertThat(it.server.idleTimeout).isEqualTo(defaultIdleTimeoutSec)
-                            assertThat(it.server.maxPayloadSizeBytes).isEqualTo(defaultMaxPayloadSizeBytes)
+                            assertThat(it.server.listenPort)
+                                    .isEqualTo(defaultListenPort)
+                            assertThat(it.server.maxPayloadSizeBytes)
+                                    .isEqualTo(defaultMaxPayloadSizeBytes)
+                            assertThat(it.server.idleTimeout)
+                                    .isEqualTo(Duration.ofSeconds(defaultIdleTimeoutSec))
 
                             val securityKeys = it.security.keys
                                     .getOrElse { fail("Should be immutableSecurityKeys") } as SecurityKeys
@@ -103,10 +109,14 @@ internal object ConfigurationValidatorTest : Spek({
                             securityKeys.keyStorePassword().use { assertThat(it).isEqualTo(KEYSTORE_PASSWORD.toCharArray()) }
                             securityKeys.trustStorePassword().use { assertThat(it).isEqualTo(TRUSTSTORE_PASSWORD.toCharArray()) }
 
-                            assertThat(it.cbs.firstRequestDelay).isEqualTo(defaultFirstReqDelaySec)
-                            assertThat(it.cbs.requestInterval).isEqualTo(defaultRequestIntervalSec)
+                            assertThat(it.cbs.firstRequestDelay)
+                                    .isEqualTo(Duration.ofSeconds(defaultFirstReqDelaySec))
+                            assertThat(it.cbs.requestInterval)
+                                    .isEqualTo(Duration.ofSeconds(defaultRequestIntervalSec))
+
+                            assertThat(it.collector.routing)
+                                    .isEqualTo(sampleRouting)
 
-                            assertThat(it.collector.routing).isEqualTo(emptyRouting)
                             assertThat(it.logLevel).isEqualTo(LogLevel.TRACE)
                         }
                 )
@@ -130,16 +140,16 @@ internal object ConfigurationValidatorTest : Spek({
                         },
                         {
                             assertThat(it.server.idleTimeout)
-                                    .isEqualTo(defaultIdleTimeoutSec)
+                                    .isEqualTo(Duration.ofSeconds(defaultIdleTimeoutSec))
 
                             assertThat(it.security.keys)
                                     .isEqualTo(None)
 
                             assertThat(it.cbs.firstRequestDelay)
-                                    .isEqualTo(defaultFirstReqDelaySec)
+                                    .isEqualTo(Duration.ofSeconds(defaultFirstReqDelaySec))
 
                             assertThat(it.collector.routing)
-                                    .isEqualTo(emptyRouting)
+                                    .isEqualTo(sampleRouting)
                         }
                 )
             }
@@ -172,42 +182,47 @@ internal object ConfigurationValidatorTest : Spek({
 })
 
 private fun partialConfiguration(listenPort: Option<Int> = Some(defaultListenPort),
-                                 idleTimeoutSec: Option<Duration> = Some(defaultIdleTimeoutSec),
+                                 idleTimeoutSec: Option<Long> = Some(defaultIdleTimeoutSec),
                                  maxPayloadSizeBytes: Option<Int> = Some(defaultMaxPayloadSizeBytes),
-                                 firstReqDelaySec: Option<Duration> = Some(defaultFirstReqDelaySec),
-                                 requestIntervalSec: Option<Duration> = Some(defaultRequestIntervalSec),
+                                 firstReqDelaySec: Option<Long> = Some(defaultFirstReqDelaySec),
+                                 requestIntervalSec: Option<Long> = Some(defaultRequestIntervalSec),
                                  sslDisable: Option<Boolean> = Some(false),
                                  keyStoreFile: Option<String> = Some(KEYSTORE),
                                  keyStorePassword: Option<String> = Some(KEYSTORE_PASSWORD),
                                  trustStoreFile: Option<String> = Some(TRUSTSTORE),
                                  trustStorePassword: Option<String> = Some(TRUSTSTORE_PASSWORD),
-                                 routing: Option<Routing> = Some(emptyRouting),
+                                 streamPublishers: Option<List<KafkaSink>> = Some(sampleStreamsDefinition),
                                  logLevel: Option<LogLevel> = Some(LogLevel.INFO)
-) =
-        PartialConfiguration(
-                listenPort = listenPort,
-                idleTimeoutSec = idleTimeoutSec,
-                maxPayloadSizeBytes = maxPayloadSizeBytes,
-                firstRequestDelaySec = firstReqDelaySec,
-                requestIntervalSec = requestIntervalSec,
-                sslDisable = sslDisable,
-                keyStoreFile = keyStoreFile,
-                keyStorePassword = keyStorePassword,
-                trustStoreFile = trustStoreFile,
-                trustStorePassword = trustStorePassword,
-                routing = routing,
-                logLevel = logLevel
-        )
-
-val defaultListenPort = 1234
-val defaultRequestIntervalSec = Duration.ofSeconds(3)
-val defaultMaxPayloadSizeBytes = 2
-val defaultIdleTimeoutSec = Duration.ofSeconds(10L)
-val defaultFirstReqDelaySec = Duration.ofSeconds(10L)
-
-val KEYSTORE = "test.ks.pkcs12"
-val KEYSTORE_PASSWORD = "changeMe"
-val TRUSTSTORE = "trust.ks.pkcs12"
-val TRUSTSTORE_PASSWORD = "changeMeToo"
-
-val emptyRouting: Routing = emptyList()
+) = PartialConfiguration(
+        listenPort = listenPort,
+        idleTimeoutSec = idleTimeoutSec,
+        maxPayloadSizeBytes = maxPayloadSizeBytes,
+        firstRequestDelaySec = firstReqDelaySec,
+        requestIntervalSec = requestIntervalSec,
+        sslDisable = sslDisable,
+        keyStoreFile = keyStoreFile,
+        keyStorePassword = keyStorePassword,
+        trustStoreFile = trustStoreFile,
+        trustStorePassword = trustStorePassword,
+        streamPublishers = streamPublishers,
+        logLevel = logLevel
+)
+
+const val defaultListenPort = 1234
+const val defaultMaxPayloadSizeBytes = 2
+const val defaultRequestIntervalSec = 3L
+const val defaultIdleTimeoutSec = 10L
+const val defaultFirstReqDelaySec = 10L
+
+const val KEYSTORE = "test.ks.pkcs12"
+const val KEYSTORE_PASSWORD = "changeMe"
+const val TRUSTSTORE = "trust.ks.pkcs12"
+const val TRUSTSTORE_PASSWORD = "changeMeToo"
+
+const val sampleSinkName = "perf3gpp"
+
+private val sampleSink = mock<KafkaSink>().also {
+    whenever(it.name()).thenReturn(sampleSinkName)
+}
+val sampleStreamsDefinition = listOf(sampleSink)
+val sampleRouting = listOf(Route(sampleSink.name(), sampleSink))
\ No newline at end of file
diff --git a/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/FileConfigurationReaderTest.kt b/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/JsonConfigurationParserTest.kt
index b4683458..ad38fd51 100644
--- a/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/FileConfigurationReaderTest.kt
+++ b/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/JsonConfigurationParserTest.kt
@@ -35,15 +35,15 @@ import kotlin.test.fail
  * @author Pawel Biniek <pawel.biniek@nokia.com>
  * @since February 2019
  */
-internal object FileConfigurationReaderTest : Spek({
-    describe("A configuration loader utility") {
-        val cut = FileConfigurationReader()
+internal object JsonConfigurationParserTest : Spek({
+    describe("A configuration parser utility") {
+        val cut = JsonConfigurationParser()
 
-        describe("partial configuration loading") {
+        describe("partial configuration parsing") {
             it("parses enumerations") {
                 val input = """{"logLevel":"ERROR"}"""
 
-                val config = cut.loadConfig(StringReader(input))
+                val config = cut.parse(StringReader(input))
                 assertThat(config.logLevel).isEqualTo(Some(LogLevel.ERROR))
             }
 
@@ -53,16 +53,16 @@ internal object FileConfigurationReaderTest : Spek({
                     "cbs.firstRequestDelaySec": 10
                 }
                 """.trimIndent()
-                val config = cut.loadConfig(StringReader(input))
+                val config = cut.parse(StringReader(input))
                 assertThat(config.listenPort).isEqualTo(Some(12003))
-                assertThat(config.firstRequestDelaySec).isEqualTo(Some(Duration.ofSeconds(10)))
+                assertThat(config.firstRequestDelaySec).isEqualTo(Some(10L))
             }
 
             it("parses disabled security configuration") {
                 val input = """{
                     "security.sslDisable": true
                 }""".trimIndent()
-                val config = cut.loadConfig(StringReader(input))
+                val config = cut.parse(StringReader(input))
 
                 assertThat(config.sslDisable.getOrElse { fail("Should be Some") }).isTrue()
             }
@@ -71,26 +71,26 @@ internal object FileConfigurationReaderTest : Spek({
                 val input = """{
                     "logLevel": something
                 }""".trimMargin()
-                val config = cut.loadConfig(input.reader())
+                val config = cut.parse(input.reader())
 
                 assertThat(config.logLevel.isEmpty())
             }
         }
 
-        describe("complete file loading") {
-            it("loads actual file") {
-                val config = cut.loadConfig(
+        describe("complete json parsing") {
+            it("parses actual json") {
+                val config = cut.parse(
                         javaClass.resourceAsStream("/sampleConfig.json"))
 
                 assertThat(config).isNotNull
                 assertThat(config.logLevel).isEqualTo(Some(LogLevel.ERROR))
 
                 assertThat(config.listenPort).isEqualTo(Some(6000))
-                assertThat(config.idleTimeoutSec).isEqualTo(Some(Duration.ofSeconds(1200)))
+                assertThat(config.idleTimeoutSec).isEqualTo(Some(1200L))
                 assertThat(config.maxPayloadSizeBytes).isEqualTo(Some(1048576))
 
-                assertThat(config.firstRequestDelaySec).isEqualTo(Some(Duration.ofSeconds(7)))
-                assertThat(config.requestIntervalSec).isEqualTo(Some(Duration.ofSeconds(900)))
+                assertThat(config.firstRequestDelaySec).isEqualTo(Some(7L))
+                assertThat(config.requestIntervalSec).isEqualTo(Some(900L))
 
                 assertThat(config.sslDisable).isEqualTo(Some(false))
                 assertThat(config.keyStoreFile).isEqualTo(Some("test.ks.pkcs12"))
@@ -101,4 +101,3 @@ internal object FileConfigurationReaderTest : Spek({
         }
     }
 })
-