gerrit.onap Code Review - cps.git/blob

   1 /*
   2  *  ============LICENSE_START=======================================================
   3  *  Copyright (C) 2024-2025 OpenInfra Foundation Europe. All rights reserved.
   4  *  ================================================================================
   5  *  Licensed under the Apache License, Version 2.0 (the 'License');
   6  *  you may not use this file except in compliance with the License.
   7  *  You may obtain a copy of the License at
   8  *
   9  *        http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  *  Unless required by applicable law or agreed to in writing, software
  12  *  distributed under the License is distributed on an 'AS IS' BASIS,
  13  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  *  See the License for the specific language governing permissions and
  15  *  limitations under the License.
  16  *
  17  *  SPDX-License-Identifier: Apache-2.0
  18  *  ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.cps.integration.functional.ncmp.inventory
  22
  23 import org.apache.kafka.clients.consumer.KafkaConsumer
  24 import org.onap.cps.events.LegacyEvent
  25 import org.onap.cps.integration.KafkaTestContainer
  26 import org.onap.cps.integration.base.CpsIntegrationSpecBase
  27 import org.onap.cps.ncmp.api.NcmpResponseStatus
  28 import org.onap.cps.ncmp.api.inventory.models.CmHandleRegistrationResponse
  29 import org.onap.cps.ncmp.api.inventory.models.CmHandleState
  30 import org.onap.cps.ncmp.api.inventory.models.DmiPluginRegistration
  31 import org.onap.cps.ncmp.api.inventory.models.LockReasonCategory
  32 import org.onap.cps.ncmp.api.inventory.models.NcmpServiceCmHandle
  33 import org.onap.cps.ncmp.events.lcm.v1.LcmEvent
  34 import org.onap.cps.ncmp.impl.NetworkCmProxyInventoryFacadeImpl
  35
  36 import java.time.Duration
  37
  38 class CmHandleCreateSpec extends CpsIntegrationSpecBase {
  39
  40     NetworkCmProxyInventoryFacadeImpl objectUnderTest
  41     def uniqueId = 'my-new-cm-handle'
  42
  43     KafkaConsumer<String, LegacyEvent> kafkaConsumer
  44
  45     def setup() {
  46         objectUnderTest = networkCmProxyInventoryFacade
  47         subscribeAndClearPreviousMessages('test-group', 'ncmp-events')
  48     }
  49
  50     def cleanup() {
  51         kafkaConsumer.unsubscribe()
  52         kafkaConsumer.close()
  53     }
  54
  55     def 'CM Handle registration.'() {
  56         given: 'DMI will return modules when requested'
  57             dmiDispatcher1.moduleNamesPerCmHandleId['ch-1'] = ['M1', 'M2']
  58             dmiDispatcher1.moduleNamesPerCmHandleId[uniqueId] = ['M1', 'M2']
  59         when: 'a CM-handle is registered for creation'
  60             def cmHandleToCreate = new NcmpServiceCmHandle(cmHandleId: uniqueId,
  61                                                            alternateId: 'fdn1',
  62                                                            moduleSetTag: 'tag1',
  63                                                            dataProducerIdentifier: 'prod1',
  64                                                            publicProperties: [color:'green'])
  65             def dmiPluginRegistration = new DmiPluginRegistration(dmiPlugin: DMI1_URL, createdCmHandles: [cmHandleToCreate])
  66             def dmiPluginRegistrationResponse = objectUnderTest.updateDmiRegistration(dmiPluginRegistration)
  67         then: 'registration gives successful response'
  68             assert dmiPluginRegistrationResponse.createdCmHandles == [CmHandleRegistrationResponse.createSuccessResponse(uniqueId)]
  69         and: 'CM-handle is initially in ADVISED state'
  70             assert CmHandleState.ADVISED == objectUnderTest.getCmHandleCompositeState(uniqueId).cmHandleState
  71         then: 'the module sync watchdog is triggered'
  72             moduleSyncWatchdog.moduleSyncAdvisedCmHandles()
  73         then: 'CM-handle goes to READY state after module sync'
  74             assert CmHandleState.READY == objectUnderTest.getCmHandleCompositeState(uniqueId).cmHandleState
  75         and: 'the CM-handle has expected modules'
  76             assert ['M1', 'M2'] == objectUnderTest.getYangResourcesModuleReferences(uniqueId).moduleName.sort()
  77         then: 'get the last 2 messages and related headers'
  78             def consumerRecords = getLatestConsumerRecordsWithMaxPollOf1Second(kafkaConsumer, 2)
  79             def messages = []
  80             def headerMaps = []
  81             consumerRecords.each { consumerRecord ->
  82                 messages.add(jsonObjectMapper.convertJsonString(consumerRecord.value().toString(), LcmEvent))
  83                 headerMaps.add(getHeadersAsMap(consumerRecord))
  84             }
  85         and: 'both messages have the correct common attributes (that did not change)'
  86             assert messages.event.every {
  87                 it.cmHandleId == uniqueId &&
  88                 it.alternateId == 'fdn1' &&
  89                 it.moduleSetTag == 'tag1' &&
  90                 it.dataProducerIdentifier == 'prod1'
  91             }
  92         and: 'the header fields are populated correctly and stored as kafka headers too'
  93             validateEventHeaders(messages[0], headerMaps[0], 'create')
  94             validateEventHeaders(messages[1], headerMaps[1], 'update')
  95         and: 'the first lcm event has no old values and the initial attributes as new values state ADVISED'
  96             with(messages[0].event) {
  97                 assert oldValues == null
  98                 assert newValues.cmHandleState.value() == 'ADVISED'
  99                 assert newValues.dataSyncEnabled == null
 100                 assert newValues.cmHandleProperties[0] == [color:'green']
 101             }
 102         and: 'the next event is about update to READY state (new value), the old value for state is ADVISED'
 103             assert messages[1].event.oldValues.cmHandleState.value() == 'ADVISED'
 104             assert messages[1].event.newValues.cmHandleState.value() == 'READY'
 105         and: 'the cm handle (public) properties have not changed and are therefore null for old and new values'
 106             assert messages[1].event.oldValues.cmHandleProperties == null
 107             assert messages[1].event.newValues.cmHandleProperties == null
 108         and: 'the data sync flag goes from undefined to false'
 109             assert messages[1].event.oldValues.dataSyncEnabled == null
 110             assert messages[1].event.newValues.dataSyncEnabled == false
 111         and: 'there are no more messages to be read'
 112             assert getLatestConsumerRecordsWithMaxPollOf1Second(kafkaConsumer, 1).size() == 0
 113         cleanup: 'deregister CM handle'
 114             deregisterCmHandle(DMI1_URL, uniqueId)
 115     }
 116
 117     def 'CM Handle registration with DMI error during module sync.'() {
 118         given: 'DMI is not available to handle requests'
 119             dmiDispatcher1.isAvailable = false
 120         when: 'a CM-handle is registered for creation'
 121             def cmHandleToCreate = new NcmpServiceCmHandle(cmHandleId: 'ch-1')
 122             def dmiPluginRegistration = new DmiPluginRegistration(dmiPlugin: DMI1_URL, createdCmHandles: [cmHandleToCreate])
 123             objectUnderTest.updateDmiRegistration(dmiPluginRegistration)
 124         and: 'the module sync watchdog is triggered'
 125             moduleSyncWatchdog.moduleSyncAdvisedCmHandles()
 126         then: 'CM-handle goes to LOCKED state with reason MODULE_SYNC_FAILED'
 127             def cmHandleCompositeState = objectUnderTest.getCmHandleCompositeState('ch-1')
 128             assert cmHandleCompositeState.cmHandleState == CmHandleState.LOCKED
 129             assert cmHandleCompositeState.lockReason.lockReasonCategory == LockReasonCategory.MODULE_SYNC_FAILED
 130         and: 'CM-handle has no modules'
 131             assert objectUnderTest.getYangResourcesModuleReferences('ch-1').empty
 132         cleanup: 'deregister CM handle'
 133             deregisterCmHandle(DMI1_URL, 'ch-1')
 134     }
 135
 136     def 'Create a CM-handle with existing moduleSetTag.'() {
 137         given: 'DMI will return modules when requested'
 138             dmiDispatcher1.moduleNamesPerCmHandleId = ['ch-1': ['M1', 'M2'], 'ch-2': ['M1', 'M3']]
 139         and: 'existing CM-handles cm-1 with moduleSetTag "A", and cm-2 with moduleSetTag "B"'
 140             registerCmHandle(DMI1_URL, 'ch-1', 'A')
 141             registerCmHandle(DMI1_URL, 'ch-2', 'B')
 142         when: 'a CM-handle is registered for creation with moduleSetTag "B"'
 143             def cmHandleToCreate = new NcmpServiceCmHandle(cmHandleId: 'ch-3', moduleSetTag: 'B')
 144             objectUnderTest.updateDmiRegistration(new DmiPluginRegistration(dmiPlugin: DMI1_URL, createdCmHandles: [cmHandleToCreate]))
 145         and: 'the module sync watchdog is triggered'
 146             moduleSyncWatchdog.moduleSyncAdvisedCmHandles()
 147         then: 'the CM-handle goes to READY state'
 148             assert CmHandleState.READY == objectUnderTest.getCmHandleCompositeState('ch-3').cmHandleState
 149         and: 'the CM-handle has expected moduleSetTag'
 150             assert objectUnderTest.getNcmpServiceCmHandle('ch-3').moduleSetTag == 'B'
 151         and: 'the CM-handle has expected modules from module set "B": M1 and M3'
 152             assert ['M1', 'M3'] == objectUnderTest.getYangResourcesModuleReferences('ch-3').moduleName.sort()
 153         cleanup: 'deregister CM handles'
 154             deregisterCmHandles(DMI1_URL, ['ch-1', 'ch-2', 'ch-3'])
 155     }
 156
 157     def 'Create CM-handles with alternate IDs.'() {
 158         given: 'DMI will return modules for all CM-handles when requested'
 159             dmiDispatcher1.moduleNamesPerCmHandleId = (1..7).collectEntries{ ['ch-'+it, ['M1']] }
 160         and: 'an existing CM-handle with an alternate ID'
 161             registerCmHandle(DMI1_URL, 'ch-1', NO_MODULE_SET_TAG, 'existing-alt-id')
 162         and: 'an existing CM-handle with no alternate ID'
 163             registerCmHandle(DMI1_URL, 'ch-2', NO_MODULE_SET_TAG, NO_ALTERNATE_ID)
 164         when: 'a batch of CM-handles is registered for creation with various alternate IDs'
 165             def cmHandlesToCreate = [
 166                 new NcmpServiceCmHandle(cmHandleId: 'ch-3', alternateId: NO_ALTERNATE_ID),
 167                 new NcmpServiceCmHandle(cmHandleId: 'ch-4', alternateId: 'unique-alt-id'),
 168                 new NcmpServiceCmHandle(cmHandleId: 'ch-5', alternateId: 'existing-alt-id'),
 169                 new NcmpServiceCmHandle(cmHandleId: 'ch-6', alternateId: 'duplicate-alt-id'),
 170                 new NcmpServiceCmHandle(cmHandleId: 'ch-7', alternateId: 'duplicate-alt-id'),
 171             ]
 172             def dmiPluginRegistration = new DmiPluginRegistration(dmiPlugin: DMI1_URL, createdCmHandles: cmHandlesToCreate)
 173             def dmiPluginRegistrationResponse = objectUnderTest.updateDmiRegistration(dmiPluginRegistration)
 174         then: 'registration gives expected responses'
 175             assert dmiPluginRegistrationResponse.createdCmHandles.sort { it.cmHandle } == [
 176                 CmHandleRegistrationResponse.createSuccessResponse('ch-3'),
 177                 CmHandleRegistrationResponse.createSuccessResponse('ch-4'),
 178                 CmHandleRegistrationResponse.createFailureResponse('ch-5', NcmpResponseStatus.CM_HANDLE_ALREADY_EXIST),
 179                 CmHandleRegistrationResponse.createSuccessResponse('ch-6'),
 180                 CmHandleRegistrationResponse.createFailureResponse('ch-7', NcmpResponseStatus.CM_HANDLE_ALREADY_EXIST),
 181             ]
 182         cleanup: 'deregister CM handles'
 183             deregisterCmHandles(DMI1_URL, (1..7).collect{ 'ch-'+it })
 184     }
 185
 186     def 'CM Handle retry after failed module sync.'() {
 187         given: 'DMI is not initially available to handle requests'
 188             dmiDispatcher1.isAvailable = false
 189         when: 'CM-handles are registered for creation'
 190             def cmHandlesToCreate = [new NcmpServiceCmHandle(cmHandleId: 'ch-1'), new NcmpServiceCmHandle(cmHandleId: 'ch-2')]
 191             def dmiPluginRegistration = new DmiPluginRegistration(dmiPlugin: DMI1_URL, createdCmHandles: cmHandlesToCreate)
 192             objectUnderTest.updateDmiRegistration(dmiPluginRegistration)
 193         and: 'the module sync watchdog is triggered'
 194             moduleSyncWatchdog.moduleSyncAdvisedCmHandles()
 195         then: 'CM-handles go to LOCKED state'
 196             assert objectUnderTest.getCmHandleCompositeState('ch-1').cmHandleState == CmHandleState.LOCKED
 197         when: 'DMI is available for retry'
 198             dmiDispatcher1.moduleNamesPerCmHandleId = ['ch-1': ['M1', 'M2'], 'ch-2': ['M1', 'M2']]
 199             dmiDispatcher1.isAvailable = true
 200         and: 'the module sync watchdog is triggered TWICE'
 201             2.times { moduleSyncWatchdog.moduleSyncAdvisedCmHandles() }
 202         then: 'Both CM-handles go to READY state'
 203             ['ch-1', 'ch-2'].each { cmHandleId ->
 204                 assert objectUnderTest.getCmHandleCompositeState(cmHandleId).cmHandleState == CmHandleState.READY
 205             }
 206         and: 'Both CM-handles have expected modules'
 207             ['ch-1', 'ch-2'].each { cmHandleId ->
 208                 assert objectUnderTest.getYangResourcesModuleReferences(cmHandleId).moduleName.sort() == ['M1', 'M2']
 209             }
 210         cleanup: 'deregister CM handles'
 211             deregisterCmHandles(DMI1_URL, ['ch-1', 'ch-2'])
 212     }
 213
 214     def subscribeAndClearPreviousMessages(consumerGroupId, topicName) {
 215         kafkaConsumer = KafkaTestContainer.getLegacyEventConsumer(consumerGroupId)
 216         kafkaConsumer.subscribe([topicName])
 217         kafkaConsumer.poll(Duration.ofMillis(500))
 218     }
 219
 220     def getHeadersAsMap(consumerRecord) {
 221         def headersAsMap = [:]
 222         consumerRecord.headers().each { header ->
 223             def value = (new String((byte[]) header.value())).substring(7) // The raw header is prefixed with encoded type
 224             headersAsMap.put(header.key(), value)
 225         }
 226         return headersAsMap
 227     }
 228
 229     def validateEventHeaders(message, headerAsMap, expectedEventType) {
 230         with(message) {
 231             assert UUID.fromString(eventId) != null
 232             assert headerAsMap.get('eventId') == eventId
 233             assert eventCorrelationId == uniqueId
 234             assert headerAsMap.get('eventCorrelationId') == eventCorrelationId
 235             assert timestampIsVeryRecent(eventTime)
 236             assert headerAsMap.get('eventTime') == eventTime
 237             assert eventSource == 'org.onap.ncmp'
 238             assert headerAsMap.get('eventSource') == eventSource
 239             assert eventType == 'org.onap.ncmp.cmhandle-lcm-event.'+expectedEventType
 240             assert headerAsMap.get('eventType') == eventType
 241             assert eventSchema == 'org.onap.ncmp:cmhandle-lcm-event'
 242             assert headerAsMap.get('eventSchema') == eventSchema
 243             assert eventSchemaVersion == '1.0'
 244             assert headerAsMap.get('eventSchemaVersion') == eventSchemaVersion
 245         }
 246         return true
 247     }
 248
 249 }