Merge "Improve error scenarios SubscriptionEventForwarder"

3 files changed, 50 insertions, 6 deletions

diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java
index 4afa051d30..4654b148c6 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java
@@ -56,7 +56,8 @@ public class SubscriptionEventForwarder {
 
     private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
 
-    private static final String DMI_AVC_SUBSCRIPTION_TOPIC_PREFIX = "ncmp-dmi-cm-avc-subscription-";
+    @Value("${app.ncmp.avc.subscription-forward-topic-prefix}")
+    private String dmiAvcSubscriptionTopicPrefix;
 
     @Value("${ncmp.timers.subscription-forwarding.dmi-response-timeout-ms:30000}")
     private int dmiResponseTimeoutInMs;
@@ -82,8 +83,15 @@ public class SubscriptionEventForwarder {
                 = DmiServiceNameOrganizer.getDmiPropertiesPerCmHandleIdPerServiceName(yangModelCmHandles);
 
         final Set<String> dmisToRespond = new HashSet<>(dmiPropertiesPerCmHandleIdPerServiceName.keySet());
-        startResponseTimeout(subscriptionEvent, dmisToRespond);
-        forwardEventToDmis(dmiPropertiesPerCmHandleIdPerServiceName, subscriptionEvent);
+        if (dmisToRespond.isEmpty()) {
+            log.info("placeholder to create full outcome response for subscriptionEventId: {}.",
+                subscriptionEvent.getEvent().getSubscription().getClientID()
+                    + subscriptionEvent.getEvent().getSubscription().getName());
+            //TODO outcome response with no cmhandles
+        } else {
+            startResponseTimeout(subscriptionEvent, dmisToRespond);
+            forwardEventToDmis(dmiPropertiesPerCmHandleIdPerServiceName, subscriptionEvent);
+        }
     }
 
     private void forwardEventToDmis(final Map<String, Map<String, Map<String, String>>> dmiNameCmHandleMap,
@@ -91,8 +99,8 @@ public class SubscriptionEventForwarder {
         dmiNameCmHandleMap.forEach((dmiName, cmHandlePropertiesMap) -> {
             subscriptionEvent.getEvent().getPredicates().setTargets(Collections.singletonList(cmHandlePropertiesMap));
             final String eventKey = createEventKey(subscriptionEvent, dmiName);
-            eventsPublisher.publishEvent(
-                DMI_AVC_SUBSCRIPTION_TOPIC_PREFIX + dmiName, eventKey, subscriptionEvent);
+            final String dmiAvcSubscriptionTopic = dmiAvcSubscriptionTopicPrefix + dmiName;
+            eventsPublisher.publishEvent(dmiAvcSubscriptionTopic, eventKey, subscriptionEvent);
         });
     }
 
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 457eb6fa99..a3dec29ede 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
@@ -113,6 +113,41 @@ class SubscriptionEventForwarderSpec extends MessagingBaseSpec {
             'wildcard' | ['CMHandle*']
     }
 
+    def 'Forward valid CM create subscription where targets are not associated to any existing CMHandles'() {
+        given: 'an event'
+            def jsonData = TestUtils.getResourceFileContent('avcSubscriptionCreationEvent.json')
+            def testEventSent = jsonObjectMapper.convertJsonString(jsonData, SubscriptionEvent.class)
+        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'
+            objectUnderTest.dmiResponseTimeoutInMs = 2000
+        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)
+        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'
+            0 * mockSubscriptionEventPublisher.publishEvent("ncmp-dmi-cm-avc-subscription-DMIName1", "SCO-9989752-cm-subscription-001-DMIName1",
+                subscriptionEvent -> {
+                    Map targets = subscriptionEvent.getEvent().getPredicates().getTargets().get(0)
+                    targets["CMHandle1"] == ["shape":"circle"]
+                    targets["CMHandle2"] == ["shape":"square"]
+                }
+            )
+            0 * mockSubscriptionEventPublisher.publishEvent("ncmp-dmi-cm-avc-subscription-DMIName2", "SCO-9989752-cm-subscription-001-DMIName2",
+                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'
+            0 * mockForwardedSubscriptionEventCache.containsKey("SCO-9989752cm-subscription-001") >> true
+            0 * mockForwardedSubscriptionEventCache.get(_)
+        and: 'the subscription id is removed from the event cache map returning the asynchronous blocking variable'
+            0 * mockForwardedSubscriptionEventCache.remove("SCO-9989752cm-subscription-001") >> {block.set(_)}
+    }
+
     static def createYangModelCmHandleWithDmiProperty(id, dmiId,propertyName, propertyValue) {
         return new YangModelCmHandle(id:"CMHandle" + id, dmiDataServiceName: "DMIName" + dmiId, dmiProperties: [new YangModelCmHandle.Property(propertyName,propertyValue)])
     }
diff --git a/cps-ncmp-service/src/test/resources/application.yml b/cps-ncmp-service/src/test/resources/application.yml
index 66194ad9e5..1016f2b033 100644
--- a/cps-ncmp-service/src/test/resources/application.yml
+++ b/cps-ncmp-service/src/test/resources/application.yml
@@ -1,5 +1,5 @@
 #  ============LICENSE_START=======================================================
-#  Copyright (C) 2021-2022 Nordix Foundation
+#  Copyright (C) 2021-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.
@@ -21,6 +21,7 @@ app:
         avc:
             subscription-topic: cm-avc-subscription
             cm-events-topic: cm-events
+            subscription-forward-topic-prefix: ${NCMP_FORWARD_CM_AVC_SUBSCRIPTION:ncmp-dmi-cm-avc-subscription-}
 
 ncmp:
     dmi: