Subscription Create Event Outcome Kafka Part

- Add subscription event outcome schema with java type for pojos
- Add subscription event outcome json for testing
- Add mapper to convert subscription response to event outcome
- Add a bean to handle subscription response outcome tasks
- Change response consumer to publish outcome for client app
- Change response timeout task to publish outcome for client app
- Change subscription persistance to read datanodes
- Add helper to extract cm handle to status mapping from data nodes
  event
- Fix code smells

Issue-ID: CPS-1507
Change-Id: I70195073490f456f014e53c1f59d1b6761d18cd4
Signed-off-by: halil.cakal <halil.cakal@est.tech>

10 files changed, 314 insertions, 53 deletions

diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/event/avc/SubscriptionEventResponseConsumerSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/event/avc/SubscriptionEventResponseConsumerSpec.groovy
index e9f66892c..80c9b69c0 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/event/avc/SubscriptionEventResponseConsumerSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/event/avc/SubscriptionEventResponseConsumerSpec.groovy
@@ -22,7 +22,9 @@ package org.onap.cps.ncmp.api.impl.event.avc
 
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.hazelcast.map.IMap
+import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.onap.cps.ncmp.api.impl.events.avcsubscription.SubscriptionEventResponseMapper
+import org.onap.cps.ncmp.api.impl.events.avcsubscription.SubscriptionEventResponseOutcome
 import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionPersistenceImpl
 import org.onap.cps.ncmp.api.kafka.MessagingBaseSpec
 import org.onap.cps.ncmp.api.models.SubscriptionEventResponse
@@ -34,24 +36,24 @@ class SubscriptionEventResponseConsumerSpec extends MessagingBaseSpec {
 
     IMap<String, Set<String>> mockForwardedSubscriptionEventCache = Mock(IMap<String, Set<String>>)
     def mockSubscriptionPersistence = Mock(SubscriptionPersistenceImpl)
-    def mockSubscriptionEventResponseMapper = Mock(SubscriptionEventResponseMapper)
+    def mockSubscriptionEventResponseMapper  = Mock(SubscriptionEventResponseMapper)
+    def mockSubscriptionEventResponseOutcome = Mock(SubscriptionEventResponseOutcome)
 
     def objectUnderTest = new SubscriptionEventResponseConsumer(mockForwardedSubscriptionEventCache,
-        mockSubscriptionPersistence, mockSubscriptionEventResponseMapper)
+        mockSubscriptionPersistence, mockSubscriptionEventResponseMapper, mockSubscriptionEventResponseOutcome)
 
+    def cmHandleToStatusMap = [CMHandle1: 'PENDING', CMHandle2: 'ACCEPTED'] as Map
+    def testEventReceived = new SubscriptionEventResponse(clientId: 'some-client-id',
+        subscriptionName: 'some-subscription-name', dmiName: 'some-dmi-name', cmHandleIdToStatus: cmHandleToStatusMap)
+    def consumerRecord = new ConsumerRecord<String, SubscriptionEventResponse>('topic-name', 0, 0, 'event-key', testEventReceived)
 
     def 'Consume Subscription Event Response where all DMIs have responded'() {
-        given: 'a subscription event response with a clientId, subscriptionName and dmiName'
-            def testEventReceived = new SubscriptionEventResponse()
-            testEventReceived.clientId = 'some-client-id'
-            testEventReceived.subscriptionName = 'some-subscription-name'
-            testEventReceived.dmiName = 'some-dmi-name'
-        and: 'notifications are enabled'
+        given: 'a subscription event response and notifications are enabled'
             objectUnderTest.notificationFeatureEnabled = true
         and: 'subscription model loader is enabled'
             objectUnderTest.subscriptionModelLoaderEnabled = true
         when: 'the valid event is consumed'
-            objectUnderTest.consumeSubscriptionEventResponse(testEventReceived)
+            objectUnderTest.consumeSubscriptionEventResponse(consumerRecord)
         then: 'the forwarded subscription event cache returns only the received dmiName existing for the subscription create event'
             1 * mockForwardedSubscriptionEventCache.containsKey('some-client-idsome-subscription-name') >> true
             1 * mockForwardedSubscriptionEventCache.get('some-client-idsome-subscription-name') >> (['some-dmi-name'] as Set)
@@ -59,20 +61,17 @@ class SubscriptionEventResponseConsumerSpec extends MessagingBaseSpec {
             1 * mockForwardedSubscriptionEventCache.get('some-client-idsome-subscription-name') >> ([] as Set)
         and: 'the subscription event is removed from the map'
             1 * mockForwardedSubscriptionEventCache.remove('some-client-idsome-subscription-name')
+        and: 'a response outcome has been created'
+            1 * mockSubscriptionEventResponseOutcome.sendResponse('some-client-id', 'some-subscription-name', true)
     }
 
     def 'Consume Subscription Event Response where another DMI has not yet responded'() {
-        given: 'a subscription event response with a clientId, subscriptionName and dmiName'
-            def testEventReceived = new SubscriptionEventResponse()
-            testEventReceived.clientId = 'some-client-id'
-            testEventReceived.subscriptionName = 'some-subscription-name'
-            testEventReceived.dmiName = 'some-dmi-name'
-        and: 'notifications are enabled'
+        given: 'a subscription event response and notifications are enabled'
             objectUnderTest.notificationFeatureEnabled = true
         and: 'subscription model loader is enabled'
             objectUnderTest.subscriptionModelLoaderEnabled = true
         when: 'the valid event is consumed'
-            objectUnderTest.consumeSubscriptionEventResponse(testEventReceived)
+            objectUnderTest.consumeSubscriptionEventResponse(consumerRecord)
         then: 'the forwarded subscription event cache returns only the received dmiName existing for the subscription create event'
             1 * mockForwardedSubscriptionEventCache.containsKey('some-client-idsome-subscription-name') >> true
             1 * mockForwardedSubscriptionEventCache.get('some-client-idsome-subscription-name') >> (['some-dmi-name', 'non-responded-dmi'] as Set)
@@ -80,5 +79,7 @@ class SubscriptionEventResponseConsumerSpec extends MessagingBaseSpec {
             1 * mockForwardedSubscriptionEventCache.get('some-client-idsome-subscription-name') >> (['non-responded-dmi'] as Set)
         and: 'the subscription event is not removed from the map'
             0 * mockForwardedSubscriptionEventCache.remove(_)
+        and: 'a response outcome has not been created'
+            0 * mockSubscriptionEventResponseOutcome.sendResponse(*_)
     }
 }
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/event/avc/SubscriptionOutcomeMapperSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/event/avc/SubscriptionOutcomeMapperSpec.groovy
new file mode 100644
index 000000000..22067745f
--- /dev/null
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/event/avc/SubscriptionOutcomeMapperSpec.groovy
@@ -0,0 +1,55 @@
+/*
+ *  ============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.event.avc
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import org.mapstruct.factory.Mappers
+import org.onap.cps.ncmp.api.models.SubscriptionEventResponse
+import org.onap.cps.ncmp.events.avc.subscription.v1.SubscriptionEventOutcome
+import org.onap.cps.ncmp.utils.TestUtils
+import org.onap.cps.utils.JsonObjectMapper
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.boot.test.context.SpringBootTest
+import spock.lang.Specification
+
+
+@SpringBootTest(classes = [JsonObjectMapper, ObjectMapper])
+class SubscriptionOutcomeMapperSpec extends Specification {
+
+    SubscriptionOutcomeMapper objectUnderTest = Mappers.getMapper(SubscriptionOutcomeMapper)
+
+    @Autowired
+    JsonObjectMapper jsonObjectMapper
+
+    def 'Map subscription event response to subscription event outcome'() {
+        given: 'a Subscription Response Event'
+            def jsonData = TestUtils.getResourceFileContent('avcSubscriptionEventResponse.json')
+            def testEventToMap = jsonObjectMapper.convertJsonString(jsonData, SubscriptionEventResponse.class)
+        and: 'a Subscription Outcome Event'
+            def jsonDataOutcome = TestUtils.getResourceFileContent('avcSubscriptionOutcomeEvent.json')
+            def testEventTarget = jsonObjectMapper.convertJsonString(jsonDataOutcome, SubscriptionEventOutcome.class)
+        when: 'the subscription response event is mapped to a subscription event outcome'
+            def result = objectUnderTest.toSubscriptionEventOutcome(testEventToMap)
+            result.setEventType(SubscriptionEventOutcome.EventType.PARTIAL_OUTCOME)
+        then: 'the resulting subscription event outcome contains the correct clientId'
+            assert result == testEventTarget
+    }
+}
\ No newline at end of file
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/SubscriptionEventResponseMapperSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/SubscriptionEventResponseMapperSpec.groovy
index 7fb817bc9..cde0d1fa0 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/SubscriptionEventResponseMapperSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/SubscriptionEventResponseMapperSpec.groovy
@@ -51,9 +51,10 @@ class SubscriptionEventResponseMapperSpec extends Specification {
         and: 'subscription name'
             assert result.subscriptionName == "cm-subscription-001"
         and: 'predicate targets '
-            assert result.predicates.targetCmHandles.cmHandleId == ["CMHandle1", "CMHandle2"]
+            assert result.predicates.targetCmHandles.cmHandleId == ["CMHandle1", "CMHandle3", "CMHandle4", "CMHandle5"]
         and: 'the status for these targets is set to expected values'
-            assert result.predicates.targetCmHandles.status == [SubscriptionStatus.ACCEPTED, SubscriptionStatus.REJECTED]
+            assert result.predicates.targetCmHandles.status == [SubscriptionStatus.ACCEPTED, SubscriptionStatus.REJECTED,
+            SubscriptionStatus.PENDING, SubscriptionStatus.PENDING]
         and: 'the topic is null'
             assert result.topic == null
     }
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventConsumerSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventConsumerSpec.groovy
index 243c31b39..cccd61b71 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventConsumerSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventConsumerSpec.groovy
@@ -21,6 +21,7 @@
 package org.onap.cps.ncmp.api.impl.events.avcsubscription
 
 import com.fasterxml.jackson.databind.ObjectMapper
+import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionPersistence
 import org.onap.cps.ncmp.api.impl.yangmodels.YangModelSubscriptionEvent
 import org.onap.cps.ncmp.api.kafka.MessagingBaseSpec
@@ -48,30 +49,32 @@ class SubscriptionEventConsumerSpec extends MessagingBaseSpec {
         given: 'an event with data category CM'
             def jsonData = TestUtils.getResourceFileContent('avcSubscriptionCreationEvent.json')
             def testEventSent = jsonObjectMapper.convertJsonString(jsonData, SubscriptionEvent.class)
+            def consumerRecord = new ConsumerRecord<String, SubscriptionEvent>('topic-name', 0, 0, 'event-key', testEventSent)
         and: 'notifications are enabled'
             objectUnderTest.notificationFeatureEnabled = true
         and: 'subscription model loader is enabled'
             objectUnderTest.subscriptionModelLoaderEnabled = true
         when: 'the valid event is consumed'
-            objectUnderTest.consumeSubscriptionEvent(testEventSent)
+            objectUnderTest.consumeSubscriptionEvent(consumerRecord)
         then: 'the event is mapped to a yangModelSubscription'
             1 * mockSubscriptionEventMapper.toYangModelSubscriptionEvent(testEventSent) >> yangModelSubscriptionEvent
         and: 'the event is persisted'
             1 * mockSubscriptionPersistence.saveSubscriptionEvent(yangModelSubscriptionEvent)
         and: 'the event is forwarded'
-            1 * mockSubscriptionEventForwarder.forwardCreateSubscriptionEvent(testEventSent)
+            1 * mockSubscriptionEventForwarder.forwardCreateSubscriptionEvent(testEventSent, consumerRecord.headers())
     }
 
     def 'Consume valid CM create message where notifications and model loader are disabled'() {
         given: 'an event with data category CM'
             def jsonData = TestUtils.getResourceFileContent('avcSubscriptionCreationEvent.json')
             def testEventSent = jsonObjectMapper.convertJsonString(jsonData, SubscriptionEvent.class)
+            def consumerRecord = new ConsumerRecord<String, SubscriptionEvent>('topic-name', 0, 0, 'event-key', testEventSent)
         and: 'notifications are disabled'
             objectUnderTest.notificationFeatureEnabled = false
         and: 'subscription model loader is disabled'
             objectUnderTest.subscriptionModelLoaderEnabled = false
         when: 'the valid event is consumed'
-            objectUnderTest.consumeSubscriptionEvent(testEventSent)
+            objectUnderTest.consumeSubscriptionEvent(consumerRecord)
         then: 'the event is not mapped to a yangModelSubscription'
             0 * mockSubscriptionEventMapper.toYangModelSubscriptionEvent(*_) >> yangModelSubscriptionEvent
         and: 'the event is not persisted'
@@ -84,10 +87,11 @@ class SubscriptionEventConsumerSpec extends MessagingBaseSpec {
         given: 'an event'
             def jsonData = TestUtils.getResourceFileContent('avcSubscriptionCreationEvent.json')
             def testEventSent = jsonObjectMapper.convertJsonString(jsonData, SubscriptionEvent.class)
+            def consumerRecord = new ConsumerRecord<String, SubscriptionEvent>('topic-name', 0, 0, 'event-key', testEventSent)
         and: 'dataCategory is set to FM'
             testEventSent.getEvent().getDataType().setDataCategory("FM")
         when: 'the valid event is consumed'
-            objectUnderTest.consumeSubscriptionEvent(testEventSent)
+            objectUnderTest.consumeSubscriptionEvent(consumerRecord)
         then: 'no exception is thrown'
             noExceptionThrown()
         and: 'the event is not mapped to a yangModelSubscription'
@@ -102,10 +106,11 @@ class SubscriptionEventConsumerSpec extends MessagingBaseSpec {
         given: 'an event'
             def jsonData = TestUtils.getResourceFileContent('avcSubscriptionCreationEvent.json')
             def testEventSent = jsonObjectMapper.convertJsonString(jsonData, SubscriptionEvent.class)
+            def consumerRecord = new ConsumerRecord<String, SubscriptionEvent>('topic-name', 0, 0, 'event-key', testEventSent)
         and: 'datastore is set to a non passthrough datastore'
             testEventSent.getEvent().getPredicates().setDatastore("operational")
         when: 'the valid event is consumed'
-            objectUnderTest.consumeSubscriptionEvent(testEventSent)
+            objectUnderTest.consumeSubscriptionEvent(consumerRecord)
         then: 'an operation not yet supported exception is thrown'
             thrown(OperationNotYetSupportedException)
     }
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarderSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarderSpec.groovy
index a3dec29ed..63ddcef55 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarderSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarderSpec.groovy
@@ -22,6 +22,7 @@ package org.onap.cps.ncmp.api.impl.events.avcsubscription
 
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.hazelcast.map.IMap
+import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.onap.cps.ncmp.api.impl.events.EventsPublisher
 import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle
 import org.onap.cps.ncmp.api.inventory.InventoryPersistence
@@ -35,6 +36,8 @@ import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.boot.test.context.SpringBootTest
 import spock.util.concurrent.BlockingVariable
 
+import java.util.concurrent.TimeUnit
+
 @SpringBootTest(classes = [ObjectMapper, JsonObjectMapper, SubscriptionEventForwarder])
 class SubscriptionEventForwarderSpec extends MessagingBaseSpec {
 
@@ -47,7 +50,8 @@ class SubscriptionEventForwarderSpec extends MessagingBaseSpec {
     EventsPublisher<SubscriptionEvent> mockSubscriptionEventPublisher = Mock(EventsPublisher<SubscriptionEvent>)
     @SpringBean
     IMap<String, Set<String>> mockForwardedSubscriptionEventCache = Mock(IMap<String, Set<String>>)
-
+    @SpringBean
+    SubscriptionEventResponseOutcome mockSubscriptionEventResponseOutcome = Mock(SubscriptionEventResponseOutcome)
     @Autowired
     JsonObjectMapper jsonObjectMapper
 
@@ -55,6 +59,7 @@ class SubscriptionEventForwarderSpec extends MessagingBaseSpec {
         given: 'an event'
             def jsonData = TestUtils.getResourceFileContent('avcSubscriptionCreationEvent.json')
             def testEventSent = jsonObjectMapper.convertJsonString(jsonData, SubscriptionEvent.class)
+            def consumerRecord = new ConsumerRecord<String, SubscriptionEvent>('topic-name', 0, 0, 'event-key', testEventSent)
         and: 'the InventoryPersistence returns private properties for the supplied CM Handles'
             1 * mockInventoryPersistence.getYangModelCmHandles(["CMHandle1", "CMHandle2", "CMHandle3"]) >> [
                 createYangModelCmHandleWithDmiProperty(1, 1,"shape","circle"),
@@ -66,44 +71,46 @@ class SubscriptionEventForwarderSpec extends MessagingBaseSpec {
         and: 'a Blocking Variable is used for the Asynchronous call with a timeout of 5 seconds'
             def block = new BlockingVariable<Object>(5)
         when: 'the valid event is forwarded'
-            objectUnderTest.forwardCreateSubscriptionEvent(testEventSent)
+            objectUnderTest.forwardCreateSubscriptionEvent(testEventSent, consumerRecord.headers())
         then: 'An asynchronous call is made to the blocking variable'
             block.get()
         then: 'the event is added to the forwarded subscription event cache'
-            1 * mockForwardedSubscriptionEventCache.put("SCO-9989752cm-subscription-001", ["DMIName1", "DMIName2"] as Set)
+            1 * mockForwardedSubscriptionEventCache.put("SCO-9989752cm-subscription-001", ["DMIName1", "DMIName2"] as Set, 600, TimeUnit.SECONDS)
         and: 'the event is forwarded twice with the CMHandle private properties and provides a valid listenable future'
             1 * mockSubscriptionEventPublisher.publishEvent("ncmp-dmi-cm-avc-subscription-DMIName1", "SCO-9989752-cm-subscription-001-DMIName1",
-                subscriptionEvent -> {
+                consumerRecord.headers(), subscriptionEvent -> {
                     Map targets = subscriptionEvent.getEvent().getPredicates().getTargets().get(0)
                     targets["CMHandle1"] == ["shape":"circle"]
                     targets["CMHandle2"] == ["shape":"square"]
                 }
             )
             1 * mockSubscriptionEventPublisher.publishEvent("ncmp-dmi-cm-avc-subscription-DMIName2", "SCO-9989752-cm-subscription-001-DMIName2",
-                subscriptionEvent -> {
+                consumerRecord.headers(), subscriptionEvent -> {
                     Map targets = subscriptionEvent.getEvent().getPredicates().getTargets().get(0)
                     targets["CMHandle3"] == ["shape":"triangle"]
                 }
             )
         and: 'a separate thread has been created where the map is polled'
             1 * mockForwardedSubscriptionEventCache.containsKey("SCO-9989752cm-subscription-001") >> true
-            1 * mockForwardedSubscriptionEventCache.get(_) >> (DMINamesInMap)
+            1 * mockForwardedSubscriptionEventCache.get(_) >> DMINamesInMap
+            1 * mockSubscriptionEventResponseOutcome.sendResponse(*_)
         and: 'the subscription id is removed from the event cache map returning the asynchronous blocking variable'
             1 * mockForwardedSubscriptionEventCache.remove("SCO-9989752cm-subscription-001") >> {block.set(_)}
         where:
             scenario                                  | DMINamesInMap
             'there are dmis which have not responded' | ["DMIName1", "DMIName2"] as Set
-            'all dmis have responded '                | [] as Set
+            'all dmis have responded'                 | [] as Set
     }
 
     def 'Forward CM create subscription where target CM Handles are #scenario'() {
         given: 'an event'
             def jsonData = TestUtils.getResourceFileContent('avcSubscriptionCreationEvent.json')
             def testEventSent = jsonObjectMapper.convertJsonString(jsonData, SubscriptionEvent.class)
+            def consumerRecord = new ConsumerRecord<String, SubscriptionEvent>('topic-name', 0, 0, 'event-key', testEventSent)
         and: 'the target CMHandles are set to #scenario'
             testEventSent.getEvent().getPredicates().setTargets(invalidTargets)
         when: 'the event is forwarded'
-            objectUnderTest.forwardCreateSubscriptionEvent(testEventSent)
+            objectUnderTest.forwardCreateSubscriptionEvent(testEventSent, consumerRecord.headers())
         then: 'an operation not yet supported exception is thrown'
             thrown(OperationNotYetSupportedException)
         where:
@@ -117,6 +124,7 @@ class SubscriptionEventForwarderSpec extends MessagingBaseSpec {
         given: 'an event'
             def jsonData = TestUtils.getResourceFileContent('avcSubscriptionCreationEvent.json')
             def testEventSent = jsonObjectMapper.convertJsonString(jsonData, SubscriptionEvent.class)
+            def consumerRecord = new ConsumerRecord<String, SubscriptionEvent>('topic-name', 0, 0, 'event-key', testEventSent)
         and: 'the InventoryPersistence returns no private properties for the supplied CM Handles'
             1 * mockInventoryPersistence.getYangModelCmHandles(["CMHandle1", "CMHandle2", "CMHandle3"]) >> []
         and: 'the thread creation delay is reduced to 2 seconds for testing'
@@ -124,7 +132,7 @@ class SubscriptionEventForwarderSpec extends MessagingBaseSpec {
         and: 'a Blocking Variable is used for the Asynchronous call with a timeout of 5 seconds'
             def block = new BlockingVariable<Object>(5)
         when: 'the valid event is forwarded'
-            objectUnderTest.forwardCreateSubscriptionEvent(testEventSent)
+            objectUnderTest.forwardCreateSubscriptionEvent(testEventSent, consumerRecord.headers())
         then: 'the event is not added to the forwarded subscription event cache'
             0 * mockForwardedSubscriptionEventCache.put("SCO-9989752cm-subscription-001", ["DMIName1", "DMIName2"] as Set)
         and: 'the event is forwarded twice with the CMHandle private properties and provides a valid listenable future'
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventResponseOutcomeSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventResponseOutcomeSpec.groovy
new file mode 100644
index 000000000..53c5cd2c7
--- /dev/null
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventResponseOutcomeSpec.groovy
@@ -0,0 +1,89 @@
+/*
+ * ============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.events.avcsubscription
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import org.mapstruct.factory.Mappers
+import org.onap.cps.ncmp.api.impl.event.avc.SubscriptionOutcomeMapper
+import org.onap.cps.ncmp.api.impl.events.EventsPublisher
+import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionPersistence
+import org.onap.cps.ncmp.api.impl.utils.DataNodeBaseSpec
+import org.onap.cps.ncmp.events.avc.subscription.v1.SubscriptionEventOutcome
+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.testcontainers.shaded.org.bouncycastle.crypto.engines.EthereumIESEngine
+
+@SpringBootTest(classes = [ObjectMapper, JsonObjectMapper, SubscriptionOutcomeMapper, SubscriptionEventResponseOutcome])
+class SubscriptionEventResponseOutcomeSpec extends DataNodeBaseSpec {
+
+    @Autowired
+    SubscriptionEventResponseOutcome objectUnderTest
+
+    @SpringBean
+    SubscriptionPersistence mockSubscriptionPersistence = Mock(SubscriptionPersistence)
+    @SpringBean
+    EventsPublisher<SubscriptionEventOutcome> mockSubscriptionEventOutcomePublisher = Mock(EventsPublisher<SubscriptionEventOutcome>)
+    @SpringBean
+    SubscriptionOutcomeMapper subscriptionOutcomeMapper = Mappers.getMapper(SubscriptionOutcomeMapper)
+
+    @Autowired
+    JsonObjectMapper jsonObjectMapper
+
+    def 'Generate response via fetching data nodes from database.'() {
+        given: 'a db call to get data nodes for subscription event'
+            1 * mockSubscriptionPersistence.getDataNodesForSubscriptionEvent() >> [dataNode4]
+        when: 'a response is generated'
+            def result = objectUnderTest.generateResponse('some-client-id', 'some-subscription-name', isFullOutcomeResponse)
+        then: 'the result will have the same values as same as in dataNode4'
+            result.eventType == eventType
+            result.getEvent().getSubscription().getClientID() == 'some-client-id'
+            result.getEvent().getSubscription().getName() == 'some-subscription-name'
+            result.getEvent().getPredicates().getPendingTargets() == ['CMHandle3']
+            result.getEvent().getPredicates().getRejectedTargets() == ['CMHandle1']
+            result.getEvent().getPredicates().getAcceptedTargets() == ['CMHandle2']
+        where: 'the following values are used'
+            scenario             | isFullOutcomeResponse || eventType
+            'is full outcome'    | true                  || SubscriptionEventOutcome.EventType.COMPLETE_OUTCOME
+            'is partial outcome' | false                 || SubscriptionEventOutcome.EventType.PARTIAL_OUTCOME
+    }
+
+    def 'Form subscription outcome message with a list of cm handle id to status mapping'() {
+        given: 'a list of collection including cm handle id to status'
+            def cmHandleIdToStatus = [['PENDING', 'CMHandle5'], ['PENDING', 'CMHandle4'], ['ACCEPTED', 'CMHandle1'], ['REJECTED', 'CMHandle3']]
+        and: 'an outcome event'
+            def jsonData = TestUtils.getResourceFileContent('avcSubscriptionOutcomeEvent.json')
+            def eventOutcome = jsonObjectMapper.convertJsonString(jsonData, SubscriptionEventOutcome.class)
+            eventOutcome.setEventType(eventType)
+        when: 'a subscription outcome message formed'
+            def result = objectUnderTest.formSubscriptionOutcomeMessage(cmHandleIdToStatus, 'SCO-9989752',
+                'cm-subscription-001', isFullOutcomeResponse)
+            result.getEvent().getPredicates().getPendingTargets().sort()
+        then: 'the result will be equal to event outcome'
+            result == eventOutcome
+        where: 'the following values are used'
+            scenario             | isFullOutcomeResponse | eventType
+            'is full outcome'    | true                  | SubscriptionEventOutcome.EventType.COMPLETE_OUTCOME
+            'is partial outcome' | false                 | SubscriptionEventOutcome.EventType.PARTIAL_OUTCOME
+    }
+}
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/lcm/LcmEventsServiceSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/lcm/LcmEventsServiceSpec.groovy
index 2d3f8ac51..edc6e3bcf 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/lcm/LcmEventsServiceSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/lcm/LcmEventsServiceSpec.groovy
@@ -35,7 +35,7 @@ class LcmEventsServiceSpec extends Specification {
     def objectUnderTest = new LcmEventsService(mockLcmEventsPublisher, mockJsonObjectMapper)
 
     def 'Create and Publish lcm event where events are #scenario'() {
-        given: 'a cm handle id and Lcm Event'
+        given: 'a cm handle id, Lcm Event, and headers'
             def cmHandleId = 'test-cm-handle-id'
             def eventId = UUID.randomUUID().toString()
             def lcmEvent = new LcmEvent(eventId: eventId, eventCorrelationId: cmHandleId)
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/subscriptions/SubscriptionPersistenceSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/subscriptions/SubscriptionPersistenceSpec.groovy
index 75760091d..a372abe6f 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/subscriptions/SubscriptionPersistenceSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/subscriptions/SubscriptionPersistenceSpec.groovy
@@ -36,23 +36,21 @@ class SubscriptionPersistenceSpec extends Specification {
     private static final String SUBSCRIPTION_REGISTRY_PARENT = "/subscription-registry";
 
     def jsonObjectMapper = new JsonObjectMapper(new ObjectMapper())
-
     def mockCpsDataService = Mock(CpsDataService)
-
     def objectUnderTest = new SubscriptionPersistenceImpl(jsonObjectMapper, mockCpsDataService)
 
+    def predicates = new YangModelSubscriptionEvent.Predicates(datastore: 'some-datastore',
+        targetCmHandles: [new YangModelSubscriptionEvent.TargetCmHandle('cmhandle1'),
+                          new YangModelSubscriptionEvent.TargetCmHandle('cmhandle2')])
+    def yangModelSubscriptionEvent = new YangModelSubscriptionEvent(clientId: 'some-client-id',
+        subscriptionName: 'some-subscription-name', tagged: true, topic: 'some-topic', predicates: predicates)
+
    def 'save a subscription event' () {
-       given: 'a yang model subscription event'
-           def predicates = new YangModelSubscriptionEvent.Predicates(datastore: 'some-datastore',
-               targetCmHandles: [new YangModelSubscriptionEvent.TargetCmHandle('cmhandle1'),
-                                 new YangModelSubscriptionEvent.TargetCmHandle('cmhandle2')])
-           def yangModelSubscriptionEvent = new YangModelSubscriptionEvent(clientId: 'some-client-id',
-                subscriptionName: 'some-subscription-name', tagged: true, topic: 'some-topic', predicates: predicates)
-       and: 'a data node that does not exist in db'
-           def dataNodeNonExist = new DataNodeBuilder().withDataspace('NCMP-Admin')
+       given: 'a data node that does not exist in db'
+           def blankDataNode = new DataNodeBuilder().withDataspace('NCMP-Admin')
                 .withAnchor('AVC-Subscriptions').withXpath('/subscription-registry').build()
        and: 'cps data service return non existing data node'
-            mockCpsDataService.getDataNodes(*_) >> [dataNodeNonExist]
+            mockCpsDataService.getDataNodes(*_) >> [blankDataNode]
        when: 'the yangModelSubscriptionEvent is saved into db'
             objectUnderTest.saveSubscriptionEvent(yangModelSubscriptionEvent)
        then: 'the cpsDataService save operation is called with the correct data'
@@ -66,20 +64,14 @@ class SubscriptionPersistenceSpec extends Specification {
    }
 
     def 'update a subscription event' () {
-        given: 'a yang model subscription event'
-            def predicates = new YangModelSubscriptionEvent.Predicates(datastore: 'some-datastore',
-                targetCmHandles: [new YangModelSubscriptionEvent.TargetCmHandle('cmhandle1'),
-                                  new YangModelSubscriptionEvent.TargetCmHandle('cmhandle2')])
-            def yangModelSubscriptionEvent = new YangModelSubscriptionEvent(clientId: 'some-client-id',
-                subscriptionName: 'some-subscription-name', tagged: true, topic: 'some-topic', predicates: predicates)
-        and: 'a data node exist in db'
+        given: 'a data node exist in db'
             def childDataNode = new DataNodeBuilder().withDataspace('NCMP-Admin')
                 .withAnchor('AVC-Subscriptions').withXpath('/subscription-registry/subscription').build()
-            def dataNodeExist = new DataNodeBuilder().withDataspace('NCMP-Admin')
+            def engagedDataNode = new DataNodeBuilder().withDataspace('NCMP-Admin')
                 .withAnchor('AVC-Subscriptions').withXpath('/subscription-registry')
                 .withChildDataNodes([childDataNode]).build()
         and: 'cps data service return existing data node'
-            mockCpsDataService.getDataNodes(*_) >> [dataNodeExist]
+            mockCpsDataService.getDataNodes(*_) >> [engagedDataNode]
         when: 'the yangModelSubscriptionEvent is saved into db'
             objectUnderTest.saveSubscriptionEvent(yangModelSubscriptionEvent)
         then: 'the cpsDataService update operation is called with the correct data'
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/utils/DataNodeBaseSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/utils/DataNodeBaseSpec.groovy
new file mode 100644
index 000000000..7474166ff
--- /dev/null
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/utils/DataNodeBaseSpec.groovy
@@ -0,0 +1,52 @@
+/*
+ * ============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.utils
+
+import org.onap.cps.spi.model.DataNodeBuilder
+import spock.lang.Specification
+
+class DataNodeBaseSpec extends Specification {
+
+    def leaves1 = [status:'PENDING', cmHandleId:'CMHandle3'] as Map
+    def dataNode1 = createDataNodeWithLeaves(leaves1)
+
+    def leaves2 = [status:'ACCEPTED', cmHandleId:'CMHandle2'] as Map
+    def dataNode2 = createDataNodeWithLeaves(leaves2)
+
+    def leaves3 = [status:'REJECTED', cmHandleId:'CMHandle1'] as Map
+    def dataNode3 = createDataNodeWithLeaves(leaves3)
+
+    def leaves4 = [datastore:'passthrough-running'] as Map
+    def dataNode4 = createDataNodeWithLeavesAndChildDataNodes(leaves4, [dataNode1, dataNode2, dataNode3])
+
+    static def createDataNodeWithLeaves(leaves) {
+        return new DataNodeBuilder().withDataspace('NCMP-Admin')
+            .withAnchor('AVC-Subscriptions').withXpath('/subscription-registry/subscription')
+            .withLeaves(leaves).build()
+    }
+
+    static def createDataNodeWithLeavesAndChildDataNodes(leaves, dataNodes) {
+        return new DataNodeBuilder().withDataspace('NCMP-Admin')
+            .withAnchor('AVC-Subscriptions').withXpath('/subscription-registry/subscription')
+            .withLeaves(leaves).withChildDataNodes(dataNodes)
+            .build()
+    }
+}
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/utils/DataNodeHelperSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/utils/DataNodeHelperSpec.groovy
new file mode 100644
index 000000000..e527ae12b
--- /dev/null
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/utils/DataNodeHelperSpec.groovy
@@ -0,0 +1,58 @@
+/*
+ * ============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.utils
+
+import org.onap.cps.spi.model.DataNodeBuilder
+
+class DataNodeHelperSpec extends DataNodeBaseSpec {
+
+    def 'Get data node leaves as expected from a nested data node.'() {
+        given: 'a nested data node'
+            def dataNode = new DataNodeBuilder().withDataspace('NCMP-Admin')
+                .withAnchor('AVC-Subscriptions').withXpath('/subscription-registry/subscription')
+                .withLeaves([clientID:'SCO-9989752', isTagged:false, subscriptionName:'cm-subscription-001'])
+                .withChildDataNodes([dataNode4]).build()
+        when: 'the nested data node is flatten and retrieves the leaves '
+            def result = DataNodeHelper.getDataNodeLeaves([dataNode])
+        then: 'the result list size is 5'
+            result.size() == 5
+        and: 'all the leaves result list are equal to given leaves of data nodes'
+            result[0] == [clientID:'SCO-9989752', isTagged:false, subscriptionName:'cm-subscription-001']
+            result[1] == [datastore:'passthrough-running']
+            result[2] == [status:'PENDING', cmHandleId:'CMHandle3']
+            result[3] == [status:'ACCEPTED', cmHandleId:'CMHandle2']
+            result[4] == [status:'REJECTED', cmHandleId:'CMHandle1']
+    }
+
+    def 'Get cm handle id to status as expected from a nested data node.'() {
+        given: 'a nested data node'
+            def dataNode = new DataNodeBuilder().withDataspace('NCMP-Admin')
+                .withAnchor('AVC-Subscriptions').withXpath('/subscription-registry/subscription')
+                .withLeaves([clientID:'SCO-9989752', isTagged:false, subscriptionName:'cm-subscription-001'])
+                .withChildDataNodes([dataNode4]).build()
+        and: 'the nested data node is flatten and retrieves the leaves '
+            def leaves = DataNodeHelper.getDataNodeLeaves([dataNode])
+        when:'cm handle id to status is retrieved'
+            def result = DataNodeHelper.getCmHandleIdToStatus(leaves);
+        then: 'the result list size is 3'
+            result.size() == 3
+    }
+}