Merge "Patch # 1: Data operation response event (NCMP → Client App) to comply with CloudEvents"

1 files changed, 18 insertions, 20 deletions

diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/NcmpAsyncBatchEventConsumerSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/NcmpAsyncDataOperationEventConsumerSpec.groovy
index 02071cd8cf..d9b9ce6db0 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/NcmpAsyncBatchEventConsumerSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/NcmpAsyncDataOperationEventConsumerSpec.groovy
@@ -28,7 +28,7 @@ import org.apache.kafka.common.header.internals.RecordHeader
 import org.apache.kafka.common.serialization.StringDeserializer
 import org.onap.cps.ncmp.api.impl.events.EventsPublisher
 import org.onap.cps.ncmp.api.kafka.MessagingBaseSpec
-import org.onap.cps.ncmp.events.async.BatchDataResponseEventV1
+import org.onap.cps.ncmp.events.async1_0_0.DataOperationEvent
 import org.onap.cps.ncmp.utils.TestUtils
 import org.onap.cps.utils.JsonObjectMapper
 import org.spockframework.spring.SpringBean
@@ -37,43 +37,41 @@ import org.springframework.boot.test.context.SpringBootTest
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy
 import org.springframework.test.annotation.DirtiesContext
 import org.testcontainers.spock.Testcontainers
-
 import java.time.Duration
 
-@SpringBootTest(classes = [EventsPublisher, NcmpAsyncBatchEventConsumer, BatchRecordFilterStrategy,JsonObjectMapper,
-                ObjectMapper])
+@SpringBootTest(classes = [EventsPublisher, NcmpAsyncDataOperationEventConsumer, DataOperationRecordFilterStrategy,JsonObjectMapper, ObjectMapper])
 @Testcontainers
 @DirtiesContext
-class NcmpAsyncBatchEventConsumerSpec extends MessagingBaseSpec {
+class NcmpAsyncDataOperationEventConsumerSpec extends MessagingBaseSpec {
 
     @SpringBean
-    EventsPublisher asyncBatchEventPublisher = new EventsPublisher<BatchDataResponseEventV1>(legacyEventKafkaTemplate, cloudEventKafkaTemplate)
+    EventsPublisher asyncDataOperationEventPublisher = new EventsPublisher<DataOperationEvent>(legacyEventKafkaTemplate, cloudEventKafkaTemplate)
 
     @SpringBean
-    NcmpAsyncBatchEventConsumer asyncBatchEventConsumer = new NcmpAsyncBatchEventConsumer(asyncBatchEventPublisher)
+    NcmpAsyncDataOperationEventConsumer asyncDataOperationEventConsumer = new NcmpAsyncDataOperationEventConsumer(asyncDataOperationEventPublisher)
 
     @Autowired
     JsonObjectMapper jsonObjectMapper
 
     @Autowired
-    RecordFilterStrategy<String, BatchDataResponseEventV1> recordFilterStrategy
+    RecordFilterStrategy<String, DataOperationEvent> recordFilterStrategy
 
     def legacyEventKafkaConsumer = new KafkaConsumer<>(eventConsumerConfigProperties('test', StringDeserializer))
     def static clientTopic = 'client-topic'
-    def static batchEventType = 'org.onap.cps.ncmp.events.async.BatchDataResponseEventV1'
+    def static dataOperationType = 'org.onap.cps.ncmp.events.async1_0_0.DataOperationEvent'
 
     def 'Consume and publish event to client specified topic'() {
         given: 'consumer subscribing to client topic'
             legacyEventKafkaConsumer.subscribe([clientTopic])
-        and: 'consumer record for batch event'
-            def consumerRecordIn = createConsumerRecord(batchEventType)
-        when: 'the batch event is consumed and published to client specified topic'
-            asyncBatchEventConsumer.consumeAndPublish(consumerRecordIn)
+        and: 'consumer record for data operation event'
+            def consumerRecordIn = createConsumerRecord(dataOperationType)
+        when: 'the data operation event is consumed and published to client specified topic'
+            asyncDataOperationEventConsumer.consumeAndPublish(consumerRecordIn)
         and: 'the client specified topic is polled'
             def consumerRecordOut = legacyEventKafkaConsumer.poll(Duration.ofMillis(1500))[0]
         then: 'verifying consumed event operationID is same as published event operationID'
-            def operationIdIn = consumerRecordIn.value.event.batchResponses[0].operationId
-            def operationIdOut = jsonObjectMapper.convertJsonString((String)consumerRecordOut.value(), BatchDataResponseEventV1.class).event.batchResponses[0].operationId
+            def operationIdIn = consumerRecordIn.value.data.responses[0].operationId
+            def operationIdOut = jsonObjectMapper.convertJsonString((String)consumerRecordOut.value(), DataOperationEvent.class).data.responses[0].operationId
             assert operationIdIn == operationIdOut
     }
 
@@ -85,14 +83,14 @@ class NcmpAsyncBatchEventConsumerSpec extends MessagingBaseSpec {
         then: 'the event is #description'
             assert result == expectedResult
         where: 'filter the event based on the eventType #eventType'
-            description                                     | eventType       || expectedResult
-            'not filtered(the consumer will see the event)' | batchEventType  || false
-            'filtered(the consumer will not see the event)' | 'wrongType'     || true
+            description                                     | eventType         || expectedResult
+            'not filtered(the consumer will see the event)' | dataOperationType || false
+            'filtered(the consumer will not see the event)' | 'wrongType'       || true
     }
 
     def createConsumerRecord(eventTypeAsString) {
-        def jsonData = TestUtils.getResourceFileContent('batchDataEvent.json')
-        def testEventSent = jsonObjectMapper.convertJsonString(jsonData, BatchDataResponseEventV1.class)
+        def jsonData = TestUtils.getResourceFileContent('dataOperationEvent.json')
+        def testEventSent = jsonObjectMapper.convertJsonString(jsonData, DataOperationEvent.class)
         def eventTarget = SerializationUtils.serialize(clientTopic)
         def eventType = SerializationUtils.serialize(eventTypeAsString)
         def eventId = SerializationUtils.serialize('12345')