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.tests.fakes.HVRANMEAS_TOPIC
28 import org.onap.dcae.collectors.veshv.tests.fakes.basicConfiguration
29 import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain
32 * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
35 object VesHvSpecification : Spek({
36 describe("VES High Volume Collector") {
37 system("should handle multiple HV RAN events") { sut ->
38 sut.configurationProvider.updateConfiguration(basicConfiguration)
39 val messages = sut.handleConnection(vesMessage(Domain.HVRANMEAS), vesMessage(Domain.HVRANMEAS))
42 .describedAs("should send all events")
47 describe("Memory management") {
49 system("should release memory for each handled and dropped message") { sut ->
50 sut.configurationProvider.updateConfiguration(basicConfiguration)
51 val validMessage = vesMessage(Domain.HVRANMEAS)
52 val msgWithInvalidDomain = vesMessage(Domain.OTHER)
53 val msgWithInvalidFrame = invalidWireFrame()
54 val expectedRefCnt = 0
56 val (handledEvents, exception) = sut.handleConnectionReturningError(
57 validMessage, msgWithInvalidDomain, msgWithInvalidFrame)
59 assertThat(handledEvents).hasSize(1)
60 assertThat(exception).isNull()
62 assertThat(validMessage.refCnt())
63 .describedAs("handled message should be released")
64 .isEqualTo(expectedRefCnt)
65 assertThat(msgWithInvalidDomain.refCnt())
66 .describedAs("message with invalid domain should be released")
67 .isEqualTo(expectedRefCnt)
68 assertThat(msgWithInvalidFrame.refCnt())
69 .describedAs("message with invalid frame should be released")
70 .isEqualTo(expectedRefCnt)
74 system("should release memory for each message with invalid payload") { sut ->
75 sut.configurationProvider.updateConfiguration(basicConfiguration)
76 val validMessage = vesMessage(Domain.HVRANMEAS)
77 val msgWithInvalidPayload = invalidVesMessage()
78 val expectedRefCnt = 0
80 val (handledEvents, exception) = sut.handleConnectionReturningError(
81 validMessage, msgWithInvalidPayload)
83 assertThat(handledEvents).hasSize(1)
84 assertThat(exception?.cause).isInstanceOf(InvalidProtocolBufferException::class.java)
86 assertThat(validMessage.refCnt())
87 .describedAs("handled message should be released")
88 .isEqualTo(expectedRefCnt)
89 assertThat(msgWithInvalidPayload.refCnt())
90 .describedAs("message with invalid payload should be released")
91 .isEqualTo(expectedRefCnt)
95 system("should release memory for each message with garbage frame") { sut ->
96 sut.configurationProvider.updateConfiguration(basicConfiguration)
97 val validMessage = vesMessage(Domain.HVRANMEAS)
98 val msgWithGarbageFrame = garbageFrame()
99 val expectedRefCnt = 0
101 val (handledEvents, exception) = sut.handleConnectionReturningError(
102 validMessage, msgWithGarbageFrame)
104 assertThat(handledEvents).hasSize(1)
105 assertThat(exception?.cause)
106 .isInstanceOf(InvalidWireFrameMarkerException::class.java)
108 assertThat(validMessage.refCnt())
109 .describedAs("handled message should be released")
110 .isEqualTo(expectedRefCnt)
111 assertThat(msgWithGarbageFrame.refCnt())
112 .describedAs("message with garbage frame should be released")
113 .isEqualTo(expectedRefCnt)
118 describe("message routing") {
119 system("should direct message to a topic by means of routing configuration") { sut ->
120 sut.configurationProvider.updateConfiguration(basicConfiguration)
122 val messages = sut.handleConnection(vesMessage(Domain.HVRANMEAS))
123 assertThat(messages).describedAs("number of routed messages").hasSize(1)
125 val msg = messages[0]
126 assertThat(msg.topic).describedAs("routed message topic").isEqualTo(HVRANMEAS_TOPIC)
127 assertThat(msg.partition).describedAs("routed message partition").isEqualTo(0)
130 system("should drop message if route was not found") { sut ->
131 sut.configurationProvider.updateConfiguration(basicConfiguration)
132 val messages = sut.handleConnection(
133 vesMessage(Domain.OTHER, "first"),
134 vesMessage(Domain.HVRANMEAS, "second"),
135 vesMessage(Domain.HEARTBEAT, "third"))
137 assertThat(messages).describedAs("number of routed messages").hasSize(1)
139 val msg = messages[0]
140 assertThat(msg.topic).describedAs("routed message topic").isEqualTo(HVRANMEAS_TOPIC)
141 assertThat(msg.message.header.eventId).describedAs("routed message eventId").isEqualTo("second")