Increase Test Coverage for DMI

Issue-ID: CPS-2203
Change-Id: Ib2fd213512db9f23220db53ed795961e747ee17c
Signed-off-by: seanbeirne <sean.beirne@est.tech>

3 files changed, 33 insertions, 4 deletions

diff --git a/src/main/java/org/onap/cps/ncmp/dmi/notifications/cmsubscription/CmNotificationSubscriptionDmiInEventConsumer.java b/src/main/java/org/onap/cps/ncmp/dmi/notifications/cmsubscription/CmNotificationSubscriptionDmiInEventConsumer.java
index ecfef6f3..da6243ca 100644
--- a/src/main/java/org/onap/cps/ncmp/dmi/notifications/cmsubscription/CmNotificationSubscriptionDmiInEventConsumer.java
+++ b/src/main/java/org/onap/cps/ncmp/dmi/notifications/cmsubscription/CmNotificationSubscriptionDmiInEventConsumer.java
@@ -91,7 +91,7 @@ public class CmNotificationSubscriptionDmiInEventConsumer {
         if (cmNotificationSubscriptionStatus.equals(CmNotificationSubscriptionStatus.ACCEPTED)) {
             cmNotificationSubscriptionDmiOutEventData.setStatusCode("1");
             cmNotificationSubscriptionDmiOutEventData.setStatusMessage("ACCEPTED");
-        } else if (cmNotificationSubscriptionStatus.equals(CmNotificationSubscriptionStatus.REJECTED)) {
+        } else {
             cmNotificationSubscriptionDmiOutEventData.setStatusCode("2");
             cmNotificationSubscriptionDmiOutEventData.setStatusMessage("REJECTED");
         }
diff --git a/src/test/groovy/org/onap/cps/ncmp/dmi/notifications/async/AsyncTaskExecutorIntegrationSpec.groovy b/src/test/groovy/org/onap/cps/ncmp/dmi/notifications/async/AsyncTaskExecutorIntegrationSpec.groovy
index 7ca2d54c..12ca05cf 100644
--- a/src/test/groovy/org/onap/cps/ncmp/dmi/notifications/async/AsyncTaskExecutorIntegrationSpec.groovy
+++ b/src/test/groovy/org/onap/cps/ncmp/dmi/notifications/async/AsyncTaskExecutorIntegrationSpec.groovy
@@ -23,6 +23,7 @@ package org.onap.cps.ncmp.dmi.notifications.async
 import com.fasterxml.jackson.databind.ObjectMapper
 import org.onap.cps.ncmp.dmi.api.kafka.MessagingBaseSpec
 import org.onap.cps.ncmp.dmi.exception.HttpClientRequestException
+import org.onap.cps.ncmp.dmi.model.DataAccessRequest
 import org.onap.cps.ncmp.event.model.DmiAsyncRequestResponseEvent
 import org.spockframework.spring.SpringBean
 import org.springframework.boot.test.context.SpringBootTest
@@ -31,6 +32,7 @@ import org.springframework.test.annotation.DirtiesContext
 import org.testcontainers.spock.Testcontainers
 
 import java.time.Duration
+import java.util.function.Supplier
 
 @SpringBootTest(classes = [AsyncTaskExecutor, DmiAsyncRequestResponseEventProducer])
 @Testcontainers
@@ -42,6 +44,7 @@ class AsyncTaskExecutorIntegrationSpec extends MessagingBaseSpec {
         new DmiAsyncRequestResponseEventProducer(kafkaTemplate)
 
     def spiedObjectMapper = Spy(ObjectMapper)
+    def mockSupplier = Mock(Supplier)
 
     def objectUnderTest = new AsyncTaskExecutor(cpsAsyncRequestResponseEventProducer)
 
@@ -83,4 +86,25 @@ class AsyncTaskExecutorIntegrationSpec extends MessagingBaseSpec {
             assert event.getEventContent().getResponseCode() == '500'
     }
 
+    def 'Execute an Async Task using asyncTaskExecutor and throw an error'() {
+        given: 'A task to be executed'
+            def requestId = '123456'
+            def operationEnum = DataAccessRequest.OperationEnum.CREATE
+            def timeOut = 100
+        when: 'AsyncTask has been executed'
+            objectUnderTest.executeAsyncTask(taskSupplierForFailingTask(), TEST_TOPIC, requestId, operationEnum, timeOut)
+            def records = kafkaConsumer.poll(Duration.ofMillis(1500))
+        then: 'the record received is the event sent'
+            def record = records.iterator().next()
+            DmiAsyncRequestResponseEvent event  = spiedObjectMapper.readValue(record.value(), DmiAsyncRequestResponseEvent)
+        and: 'the status & code matches expected'
+            assert event.getEventContent().getResponseStatus() == 'Internal Server Error'
+            assert event.getEventContent().getResponseCode() == '500'
+
+    }
+
+    def taskSupplierForFailingTask() {
+        return () -> { throw new RuntimeException('original exception message') }
+    }
+
 }
\ No newline at end of file
diff --git a/src/test/groovy/org/onap/cps/ncmp/dmi/notifications/cmsubscription/CmNotificationSubscriptionDmiInEventConsumerSpec.groovy b/src/test/groovy/org/onap/cps/ncmp/dmi/notifications/cmsubscription/CmNotificationSubscriptionDmiInEventConsumerSpec.groovy
index 47953439..aa331c4b 100644
--- a/src/test/groovy/org/onap/cps/ncmp/dmi/notifications/cmsubscription/CmNotificationSubscriptionDmiInEventConsumerSpec.groovy
+++ b/src/test/groovy/org/onap/cps/ncmp/dmi/notifications/cmsubscription/CmNotificationSubscriptionDmiInEventConsumerSpec.groovy
@@ -73,23 +73,28 @@ class CmNotificationSubscriptionDmiInEventConsumerSpec extends MessagingBaseSpec
             objectUnderTest.dmiName = 'test-ncmp-dmi'
             objectUnderTest.cmNotificationSubscriptionResponseTopic = testTopic
             def correlationId = 'test-subscriptionId#test-ncmp-dmi'
-            def cmSubscriptionDmiOutEventData = new Data(statusCode: '1', statusMessage: 'ACCEPTED')
+            def cmSubscriptionDmiOutEventData = new Data(statusCode: subscriptionStatusCode, statusMessage: subscriptionStatusMessage)
             def subscriptionEventResponse =
                     new CmNotificationSubscriptionDmiOutEvent().withData(cmSubscriptionDmiOutEventData)
         and: 'consumer has a subscription'
             kafkaConsumer.subscribe([testTopic] as List<String>)
         when: 'an event is published'
             def eventKey = UUID.randomUUID().toString()
-            objectUnderTest.createAndSendCmNotificationSubscriptionDmiOutEvent(eventKey, "subscriptionCreatedStatus", correlationId, CmNotificationSubscriptionStatus.ACCEPTED)
+            objectUnderTest.createAndSendCmNotificationSubscriptionDmiOutEvent(eventKey, "subscriptionCreatedStatus", correlationId, subscriptionAcceptanceType)
         and: 'topic is polled'
             def records = kafkaConsumer.poll(Duration.ofMillis(1500))
-        then: 'poll returns one record'
+        then: 'poll returns one record and close kafkaConsumer'
             assert records.size() == 1
             def record = records.iterator().next()
+            kafkaConsumer.close()
         and: 'the record value matches the expected event value'
             def expectedValue = objectMapper.writeValueAsString(subscriptionEventResponse)
             assert expectedValue == record.value
             assert eventKey == record.key
+        where: 'given #scenario'
+            scenario                   | subscriptionAcceptanceType                 | subscriptionStatusCode | subscriptionStatusMessage
+            'Subscription is Accepted' | CmNotificationSubscriptionStatus.ACCEPTED  | '1'                    | 'ACCEPTED'
+            'Subscription is Rejected' | CmNotificationSubscriptionStatus.REJECTED  | '2'                    | 'REJECTED'
     }
 
     def 'Consume valid message.'() {