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 @RunWith(SpringRunner::class)
  48 @DataJpaTest
  49 @DirtiesContext
  50 @ContextConfiguration(
  51     classes = [BluePrintMessageLibConfiguration::class,
  52         BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class,
  53         MessagePrioritizationConfiguration::class, TestDatabaseConfiguration::class]
  54 )
  55 @TestPropertySource(
  56     properties =
  57     [
  58         "spring.jpa.show-sql=true",
  59         "spring.jpa.properties.hibernate.show_sql=true",
  60         "spring.jpa.hibernate.naming.physical-strategy=org.hibernate.boot.model.naming.PhysicalNamingStrategyStandardImpl",
  61
  62         "blueprintsprocessor.messageconsumer.prioritize-input.type=kafka-streams-basic-auth",
  63         "blueprintsprocessor.messageconsumer.prioritize-input.bootstrapServers=127.0.0.1:9092",
  64         "blueprintsprocessor.messageconsumer.prioritize-input.applicationId=test-prioritize-application",
  65         "blueprintsprocessor.messageconsumer.prioritize-input.topic=prioritize-input-topic",
  66
  67         // To send initial test message
  68         "blueprintsprocessor.messageproducer.prioritize-input.type=kafka-basic-auth",
  69         "blueprintsprocessor.messageproducer.prioritize-input.bootstrapServers=127.0.0.1:9092",
  70         "blueprintsprocessor.messageproducer.prioritize-input.topic=prioritize-input-topic"
  71     ]
  72 )
  73 open class MessagePrioritizationConsumerTest {
  74
  75     @Autowired
  76     lateinit var applicationContext: ApplicationContext
  77
  78     @Autowired
  79     lateinit var prioritizationMessageRepository: PrioritizationMessageRepository
  80
  81     @Autowired
  82     lateinit var bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService
  83
  84     @Before
  85     fun setup() {
  86         BluePrintDependencyService.inject(applicationContext)
  87     }
  88
  89     @Test
  90     fun testBluePrintKafkaJDBCKeyStore() {
  91         runBlocking {
  92             assertNotNull(prioritizationMessageRepository, "failed to get prioritizationMessageRepository")
  93
  94             val messagePrioritizationService: MessagePrioritizationStateService = BluePrintDependencyService
  95                 .instance(MessagePrioritizationStateService::class)
  96             assertNotNull(messagePrioritizationService, "failed to get messagePrioritizationService")
  97
  98             MessagePrioritizationSample.sampleMessages(MessageState.NEW.name, 1).forEach {
  99                 val message = messagePrioritizationService.saveMessage(it)
 100                 val repoResult = messagePrioritizationService.getMessage(message.id)
 101                 assertNotNull(repoResult, "failed to get inserted message.")
 102             }
 103         }
 104     }
 105
 106     @Test
 107     fun testStartConsuming() {
 108         runBlocking {
 109             val configuration = MessagePrioritizationSample.samplePrioritizationConfiguration()
 110
 111             val streamingConsumerService = bluePrintMessageLibPropertyService
 112                 .blueprintMessageConsumerService(configuration.inputTopicSelector)
 113             assertNotNull(streamingConsumerService, "failed to get blueprintMessageConsumerService")
 114
 115             val spyStreamingConsumerService = spyk(streamingConsumerService)
 116             coEvery { spyStreamingConsumerService.consume(any(), any()) } returns Unit
 117             coEvery { spyStreamingConsumerService.shutDown() } returns Unit
 118             val messagePrioritizationConsumer = MessagePrioritizationConsumer(bluePrintMessageLibPropertyService)
 119             val spyMessagePrioritizationConsumer = spyk(messagePrioritizationConsumer)
 120
 121             // Test Topology
 122             val kafkaStreamConsumerFunction = spyMessagePrioritizationConsumer.kafkaStreamConsumerFunction(configuration)
 123             val messageConsumerProperties = bluePrintMessageLibPropertyService
 124                 .messageConsumerProperties("blueprintsprocessor.messageconsumer.prioritize-input")
 125             val topology = kafkaStreamConsumerFunction.createTopology(messageConsumerProperties, null)
 126             assertNotNull(topology, "failed to get create topology")
 127
 128             every { spyMessagePrioritizationConsumer.consumerService(any()) } returns spyStreamingConsumerService
 129             spyMessagePrioritizationConsumer.startConsuming(configuration)
 130             spyMessagePrioritizationConsumer.shutDown()
 131         }
 132     }
 133
 134     /** Integration Kafka Testing, Enable and use this test case only for local desktop testing with real kafka broker */
 135     // @Test
 136     fun testMessagePrioritizationConsumer() {
 137         runBlocking {
 138             val messagePrioritizationConsumer = MessagePrioritizationConsumer(bluePrintMessageLibPropertyService)
 139             messagePrioritizationConsumer.startConsuming(MessagePrioritizationSample.samplePrioritizationConfiguration())
 140
 141             /** Send sample message with every 1 sec */
 142             val blueprintMessageProducerService = bluePrintMessageLibPropertyService
 143                 .blueprintMessageProducerService("prioritize-input") as KafkaBasicAuthMessageProducerService
 144             launch {
 145                 MessagePrioritizationSample.sampleMessages(MessageState.NEW.name, 2).forEach {
 146                     delay(100)
 147                     val headers: MutableMap<String, String> = hashMapOf()
 148                     headers["id"] = it.id
 149                     blueprintMessageProducerService.sendMessageNB(
 150                         message = it.asJsonString(false),
 151                         headers = headers
 152                     )
 153                 }
 154
 155                 MessagePrioritizationSample
 156                     .sampleMessageWithSameCorrelation("same-group", MessageState.NEW.name, 2)
 157                     .forEach {
 158                         delay(100)
 159                         val headers: MutableMap<String, String> = hashMapOf()
 160                         headers["id"] = it.id
 161                         blueprintMessageProducerService.sendMessageNB(
 162                             message = it.asJsonString(false),
 163                             headers = headers
 164                         )
 165                     }
 166
 167                 MessagePrioritizationSample
 168                     .sampleMessageWithDifferentTypeSameCorrelation("group-typed", MessageState.NEW.name, 3)
 169                     .forEach {
 170                         delay(2000)
 171                         val headers: MutableMap<String, String> = hashMapOf()
 172                         headers["id"] = it.id
 173                         blueprintMessageProducerService.sendMessageNB(
 174                             message = it.asJsonString(false),
 175                             headers = headers
 176                         )
 177                     }
 178             }
 179             delay(10000)
 180             messagePrioritizationConsumer.shutDown()
 181         }
 182     }
 183 }