gerrit.onap Code Review - ccsdk/cds.git/blob

   1 /*
   2  * Copyright © 2018-2019 AT&T Intellectual Property.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization
  18
  19 import io.mockk.coEvery
  20 import io.mockk.every
  21 import io.mockk.spyk
  22 import kotlinx.coroutines.delay
  23 import kotlinx.coroutines.launch
  24 import kotlinx.coroutines.runBlocking
  25 import org.junit.Before
  26 import org.junit.runner.RunWith
  27 import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
  28 import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
  29 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.PrioritizationMessageRepository
  30 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service.MessagePrioritizationStateService
  31 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.utils.MessagePrioritizationSample
  32 import org.onap.ccsdk.cds.blueprintsprocessor.message.BluePrintMessageLibConfiguration
  33 import org.onap.ccsdk.cds.blueprintsprocessor.message.service.BluePrintMessageLibPropertyService
  34 import org.onap.ccsdk.cds.blueprintsprocessor.message.service.KafkaBasicAuthMessageProducerService
  35 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonString
  36 import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService
  37 import org.springframework.beans.factory.annotation.Autowired
  38 import org.springframework.boot.test.autoconfigure.orm.jpa.DataJpaTest
  39 import org.springframework.context.ApplicationContext
  40 import org.springframework.test.annotation.DirtiesContext
  41 import org.springframework.test.context.ContextConfiguration
  42 import org.springframework.test.context.TestPropertySource
  43 import org.springframework.test.context.junit4.SpringRunner
  44 import kotlin.test.Test
  45 import kotlin.test.assertNotNull
  46
  47
  48 @RunWith(SpringRunner::class)
  49 @DataJpaTest
  50 @DirtiesContext
  51 @ContextConfiguration(classes = [BluePrintMessageLibConfiguration::class,
  52     BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class,
  53     MessagePrioritizationConfiguration::class, TestDatabaseConfiguration::class])
  54 @TestPropertySource(properties =
  55 [
  56     "spring.jpa.show-sql=true",
  57     "spring.jpa.properties.hibernate.show_sql=true",
  58     "spring.jpa.hibernate.naming.physical-strategy=org.hibernate.boot.model.naming.PhysicalNamingStrategyStandardImpl",
  59
  60     "blueprintsprocessor.messageconsumer.prioritize-input.type=kafka-streams-basic-auth",
  61     "blueprintsprocessor.messageconsumer.prioritize-input.bootstrapServers=127.0.0.1:9092",
  62     "blueprintsprocessor.messageconsumer.prioritize-input.applicationId=test-prioritize-application",
  63     "blueprintsprocessor.messageconsumer.prioritize-input.topic=prioritize-input-topic",
  64
  65     // To send initial test message
  66     "blueprintsprocessor.messageproducer.prioritize-input.type=kafka-basic-auth",
  67     "blueprintsprocessor.messageproducer.prioritize-input.bootstrapServers=127.0.0.1:9092",
  68     "blueprintsprocessor.messageproducer.prioritize-input.topic=prioritize-input-topic"
  69 ])
  70 open class MessagePrioritizationConsumerTest {
  71
  72     @Autowired
  73     lateinit var applicationContext: ApplicationContext
  74
  75     @Autowired
  76     lateinit var prioritizationMessageRepository: PrioritizationMessageRepository
  77
  78     @Autowired
  79     lateinit var bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService
  80
  81     @Before
  82     fun setup() {
  83         BluePrintDependencyService.inject(applicationContext)
  84     }
  85
  86     @Test
  87     fun testBluePrintKafkaJDBCKeyStore() {
  88         runBlocking {
  89             assertNotNull(prioritizationMessageRepository, "failed to get prioritizationMessageRepository")
  90
  91             val messagePrioritizationService: MessagePrioritizationStateService = BluePrintDependencyService
  92                     .instance(MessagePrioritizationStateService::class)
  93             assertNotNull(messagePrioritizationService, "failed to get messagePrioritizationService")
  94
  95             MessagePrioritizationSample.sampleMessages(MessageState.NEW.name, 1).forEach {
  96                 val message = messagePrioritizationService.saveMessage(it)
  97                 val repoResult = messagePrioritizationService.getMessage(message.id)
  98                 assertNotNull(repoResult, "failed to get inserted message.")
  99             }
 100         }
 101     }
 102
 103     @Test
 104     fun testStartConsuming() {
 105         runBlocking {
 106             val configuration = MessagePrioritizationSample.samplePrioritizationConfiguration()
 107
 108             val streamingConsumerService = bluePrintMessageLibPropertyService
 109                     .blueprintMessageConsumerService(configuration.inputTopicSelector)
 110             assertNotNull(streamingConsumerService, "failed to get blueprintMessageConsumerService")
 111
 112             val spyStreamingConsumerService = spyk(streamingConsumerService)
 113             coEvery { spyStreamingConsumerService.consume(any(), any()) } returns Unit
 114             coEvery { spyStreamingConsumerService.shutDown() } returns Unit
 115             val messagePrioritizationConsumer = MessagePrioritizationConsumer(bluePrintMessageLibPropertyService)
 116             val spyMessagePrioritizationConsumer = spyk(messagePrioritizationConsumer)
 117
 118
 119             // Test Topology
 120             val kafkaStreamConsumerFunction = spyMessagePrioritizationConsumer.kafkaStreamConsumerFunction(configuration)
 121             val messageConsumerProperties = bluePrintMessageLibPropertyService
 122                     .messageConsumerProperties("blueprintsprocessor.messageconsumer.prioritize-input")
 123             val topology = kafkaStreamConsumerFunction.createTopology(messageConsumerProperties, null)
 124             assertNotNull(topology, "failed to get create topology")
 125
 126             every { spyMessagePrioritizationConsumer.consumerService(any()) } returns spyStreamingConsumerService
 127             spyMessagePrioritizationConsumer.startConsuming(configuration)
 128             spyMessagePrioritizationConsumer.shutDown()
 129         }
 130     }
 131
 132     /** Integration Kafka Testing, Enable and use this test case only for local desktop testing with real kafka broker */
 133     //@Test
 134     fun testMessagePrioritizationConsumer() {
 135         runBlocking {
 136             val messagePrioritizationConsumer = MessagePrioritizationConsumer(bluePrintMessageLibPropertyService)
 137             messagePrioritizationConsumer.startConsuming(MessagePrioritizationSample.samplePrioritizationConfiguration())
 138
 139             /** Send sample message with every 1 sec */
 140             val blueprintMessageProducerService = bluePrintMessageLibPropertyService
 141                     .blueprintMessageProducerService("prioritize-input") as KafkaBasicAuthMessageProducerService
 142             launch {
 143              MessagePrioritizationSample.sampleMessages(MessageState.NEW.name, 2).forEach {
 144                     delay(100)
 145                     val headers: MutableMap<String, String> = hashMapOf()
 146                     headers["id"] = it.id
 147                     blueprintMessageProducerService.sendMessageNB(message = it.asJsonString(false),
 148                             headers = headers)
 149                 }
 150
 151                 MessagePrioritizationSample
 152                         .sampleMessageWithSameCorrelation("same-group", MessageState.NEW.name, 2)
 153                         .forEach {
 154                             delay(100)
 155                             val headers: MutableMap<String, String> = hashMapOf()
 156                             headers["id"] = it.id
 157                             blueprintMessageProducerService.sendMessageNB(message = it.asJsonString(false),
 158                                     headers = headers)
 159                         }
 160
 161                 MessagePrioritizationSample
 162                         .sampleMessageWithDifferentTypeSameCorrelation("group-typed", MessageState.NEW.name, 3)
 163                         .forEach {
 164                             delay(2000)
 165                             val headers: MutableMap<String, String> = hashMapOf()
 166                             headers["id"] = it.id
 167                             blueprintMessageProducerService.sendMessageNB(message = it.asJsonString(false),
 168                                     headers = headers)
 169                         }
 170             }
 171             delay(10000)
 172             messagePrioritizationConsumer.shutDown()
 173         }
 174     }
 175 }