Extract transforming logic from validator

Change-Id: Ic019b1796e17d24f14f41a817af6e5ecd8c7244b
Issue-ID: DCAEGEN2-1416
Signed-off-by: Filip Krzywka <filip.krzywka@nokia.com>

5 files changed, 401 insertions, 203 deletions

diff --git a/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/CollectorConfigurationTest.kt b/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/CollectorConfigurationTest.kt
deleted file mode 100644
index dbdf4ad0..00000000
--- a/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/CollectorConfigurationTest.kt
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * dcaegen2-collectors-veshv
- * ================================================================================
- * Copyright (C) 2019 NOKIA
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============LICENSE_END=========================================================
- */
-package org.onap.dcae.collectors.veshv.config.impl
-
-import com.nhaarman.mockitokotlin2.mock
-import com.nhaarman.mockitokotlin2.whenever
-import org.assertj.core.api.Assertions.assertThat
-import org.jetbrains.spek.api.Spek
-import org.jetbrains.spek.api.dsl.describe
-import org.jetbrains.spek.api.dsl.it
-import org.jetbrains.spek.api.dsl.on
-import org.onap.dcae.collectors.veshv.config.api.model.CollectorConfiguration
-import org.onap.dcae.collectors.veshv.config.api.model.CollectorConfiguration.Companion.DEFAULT_MAX_PAYLOAD_SIZE
-import org.onap.dcae.collectors.veshv.config.api.model.Route
-import org.onap.dcaegen2.services.sdk.model.streams.dmaap.KafkaSink
-
-/**
- * @author Jakub Dudycz <jakub.dudycz@nokia.com>
- * @since May 2018
- */
-internal object CollectorConfigurationTest : Spek({
-
-    describe("CollectorConfiguration") {
-        describe("calculating maxPayloadSizeBytes") {
-            on("defined routes") {
-                val sampleRouting = listOf(
-                        Route(sink1.name(), sink1),
-                        Route(sink2.name(), sink2),
-                        Route(sink3.name(), sink3)
-                )
-                val configuration = CollectorConfiguration(sampleRouting)
-
-                it("should use the highest value among all routes") {
-                    assertThat(configuration.maxPayloadSizeBytes)
-                            .isEqualTo(highestMaxPayloadSize)
-                }
-            }
-
-            on("empty routing") {
-                val configuration = CollectorConfiguration(emptyList())
-
-                it("should use default value") {
-                    assertThat(configuration.maxPayloadSizeBytes)
-                            .isEqualTo(DEFAULT_MAX_PAYLOAD_SIZE)
-                }
-            }
-        }
-    }
-})
-
-private const val highestMaxPayloadSize = 3
-
-private val sink1 = mock<KafkaSink>().also {
-    whenever(it.name()).thenReturn("")
-    whenever(it.maxPayloadSizeBytes()).thenReturn(1)
-}
-
-private val sink2 = mock<KafkaSink>().also {
-    whenever(it.name()).thenReturn("")
-    whenever(it.maxPayloadSizeBytes()).thenReturn(2)
-}
-
-private val sink3 = mock<KafkaSink>().also {
-    whenever(it.name()).thenReturn("")
-    whenever(it.maxPayloadSizeBytes()).thenReturn(highestMaxPayloadSize)
-}
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 cb8d5005..ca09d84f 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
@@ -27,7 +27,6 @@ import org.jetbrains.spek.api.dsl.it
 import org.onap.dcae.collectors.veshv.utils.logging.LogLevel
 import java.io.InputStreamReader
 import java.io.Reader
-import java.time.Duration
 
 /**
  * @author Pawel Biniek <pawel.biniek@nokia.com>
@@ -37,14 +36,14 @@ internal object ConfigurationMergerTest : Spek({
     describe("Merges partial configurations into one") {
         it("merges single parameter into empty config") {
             val actual = PartialConfiguration()
-            val diff = PartialConfiguration(logLevel = Some(LogLevel.INFO))
+            val diff = PartialConfiguration(logLevel = Some(LogLevel.WARN))
 
             val result = ConfigurationMerger().merge(actual, diff)
 
-            assertThat(result.logLevel).isEqualTo(Some(LogLevel.INFO))
+            assertThat(result.logLevel).isEqualTo(Some(LogLevel.WARN))
         }
 
-        val someListenPort = Some(45)
+        val someListenPort = Some(defaultListenPort)
         it("merges single embedded parameter into empty config") {
             val actual = PartialConfiguration()
             val diff = PartialConfiguration(listenPort = someListenPort)
@@ -58,11 +57,11 @@ internal object ConfigurationMergerTest : Spek({
             val actual = JsonConfigurationParser().parse(
                     InputStreamReader(
                             JsonConfigurationParserTest.javaClass.getResourceAsStream("/sampleConfig.json")) as Reader)
-            val diff = PartialConfiguration(logLevel = Some(LogLevel.INFO))
+            val diff = PartialConfiguration(logLevel = Some(LogLevel.WARN))
 
             val result = ConfigurationMerger().merge(actual, diff)
 
-            assertThat(result.logLevel).isEqualTo(Some(LogLevel.INFO))
+            assertThat(result.logLevel).isEqualTo(Some(LogLevel.WARN))
         }
 
         it("merges single embedded parameter into full config") {
@@ -74,7 +73,6 @@ internal object ConfigurationMergerTest : Spek({
             val result = ConfigurationMerger().merge(actual, diff)
 
             assertThat(result.listenPort).isEqualTo(someListenPort)
-            assertThat(result.idleTimeoutSec.isEmpty()).isFalse()
             assertThat(result.idleTimeoutSec).isEqualTo(Some(1200L))
         }
 
diff --git a/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationTransformerTest.kt b/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationTransformerTest.kt
new file mode 100644
index 00000000..42919e4d
--- /dev/null
+++ b/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationTransformerTest.kt
@@ -0,0 +1,218 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcae.collectors.veshv.config.impl
+
+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.given
+import org.jetbrains.spek.api.dsl.it
+import org.jetbrains.spek.api.dsl.on
+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
+
+
+internal object ConfigurationTransformerTest : Spek({
+    describe("ConfigurationTransformer") {
+        val cut = ConfigurationTransformer()
+
+        describe("transforming partial configuration to final") {
+            val config = ValidatedPartialConfiguration(
+                    listenPort = defaultListenPort,
+                    idleTimeoutSec = defaultIdleTimeoutSec,
+                    cbsConfiguration = ValidatedCbsConfiguration(
+                            firstRequestDelaySec = defaultFirstReqDelaySec,
+                            requestIntervalSec = defaultRequestIntervalSec
+                    ),
+                    securityConfiguration = Some(ValidatedSecurityPaths(
+                            keyStoreFile = KEYSTORE,
+                            keyStorePasswordFile = KEYSTORE_PASS_FILE,
+                            trustStoreFile = TRUSTSTORE,
+                            trustStorePasswordFile = TRUSTSTORE_PASS_FILE
+                    )),
+                    streamPublishers = sampleStreamsDefinition,
+                    logLevel = Some(LogLevel.TRACE)
+            )
+
+            given("transformed configuration") {
+                val result = cut.toFinalConfiguration(config)
+
+                it("should create server configuration") {
+                    assertThat(result.server.listenPort).isEqualTo(defaultListenPort)
+                    assertThat(result.server.idleTimeout)
+                            .describedAs("idleTimeout transformed from number to duration")
+                            .isEqualTo(Duration.ofSeconds(defaultIdleTimeoutSec))
+                }
+
+                it("should create CBS configuration") {
+                    assertThat(result.cbs.firstRequestDelay)
+                            .describedAs("firstRequestDelay transformed from number to duration")
+                            .isEqualTo(Duration.ofSeconds(defaultFirstReqDelaySec))
+                    assertThat(result.cbs.requestInterval)
+                            .describedAs("requestInterval transformed from number to duration")
+                            .isEqualTo(Duration.ofSeconds(defaultRequestIntervalSec))
+                }
+
+                it("should create collector configuration") {
+                    assertThat(result.collector.routing)
+                            .describedAs("routing transformed from kafka sinks to routes")
+                            .isEqualTo(sampleRouting)
+
+                    assertThat(result.collector.maxPayloadSizeBytes)
+                            .describedAs("maxPayloadSizeBytes calculated from kafka sinks")
+                            .isEqualTo(DEFAULT_MAX_PAYLOAD_SIZE_BYTES)
+                }
+
+                it("should use specified log level") {
+                    assertThat(result.logLevel)
+                            .describedAs("logLevel was not transformed when present")
+                            .isEqualTo(LogLevel.TRACE)
+                }
+
+                it("should create security keys") {
+                    result.security.keys.fold({ fail("Should be Some") }, {
+                        assertThat(it.keyStore().path()).isEqualTo(File(KEYSTORE).toPath())
+                        assertThat(it.trustStore().path()).isEqualTo(File(TRUSTSTORE).toPath())
+                        it.keyStorePassword().use { assertThat(it).isEqualTo(KEYSTORE_PASSWORD.toCharArray()) }
+                        it.trustStorePassword().use { assertThat(it).isEqualTo(TRUSTSTORE_PASSWORD.toCharArray()) }
+                    })
+                }
+            }
+        }
+
+        describe("transforming configuration with empty log level") {
+            val config = validatedConfiguration(
+                    logLevel = None
+            )
+
+            it("should use default log level") {
+                val result = cut.toFinalConfiguration(config)
+
+                assertThat(result.logLevel).isEqualTo(DEFAULT_LOG_LEVEL)
+            }
+        }
+
+        describe("transforming configuration with security disabled") {
+            val config = validatedConfiguration(
+                    sslDisable = Some(true),
+                    keyStoreFile = "",
+                    keyStorePasswordFile = "",
+                    trustStoreFile = "",
+                    trustStorePasswordFile = ""
+            )
+
+            it("should create valid configuration with empty security keys") {
+                val result = cut.toFinalConfiguration(config)
+
+                assertThat(result.security.keys).isEqualTo(None)
+            }
+        }
+
+        describe("transforming configuration with ssl disable missing") {
+            val config = validatedConfiguration(
+                    sslDisable = None
+            )
+
+            it("should create configuration with ssl enabled") {
+                val result = cut.toFinalConfiguration(config)
+                val securityKeys = result.security.keys
+                        .getOrElse { fail("Should be immutableSecurityKeys") } as SecurityKeys
+                assertThat(securityKeys.keyStore().path()).isEqualTo(File(KEYSTORE).toPath())
+                assertThat(securityKeys.trustStore().path()).isEqualTo(File(TRUSTSTORE).toPath())
+                securityKeys.keyStorePassword().use { assertThat(it).isEqualTo(KEYSTORE_PASSWORD.toCharArray()) }
+                securityKeys.trustStorePassword().use { assertThat(it).isEqualTo(TRUSTSTORE_PASSWORD.toCharArray()) }
+            }
+        }
+
+        describe("calculating maxPayloadSizeBytes") {
+            on("defined routes") {
+                val highestMaxPayloadSize = 3
+                val sink1 = mock<KafkaSink>().also {
+                    whenever(it.name()).thenReturn("1")
+                    whenever(it.maxPayloadSizeBytes()).thenReturn(1)
+                }
+                val sink2 = mock<KafkaSink>().also {
+                    whenever(it.name()).thenReturn("2")
+                    whenever(it.maxPayloadSizeBytes()).thenReturn(highestMaxPayloadSize)
+                }
+                val config = validatedConfiguration(
+                        streamPublishers = listOf(sink1, sink2)
+                )
+
+                val result = cut.toFinalConfiguration(config)
+
+                it("should use the highest value among all routes") {
+                    assertThat(result.collector.maxPayloadSizeBytes)
+                            .isEqualTo(highestMaxPayloadSize)
+                }
+            }
+
+            on("empty routing") {
+                val config = validatedConfiguration(
+                        streamPublishers = emptyList()
+                )
+
+                val result = cut.toFinalConfiguration(config)
+
+                it("should use default value") {
+                    assertThat(result.collector.maxPayloadSizeBytes)
+                            .isEqualTo(DEFAULT_MAX_PAYLOAD_SIZE_BYTES)
+                }
+            }
+        }
+
+    }
+})
+
+private fun validatedConfiguration(listenPort: Int = defaultListenPort,
+                                   idleTimeoutSec: Long = defaultIdleTimeoutSec,
+                                   firstReqDelaySec: Long = defaultFirstReqDelaySec,
+                                   requestIntervalSec: Long = defaultRequestIntervalSec,
+                                   sslDisable: Option<Boolean> = Some(false),
+                                   keyStoreFile: String = KEYSTORE,
+                                   keyStorePasswordFile: String = KEYSTORE_PASS_FILE,
+                                   trustStoreFile: String = TRUSTSTORE,
+                                   trustStorePasswordFile: String = TRUSTSTORE_PASS_FILE,
+                                   streamPublishers: List<KafkaSink> = sampleStreamsDefinition,
+                                   logLevel: Option<LogLevel> = Some(LogLevel.INFO)
+): ValidatedPartialConfiguration = PartialConfiguration(
+        listenPort = Some(listenPort),
+        idleTimeoutSec = Some(idleTimeoutSec),
+        firstRequestDelaySec = Some(firstReqDelaySec),
+        requestIntervalSec = Some(requestIntervalSec),
+        streamPublishers = Some(streamPublishers),
+        sslDisable = sslDisable,
+        keyStoreFile = Some(keyStoreFile),
+        keyStorePasswordFile = Some(keyStorePasswordFile),
+        trustStoreFile = Some(trustStoreFile),
+        trustStorePasswordFile = Some(trustStorePasswordFile),
+        logLevel = logLevel
+).unsafeAsValidated()
+
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 5495c865..26a9cc57 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
@@ -22,101 +22,84 @@ package org.onap.dcae.collectors.veshv.config.impl
 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.assertj.core.api.Assertions.*
 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.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.config.api.model.ValidationException
 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.nio.file.Paths
-import java.time.Duration
 
 internal object ConfigurationValidatorTest : Spek({
     describe("ConfigurationValidator") {
         val cut = ConfigurationValidator()
 
         describe("validating partial configuration with missing fields") {
-            val config = PartialConfiguration(
-                    listenPort = Some(1)
-            )
-
-            it("should return ValidationError") {
-                val result = cut.validate(config)
-                assertThat(result.isLeft()).isTrue()
-            }
-        }
+            val config = PartialConfiguration(listenPort = Some(5))
 
-        describe("validating configuration with empty log level") {
-            val config = partialConfiguration(
-                    logLevel = None
-            )
-
-            it("should use default log level") {
+            it("should return ValidationException with missing required fields description") {
                 val result = cut.validate(config)
-                result.fold(
-                        {
-                            fail("Configuration should have been created successfully")
-                        },
-                        {
-                            assertThat(it.logLevel).isEqualTo(DEFAULT_LOG_LEVEL)
-                        }
-                )
+                result.fold({
+                    assertThat(it.message).doesNotContain(PartialConfiguration::listenPort.name)
+
+                    assertThat(it.message).contains(PartialConfiguration::idleTimeoutSec.name)
+                    assertThat(it.message).contains(PartialConfiguration::firstRequestDelaySec.name)
+                    assertThat(it.message).contains(PartialConfiguration::requestIntervalSec.name)
+                    assertThat(it.message).contains(PartialConfiguration::streamPublishers.name)
+                    assertThat(it.message).contains(PartialConfiguration::keyStoreFile.name)
+                    assertThat(it.message).contains(PartialConfiguration::keyStorePasswordFile.name)
+                    assertThat(it.message).contains(PartialConfiguration::trustStoreFile.name)
+                    assertThat(it.message).contains(PartialConfiguration::trustStorePasswordFile.name)
+
+                    assertThat(it.message).doesNotContain(PartialConfiguration::logLevel.name)
+                    assertThat(it.message).doesNotContain(PartialConfiguration::sslDisable.name)
+                }, { fail("Should be ValidationException") })
             }
         }
 
-        describe("validating complete configuration") {
+        describe("validating complete valid configuration") {
             val config = PartialConfiguration(
                     listenPort = Some(defaultListenPort),
                     idleTimeoutSec = Some(defaultIdleTimeoutSec),
                     firstRequestDelaySec = Some(defaultFirstReqDelaySec),
                     requestIntervalSec = Some(defaultRequestIntervalSec),
                     sslDisable = Some(false),
-                    keyStoreFile = Some(keyStore),
-                    keyStorePasswordFile = Some(keyStorePassFile),
-                    trustStoreFile = Some(trustStore),
-                    trustStorePasswordFile = Some(trustStorePassFile),
+                    keyStoreFile = Some(KEYSTORE),
+                    keyStorePasswordFile = Some(KEYSTORE_PASSWORD),
+                    trustStoreFile = Some(TRUSTSTORE),
+                    trustStorePasswordFile = Some(TRUSTSTORE_PASSWORD),
                     streamPublishers = Some(sampleStreamsDefinition),
                     logLevel = Some(LogLevel.TRACE)
             )
 
-            it("should create valid configuration") {
+            it("should create validated configuration") {
                 val result = cut.validate(config)
                 result.fold(
                         {
                             fail("Configuration should have been created successfully")
                         },
                         {
-                            assertThat(it.server.listenPort)
+                            assertThat(it.listenPort)
                                     .isEqualTo(defaultListenPort)
-                            assertThat(it.server.idleTimeout)
-                                    .isEqualTo(Duration.ofSeconds(defaultIdleTimeoutSec))
-
-                            val securityKeys = it.security.keys
-                                    .getOrElse { fail("Should be immutableSecurityKeys") } as SecurityKeys
-                            assertThat(securityKeys.keyStore().path()).isEqualTo(File(keyStore).toPath())
-                            assertThat(securityKeys.trustStore().path()).isEqualTo(File(trustStore).toPath())
-                            securityKeys.keyStorePassword().use { assertThat(it).isEqualTo(keyStorePass.toCharArray()) }
-                            securityKeys.trustStorePassword().use { assertThat(it).isEqualTo(trustStorePass.toCharArray()) }
-
-                            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.maxPayloadSizeBytes)
-                                    .isEqualTo(sampleMaxPayloadSize)
-
-                            assertThat(it.logLevel).isEqualTo(LogLevel.TRACE)
+                            assertThat(it.idleTimeoutSec)
+                                    .isEqualTo(defaultIdleTimeoutSec)
+
+                            it.securityConfiguration.fold({
+                                fail("Should have been validated successfully")
+                            }, {
+                                assertThat(it.keyStoreFile).isEqualTo(KEYSTORE)
+                                assertThat(it.keyStorePasswordFile).isEqualTo(KEYSTORE_PASSWORD)
+                                assertThat(it.trustStoreFile).isEqualTo(TRUSTSTORE)
+                                assertThat(it.trustStorePasswordFile).isEqualTo(TRUSTSTORE_PASSWORD)
+                            })
+
+                            assertThat(it.cbsConfiguration.firstRequestDelaySec).isEqualTo(defaultFirstReqDelaySec)
+                            assertThat(it.cbsConfiguration.requestIntervalSec).isEqualTo(defaultRequestIntervalSec)
+
+                            assertThat(it.streamPublishers).isEqualTo(sampleStreamsDefinition)
+
+                            assertThat(it.logLevel).isEqualTo(Some(LogLevel.TRACE))
                         }
                 )
             }
@@ -126,29 +109,26 @@ internal object ConfigurationValidatorTest : Spek({
             val config = partialConfiguration(
                     sslDisable = Some(true),
                     keyStoreFile = Some(""),
-                    keyStorePassword = Some(""),
-                    trustStoreFile = Some(""),
-                    trustStorePassword = Some("")
+                    keyStorePasswordFile = None,
+                    trustStoreFile = None,
+                    trustStorePasswordFile = Some("")
             )
 
-            it("should create valid configuration") {
+            it("should return validated configuration regardless of security keys presence") {
                 val result = cut.validate(config)
                 result.fold(
                         {
                             fail("Configuration should have been created successfully but was $it")
                         },
                         {
-                            assertThat(it.server.idleTimeout)
-                                    .isEqualTo(Duration.ofSeconds(defaultIdleTimeoutSec))
+                            assertThat(it.idleTimeoutSec).isEqualTo(defaultIdleTimeoutSec)
 
-                            assertThat(it.security.keys)
-                                    .isEqualTo(None)
+                            assertThat(it.securityConfiguration.isEmpty()).isTrue()
 
-                            assertThat(it.cbs.firstRequestDelay)
-                                    .isEqualTo(Duration.ofSeconds(defaultFirstReqDelaySec))
+                            assertThat(it.cbsConfiguration.firstRequestDelaySec).isEqualTo(defaultFirstReqDelaySec)
+                            assertThat(it.cbsConfiguration.requestIntervalSec).isEqualTo(defaultRequestIntervalSec)
 
-                            assertThat(it.collector.routing)
-                                    .isEqualTo(sampleRouting)
+                            assertThat(it.streamPublishers).isEqualTo(sampleStreamsDefinition)
                         }
                 )
             }
@@ -159,24 +139,81 @@ internal object ConfigurationValidatorTest : Spek({
                     sslDisable = None
             )
 
-            it("should create valid configuration with ssl enabled") {
+            it("should return validated configuration") {
                 val result = cut.validate(config)
                 result.fold(
                         {
                             fail("Configuration should have been created successfully but was $it")
                         },
                         {
-                            val securityKeys = it.security.keys
-                                    .getOrElse { fail("Should be immutableSecurityKeys") } as SecurityKeys
-                            assertThat(securityKeys.keyStore().path()).isEqualTo(File(keyStore).toPath())
-                            assertThat(securityKeys.trustStore().path()).isEqualTo(File(trustStore).toPath())
-                            securityKeys.keyStorePassword().use { assertThat(it).isEqualTo(keyStorePass.toCharArray()) }
-                            securityKeys.trustStorePassword().use { assertThat(it).isEqualTo(trustStorePass.toCharArray()) }
+                            it.securityConfiguration.fold({
+                                fail("Should have been validated successfully")
+                            }, {
+                                assertThat(it.keyStoreFile).isEqualTo(KEYSTORE)
+                                assertThat(it.keyStorePasswordFile).isEqualTo(KEYSTORE_PASSWORD)
+                                assertThat(it.trustStoreFile).isEqualTo(TRUSTSTORE)
+                                assertThat(it.trustStorePasswordFile).isEqualTo(TRUSTSTORE_PASSWORD)
+                            })
+
                         }
                 )
             }
         }
 
+        describe("validating configuration with ssl enabled, but not all required security fields set") {
+            val config = partialConfiguration(
+                    sslDisable = Some(false),
+                    keyStoreFile = Some(KEYSTORE),
+                    keyStorePasswordFile = None,
+                    trustStoreFile = None,
+                    trustStorePasswordFile = Some(TRUSTSTORE_PASSWORD)
+            )
+
+            it("should return validated configuration") {
+                val result = cut.validate(config)
+
+                assertThat(result.isLeft())
+                        .describedAs("security validation result")
+                        .isTrue()
+            }
+        }
+
+        describe("validating CBS configuration from partial") {
+            given("valid CBS configuration") {
+                val config = partialConfiguration()
+
+                it("should returned validated config") {
+                    val result = cut.validatedCbsConfiguration(config)
+
+                    assertThat(result.firstRequestDelaySec).isEqualTo(defaultFirstReqDelaySec)
+                    assertThat(result.requestIntervalSec).isEqualTo(defaultRequestIntervalSec)
+                }
+
+            }
+
+            given("missing firstReqDelaySec") {
+                val config = partialConfiguration(
+                        firstReqDelaySec = None
+                )
+
+                it("should throw validation exception") {
+                    assertThatExceptionOfType(ValidationException::class.java).isThrownBy {
+                        cut.validatedCbsConfiguration(config)
+                    }.withMessageContaining(PartialConfiguration::firstRequestDelaySec.name)
+                }
+            }
+
+            given("missing requestIntervalSec") {
+                val config = partialConfiguration(
+                        requestIntervalSec = None)
+
+                it("should throw validation exception") {
+                    assertThatExceptionOfType(ValidationException::class.java).isThrownBy {
+                        cut.validatedCbsConfiguration(config)
+                    }.withMessageContaining(PartialConfiguration::requestIntervalSec.name)
+                }
+            }
+        }
     }
 })
 
@@ -185,10 +222,10 @@ private fun partialConfiguration(listenPort: Option<Int> = Some(defaultListenPor
                                  firstReqDelaySec: Option<Long> = Some(defaultFirstReqDelaySec),
                                  requestIntervalSec: Option<Long> = Some(defaultRequestIntervalSec),
                                  sslDisable: Option<Boolean> = Some(false),
-                                 keyStoreFile: Option<String> = Some(keyStore),
-                                 keyStorePassword: Option<String> = Some(keyStorePassFile),
-                                 trustStoreFile: Option<String> = Some(trustStore),
-                                 trustStorePassword: Option<String> = Some(trustStorePassFile),
+                                 keyStoreFile: Option<String> = Some(KEYSTORE),
+                                 keyStorePasswordFile: Option<String> = Some(KEYSTORE_PASSWORD),
+                                 trustStoreFile: Option<String> = Some(TRUSTSTORE),
+                                 trustStorePasswordFile: Option<String> = Some(TRUSTSTORE_PASSWORD),
                                  streamPublishers: Option<List<KafkaSink>> = Some(sampleStreamsDefinition),
                                  logLevel: Option<LogLevel> = Some(LogLevel.INFO)
 ) = PartialConfiguration(
@@ -198,35 +235,9 @@ private fun partialConfiguration(listenPort: Option<Int> = Some(defaultListenPor
         requestIntervalSec = requestIntervalSec,
         sslDisable = sslDisable,
         keyStoreFile = keyStoreFile,
-        keyStorePasswordFile = keyStorePassword,
+        keyStorePasswordFile = keyStorePasswordFile,
         trustStoreFile = trustStoreFile,
-        trustStorePasswordFile = trustStorePassword,
+        trustStorePasswordFile = trustStorePasswordFile,
         streamPublishers = streamPublishers,
         logLevel = logLevel
 )
-
-private fun resourcePathAsString(resource: String) =
-        Paths.get(ConfigurationValidatorTest::class.java.getResource(resource).toURI()).toString()
-
-private const val defaultListenPort = 1234
-private const val defaultRequestIntervalSec = 3L
-private const val defaultIdleTimeoutSec = 10L
-private const val defaultFirstReqDelaySec = 10L
-
-private const val keyStore = "test.ks.pkcs12"
-private const val trustStore = "trust.ks.pkcs12"
-private const val keyStorePass = "change.me"
-private const val trustStorePass = "change.me.too"
-private val keyStorePassFile = resourcePathAsString("/test.ks.pass")
-private val trustStorePassFile = resourcePathAsString("/trust.ks.pass")
-
-private const val sampleSinkName = "perf3gpp"
-const val sampleMaxPayloadSize = 1024
-
-private val sink = mock<KafkaSink>().also {
-    whenever(it.name()).thenReturn(sampleSinkName)
-    whenever(it.maxPayloadSizeBytes()).thenReturn(sampleMaxPayloadSize)
-}
-
-private val sampleStreamsDefinition = listOf(sink)
-private val sampleRouting = listOf(Route(sink.name(), sink))
diff --git a/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/test_constants.kt b/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/test_constants.kt
new file mode 100644
index 00000000..f07af079
--- /dev/null
+++ b/sources/hv-collector-configuration/src/test/kotlin/org/onap/dcae/collectors/veshv/config/impl/test_constants.kt
@@ -0,0 +1,54 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcae.collectors.veshv.config.impl
+
+import com.nhaarman.mockitokotlin2.mock
+import com.nhaarman.mockitokotlin2.whenever
+import org.onap.dcae.collectors.veshv.config.api.model.Route
+import org.onap.dcae.collectors.veshv.utils.logging.LogLevel
+import org.onap.dcaegen2.services.sdk.model.streams.dmaap.KafkaSink
+import java.nio.file.Paths
+
+private fun resourcePathAsString(resource: String) =
+        Paths.get(ConfigurationValidatorTest::class.java.getResource(resource).toURI()).toString()
+
+internal val DEFAULT_LOG_LEVEL = LogLevel.INFO
+
+internal const val defaultListenPort = 1234
+internal const val defaultRequestIntervalSec = 3L
+internal const val defaultIdleTimeoutSec = 10L
+internal const val defaultFirstReqDelaySec = 10L
+
+internal const val KEYSTORE = "test.ks.pkcs12"
+internal const val KEYSTORE_PASSWORD = "change.me"
+internal const val TRUSTSTORE = "trust.ks.pkcs12"
+internal const val TRUSTSTORE_PASSWORD = "change.me.too"
+internal val KEYSTORE_PASS_FILE = resourcePathAsString("/test.ks.pass")
+internal val TRUSTSTORE_PASS_FILE = resourcePathAsString("/trust.ks.pass")
+
+internal const val DEFAULT_MAX_PAYLOAD_SIZE_BYTES = 1024 * 1024
+
+private val sampleSink = mock<KafkaSink>().also {
+    whenever(it.name()).thenReturn("perf3gpp")
+    whenever(it.maxPayloadSizeBytes()).thenReturn(DEFAULT_MAX_PAYLOAD_SIZE_BYTES)
+}
+
+internal val sampleStreamsDefinition = listOf(sampleSink)
+internal val sampleRouting = listOf(Route(sampleSink.name(), sampleSink))
\ No newline at end of file