Patch #2 : Introduce kafka template for cloud events

[cps.git] / cps-ncmp-service / src / test / groovy / org / onap / cps / ncmp / api / impl / async / CpsAsyncRequestResponseEventIntegrationSpec.groovy

diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/CpsAsyncRequestResponseEventIntegrationSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/CpsAsyncRequestResponseEventIntegrationSpec.groovy
index 31f179a..fe7b3f1 100644 (file)
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/CpsAsyncRequestResponseEventIntegrationSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/CpsAsyncRequestResponseEventIntegrationSpec.groovy
@@ -1,6 +1,6 @@
 /*
  * ============LICENSE_START=======================================================
- * Copyright (c) 2022 Nordix Foundation.
+ * Copyright (c) 2022-2023 Nordix Foundation.
  *  ================================================================================
  *  Licensed under the Apache License, Version 2.0 (the "License");
  *  you may not use this file except in compliance with the License.
@@ -22,8 +22,10 @@ package org.onap.cps.ncmp.api.impl.async
 
 import com.fasterxml.jackson.databind.ObjectMapper
 import org.apache.kafka.clients.consumer.KafkaConsumer
+import org.apache.kafka.common.serialization.StringDeserializer
 import org.mapstruct.factory.Mappers
-import org.onap.cps.ncmp.api.utils.MessagingSpec
+import org.onap.cps.ncmp.api.impl.events.EventsPublisher
+import org.onap.cps.ncmp.api.kafka.MessagingBaseSpec
 import org.onap.cps.ncmp.event.model.DmiAsyncRequestResponseEvent
 import org.onap.cps.ncmp.event.model.NcmpAsyncRequestResponseEvent
 import org.onap.cps.ncmp.utils.TestUtils
@@ -33,17 +35,17 @@ import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.boot.test.context.SpringBootTest
 import org.springframework.test.annotation.DirtiesContext
 import org.testcontainers.spock.Testcontainers
-
 import java.time.Duration
 
-@SpringBootTest(classes = [NcmpAsyncRequestResponseEventProducer, NcmpAsyncRequestResponseEventConsumer, ObjectMapper, JsonObjectMapper])
+@SpringBootTest(classes = [EventsPublisher, NcmpAsyncRequestResponseEventConsumer, ObjectMapper, JsonObjectMapper])
 @Testcontainers
 @DirtiesContext
-class NcmpAsyncRequestResponseEventProducerIntegrationSpec extends MessagingSpec {
+class NcmpAsyncRequestResponseEventProducerIntegrationSpec extends MessagingBaseSpec {
 
     @SpringBean
-    NcmpAsyncRequestResponseEventProducer cpsAsyncRequestResponseEventProducerService =
-        new NcmpAsyncRequestResponseEventProducer(kafkaTemplate);
+    EventsPublisher cpsAsyncRequestResponseEventPublisher =
+        new EventsPublisher<NcmpAsyncRequestResponseEvent>(legacyEventKafkaTemplate, cloudEventKafkaTemplate);
+
 
     @SpringBean
     NcmpAsyncRequestResponseEventMapper ncmpAsyncRequestResponseEventMapper =
@@ -51,24 +53,24 @@ class NcmpAsyncRequestResponseEventProducerIntegrationSpec extends MessagingSpec
 
     @SpringBean
     NcmpAsyncRequestResponseEventConsumer ncmpAsyncRequestResponseEventConsumer =
-            new NcmpAsyncRequestResponseEventConsumer(cpsAsyncRequestResponseEventProducerService,
+            new NcmpAsyncRequestResponseEventConsumer(cpsAsyncRequestResponseEventPublisher,
                     ncmpAsyncRequestResponseEventMapper)
 
     @Autowired
     JsonObjectMapper jsonObjectMapper
 
-    def kafkaConsumer = new KafkaConsumer<>(consumerConfigProperties('test'))
+    def legacyEventKafkaConsumer = new KafkaConsumer<>(eventConsumerConfigProperties('test', StringDeserializer))
 
     def 'Consume and forward valid message'() {
         given: 'consumer has a subscription'
-            kafkaConsumer.subscribe(['test-topic'] as List<String>)
+            legacyEventKafkaConsumer.subscribe(['test-topic'] as List<String>)
         and: 'an event is sent'
             def jsonData = TestUtils.getResourceFileContent('dmiAsyncRequestResponseEvent.json')
             def testEventSent = jsonObjectMapper.convertJsonString(jsonData, DmiAsyncRequestResponseEvent.class)
         when: 'the event is consumed'
             ncmpAsyncRequestResponseEventConsumer.consumeAndForward(testEventSent)
         and: 'the topic is polled'
-            def records = kafkaConsumer.poll(Duration.ofMillis(1500))
+            def records = legacyEventKafkaConsumer.poll(Duration.ofMillis(1500))
         then: 'poll returns one record'
             assert records.size() == 1
         and: 'consumed forwarded event id is the same as sent event id'