diff options
Diffstat (limited to 'sources/hv-collector-dcae-app-simulator/src/test/kotlin/org')
5 files changed, 658 insertions, 0 deletions
diff --git a/sources/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/impl/ConsumerTest.kt b/sources/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/impl/ConsumerTest.kt new file mode 100644 index 00000000..08558d76 --- /dev/null +++ b/sources/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/impl/ConsumerTest.kt @@ -0,0 +1,82 @@ +/* + * ============LICENSE_START======================================================= + * dcaegen2-collectors-veshv + * ================================================================================ + * Copyright (C) 2018 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.simulators.dcaeapp.impl + +import org.apache.kafka.clients.consumer.ConsumerRecord +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 reactor.kafka.receiver.ReceiverRecord + + +/** + * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> + * @since August 2018 + */ +internal class ConsumerTest : Spek({ + + lateinit var cut: Consumer + + beforeEachTest { + cut = Consumer() + } + + describe("Consumer which holds the state of received Kafka records") { + it("should contain empty state in the beginning") { + assertEmptyState(cut) + } + + describe("update") { + val value = byteArrayOf(2) + + beforeEachTest { + cut.update(receiverRecord( + topic = "topic", + key = byteArrayOf(1), + value = value + )).unsafeRunSync() + } + + it("should contain one message if it was updated once") { + assertState(cut, value) + } + + it("should contain empty state message if it was reset after update") { + cut.reset().unsafeRunSync() + assertEmptyState(cut) + } + } + } +}) + +fun assertEmptyState(cut: Consumer) { + assertState(cut) +} + +fun assertState(cut: Consumer, vararg values: ByteArray) { + assertThat(cut.currentState().consumedMessages) + .containsOnly(*values) + assertThat(cut.currentState().messagesCount) + .isEqualTo(values.size) +} + +fun receiverRecord(topic: String, key: ByteArray, value: ByteArray) = + ReceiverRecord(ConsumerRecord(topic, 1, 100L, key, value), null) diff --git a/sources/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/impl/DcaeAppSimulatorTest.kt b/sources/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/impl/DcaeAppSimulatorTest.kt new file mode 100644 index 00000000..e1641cbb --- /dev/null +++ b/sources/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/impl/DcaeAppSimulatorTest.kt @@ -0,0 +1,183 @@ +/* + * ============LICENSE_START======================================================= + * dcaegen2-collectors-veshv + * ================================================================================ + * Copyright (C) 2018 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.simulators.dcaeapp.impl + +import arrow.core.Left +import arrow.core.None +import arrow.core.Some +import arrow.effects.IO +import com.google.protobuf.ByteString +import com.nhaarman.mockitokotlin2.any +import com.nhaarman.mockitokotlin2.eq +import com.nhaarman.mockitokotlin2.mock +import com.nhaarman.mockitokotlin2.never +import com.nhaarman.mockitokotlin2.verify +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.mockito.ArgumentMatchers.anySet +import org.onap.ves.VesEventOuterClass.CommonEventHeader +import org.onap.ves.VesEventOuterClass.VesEvent +import java.util.concurrent.ConcurrentLinkedQueue + +/** + * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> + * @since August 2018 + */ +internal class DcaeAppSimulatorTest : Spek({ + lateinit var consumerFactory: ConsumerFactory + lateinit var messageStreamValidation: MessageStreamValidation + lateinit var consumer: Consumer + lateinit var cut: DcaeAppSimulator + + beforeEachTest { + consumerFactory = mock() + messageStreamValidation = mock() + consumer = mock() + cut = DcaeAppSimulator(consumerFactory, messageStreamValidation) + + whenever(consumerFactory.createConsumerForTopics(anySet())).thenReturn(IO.just(consumer)) + } + + fun consumerState(vararg messages: ByteArray) = ConsumerState(ConcurrentLinkedQueue(messages.toList())) + + describe("listenToTopics") { + val topics = setOf("perf3gpp", "faults") + + it("should fail when topic list is empty") { + val result = cut.listenToTopics(setOf()).attempt().unsafeRunSync() + assertThat(result.isLeft()).isTrue() + } + + it("should fail when topic list contains empty strings") { + val result = cut.listenToTopics(setOf("perf3gpp", " ", "faults")).attempt().unsafeRunSync() + assertThat(result.isLeft()).isTrue() + } + + it("should subscribe to given topics") { + cut.listenToTopics(topics).unsafeRunSync() + verify(consumerFactory).createConsumerForTopics(topics) + } + + it("should subscribe to given topics when called with comma separated list") { + cut.listenToTopics("perf3gpp,faults").unsafeRunSync() + verify(consumerFactory).createConsumerForTopics(topics) + } + + it("should handle errors") { + // given + val error = RuntimeException("WTF") + whenever(consumerFactory.createConsumerForTopics(anySet())) + .thenReturn(IO.raiseError(error)) + + // when + val result = cut.listenToTopics("perf3gpp").attempt().unsafeRunSync() + + // then + assertThat(result).isEqualTo(Left(error)) + } + } + + describe("state") { + + it("should return None when topics hasn't been initialized") { + assertThat(cut.state()).isEqualTo(None) + } + + describe("when topics are initialized") { + beforeEachTest { + cut.listenToTopics("perf3gpp").unsafeRunSync() + } + + it("should return some state when it has been set") { + val state = consumerState() + whenever(consumer.currentState()).thenReturn(state) + + assertThat(cut.state()).isEqualTo(Some(state)) + } + } + } + + describe("resetState") { + it("should do nothing when topics hasn't been initialized") { + cut.resetState().unsafeRunSync() + verify(consumer, never()).reset() + } + + describe("when topics are initialized") { + beforeEachTest { + cut.listenToTopics("perf3gpp").unsafeRunSync() + } + + it("should reset the state") { + // given + whenever(consumer.reset()).thenReturn(IO.unit) + + // when + cut.resetState().unsafeRunSync() + + // then + verify(consumer).reset() + } + } + } + + describe("validate") { + beforeEachTest { + whenever(messageStreamValidation.validate(any(), any())).thenReturn(IO.just(true)) + } + + it("should use empty list when consumer is unavailable") { + // when + val result = cut.validate("['The JSON']".byteInputStream()).unsafeRunSync() + + // then + verify(messageStreamValidation).validate(any(), eq(emptyList())) + assertThat(result).isTrue() + } + + it("should delegate to MessageStreamValidation") { + // given + cut.listenToTopics("perf3gpp").unsafeRunSync() + whenever(consumer.currentState()).thenReturn(consumerState(vesEvent().toByteArray())) + + // when + val result = cut.validate("['The JSON']".byteInputStream()).unsafeRunSync() + + // then + verify(messageStreamValidation).validate(any(), any()) + assertThat(result).isTrue() + } + } +}) + + +private const val DUMMY_EVENT_ID = "aaa" +private const val DUMMY_PAYLOAD = "payload" + +private fun vesEvent(eventId: String = DUMMY_EVENT_ID, payload: String = DUMMY_PAYLOAD): VesEvent { + return VesEvent.newBuilder() + .setCommonEventHeader(CommonEventHeader.newBuilder() + .setEventId(eventId)) + .setEventFields(ByteString.copyFrom(payload.toByteArray())) + .build() +} diff --git a/sources/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/impl/MessageStreamValidationTest.kt b/sources/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/impl/MessageStreamValidationTest.kt new file mode 100644 index 00000000..a631be76 --- /dev/null +++ b/sources/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/impl/MessageStreamValidationTest.kt @@ -0,0 +1,214 @@ +/* + * ============LICENSE_START======================================================= + * dcaegen2-collectors-veshv + * ================================================================================ + * Copyright (C) 2018 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.simulators.dcaeapp.impl + +import arrow.core.Either +import arrow.core.Right +import com.google.protobuf.ByteString +import com.nhaarman.mockitokotlin2.any +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.mockito.ArgumentMatchers.anyList +import org.onap.dcae.collectors.veshv.domain.WireFrameMessage +import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageGenerator +import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageParameters +import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageParametersParser +import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageType +import org.onap.ves.VesEventOuterClass.CommonEventHeader +import org.onap.ves.VesEventOuterClass.VesEvent +import reactor.core.publisher.Flux +import javax.json.stream.JsonParsingException + +/** + * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com> + * @since August 2018 + */ +internal class MessageStreamValidationTest : Spek({ + lateinit var messageParametersParser: MessageParametersParser + lateinit var messageGenerator: MessageGenerator + lateinit var cut: MessageStreamValidation + + beforeEachTest { + messageParametersParser = mock() + messageGenerator = mock() + cut = MessageStreamValidation(messageGenerator, messageParametersParser) + } + + fun givenParsedMessageParameters(vararg params: MessageParameters) { + whenever(messageParametersParser.parse(any())).thenReturn(Right(params.toList())) + } + + describe("validate") { + + it("should return error when JSON is invalid") { + // when + val result = cut.validate("[{invalid json}]".byteInputStream(), listOf()).attempt().unsafeRunSync() + + // then + when(result) { + is Either.Left -> assertThat(result.a).isInstanceOf(JsonParsingException::class.java) + else -> fail("validation should fail") + } + } + + it("should return error when message param list is empty") { + // given + givenParsedMessageParameters() + + // when + val result = cut.validate(sampleJsonAsStream(), listOf()).attempt().unsafeRunSync() + + // then + assertThat(result.isLeft()).isTrue() + } + + describe("when validating headers only") { + it("should return true when messages are the same") { + // given + val jsonAsStream = sampleJsonAsStream() + val event = vesEvent() + val generatedWireProtocolFrame = WireFrameMessage(event.toByteArray()) + val receivedMessageBytes = event.toByteArray() + + givenParsedMessageParameters(MessageParameters(event.commonEventHeader, MessageType.VALID, 1)) + whenever(messageGenerator.createMessageFlux(anyList())).thenReturn(Flux.just(generatedWireProtocolFrame)) + + // when + val result = cut.validate(jsonAsStream, listOf(receivedMessageBytes)).unsafeRunSync() + + // then + assertThat(result).isTrue() + } + + it("should return true when messages differ with payload only") { + // given + val jsonAsStream = sampleJsonAsStream() + val generatedEvent = vesEvent(payload = "payload A") + val receivedEvent = vesEvent(payload = "payload B") + val generatedWireProtocolFrame = WireFrameMessage(generatedEvent.toByteArray()) + val receivedMessageBytes = receivedEvent.toByteArray() + + givenParsedMessageParameters(MessageParameters(generatedEvent.commonEventHeader, MessageType.VALID, 1)) + whenever(messageGenerator.createMessageFlux(anyList())).thenReturn(Flux.just(generatedWireProtocolFrame)) + + // when + val result = cut.validate(jsonAsStream, listOf(receivedMessageBytes)).unsafeRunSync() + + // then + assertThat(result).isTrue() + } + + it("should return false when messages are different") { + // given + val jsonAsStream = sampleJsonAsStream() + val generatedEvent = vesEvent() + val receivedEvent = vesEvent(eventId = "bbb") + val generatedWireProtocolFrame = WireFrameMessage(generatedEvent.toByteArray()) + val receivedMessageBytes = receivedEvent.toByteArray() + + givenParsedMessageParameters(MessageParameters(generatedEvent.commonEventHeader, MessageType.VALID, 1)) + whenever(messageGenerator.createMessageFlux(anyList())).thenReturn(Flux.just(generatedWireProtocolFrame)) + + // when + val result = cut.validate(jsonAsStream, listOf(receivedMessageBytes)).unsafeRunSync() + + // then + assertThat(result).isFalse() + } + } + + describe("when validating whole messages") { + it("should return true when messages are the same") { + // given + val jsonAsStream = sampleJsonAsStream() + val event = vesEvent() + val generatedWireProtocolFrame = WireFrameMessage(event.toByteArray()) + val receivedMessageBytes = event.toByteArray() + + givenParsedMessageParameters(MessageParameters(event.commonEventHeader, MessageType.FIXED_PAYLOAD, 1)) + whenever(messageGenerator.createMessageFlux(anyList())).thenReturn(Flux.just(generatedWireProtocolFrame)) + + // when + val result = cut.validate(jsonAsStream, listOf(receivedMessageBytes)).unsafeRunSync() + + // then + assertThat(result).isTrue() + } + + it("should return false when messages differ with payload only") { + // given + val jsonAsStream = sampleJsonAsStream() + val generatedEvent = vesEvent(payload = "payload A") + val receivedEvent = vesEvent(payload = "payload B") + val generatedWireProtocolFrame = WireFrameMessage(generatedEvent.toByteArray()) + val receivedMessageBytes = receivedEvent.toByteArray() + + givenParsedMessageParameters(MessageParameters(generatedEvent.commonEventHeader, MessageType.FIXED_PAYLOAD, 1)) + whenever(messageGenerator.createMessageFlux(anyList())).thenReturn(Flux.just(generatedWireProtocolFrame)) + + // when + val result = cut.validate(jsonAsStream, listOf(receivedMessageBytes)).unsafeRunSync() + + // then + assertThat(result).isFalse() + } + + it("should return false when messages are different") { + // given + val jsonAsStream = sampleJsonAsStream() + val generatedEvent = vesEvent() + val receivedEvent = vesEvent("bbb") + val generatedWireProtocolFrame = WireFrameMessage(generatedEvent.toByteArray()) + val receivedMessageBytes = receivedEvent.toByteArray() + + givenParsedMessageParameters(MessageParameters(generatedEvent.commonEventHeader, MessageType.FIXED_PAYLOAD, 1)) + whenever(messageGenerator.createMessageFlux(anyList())).thenReturn(Flux.just(generatedWireProtocolFrame)) + + // when + val result = cut.validate(jsonAsStream, listOf(receivedMessageBytes)).unsafeRunSync() + + // then + assertThat(result).isFalse() + } + } + } +}) + + + +private const val DUMMY_EVENT_ID = "aaa" +private const val DUMMY_PAYLOAD = "payload" + +private fun vesEvent(eventId: String = DUMMY_EVENT_ID, payload: String = DUMMY_PAYLOAD): VesEvent { + return VesEvent.newBuilder() + .setCommonEventHeader(CommonEventHeader.newBuilder() + .setEventId(eventId)) + .setEventFields(ByteString.copyFrom(payload.toByteArray())) + .build() +} + +private const val sampleJsonArray = """["headersOnly"]""" + +private fun sampleJsonAsStream() = sampleJsonArray.byteInputStream() diff --git a/sources/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/impl/adapters/KafkaSourceTest.kt b/sources/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/impl/adapters/KafkaSourceTest.kt new file mode 100644 index 00000000..de74f628 --- /dev/null +++ b/sources/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/impl/adapters/KafkaSourceTest.kt @@ -0,0 +1,54 @@ +/* + * ============LICENSE_START======================================================= + * dcaegen2-collectors-veshv + * ================================================================================ + * Copyright (C) 2018 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.simulators.dcaeapp.impl.adapters + +import org.apache.kafka.clients.consumer.ConsumerConfig +import org.apache.kafka.common.serialization.ByteArrayDeserializer +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 + +/** + * @author Piotr Jaszczyk <piotr.jaszczyk></piotr.jaszczyk>@nokia.com> + * @since August 2018 + */ +internal class KafkaSourceTest : Spek({ + val servers = "kafka1:9080,kafka2:9080" + val topics = setOf("topic1", "topic2") + + describe("receiver options") { + val options = KafkaSource.createReceiverOptions(servers, topics)!!.toImmutable() + + fun verifyProperty(key: String, expectedValue: Any) { + it("should have $key option set") { + assertThat(options.consumerProperty(key)) + .isEqualTo(expectedValue) + } + } + + verifyProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers) + verifyProperty(ConsumerConfig.CLIENT_ID_CONFIG, "hv-collector-dcae-app-simulator") + verifyProperty(ConsumerConfig.GROUP_ID_CONFIG, "hv-collector-simulators") + verifyProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer::class.java) + verifyProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer::class.java) + verifyProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + } +})
\ No newline at end of file diff --git a/sources/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/impl/config/ArgDcaeAppSimConfigurationTest.kt b/sources/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/impl/config/ArgDcaeAppSimConfigurationTest.kt new file mode 100644 index 00000000..7137fe12 --- /dev/null +++ b/sources/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/impl/config/ArgDcaeAppSimConfigurationTest.kt @@ -0,0 +1,125 @@ +/* + * ============LICENSE_START======================================================= + * dcaegen2-collectors-veshv + * ================================================================================ + * Copyright (C) 2018 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.simulators.dcaeapp.impl.config + +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.given +import org.jetbrains.spek.api.dsl.it +import org.onap.dcae.collectors.veshv.tests.utils.parseExpectingFailure +import org.onap.dcae.collectors.veshv.tests.utils.parseExpectingSuccess +import org.onap.dcae.collectors.veshv.utils.commandline.WrongArgumentError + + +internal class ArgDcaeAppSimConfigurationTest : Spek({ + + lateinit var cut: ArgDcaeAppSimConfiguration + val listenPort = "1234" + val kafkaBootstrapServers = "localhosting:123,localhostinger:12345" + val kafkaTopics = "top1,top2" + + beforeEachTest { + cut = ArgDcaeAppSimConfiguration() + } + + describe("parsing arguments") { + lateinit var result: DcaeAppSimConfiguration + + given("all parameters are present in the long form") { + + beforeEachTest { + result = cut.parseExpectingSuccess( + "--listen-port", listenPort, + "--kafka-bootstrap-servers", kafkaBootstrapServers, + "--kafka-topics", kafkaTopics + ) + } + + it("should set proper port") { + assertThat(result.apiPort).isEqualTo(listenPort.toInt()) + } + + + it("should set proper kafka bootstrap servers") { + assertThat(result.kafkaBootstrapServers).isEqualTo(kafkaBootstrapServers) + } + + it("should set proper kafka topics") { + assertThat(result.kafkaTopics).isEqualTo( + setOf("top1", "top2") + ) + } + } + + given("some parameters are present in the short form") { + + beforeEachTest { + result = cut.parseExpectingSuccess( + "-p", listenPort, + "--kafka-bootstrap-servers", kafkaBootstrapServers, + "-f", kafkaTopics) + } + + it("should set proper port") { + assertThat(result.apiPort).isEqualTo(listenPort.toInt()) + } + + it("should set proper kafka bootstrap servers") { + assertThat(result.kafkaBootstrapServers).isEqualTo(kafkaBootstrapServers) + } + + it("should set proper kafka topics") { + assertThat(result.kafkaTopics).isEqualTo( + setOf("top1", "top2") + ) + } + } + + describe("required parameter is absent") { + given("kafka topics are missing") { + it("should throw exception") { + assertThat(cut.parseExpectingFailure( + "-p", listenPort, + "-s", kafkaBootstrapServers + )).isInstanceOf(WrongArgumentError::class.java) + } + } + + given("kafka bootstrap servers is missing") { + it("should throw exception") { + assertThat(cut.parseExpectingFailure( + "-p", listenPort, + "-f", kafkaTopics + )).isInstanceOf(WrongArgumentError::class.java) + } + } + + given("listen port is missing") { + it("should throw exception") { + assertThat(cut.parseExpectingFailure( + "-p", listenPort, + "-s", kafkaBootstrapServers + )).isInstanceOf(WrongArgumentError::class.java) + } + } + } + } +})
\ No newline at end of file |