[cps.git] /

/*
 *  ============LICENSE_START=======================================================
 *  Copyright (C) 2024-2026 OpenInfra Foundation Europe. All rights reserved.
 *  ================================================================================
 *  Licensed under the Apache License, Version 2.0 (the 'License');
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an 'AS IS' BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 *
 *  SPDX-License-Identifier: Apache-2.0
 *  ============LICENSE_END=========================================================
 */

package org.onap.cps.integration.functional.ncmp.inventory

import org.apache.kafka.clients.consumer.KafkaConsumer
import org.onap.cps.events.LegacyEvent
import org.onap.cps.integration.KafkaTestContainer
import org.onap.cps.integration.base.CpsIntegrationSpecBase
import org.onap.cps.ncmp.api.NcmpResponseStatus
import org.onap.cps.ncmp.api.inventory.models.CmHandleRegistrationResponse
import org.onap.cps.ncmp.api.inventory.models.DmiPluginRegistration
import org.onap.cps.ncmp.api.inventory.models.NcmpServiceCmHandle
import org.onap.cps.ncmp.events.lcm.LcmEventV1
import org.onap.cps.ncmp.events.lcm.LcmEventV2
import org.onap.cps.ncmp.impl.NetworkCmProxyInventoryFacadeImpl
import org.onap.cps.ncmp.impl.inventory.sync.lcm.LcmEventProducer
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.test.util.ReflectionTestUtils

import java.time.Duration

class CmHandleUpdateSpec extends CpsIntegrationSpecBase {

    NetworkCmProxyInventoryFacadeImpl objectUnderTest

    KafkaConsumer<String, LegacyEvent> kafkaConsumer

    @Autowired
    LcmEventProducer lcmEventProducer

    def setup() {
        objectUnderTest = networkCmProxyInventoryFacade
        subscribeAndClearPreviousMessages('test-group-for-update', 'ncmp-events')
    }

    def cleanup() {
        kafkaConsumer.unsubscribe()
        kafkaConsumer.close()
    }

    def 'Update of CM-handle with new or unchanged alternate ID succeeds.'() {
        given: 'DMI will return modules when requested'
            dmiDispatcher1.moduleNamesPerCmHandleId = ['ch-1': ['M1', 'M2']]
        and: 'existing CM-handle with alternate ID: #oldAlternateId'
            registerCmHandle(DMI1_URL, 'ch-1', NO_MODULE_SET_TAG, oldAlternateId)
        when: 'CM-handle is registered for update with new alternate ID: #newAlternateId'
            def cmHandleToUpdate = new NcmpServiceCmHandle(cmHandleId: 'ch-1', alternateId: newAlternateId)
            def dmiPluginRegistrationResponse =
                    objectUnderTest.updateDmiRegistration(new DmiPluginRegistration(dmiPlugin: DMI1_URL, updatedCmHandles: [cmHandleToUpdate]))
        then: 'registration gives successful response'
            assert dmiPluginRegistrationResponse.updatedCmHandles == [CmHandleRegistrationResponse.createSuccessResponse('ch-1')]
        and: 'the CM-handle has expected alternate ID'
            assert objectUnderTest.getNcmpServiceCmHandle('ch-1').alternateId == expectedAlternateId
        cleanup: 'deregister CM handles'
            deregisterCmHandle(DMI1_URL, 'ch-1')
        where: 'following alternate ids are used'
            oldAlternateId | newAlternateId || expectedAlternateId
            ''             | ''             || ''
            ''             | 'new'          || 'new'
            'old'          | 'old'          || 'old'
            'old'          | null           || 'old'
            'old'          | ''             || 'old'
            'old'          | '  '           || 'old'
    }

    def 'Update of CM-handle with previously set alternate ID fails.'() {
        given: 'DMI will return modules when requested'
            dmiDispatcher1.moduleNamesPerCmHandleId = ['ch-1': ['M1', 'M2']]
        and: 'existing CM-handle with alternate ID'
            registerCmHandle(DMI1_URL, 'ch-1', NO_MODULE_SET_TAG, 'original')
        when: 'a CM-handle is registered for update with new alternate ID'
            def cmHandleToUpdate = new NcmpServiceCmHandle(cmHandleId: 'ch-1', alternateId: 'new')
            def dmiPluginRegistrationResponse =
                    objectUnderTest.updateDmiRegistration(new DmiPluginRegistration(dmiPlugin: DMI1_URL, updatedCmHandles: [cmHandleToUpdate]))
        then: 'registration gives failure response, due to cm-handle already existing'
            assert dmiPluginRegistrationResponse.updatedCmHandles == [CmHandleRegistrationResponse.createFailureResponse('ch-1', NcmpResponseStatus.CM_HANDLE_ALREADY_EXIST)]
        and: 'the CM-handle still has the old alternate ID'
            assert objectUnderTest.getNcmpServiceCmHandle('ch-1').alternateId == 'original'
        cleanup: 'deregister CM handles'
            deregisterCmHandle(DMI1_URL, 'ch-1')
    }

    def 'CM Handle registration to verify changes in data producer identifier'() {
        given: 'DMI will return modules when requested'
            def cmHandleId = 'ch-id-for-update'
            dmiDispatcher1.moduleNamesPerCmHandleId[cmHandleId] = ['M1', 'M2']
        when: 'a CM-handle is registered for creation'
            def cmHandleToCreate = new NcmpServiceCmHandle(cmHandleId: cmHandleId)
            def dmiPluginRegistration = new DmiPluginRegistration(dmiPlugin: DMI1_URL, createdCmHandles: [cmHandleToCreate])
            def dmiPluginRegistrationResponse = objectUnderTest.updateDmiRegistration(dmiPluginRegistration)
        then: 'registration gives successful response'
            assert dmiPluginRegistrationResponse.createdCmHandles == [CmHandleRegistrationResponse.createSuccessResponse(cmHandleId)]
        then: 'the module sync watchdog is triggered'
            moduleSyncWatchdog.moduleSyncAdvisedCmHandles()
        and: 'flush and check there are 2 cm handle registration events (state transition from NONE to ADVISED and ADVISED to READY)'
            assert getLatestConsumerRecordsWithMaxPollOf1Second(kafkaConsumer, 2).size() == 2
        and: 'cm handle updated with the data producer identifier'
            def cmHandleToUpdate = new NcmpServiceCmHandle(cmHandleId: cmHandleId, dataProducerIdentifier: 'my-data-producer-id')
            def dmiPluginRegistrationForUpdate = new DmiPluginRegistration(dmiPlugin: DMI1_URL, updatedCmHandles: [cmHandleToUpdate])
            def dmiPluginRegistrationResponseForUpdate = objectUnderTest.updateDmiRegistration(dmiPluginRegistrationForUpdate)
        then: 'registration gives successful response'
            assert dmiPluginRegistrationResponseForUpdate.updatedCmHandles == [CmHandleRegistrationResponse.createSuccessResponse(cmHandleId)]
        and: 'get the latest message'
            def consumerRecords = getLatestConsumerRecordsWithMaxPollOf1Second(kafkaConsumer, 1)
        and: 'the message has the updated data producer identifier'
            def notificationMessages = []
            for (def consumerRecord : consumerRecords) {
                notificationMessages.add(jsonObjectMapper.convertJsonString(consumerRecord.value().toString(), LcmEventV1))
            }
            assert notificationMessages[0].event.cmHandleId.contains(cmHandleId)
            assert notificationMessages[0].event.dataProducerIdentifier == 'my-data-producer-id'
        cleanup: 'deregister CM handle'
            deregisterCmHandle(DMI1_URL, cmHandleId)
    }

    def 'CM Handle registration to verify changes in data producer identifier using V2 events'() {
        given: 'event schema version is set to v2'
            def originalEventSchemaVersion = ReflectionTestUtils.getField(lcmEventProducer, 'eventSchemaVersion')
            ReflectionTestUtils.setField(lcmEventProducer, 'eventSchemaVersion', 'v2')
        and: 'DMI will return modules when requested'
            def cmHandleId = 'ch-id-for-update-v2'
            dmiDispatcher1.moduleNamesPerCmHandleId[cmHandleId] = ['M1', 'M2']
        when: 'a CM-handle is registered for creation'
            def cmHandleToCreate = new NcmpServiceCmHandle(cmHandleId: cmHandleId)
            def dmiPluginRegistration = new DmiPluginRegistration(dmiPlugin: DMI1_URL, createdCmHandles: [cmHandleToCreate])
            def dmiPluginRegistrationResponse = objectUnderTest.updateDmiRegistration(dmiPluginRegistration)
        then: 'registration gives successful response'
            assert dmiPluginRegistrationResponse.createdCmHandles == [CmHandleRegistrationResponse.createSuccessResponse(cmHandleId)]
        then: 'the module sync watchdog is triggered'
            moduleSyncWatchdog.moduleSyncAdvisedCmHandles()
        and: 'flush and check there are 2 cm handle registration events (state transition from NONE to ADVISED and ADVISED to READY)'
            assert getLatestConsumerRecordsWithMaxPollOf1Second(kafkaConsumer, 2).size() == 2
        and: 'cm handle updated with the data producer identifier'
            def cmHandleToUpdate = new NcmpServiceCmHandle(cmHandleId: cmHandleId, dataProducerIdentifier: 'my-data-producer-id')
            def dmiPluginRegistrationForUpdate = new DmiPluginRegistration(dmiPlugin: DMI1_URL, updatedCmHandles: [cmHandleToUpdate])
            def dmiPluginRegistrationResponseForUpdate = objectUnderTest.updateDmiRegistration(dmiPluginRegistrationForUpdate)
        then: 'registration gives successful response'
            assert dmiPluginRegistrationResponseForUpdate.updatedCmHandles == [CmHandleRegistrationResponse.createSuccessResponse(cmHandleId)]
        and: 'get the latest message'
            def consumerRecords = getLatestConsumerRecordsWithMaxPollOf1Second(kafkaConsumer, 1)
        and: 'the V2 message has the updated data producer identifier in newValues'
            def notificationMessages = []
            for (def consumerRecord : consumerRecords) {
                notificationMessages.add(jsonObjectMapper.convertJsonString(consumerRecord.value().toString(), LcmEventV2))
            }
            assert notificationMessages[0].event.cmHandleId.contains(cmHandleId)
            assert notificationMessages[0].event.newValues['dataProducerIdentifier'] == 'my-data-producer-id'
        cleanup: 'restore original event schema version and deregister CM handle'
            ReflectionTestUtils.setField(lcmEventProducer, 'eventSchemaVersion', originalEventSchemaVersion)
            deregisterCmHandle(DMI1_URL, cmHandleId)
    }

    def subscribeAndClearPreviousMessages(consumerGroupId, topicName) {
        kafkaConsumer = KafkaTestContainer.getLegacyEventConsumer(consumerGroupId)
        kafkaConsumer.subscribe([topicName])
        kafkaConsumer.poll(Duration.ofMillis(500))
    }
}