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.tests.component
22 import com.google.protobuf.InvalidProtocolBufferException
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.onap.dcae.collectors.veshv.domain.exceptions.InvalidWireFrameMarkerException
27 import org.onap.dcae.collectors.veshv.domain.exceptions.WireFrameDecodingException
28 import org.onap.dcae.collectors.veshv.tests.fakes.HVRANMEAS_TOPIC
29 import org.onap.dcae.collectors.veshv.tests.fakes.basicConfiguration
30 import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain
33 * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
36 object VesHvSpecification : Spek({
37 describe("VES High Volume Collector") {
38 system("should handle multiple HV RAN events") { sut ->
39 sut.configurationProvider.updateConfiguration(basicConfiguration)
40 val messages = sut.handleConnection(vesMessage(Domain.HVRANMEAS), vesMessage(Domain.HVRANMEAS))
43 .describedAs("should send all events")
48 describe("Memory management") {
50 system("should release memory for each handled and dropped message") { sut ->
51 sut.configurationProvider.updateConfiguration(basicConfiguration)
52 val validMessage = vesMessage(Domain.HVRANMEAS)
53 val msgWithInvalidDomain = vesMessage(Domain.OTHER)
54 val msgWithInvalidFrame = invalidWireFrame()
55 val expectedRefCnt = 0
57 val (handledEvents, exception) = sut.handleConnectionReturningError(
58 validMessage, msgWithInvalidDomain, msgWithInvalidFrame)
60 assertThat(handledEvents).hasSize(1)
61 assertThat(exception).isNull()
63 assertThat(validMessage.refCnt())
64 .describedAs("handled message should be released")
65 .isEqualTo(expectedRefCnt)
66 assertThat(msgWithInvalidDomain.refCnt())
67 .describedAs("message with invalid domain should be released")
68 .isEqualTo(expectedRefCnt)
69 assertThat(msgWithInvalidFrame.refCnt())
70 .describedAs("message with invalid frame should be released")
71 .isEqualTo(expectedRefCnt)
75 system("should release memory for each message with invalid payload") { sut ->
76 sut.configurationProvider.updateConfiguration(basicConfiguration)
77 val validMessage = vesMessage(Domain.HVRANMEAS)
78 val msgWithInvalidPayload = invalidVesMessage()
79 val expectedRefCnt = 0
81 val (handledEvents, exception) = sut.handleConnectionReturningError(
82 validMessage, msgWithInvalidPayload)
84 assertThat(handledEvents).hasSize(1)
85 assertThat(exception?.cause).isInstanceOf(InvalidProtocolBufferException::class.java)
87 assertThat(validMessage.refCnt())
88 .describedAs("handled message should be released")
89 .isEqualTo(expectedRefCnt)
90 assertThat(msgWithInvalidPayload.refCnt())
91 .describedAs("message with invalid payload should be released")
92 .isEqualTo(expectedRefCnt)
96 system("should release memory for each message with garbage frame") { sut ->
97 sut.configurationProvider.updateConfiguration(basicConfiguration)
98 val validMessage = vesMessage(Domain.HVRANMEAS)
99 val msgWithGarbageFrame = garbageFrame()
100 val expectedRefCnt = 0
102 val (handledEvents, exception) = sut.handleConnectionReturningError(
103 validMessage, msgWithGarbageFrame)
105 assertThat(handledEvents).hasSize(1)
106 assertThat(exception?.cause)
107 .isInstanceOf(InvalidWireFrameMarkerException::class.java)
109 assertThat(validMessage.refCnt())
110 .describedAs("handled message should be released")
111 .isEqualTo(expectedRefCnt)
112 assertThat(msgWithGarbageFrame.refCnt())
113 .describedAs("message with garbage frame should be released")
114 .isEqualTo(expectedRefCnt)
119 describe("message routing") {
120 system("should direct message to a topic by means of routing configuration") { sut ->
121 sut.configurationProvider.updateConfiguration(basicConfiguration)
123 val messages = sut.handleConnection(vesMessage(Domain.HVRANMEAS))
124 assertThat(messages).describedAs("number of routed messages").hasSize(1)
126 val msg = messages[0]
127 assertThat(msg.topic).describedAs("routed message topic").isEqualTo(HVRANMEAS_TOPIC)
128 assertThat(msg.partition).describedAs("routed message partition").isEqualTo(1)
131 system("should drop message if route was not found") { sut ->
132 sut.configurationProvider.updateConfiguration(basicConfiguration)
133 val messages = sut.handleConnection(
134 vesMessage(Domain.OTHER, "first"),
135 vesMessage(Domain.HVRANMEAS, "second"),
136 vesMessage(Domain.HEARTBEAT, "third"))
138 assertThat(messages).describedAs("number of routed messages").hasSize(1)
140 val msg = messages[0]
141 assertThat(msg.topic).describedAs("routed message topic").isEqualTo(HVRANMEAS_TOPIC)
142 assertThat(msg.message.header.eventId).describedAs("routed message eventId").isEqualTo("second")