hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/VesHvSpecification.kt

   1 /*
   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
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  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=========================================================
  19  */
  20 package org.onap.dcae.collectors.veshv.tests.component
  21
  22 import org.assertj.core.api.Assertions.assertThat
  23 import org.jetbrains.spek.api.Spek
  24 import org.jetbrains.spek.api.dsl.describe
  25 import org.onap.dcae.collectors.veshv.tests.fakes.HVRANMEAS_TOPIC
  26 import org.onap.dcae.collectors.veshv.tests.fakes.basicConfiguration
  27 import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain
  28
  29 /**
  30  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  31  * @since May 2018
  32  */
  33 object VesHvSpecification : Spek({
  34     describe("VES High Volume Collector") {
  35         system("should handle multiple HV RAN events") { sut ->
  36             sut.configurationProvider.updateConfiguration(basicConfiguration)
  37             val messages = sut.handleConnection(vesMessage(Domain.HVRANMEAS), vesMessage(Domain.HVRANMEAS))
  38
  39             assertThat(messages)
  40                     .describedAs("should send all events")
  41                     .hasSize(2)
  42         }
  43
  44         system("should release memory for each incoming message") { sut ->
  45             sut.configurationProvider.updateConfiguration(basicConfiguration)
  46             val msgWithInvalidDomain = vesMessage(Domain.OTHER)
  47             val msgWithInvalidPayload = invalidVesMessage()
  48             val msgWithInvalidFrame = invalidWireFrame()
  49             val validMessage = vesMessage(Domain.HVRANMEAS)
  50             val refCntBeforeSending = msgWithInvalidDomain.refCnt()
  51
  52             sut.handleConnection(msgWithInvalidDomain, msgWithInvalidPayload, msgWithInvalidFrame, validMessage)
  53
  54             assertThat(msgWithInvalidDomain.refCnt())
  55                     .describedAs("message with invalid domain should be released")
  56                     .isEqualTo(refCntBeforeSending)
  57             assertThat(msgWithInvalidPayload.refCnt())
  58                     .describedAs("message with invalid payload should be released")
  59                     .isEqualTo(refCntBeforeSending)
  60             assertThat(msgWithInvalidFrame.refCnt())
  61                     .describedAs("message with invalid frame should be released")
  62                     .isEqualTo(refCntBeforeSending)
  63             assertThat(validMessage.refCnt())
  64                     .describedAs("handled message should be released")
  65                     .isEqualTo(refCntBeforeSending)
  66         }
  67     }
  68
  69     describe("message routing") {
  70         system("should direct message to a topic by means of routing configuration") { sut ->
  71             sut.configurationProvider.updateConfiguration(basicConfiguration)
  72
  73             val messages = sut.handleConnection(vesMessage(Domain.HVRANMEAS))
  74             assertThat(messages).describedAs("number of routed messages").hasSize(1)
  75
  76             val msg = messages[0]
  77             assertThat(msg.topic).describedAs("routed message topic").isEqualTo(HVRANMEAS_TOPIC)
  78             assertThat(msg.partition).describedAs("routed message partition").isEqualTo(1)
  79         }
  80
  81         system("should drop message if route was not found") { sut ->
  82             sut.configurationProvider.updateConfiguration(basicConfiguration)
  83             val messages = sut.handleConnection(
  84                     vesMessage(Domain.OTHER, "first"),
  85                     vesMessage(Domain.HVRANMEAS, "second"),
  86                     vesMessage(Domain.HEARTBEAT, "third"))
  87
  88             assertThat(messages).describedAs("number of routed messages").hasSize(1)
  89
  90             val msg = messages[0]
  91             assertThat(msg.topic).describedAs("routed message topic").isEqualTo(HVRANMEAS_TOPIC)
  92             assertThat(msg.message.header.eventId).describedAs("routed message eventId").isEqualTo("second")
  93         }
  94     }
  95 })