Merge "Kafka consumer can not be turned off"

3 files changed, 28 insertions, 13 deletions

diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventConsumer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventConsumer.java
index 4e5c57ba5..a9e7164fd 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventConsumer.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventConsumer.java
@@ -24,6 +24,7 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.onap.cps.ncmp.event.model.DmiAsyncRequestResponseEvent;
 import org.onap.cps.ncmp.event.model.NcmpAsyncRequestResponseEvent;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Component;
 
@@ -33,6 +34,7 @@ import org.springframework.stereotype.Component;
 @Component
 @Slf4j
 @RequiredArgsConstructor
+@ConditionalOnProperty(name = "notification.enabled", havingValue = "true", matchIfMissing = true)
 public class NcmpAsyncRequestResponseEventConsumer {
 
     private final NcmpAsyncRequestResponseEventProducer ncmpAsyncRequestResponseEventProducer;
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/event/NcmpEventsService.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/event/NcmpEventsService.java
index 6804ac0f0..7b5ceb57a 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/event/NcmpEventsService.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/event/NcmpEventsService.java
@@ -47,17 +47,23 @@ public class NcmpEventsService {
     @Value("${app.ncmp.events.topic:ncmp-events}")
     private String topicName;
 
+    @Value("${notification.enabled:true}")
+    private boolean notificationsEnabled;
+
     /**
      * Publish the NcmpEvent to the public topic.
      *
      * @param cmHandleId Cm Handle Id
      */
     public void publishNcmpEvent(final String cmHandleId) {
-
-        final NcmpServiceCmHandle ncmpServiceCmHandle = YangDataConverter.convertYangModelCmHandleToNcmpServiceCmHandle(
-                inventoryPersistence.getYangModelCmHandle(cmHandleId));
-        final NcmpEvent ncmpEvent = ncmpEventsCreator.populateNcmpEvent(cmHandleId, ncmpServiceCmHandle);
-        ncmpEventsPublisher.publishEvent(topicName, cmHandleId, ncmpEvent);
-
+        if (notificationsEnabled) {
+            final NcmpServiceCmHandle ncmpServiceCmHandle =
+                YangDataConverter.convertYangModelCmHandleToNcmpServiceCmHandle(
+                    inventoryPersistence.getYangModelCmHandle(cmHandleId));
+            final NcmpEvent ncmpEvent = ncmpEventsCreator.populateNcmpEvent(cmHandleId, ncmpServiceCmHandle);
+            ncmpEventsPublisher.publishEvent(topicName, cmHandleId, ncmpEvent);
+        } else {
+            log.debug("Notifications disabled.");
+        }
     }
 }
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/event/NcmpEventsServiceSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/event/NcmpEventsServiceSpec.groovy
index e265fef05..52806a867 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/event/NcmpEventsServiceSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/event/NcmpEventsServiceSpec.groovy
@@ -30,17 +30,25 @@ class NcmpEventsServiceSpec extends Specification {
 
     def mockInventoryPersistence = Mock(InventoryPersistence)
     def mockNcmpEventsPublisher = Mock(NcmpEventsPublisher)
-    def mockNcmpEventsMapper = Mock(NcmpEventsCreator)
+    def mockNcmpEventsCreator = Mock(NcmpEventsCreator)
 
-    def objectUnderTest = new NcmpEventsService(mockInventoryPersistence, mockNcmpEventsPublisher, mockNcmpEventsMapper)
+    def objectUnderTest = new NcmpEventsService(mockInventoryPersistence, mockNcmpEventsPublisher, mockNcmpEventsCreator)
 
-    def 'Create and Publish event for #operation'() {
+    def 'Create and Publish ncmp event where events are #scenario'() {
         given: 'a cm handle id and operation and responses are mocked'
             mockResponses('test-cm-handle-id', 'test-topic')
+        and: 'notifications enabled is #notificationsEnabled'
+            objectUnderTest.notificationsEnabled = notificationsEnabled
         when: 'service is called to publish ncmp event'
             objectUnderTest.publishNcmpEvent('test-cm-handle-id')
-        then: 'no exception is thrown'
-            noExceptionThrown()
+        then: 'creator is called #expectedTimesMethodCalled times'
+            expectedTimesMethodCalled * mockNcmpEventsCreator.populateNcmpEvent('test-cm-handle-id', _)
+        and: 'publisher is called #expectedTimesMethodCalled times'
+            expectedTimesMethodCalled * mockNcmpEventsPublisher.publishEvent(*_)
+        where: 'the following values are used'
+            scenario   | notificationsEnabled|| expectedTimesMethodCalled
+            'enabled'  | true                || 1
+            'disabled' | false               || 0
     }
 
     def mockResponses(cmHandleId, topicName) {
@@ -50,9 +58,8 @@ class NcmpEventsServiceSpec extends Specification {
         def ncmpServiceCmhandle = YangDataConverter.convertYangModelCmHandleToNcmpServiceCmHandle(yangModelCmHandle)
 
         mockInventoryPersistence.getYangModelCmHandle(cmHandleId) >> yangModelCmHandle
-        mockNcmpEventsMapper.populateNcmpEvent(cmHandleId, ncmpServiceCmhandle) >> ncmpEvent
+        mockNcmpEventsCreator.populateNcmpEvent(cmHandleId, ncmpServiceCmhandle) >> ncmpEvent
         mockNcmpEventsPublisher.publishEvent(topicName, cmHandleId, ncmpEvent) >> {}
     }
 
-
 }