ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/selfservice/api/messaginglib/MessagingControllerTest.kt

   1 /*
   2  * Copyright © 2019 Bell Canada
   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 package org.onap.ccsdk.cds.blueprintsprocessor.selfservice.api.messaginglib
  17
  18 import com.fasterxml.jackson.databind.node.ObjectNode
  19 import kotlinx.coroutines.reactive.awaitSingle
  20 import kotlinx.coroutines.runBlocking
  21 import org.apache.commons.lang.builder.ToStringBuilder
  22 import org.apache.kafka.clients.CommonClientConfigs
  23 import org.apache.kafka.clients.consumer.ConsumerConfig
  24 import org.apache.kafka.common.serialization.StringDeserializer
  25 import org.junit.After
  26 import org.junit.Before
  27 import org.junit.Ignore
  28 import org.junit.Test
  29 import org.junit.runner.RunWith
  30 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ActionIdentifiers
  31 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.CommonHeader
  32 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceInput
  33 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.StepData
  34 import org.onap.ccsdk.cds.blueprintsprocessor.selfservice.api.MessagingController
  35 import org.onap.ccsdk.cds.controllerblueprints.core.deleteDir
  36 import org.onap.ccsdk.cds.controllerblueprints.core.utils.JacksonUtils
  37 import org.slf4j.LoggerFactory
  38 import org.springframework.beans.factory.annotation.Autowired
  39 import org.springframework.beans.factory.annotation.Value
  40 import org.springframework.boot.autoconfigure.EnableAutoConfiguration
  41 import org.springframework.boot.autoconfigure.security.SecurityProperties
  42 import org.springframework.boot.test.autoconfigure.web.reactive.WebFluxTest
  43 import org.springframework.context.annotation.Bean
  44 import org.springframework.context.annotation.ComponentScan
  45 import org.springframework.context.annotation.Configuration
  46 import org.springframework.core.io.ByteArrayResource
  47 import org.springframework.http.client.MultipartBodyBuilder
  48 import org.springframework.kafka.annotation.EnableKafka
  49 import org.springframework.kafka.annotation.KafkaListener
  50 import org.springframework.kafka.annotation.PartitionOffset
  51 import org.springframework.kafka.annotation.TopicPartition
  52 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory
  53 import org.springframework.kafka.core.ConsumerFactory
  54 import org.springframework.kafka.core.DefaultKafkaConsumerFactory
  55 import org.springframework.kafka.core.KafkaTemplate
  56 import org.springframework.kafka.support.serializer.JsonDeserializer
  57 import org.springframework.kafka.test.context.EmbeddedKafka
  58 import org.springframework.test.annotation.DirtiesContext
  59 import org.springframework.test.context.ContextConfiguration
  60 import org.springframework.test.context.TestPropertySource
  61 import org.springframework.test.context.junit4.SpringRunner
  62 import org.springframework.test.web.reactive.server.WebTestClient
  63 import org.springframework.test.web.reactive.server.returnResult
  64 import org.springframework.web.reactive.function.BodyInserters
  65 import java.io.File
  66 import java.nio.file.Files
  67 import java.nio.file.Paths
  68 import kotlin.test.assertNotNull
  69 //FIXME("testReceive method is failing in server build, It is not stable, may be timing issue.")
  70 @Ignore
  71 @RunWith(SpringRunner::class)
  72 @EnableAutoConfiguration
  73 @ContextConfiguration(classes = [MessagingControllerTest::class, SecurityProperties::class])
  74 @ComponentScan(basePackages = ["org.onap.ccsdk.cds.blueprintsprocessor", "org.onap.ccsdk.cds.controllerblueprints"])
  75 @TestPropertySource(locations = ["classpath:application-test.properties"])
  76 @DirtiesContext
  77 @EmbeddedKafka(ports = [9092])
  78 @WebFluxTest
  79 class MessagingControllerTest {
  80
  81     private val log = LoggerFactory.getLogger(MessagingControllerTest::class.java)!!
  82
  83     @Autowired
  84     lateinit var controller: MessagingController
  85
  86     @Value("\${blueprintsprocessor.messageclient.self-service-api.consumerTopic}")
  87     lateinit var topicUsedForConsumer: String
  88
  89     @Autowired
  90     lateinit var kt: KafkaTemplate<String, ExecutionServiceInput>
  91
  92     @Autowired
  93     lateinit var webTestClient: WebTestClient
  94
  95     var event: ExecutionServiceInput? = null
  96
  97     @Before
  98     fun setup() {
  99         deleteDir("target", "blueprints")
 100         uploadBluePrint()
 101     }
 102
 103     @After
 104     fun clean() {
 105         deleteDir("target", "blueprints")
 106     }
 107
 108     @Test
 109     fun testReceive() {
 110         val samplePayload = "{\n" +
 111                 "    \"resource-assignment-request\": {\n" +
 112                 "      \"artifact-name\": [\"hostname\"],\n" +
 113                 "      \"store-result\": true,\n" +
 114                 "      \"resource-assignment-properties\" : {\n" +
 115                 "        \"hostname\": \"demo123\"\n" +
 116                 "      }\n" +
 117                 "    }\n" +
 118                 "  }"
 119
 120         kt.defaultTopic = topicUsedForConsumer
 121
 122         val input = ExecutionServiceInput().apply {
 123             commonHeader = CommonHeader().apply {
 124                 originatorId = "1"
 125                 requestId = "1234"
 126                 subRequestId = "1234-1234"
 127             }
 128
 129             actionIdentifiers = ActionIdentifiers().apply {
 130                 blueprintName = "golden"
 131                 blueprintVersion = "1.0.0"
 132                 actionName = "resource-assignment"
 133                 mode = "sync"
 134             }
 135
 136             stepData = StepData().apply {
 137                 name = "resource-assignment"
 138             }
 139
 140             payload = JacksonUtils.jsonNode(samplePayload) as ObjectNode
 141         }
 142
 143         kt.sendDefault(input)
 144         log.info("test-sender sent message='{}'", ToStringBuilder.reflectionToString(input))
 145
 146         Thread.sleep(1000)
 147
 148         assertNotNull(event)
 149     }
 150
 151     @KafkaListener(topicPartitions = [TopicPartition(topic = "\${blueprintsprocessor.messageclient.self-service-api.topic}", partitionOffsets = [PartitionOffset(partition = "0", initialOffset = "0")])])
 152     fun receivedEventFromBluePrintProducer(receivedEvent: ExecutionServiceInput) {
 153         event = receivedEvent
 154     }
 155
 156     private fun uploadBluePrint() {
 157         runBlocking {
 158             val body = MultipartBodyBuilder().apply {
 159                 part("file", object : ByteArrayResource(Files.readAllBytes(loadCbaArchive().toPath())) {
 160                     override fun getFilename(): String {
 161                         return "test-cba.zip"
 162                     }
 163                 })
 164             }.build()
 165
 166             webTestClient
 167                     .post()
 168                     .uri("/api/v1/execution-service/upload")
 169                     .body(BodyInserters.fromMultipartData(body))
 170                     .exchange()
 171                     .expectStatus().isOk
 172                     .returnResult<String>()
 173                     .responseBody
 174                     .awaitSingle()
 175         }
 176     }
 177
 178     private fun loadCbaArchive():File {
 179         return Paths.get("./src/test/resources/cba-for-kafka-integration_enriched.zip").toFile()
 180     }
 181
 182     @Configuration
 183     @EnableKafka
 184     open class ConsumerConfiguration {
 185
 186         @Value("\${blueprintsprocessor.messageclient.self-service-api.bootstrapServers}")
 187         lateinit var bootstrapServers: String
 188
 189         @Value("\${blueprintsprocessor.messageclient.self-service-api.groupId}")
 190         lateinit var groupId:String
 191
 192         @Bean
 193         open fun consumerFactory2(): ConsumerFactory<String, ExecutionServiceInput>? {
 194             val configProperties = hashMapOf<String, Any>()
 195             configProperties[CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG] = bootstrapServers
 196             configProperties[ConsumerConfig.GROUP_ID_CONFIG] = groupId
 197             configProperties[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java.name
 198             configProperties[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = JsonDeserializer::class.java.name
 199             configProperties[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "earliest"
 200             configProperties[ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG] = 1000
 201
 202             return DefaultKafkaConsumerFactory(configProperties, StringDeserializer(),
 203                     JsonDeserializer(ExecutionServiceInput::class.java))
 204         }
 205
 206         @Bean
 207         open fun listenerFactory(): ConcurrentKafkaListenerContainerFactory<String, ExecutionServiceInput> {
 208             val factory = ConcurrentKafkaListenerContainerFactory<String, ExecutionServiceInput>()
 209             factory.consumerFactory = consumerFactory2()
 210             return factory
 211         }
 212     }
 213 }
 214
 215