summaryrefslogtreecommitdiffstats
path: root/cps-ncmp-service/src/test/groovy
diff options
context:
space:
mode:
authorraviteja.karumuri <raviteja.karumuri@est.tech>2023-04-13 23:44:41 +0100
committerraviteja.karumuri <raviteja.karumuri@est.tech>2023-05-23 17:53:05 +0100
commite9ac24f8ad866f81d1ceace6d0b8f010930e9f74 (patch)
tree9bf8b9135b190eea558c51542a776172a41df24a /cps-ncmp-service/src/test/groovy
parent9b059e3ad4edc1ea8d2f62ec4bf82543a1e157c2 (diff)
NCMP : forward bulk response messages to client topic
Issue-ID: CPS-1557 Signed-off-by: raviteja.karumuri <raviteja.karumuri@est.tech> Change-Id: I0ea040e98987f992f46105afea0a171c4031d64f
Diffstat (limited to 'cps-ncmp-service/src/test/groovy')
-rw-r--r--cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/NcmpAsyncBatchEventConsumerSpec.groovy104
1 files changed, 104 insertions, 0 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/NcmpAsyncBatchEventConsumerSpec.groovy
new file mode 100644
index 0000000000..65c43a011d
--- /dev/null
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/NcmpAsyncBatchEventConsumerSpec.groovy
@@ -0,0 +1,104 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 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.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.api.impl.async
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import org.apache.commons.lang3.SerializationUtils
+import org.apache.kafka.clients.consumer.ConsumerRecord
+import org.apache.kafka.clients.consumer.KafkaConsumer
+import org.apache.kafka.common.header.internals.RecordHeader
+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.utils.TestUtils
+import org.onap.cps.utils.JsonObjectMapper
+import org.spockframework.spring.SpringBean
+import org.springframework.beans.factory.annotation.Autowired
+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])
+@Testcontainers
+@DirtiesContext
+class NcmpAsyncBatchEventConsumerSpec extends MessagingBaseSpec {
+
+ @SpringBean
+ EventsPublisher asyncBatchEventPublisher = new EventsPublisher<BatchDataResponseEventV1>(kafkaTemplate)
+
+ @SpringBean
+ NcmpAsyncBatchEventConsumer asyncBatchEventConsumer = new NcmpAsyncBatchEventConsumer(asyncBatchEventPublisher)
+
+ @Autowired
+ JsonObjectMapper jsonObjectMapper
+
+ @Autowired
+ RecordFilterStrategy<Object, Object> recordFilterStrategy
+
+ def kafkaConsumer = new KafkaConsumer<>(consumerConfigProperties('test'))
+ def static clientTopic = 'client-topic'
+ def static batchEventType = 'org.onap.cps.ncmp.events.async.BatchDataResponseEventV1'
+
+ def 'Consume and publish event to client specified topic'() {
+ given: 'consumer subscribing to client topic'
+ kafkaConsumer.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: 'the client specified topic is polled'
+ def consumerRecordOut = kafkaConsumer.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
+ assert operationIdIn == operationIdOut
+ }
+
+ def 'Filter an event with type #eventType'() {
+ given: 'consumer record for event with type #eventType'
+ def consumerRecord = createConsumerRecord(eventType)
+ when: 'while consuming the topic ncmp-async-m2m it executes the filter strategy'
+ def result = recordFilterStrategy.filter(consumerRecord)
+ 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
+ }
+
+ def createConsumerRecord(eventTypeAsString) {
+ def jsonData = TestUtils.getResourceFileContent('batchDataEvent.json')
+ def testEventSent = jsonObjectMapper.convertJsonString(jsonData, BatchDataResponseEventV1.class)
+ def eventTarget = SerializationUtils.serialize(clientTopic)
+ def eventType = SerializationUtils.serialize(eventTypeAsString)
+ def eventId = SerializationUtils.serialize('12345')
+ def consumerRecord = new ConsumerRecord<String, Object>(clientTopic, 1, 1L, '123', testEventSent)
+ consumerRecord.headers().add(new RecordHeader('eventId', eventId))
+ consumerRecord.headers().add(new RecordHeader('eventTarget', eventTarget))
+ consumerRecord.headers().add(new RecordHeader('eventType', eventType))
+ return consumerRecord
+ }
+}