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.ves.message.generator.impl
22 import com.google.protobuf.ByteString
23 import com.google.protobuf.InvalidProtocolBufferException
24 import org.assertj.core.api.Assertions.assertThat
25 import org.assertj.core.api.Assertions.assertThatExceptionOfType
26 import org.jetbrains.spek.api.Spek
27 import org.jetbrains.spek.api.dsl.describe
28 import org.jetbrains.spek.api.dsl.given
29 import org.jetbrains.spek.api.dsl.it
30 import org.jetbrains.spek.api.dsl.on
31 import org.onap.dcae.collectors.veshv.domain.ByteData
32 import org.onap.dcae.collectors.veshv.domain.WireFrameMessage
33 import org.onap.dcae.collectors.veshv.domain.VesEventDomain.PERF3GPP
34 import org.onap.dcae.collectors.veshv.domain.VesEventDomain.FAULT
35 import org.onap.dcae.collectors.veshv.domain.VesEventDomain.HEARTBEAT
36 import org.onap.dcae.collectors.veshv.tests.utils.commonHeader
37 import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageGenerator
38 import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageParameters
39 import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageType
40 import org.onap.ves.VesEventOuterClass.CommonEventHeader
41 import org.onap.ves.VesEventOuterClass.VesEvent
42 import reactor.test.test
45 * @author Jakub Dudycz <jakub.dudycz@nokia.com>
48 object MessageGeneratorImplTest : Spek({
49 describe("message factory") {
50 val maxPayloadSizeBytes = 1024
51 val generator = MessageGeneratorImpl(PayloadGenerator(), maxPayloadSizeBytes)
52 given("single message parameters") {
54 on("messages amount not specified in parameters") {
55 it("should create infinite flux") {
58 .createMessageFlux(listOf(MessageParameters(
59 commonHeader(PERF3GPP),
64 .expectNextCount(limit)
69 on("messages amount = 0 specified in parameters") {
70 it("should create empty message flux") {
72 .createMessageFlux(listOf(MessageParameters(
73 commonHeader(PERF3GPP),
82 on("messages amount specified in parameters") {
83 it("should create message flux of specified size") {
85 .createMessageFlux(listOf(MessageParameters(
86 commonHeader(PERF3GPP),
96 on("message type requesting valid message") {
97 it("should create flux of valid messages with given domain") {
99 .createMessageFlux(listOf(MessageParameters(
106 assertThat(it.isValid()).isTrue()
107 assertThat(it.payloadSize).isLessThan(maxPayloadSizeBytes)
108 assertThat(extractCommonEventHeader(it.payload).domain).isEqualTo(FAULT.domainName)
114 on("message type requesting too big payload") {
115 it("should create flux of messages with given domain and payload exceeding threshold") {
118 .createMessageFlux(listOf(MessageParameters(
119 commonHeader(PERF3GPP),
120 MessageType.TOO_BIG_PAYLOAD,
125 assertThat(it.isValid()).isTrue()
126 assertThat(it.payloadSize).isGreaterThan(maxPayloadSizeBytes)
127 assertThat(extractCommonEventHeader(it.payload).domain).isEqualTo(PERF3GPP.domainName)
133 on("message type requesting invalid GPB data ") {
134 it("should create flux of messages with invalid payload") {
136 .createMessageFlux(listOf(MessageParameters(
137 commonHeader(PERF3GPP),
138 MessageType.INVALID_GPB_DATA,
143 assertThat(it.isValid()).isTrue()
144 assertThat(it.payloadSize).isLessThan(maxPayloadSizeBytes)
145 assertThatExceptionOfType(InvalidProtocolBufferException::class.java)
146 .isThrownBy { extractCommonEventHeader(it.payload) }
152 on("message type requesting invalid wire frame ") {
153 it("should create flux of messages with invalid version") {
155 .createMessageFlux(listOf(MessageParameters(
156 commonHeader(PERF3GPP),
157 MessageType.INVALID_WIRE_FRAME,
162 assertThat(it.isValid()).isFalse()
163 assertThat(it.payloadSize).isLessThan(maxPayloadSizeBytes)
164 assertThat(extractCommonEventHeader(it.payload).domain).isEqualTo(PERF3GPP.domainName)
165 assertThat(it.versionMajor).isNotEqualTo(WireFrameMessage.SUPPORTED_VERSION_MINOR)
171 on("message type requesting fixed payload") {
172 it("should create flux of valid messages with fixed payload") {
174 .createMessageFlux(listOf(MessageParameters(
176 MessageType.FIXED_PAYLOAD,
181 assertThat(it.isValid()).isTrue()
182 assertThat(it.payloadSize).isLessThan(maxPayloadSizeBytes)
183 assertThat(extractEventFields(it.payload).size()).isEqualTo(MessageGenerator.FIXED_PAYLOAD_SIZE)
184 assertThat(extractCommonEventHeader(it.payload).domain).isEqualTo(FAULT.domainName)
190 given("list of message parameters") {
191 it("should create concatenated flux of messages") {
192 val singleFluxSize = 5L
193 val messageParameters = listOf(
194 MessageParameters(commonHeader(PERF3GPP), MessageType.VALID, singleFluxSize),
195 MessageParameters(commonHeader(FAULT), MessageType.TOO_BIG_PAYLOAD, singleFluxSize),
196 MessageParameters(commonHeader(HEARTBEAT), MessageType.VALID, singleFluxSize)
198 generator.createMessageFlux(messageParameters)
201 assertThat(it.payloadSize).isLessThan(maxPayloadSizeBytes)
202 assertThat(extractCommonEventHeader(it.payload).domain).isEqualTo(PERF3GPP.domainName)
204 .expectNextCount(singleFluxSize - 1)
206 assertThat(it.payloadSize).isGreaterThan(maxPayloadSizeBytes)
207 assertThat(extractCommonEventHeader(it.payload).domain).isEqualTo(FAULT.domainName)
209 .expectNextCount(singleFluxSize - 1)
211 assertThat(it.payloadSize).isLessThan(maxPayloadSizeBytes)
212 assertThat(extractCommonEventHeader(it.payload).domain).isEqualTo(HEARTBEAT.domainName)
214 .expectNextCount(singleFluxSize - 1)
221 fun extractCommonEventHeader(bytes: ByteData): CommonEventHeader =
222 VesEvent.parseFrom(bytes.unsafeAsArray()).commonEventHeader
225 fun extractEventFields(bytes: ByteData): ByteString =
226 VesEvent.parseFrom(bytes.unsafeAsArray()).eventFields