diff options
author | 2024-06-04 15:10:48 +0000 | |
---|---|---|
committer | 2024-06-04 15:10:48 +0000 | |
commit | c056bee2fc0f52055464839248abfa4d16c8f89c (patch) | |
tree | d035a2c954a5b5841aa163064356f901ce58961b | |
parent | daa92d74b41f4b3e1ee038532d2893bdec799bc0 (diff) | |
parent | 44504115d9fdd04b866169af9d4789fdb5ce8f90 (diff) |
Merge "CM SUBSCRIPTION: Add delete use case #1 (Client-NCMP)"
5 files changed, 114 insertions, 28 deletions
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/consumer/CmNotificationSubscriptionNcmpInEventConsumer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/consumer/CmNotificationSubscriptionNcmpInEventConsumer.java index 2c544b7b6..fb3388c11 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/consumer/CmNotificationSubscriptionNcmpInEventConsumer.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/consumer/CmNotificationSubscriptionNcmpInEventConsumer.java @@ -23,11 +23,13 @@ package org.onap.cps.ncmp.api.impl.events.cmsubscription.consumer; import static org.onap.cps.ncmp.api.impl.events.mapper.CloudEventMapper.toTargetEvent; import io.cloudevents.CloudEvent; +import java.util.List; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.onap.cps.ncmp.api.impl.events.cmsubscription.service.CmNotificationSubscriptionHandlerService; import org.onap.cps.ncmp.events.cmnotificationsubscription_merge1_0_0.client_to_ncmp.CmNotificationSubscriptionNcmpInEvent; +import org.onap.cps.ncmp.events.cmnotificationsubscription_merge1_0_0.client_to_ncmp.Predicate; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component; @@ -53,10 +55,16 @@ public class CmNotificationSubscriptionNcmpInEventConsumer { cmNotificationSubscriptionNcmpInEvent.getData().getSubscriptionId()); final String subscriptionId = cmNotificationSubscriptionNcmpInEvent.getData().getSubscriptionId(); + final List<Predicate> predicates = cmNotificationSubscriptionNcmpInEvent.getData().getPredicates(); if ("subscriptionCreateRequest".equals(cloudEvent.getType())) { - log.info("Subscription for source {} with subscription id {} ...", cloudEvent.getSource(), subscriptionId); - cmNotificationSubscriptionHandlerService.processSubscriptionCreateRequest( - cmNotificationSubscriptionNcmpInEvent); + log.info("Subscription create request for source {} with subscription id {} ...", + cloudEvent.getSource(), subscriptionId); + cmNotificationSubscriptionHandlerService.processSubscriptionCreateRequest(subscriptionId, predicates); + } + if ("subscriptionDeleteRequest".equals(cloudEvent.getType())) { + log.info("Subscription delete request for source {} with subscription id {} ...", + cloudEvent.getSource(), subscriptionId); + cmNotificationSubscriptionHandlerService.processSubscriptionDeleteRequest(subscriptionId, predicates); } } }
\ No newline at end of file diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/service/CmNotificationSubscriptionHandlerService.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/service/CmNotificationSubscriptionHandlerService.java index 536693ee4..1c52ffa79 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/service/CmNotificationSubscriptionHandlerService.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/service/CmNotificationSubscriptionHandlerService.java @@ -20,16 +20,25 @@ package org.onap.cps.ncmp.api.impl.events.cmsubscription.service; -import org.onap.cps.ncmp.events.cmnotificationsubscription_merge1_0_0.client_to_ncmp.CmNotificationSubscriptionNcmpInEvent; +import java.util.List; +import org.onap.cps.ncmp.events.cmnotificationsubscription_merge1_0_0.client_to_ncmp.Predicate; public interface CmNotificationSubscriptionHandlerService { /** - * Process cm notification subscription request. + * Process cm notification subscription create request. * - * @param cmNotificationSubscriptionNcmpInEvent CM Notification Subscription event + * @param subscriptionId subscription id + * @param predicates subscription predicates */ - void processSubscriptionCreateRequest( - final CmNotificationSubscriptionNcmpInEvent cmNotificationSubscriptionNcmpInEvent); + void processSubscriptionCreateRequest(final String subscriptionId, final List<Predicate> predicates); -} + /** + * Process cm notification subscription delete request. + * + * @param subscriptionId subscription id + * @param predicates subscription predicates + */ + void processSubscriptionDeleteRequest(final String subscriptionId, final List<Predicate> predicates); + +}
\ No newline at end of file diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/service/CmNotificationSubscriptionHandlerServiceImpl.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/service/CmNotificationSubscriptionHandlerServiceImpl.java index 128c6751c..08e3c9552 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/service/CmNotificationSubscriptionHandlerServiceImpl.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/service/CmNotificationSubscriptionHandlerServiceImpl.java @@ -33,7 +33,6 @@ import org.onap.cps.ncmp.api.impl.events.cmsubscription.DmiCmNotificationSubscri import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.CmNotificationSubscriptionStatus; import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.DmiCmNotificationSubscriptionDetails; import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.DmiCmNotificationSubscriptionPredicate; -import org.onap.cps.ncmp.events.cmnotificationsubscription_merge1_0_0.client_to_ncmp.CmNotificationSubscriptionNcmpInEvent; import org.onap.cps.ncmp.events.cmnotificationsubscription_merge1_0_0.client_to_ncmp.Predicate; import org.onap.cps.ncmp.events.cmnotificationsubscription_merge1_0_0.ncmp_to_dmi.CmNotificationSubscriptionDmiInEvent; import org.onap.cps.ncmp.events.cmsubscription_merge1_0_0.ncmp_to_client.CmNotificationSubscriptionNcmpOutEvent; @@ -50,27 +49,32 @@ public class CmNotificationSubscriptionHandlerServiceImpl implements CmNotificat private final DmiCmNotificationSubscriptionCacheHandler dmiCmNotificationSubscriptionCacheHandler; @Override - public void processSubscriptionCreateRequest( - final CmNotificationSubscriptionNcmpInEvent cmNotificationSubscriptionNcmpInEvent) { - final String subscriptionId = cmNotificationSubscriptionNcmpInEvent.getData().getSubscriptionId(); - final List<Predicate> predicates = cmNotificationSubscriptionNcmpInEvent.getData().getPredicates(); - + public void processSubscriptionCreateRequest(final String subscriptionId, final List<Predicate> predicates) { if (cmNotificationSubscriptionPersistenceService.isUniqueSubscriptionId(subscriptionId)) { dmiCmNotificationSubscriptionCacheHandler.add(subscriptionId, predicates); handleCmNotificationSubscriptionDelta(subscriptionId); - scheduleCmNotificationSubscriptionNcmpOutEventResponse(subscriptionId); + scheduleCmNotificationSubscriptionNcmpOutEventResponse(subscriptionId, + "subscriptionCreateResponse"); } else { rejectAndPublishCmNotificationSubscriptionCreateRequest(subscriptionId, predicates); } } - private void scheduleCmNotificationSubscriptionNcmpOutEventResponse(final String subscriptionId) { + @Override + public void processSubscriptionDeleteRequest(final String subscriptionId, final List<Predicate> predicates) { + dmiCmNotificationSubscriptionCacheHandler.add(subscriptionId, predicates); + sendSubscriptionDeleteRequestToDmi(subscriptionId); + scheduleCmNotificationSubscriptionNcmpOutEventResponse(subscriptionId, "subscriptionDeleteResponse"); + } + + private void scheduleCmNotificationSubscriptionNcmpOutEventResponse(final String subscriptionId, + final String eventType) { cmNotificationSubscriptionEventsHandler.publishCmNotificationSubscriptionNcmpOutEvent(subscriptionId, - "subscriptionCreateResponse", null, true); + eventType, null, true); } private void rejectAndPublishCmNotificationSubscriptionCreateRequest(final String subscriptionId, - final List<Predicate> predicates) { + final List<Predicate> predicates) { final Set<String> subscriptionTargetFilters = predicates.stream().flatMap(predicate -> predicate.getTargetFilter().stream()) .collect(Collectors.toSet()); @@ -99,8 +103,9 @@ public class CmNotificationSubscriptionHandlerServiceImpl implements CmNotificat } private void publishCmNotificationSubscriptionDmiInEventPerDmi(final String subscriptionId, - final String dmiPluginName, - final List<DmiCmNotificationSubscriptionPredicate> dmiCmNotificationSubscriptionPredicates) { + final String dmiPluginName, + final List<DmiCmNotificationSubscriptionPredicate> + dmiCmNotificationSubscriptionPredicates) { final CmNotificationSubscriptionDmiInEvent cmNotificationSubscriptionDmiInEvent = cmNotificationSubscriptionMappersHandler.toCmNotificationSubscriptionDmiInEvent( dmiCmNotificationSubscriptionPredicates); @@ -109,9 +114,21 @@ public class CmNotificationSubscriptionHandlerServiceImpl implements CmNotificat } private void acceptAndPublishCmNotificationSubscriptionNcmpOutEventPerDmi(final String subscriptionId, - final String dmiPluginName) { + final String dmiPluginName) { dmiCmNotificationSubscriptionCacheHandler.updateDmiCmNotificationSubscriptionStatusPerDmi(subscriptionId, dmiPluginName, CmNotificationSubscriptionStatus.ACCEPTED); dmiCmNotificationSubscriptionCacheHandler.persistIntoDatabasePerDmi(subscriptionId, dmiPluginName); } -} + + private void sendSubscriptionDeleteRequestToDmi(final String subscriptionId) { + final Map<String, DmiCmNotificationSubscriptionDetails> dmiCmNotificationSubscriptionDetailsMap = + dmiCmNotificationSubscriptionCacheHandler.get(subscriptionId); + dmiCmNotificationSubscriptionDetailsMap.forEach((dmiPluginName, dmiCmNotificationSubscriptionDetails) -> { + final CmNotificationSubscriptionDmiInEvent cmNotificationSubscriptionDmiInEvent = + cmNotificationSubscriptionMappersHandler.toCmNotificationSubscriptionDmiInEvent( + dmiCmNotificationSubscriptionDetails.getDmiCmNotificationSubscriptionPredicates()); + cmNotificationSubscriptionEventsHandler.publishCmNotificationSubscriptionDmiInEvent(subscriptionId, + dmiPluginName, "subscriptionDeleteRequest", cmNotificationSubscriptionDmiInEvent); + }); + } +}
\ No newline at end of file diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmNotificationSubscriptionNcmpInEventConsumerSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmNotificationSubscriptionNcmpInEventConsumerSpec.groovy index f07f3c1e6..01a92c02f 100644 --- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmNotificationSubscriptionNcmpInEventConsumerSpec.groovy +++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmNotificationSubscriptionNcmpInEventConsumerSpec.groovy @@ -78,11 +78,34 @@ class CmNotificationSubscriptionNcmpInEventConsumerSpec extends MessagingBaseSpe def loggingEvent = getLoggingEvent() assert loggingEvent.level == Level.INFO and: 'the log indicates the task completed successfully' - assert loggingEvent.formattedMessage == 'Subscription for source some-resource with subscription id test-id ...' + assert loggingEvent.formattedMessage == 'Subscription create request for source some-resource with subscription id test-id ...' and: 'the subscription handler service is called once' - 1 * mockCmNotificationSubscriptionHandlerService.processSubscriptionCreateRequest(_) + 1 * mockCmNotificationSubscriptionHandlerService.processSubscriptionCreateRequest('test-id',_) } + def 'Consume valid CmNotificationSubscriptionNcmpInEvent delete message'() { + given: 'a cmNotificationSubscription event' + def jsonData = TestUtils.getResourceFileContent('cmSubscription/cmNotificationSubscriptionNcmpInEvent.json') + def testEventSent = jsonObjectMapper.convertJsonString(jsonData, CmNotificationSubscriptionNcmpInEvent.class) + def testCloudEventSent = CloudEventBuilder.v1() + .withData(objectMapper.writeValueAsBytes(testEventSent)) + .withId('sub-id') + .withType('subscriptionDeleteRequest') + .withSource(URI.create('some-resource')) + .withExtension('correlationid', 'test-cmhandle1').build() + def consumerRecord = new ConsumerRecord<String, CloudEvent>('topic-name', 0, 0, 'event-key', testCloudEventSent) + when: 'the valid event is consumed' + objectUnderTest.consumeSubscriptionEvent(consumerRecord) + then: 'an event is logged with level INFO' + def loggingEvent = getLoggingEvent() + assert loggingEvent.level == Level.INFO + and: 'the log indicates the task completed successfully' + assert loggingEvent.formattedMessage == 'Subscription delete request for source some-resource with subscription id test-id ...' + and: 'the subscription handler service is called once' + 1 * mockCmNotificationSubscriptionHandlerService.processSubscriptionDeleteRequest('test-id',_) + } + + def getLoggingEvent() { return logger.list[1] } diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/cmsubscription/service/CmNotificationSubscriptionHandlerServiceImplSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/cmsubscription/service/CmNotificationSubscriptionHandlerServiceImplSpec.groovy index 9156ae910..982150ec0 100644 --- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/cmsubscription/service/CmNotificationSubscriptionHandlerServiceImplSpec.groovy +++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/cmsubscription/service/CmNotificationSubscriptionHandlerServiceImplSpec.groovy @@ -57,6 +57,9 @@ class CmNotificationSubscriptionHandlerServiceImplSpec extends Specification{ def testEventConsumed = jsonObjectMapper.convertJsonString(jsonData, CmNotificationSubscriptionNcmpInEvent.class) def testListOfDeltaPredicates = [new DmiCmNotificationSubscriptionPredicate(['ch1'].toSet(), DatastoreType.PASSTHROUGH_OPERATIONAL, ['/a/b'].toSet())] mockCmNotificationSubscriptionPersistenceService.isUniqueSubscriptionId("test-id") >> true + and: 'relevant details is extracted from the event' + def subscriptionId = testEventConsumed.getData().getSubscriptionId() + def predicates = testEventConsumed.getData().getPredicates() and: 'the cache handler returns for relevant subscription id' 1 * mockDmiCmNotificationSubscriptionCacheHandler.get("test-id") >> testSubscriptionDetailsMap and: 'the delta predicates is returned' @@ -66,7 +69,7 @@ class CmNotificationSubscriptionHandlerServiceImplSpec extends Specification{ 1 * mockCmNotificationSubscriptionMappersHandler .toCmNotificationSubscriptionDmiInEvent(testListOfDeltaPredicates) >> testDmiInEvent when: 'the valid and unique event is consumed' - objectUnderTest.processSubscriptionCreateRequest(testEventConsumed) + objectUnderTest.processSubscriptionCreateRequest(subscriptionId, predicates) then: 'the subscription cache handler is called once' 1 * mockDmiCmNotificationSubscriptionCacheHandler.add('test-id',_) and: 'the events handler method to publish DMI event is called correct number of times with the correct parameters' @@ -88,7 +91,7 @@ class CmNotificationSubscriptionHandlerServiceImplSpec extends Specification{ and: 'the delta predicates is returned' 1 * mockCmNotificationSubscriptionDelta.getDelta(_) >> noDeltaPredicates when: 'the valid and unique event is consumed' - objectUnderTest.processSubscriptionCreateRequest(testEventConsumed) + objectUnderTest.processSubscriptionCreateRequest('test-id', noDeltaPredicates) then: 'the subscription cache handler is called once' 1 * mockDmiCmNotificationSubscriptionCacheHandler.add('test-id', _) and: 'the subscription details are updated in the cache' @@ -103,16 +106,42 @@ class CmNotificationSubscriptionHandlerServiceImplSpec extends Specification{ def jsonData = TestUtils.getResourceFileContent('cmSubscription/cmNotificationSubscriptionNcmpInEvent.json') def testEventConsumed = jsonObjectMapper.convertJsonString(jsonData, CmNotificationSubscriptionNcmpInEvent.class) mockCmNotificationSubscriptionPersistenceService.isUniqueSubscriptionId('test-id') >> false + and: 'relevant details is extracted from the event' + def subscriptionId = testEventConsumed.getData().getSubscriptionId() + def predicates = testEventConsumed.getData().getPredicates() and: 'the NCMP out in event mapper returns an event for rejected request' def testNcmpOutEvent = new CmNotificationSubscriptionNcmpOutEvent() 1 * mockCmNotificationSubscriptionMappersHandler.toCmNotificationSubscriptionNcmpOutEventForRejectedRequest( "test-id",_) >> testNcmpOutEvent when: 'the valid but non-unique event is consumed' - objectUnderTest.processSubscriptionCreateRequest(testEventConsumed) + objectUnderTest.processSubscriptionCreateRequest(subscriptionId, predicates) then: 'the events handler method to publish DMI event is never called' 0 * mockCmNotificationSubscriptionEventsHandler.publishCmNotificationSubscriptionDmiInEvent(_,_,_,_) and: 'the events handler method to publish NCMP out event is called once' 1 * mockCmNotificationSubscriptionEventsHandler.publishCmNotificationSubscriptionNcmpOutEvent( 'test-id', 'subscriptionCreateResponse', testNcmpOutEvent, false) } + + def 'Consume valid CmNotificationSubscriptionNcmpInEvent delete message'() { + given: 'a cmNotificationSubscriptionNcmp in event for delete' + def jsonData = TestUtils.getResourceFileContent('cmSubscription/cmNotificationSubscriptionNcmpInEvent.json') + def testEventConsumed = jsonObjectMapper.convertJsonString(jsonData, CmNotificationSubscriptionNcmpInEvent.class) + and: 'relevant details is extracted from the event' + def subscriptionId = testEventConsumed.getData().getSubscriptionId() + def predicates = testEventConsumed.getData().getPredicates() + and: 'the cache handler returns for relevant subscription id' + 1 * mockDmiCmNotificationSubscriptionCacheHandler.get('test-id') >> testSubscriptionDetailsMap + when: 'the valid and unique event is consumed' + objectUnderTest.processSubscriptionDeleteRequest(subscriptionId, predicates) + then: 'the subscription cache handler is called once' + 1 * mockDmiCmNotificationSubscriptionCacheHandler.add('test-id', predicates) + and: 'the mapper handler to get DMI in event is called once' + 1 * mockCmNotificationSubscriptionMappersHandler.toCmNotificationSubscriptionDmiInEvent(_) + and: 'the events handler method to publish DMI event is called correct number of times with the correct parameters' + testSubscriptionDetailsMap.size() * mockCmNotificationSubscriptionEventsHandler.publishCmNotificationSubscriptionDmiInEvent( + 'test-id', 'dmi-1', 'subscriptionDeleteRequest', _) + and: 'we schedule to send the response after configured time from the cache' + 1 * mockCmNotificationSubscriptionEventsHandler.publishCmNotificationSubscriptionNcmpOutEvent( + 'test-id', 'subscriptionDeleteResponse', null, true) + } } |