2 * ============LICENSE_START=======================================================
3 * dcaegen2-collectors-veshv
4 * ================================================================================
5 * Copyright (C) 2018 NOKIA
6 * ================================================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ============LICENSE_END=========================================================
20 package org.onap.dcae.collectors.veshv.simulators.dcaeapp.impl
22 import org.apache.kafka.clients.consumer.ConsumerRecord
23 import org.assertj.core.api.Assertions.assertThat
24 import org.jetbrains.spek.api.Spek
25 import org.jetbrains.spek.api.dsl.describe
26 import org.jetbrains.spek.api.dsl.it
27 import reactor.kafka.receiver.ReceiverRecord
31 * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
34 internal class ConsumerTest : Spek({
36 lateinit var cut: Consumer
42 describe("Consumer which holds the state of received Kafka records") {
43 it("should contain empty state in the beginning") {
48 val value = byteArrayOf(2)
51 cut.update(receiverRecord(
58 it("should contain one message if it was updated once") {
59 assertState(cut, value)
62 it("should contain empty state message if it was reset after update") {
63 cut.reset().unsafeRunSync()
70 fun assertEmptyState(cut: Consumer) {
74 fun assertState(cut: Consumer, vararg values: ByteArray) {
75 assertThat(cut.currentState().consumedMessages)
76 .containsOnly(*values)
77 assertThat(cut.currentState().messagesCount)
78 .isEqualTo(values.size)
81 fun receiverRecord(topic: String, key: ByteArray, value: ByteArray) =
82 ReceiverRecord(ConsumerRecord(topic, 1, 100L, key, value), null)