diff options
Diffstat (limited to 'integration-test')
13 files changed, 286 insertions, 67 deletions
diff --git a/integration-test/src/test/groovy/org/onap/cps/integration/base/CpsIntegrationSpecBase.groovy b/integration-test/src/test/groovy/org/onap/cps/integration/base/CpsIntegrationSpecBase.groovy index 587cbae619..759eccd966 100644 --- a/integration-test/src/test/groovy/org/onap/cps/integration/base/CpsIntegrationSpecBase.groovy +++ b/integration-test/src/test/groovy/org/onap/cps/integration/base/CpsIntegrationSpecBase.groovy @@ -53,6 +53,7 @@ import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMock import org.springframework.boot.test.context.SpringBootTest import org.springframework.context.annotation.ComponentScan import org.springframework.data.jpa.repository.config.EnableJpaRepositories +import org.springframework.test.context.ActiveProfiles import org.springframework.test.web.servlet.MockMvc import org.testcontainers.spock.Testcontainers import spock.lang.Shared @@ -61,6 +62,7 @@ import spock.util.concurrent.PollingConditions import java.time.OffsetDateTime import java.time.format.DateTimeFormatter +import java.util.concurrent.BlockingQueue import static org.onap.cps.ncmp.impl.inventory.NcmpPersistence.NCMP_DATASPACE_NAME import static org.onap.cps.ncmp.impl.inventory.NcmpPersistence.NCMP_DMI_REGISTRY_ANCHOR @@ -73,6 +75,7 @@ import static org.onap.cps.ncmp.impl.inventory.NcmpPersistence.NCMP_DMI_REGISTRY @EnableJpaRepositories(basePackageClasses = [DataspaceRepository]) @ComponentScan(basePackages = ['org.onap.cps']) @EntityScan('org.onap.cps.ri.models') +@ActiveProfiles('module-sync-delayed') abstract class CpsIntegrationSpecBase extends Specification { @Shared @@ -118,6 +121,9 @@ abstract class CpsIntegrationSpecBase extends Specification { ModuleSyncWatchdog moduleSyncWatchdog @Autowired + BlockingQueue<DataNode> moduleSyncWorkQueue + + @Autowired JsonObjectMapper jsonObjectMapper @Autowired @@ -244,26 +250,39 @@ abstract class CpsIntegrationSpecBase extends Specification { } def registerCmHandle(dmiPlugin, cmHandleId, moduleSetTag, alternateId) { - def cmHandleToCreate = new NcmpServiceCmHandle(cmHandleId: cmHandleId, moduleSetTag: moduleSetTag, alternateId: alternateId) - networkCmProxyInventoryFacade.updateDmiRegistrationAndSyncModule(new DmiPluginRegistration(dmiPlugin: dmiPlugin, createdCmHandles: [cmHandleToCreate])) + registerCmHandleWithoutWaitForReady(dmiPlugin, cmHandleId, moduleSetTag, alternateId) + moduleSyncWatchdog.moduleSyncAdvisedCmHandles() new PollingConditions().within(MODULE_SYNC_WAIT_TIME_IN_SECONDS, () -> { CmHandleState.READY == networkCmProxyInventoryFacade.getCmHandleCompositeState(cmHandleId).cmHandleState }) } + def registerCmHandleWithoutWaitForReady(dmiPlugin, cmHandleId, moduleSetTag, alternateId) { + def cmHandleToCreate = new NcmpServiceCmHandle(cmHandleId: cmHandleId, moduleSetTag: moduleSetTag, alternateId: alternateId) + networkCmProxyInventoryFacade.updateDmiRegistration(new DmiPluginRegistration(dmiPlugin: dmiPlugin, createdCmHandles: [cmHandleToCreate])) + } + + def registerSequenceOfCmHandlesWithoutWaitForReady(dmiPlugin, moduleSetTag, numberOfCmHandles) { + def cmHandles = [] + (1..numberOfCmHandles).each { + def cmHandle = new NcmpServiceCmHandle(cmHandleId: 'ch-'+it, moduleSetTag: moduleSetTag, alternateId: NO_ALTERNATE_ID) + cmHandles.add(cmHandle) + } + networkCmProxyInventoryFacade.updateDmiRegistration(new DmiPluginRegistration(dmiPlugin: dmiPlugin, createdCmHandles: cmHandles)) + } + def deregisterCmHandle(dmiPlugin, cmHandleId) { deregisterCmHandles(dmiPlugin, [cmHandleId]) } def deregisterCmHandles(dmiPlugin, cmHandleIds) { - networkCmProxyInventoryFacade.updateDmiRegistrationAndSyncModule(new DmiPluginRegistration(dmiPlugin: dmiPlugin, removedCmHandles: cmHandleIds)) + networkCmProxyInventoryFacade.updateDmiRegistration(new DmiPluginRegistration(dmiPlugin: dmiPlugin, removedCmHandles: cmHandleIds)) } - def overrideCmHandleLastUpdateTime(cmHandleId, newUpdateTime) { - String ISO_TIMESTAMP_PATTERN = "yyyy-MM-dd'T'HH:mm:ss.SSSZ"; - DateTimeFormatter ISO_TIMESTAMP_FORMATTER = DateTimeFormatter.ofPattern(ISO_TIMESTAMP_PATTERN); - def jsonForUpdate = '{ "state": { "last-update-time": "%s" } }'.formatted(ISO_TIMESTAMP_FORMATTER.format(newUpdateTime)) - cpsDataService.updateNodeLeaves(NCMP_DATASPACE_NAME, NCMP_DMI_REGISTRY_ANCHOR, - NCMP_DMI_REGISTRY_PARENT + "/cm-handles[@id='${cmHandleId}']", jsonForUpdate, now, ContentType.JSON) + def deregisterSequenceOfCmHandles(dmiPlugin, numberOfCmHandles) { + def cmHandleIds = [] + (1..numberOfCmHandles).each { cmHandleIds.add('ch-'+it) } + networkCmProxyInventoryFacade.updateDmiRegistration(new DmiPluginRegistration(dmiPlugin: dmiPlugin, removedCmHandles: cmHandleIds)) } + } diff --git a/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleCreateSpec.groovy b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleCreateSpec.groovy index 10a9f15e21..00ce38fa2d 100644 --- a/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleCreateSpec.groovy +++ b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleCreateSpec.groovy @@ -20,7 +20,7 @@ package org.onap.cps.integration.functional.ncmp -import org.apache.kafka.common.TopicPartition +import org.apache.kafka.clients.consumer.KafkaConsumer import org.apache.kafka.common.serialization.StringDeserializer import org.onap.cps.integration.KafkaTestContainer import org.onap.cps.integration.base.CpsIntegrationSpecBase @@ -35,54 +35,69 @@ import org.onap.cps.ncmp.impl.inventory.models.LockReasonCategory import spock.util.concurrent.PollingConditions import java.time.Duration -import java.time.OffsetDateTime class CmHandleCreateSpec extends CpsIntegrationSpecBase { NetworkCmProxyInventoryFacade objectUnderTest + def uniqueId = 'ch-unique-id-for-create-test' - def kafkaConsumer = KafkaTestContainer.getConsumer('ncmp-group', StringDeserializer.class) + static KafkaConsumer kafkaConsumer def setup() { objectUnderTest = networkCmProxyInventoryFacade + subscribeAndClearPreviousMessages() } - def 'CM Handle registration is successful.'() { + def cleanupSpec() { + kafkaConsumer.unsubscribe() + kafkaConsumer.close() + } + + def 'CM Handle registration.'() { given: 'DMI will return modules when requested' dmiDispatcher1.moduleNamesPerCmHandleId['ch-1'] = ['M1', 'M2'] - - and: 'consumer subscribed to topic' - kafkaConsumer.subscribe(['ncmp-events']) + dmiDispatcher1.moduleNamesPerCmHandleId[uniqueId] = ['M1', 'M2'] when: 'a CM-handle is registered for creation' - def cmHandleToCreate = new NcmpServiceCmHandle(cmHandleId: 'ch-1') + def cmHandleToCreate = new NcmpServiceCmHandle(cmHandleId: uniqueId) def dmiPluginRegistration = new DmiPluginRegistration(dmiPlugin: DMI1_URL, createdCmHandles: [cmHandleToCreate]) - def dmiPluginRegistrationResponse = objectUnderTest.updateDmiRegistrationAndSyncModule(dmiPluginRegistration) + def dmiPluginRegistrationResponse = objectUnderTest.updateDmiRegistration(dmiPluginRegistration) then: 'registration gives successful response' - assert dmiPluginRegistrationResponse.createdCmHandles == [CmHandleRegistrationResponse.createSuccessResponse('ch-1')] + assert dmiPluginRegistrationResponse.createdCmHandles == [CmHandleRegistrationResponse.createSuccessResponse(uniqueId)] and: 'CM-handle is initially in ADVISED state' - assert CmHandleState.ADVISED == objectUnderTest.getCmHandleCompositeState('ch-1').cmHandleState + assert CmHandleState.ADVISED == objectUnderTest.getCmHandleCompositeState(uniqueId).cmHandleState + + then: 'the module sync watchdog is triggered' + moduleSyncWatchdog.moduleSyncAdvisedCmHandles() - and: 'CM-handle goes to READY state after module sync' + then: 'CM-handle goes to READY state after module sync' new PollingConditions().within(MODULE_SYNC_WAIT_TIME_IN_SECONDS, () -> { - assert CmHandleState.READY == objectUnderTest.getCmHandleCompositeState('ch-1').cmHandleState + assert CmHandleState.READY == objectUnderTest.getCmHandleCompositeState(uniqueId).cmHandleState }) - and: 'the messages is polled' - def message = kafkaConsumer.poll(Duration.ofMillis(10000)) - def records = message.records(new TopicPartition('ncmp-events', 0)) + and: 'the CM-handle has expected modules' + assert ['M1', 'M2'] == objectUnderTest.getYangResourcesModuleReferences(uniqueId).moduleName.sort() - and: 'the newest lcm event notification is received with READY state' - def notificationMessage = jsonObjectMapper.convertJsonString(records.last().value().toString(), LcmEvent) - assert notificationMessage.event.newValues.cmHandleState.value() == 'READY' + then: 'get the latest messages' + def consumerRecords = getLatestConsumerRecords() - and: 'the CM-handle has expected modules' - assert ['M1', 'M2'] == objectUnderTest.getYangResourcesModuleReferences('ch-1').moduleName.sort() + and: 'both converted messages are for the correct cm handle' + def notificationMessages = [] + for (def consumerRecord : consumerRecords) { + notificationMessages.add(jsonObjectMapper.convertJsonString(consumerRecord.value().toString(), LcmEvent)) + } + assert notificationMessages.event.cmHandleId == [ uniqueId, uniqueId ] + + and: 'the oldest event is about the update to ADVISED state' + notificationMessages[0].event.newValues.cmHandleState.value() == 'ADVISED' + + and: 'the next event is about update to READY state' + notificationMessages[1].event.newValues.cmHandleState.value() == 'READY' cleanup: 'deregister CM handle' - deregisterCmHandle(DMI1_URL, 'ch-1') + deregisterCmHandle(DMI1_URL, uniqueId) } def 'CM Handle goes to LOCKED state when DMI gives error during module sync.'() { @@ -92,7 +107,10 @@ class CmHandleCreateSpec extends CpsIntegrationSpecBase { when: 'a CM-handle is registered for creation' def cmHandleToCreate = new NcmpServiceCmHandle(cmHandleId: 'ch-1') def dmiPluginRegistration = new DmiPluginRegistration(dmiPlugin: DMI1_URL, createdCmHandles: [cmHandleToCreate]) - objectUnderTest.updateDmiRegistrationAndSyncModule(dmiPluginRegistration) + objectUnderTest.updateDmiRegistration(dmiPluginRegistration) + + and: 'the module sync watchdog is triggered' + moduleSyncWatchdog.moduleSyncAdvisedCmHandles() then: 'CM-handle goes to LOCKED state with reason MODULE_SYNC_FAILED' new PollingConditions().within(MODULE_SYNC_WAIT_TIME_IN_SECONDS, () -> { @@ -117,7 +135,10 @@ class CmHandleCreateSpec extends CpsIntegrationSpecBase { when: 'a CM-handle is registered for creation with moduleSetTag "B"' def cmHandleToCreate = new NcmpServiceCmHandle(cmHandleId: 'ch-3', moduleSetTag: 'B') - objectUnderTest.updateDmiRegistrationAndSyncModule(new DmiPluginRegistration(dmiPlugin: DMI1_URL, createdCmHandles: [cmHandleToCreate])) + objectUnderTest.updateDmiRegistration(new DmiPluginRegistration(dmiPlugin: DMI1_URL, createdCmHandles: [cmHandleToCreate])) + + and: 'the module sync watchdog is triggered' + moduleSyncWatchdog.moduleSyncAdvisedCmHandles() then: 'the CM-handle goes to READY state' new PollingConditions().within(MODULE_SYNC_WAIT_TIME_IN_SECONDS, () -> { @@ -151,7 +172,7 @@ class CmHandleCreateSpec extends CpsIntegrationSpecBase { new NcmpServiceCmHandle(cmHandleId: 'ch-7', alternateId: 'duplicate-alt-id'), ] def dmiPluginRegistration = new DmiPluginRegistration(dmiPlugin: DMI1_URL, createdCmHandles: cmHandlesToCreate) - def dmiPluginRegistrationResponse = objectUnderTest.updateDmiRegistrationAndSyncModule(dmiPluginRegistration) + def dmiPluginRegistrationResponse = objectUnderTest.updateDmiRegistration(dmiPluginRegistration) then: 'registration gives expected responses' assert dmiPluginRegistrationResponse.createdCmHandles.sort { it.cmHandle } == [ @@ -173,7 +194,11 @@ class CmHandleCreateSpec extends CpsIntegrationSpecBase { when: 'CM-handles are registered for creation' def cmHandlesToCreate = [new NcmpServiceCmHandle(cmHandleId: 'ch-1'), new NcmpServiceCmHandle(cmHandleId: 'ch-2')] def dmiPluginRegistration = new DmiPluginRegistration(dmiPlugin: DMI1_URL, createdCmHandles: cmHandlesToCreate) - objectUnderTest.updateDmiRegistrationAndSyncModule(dmiPluginRegistration) + objectUnderTest.updateDmiRegistration(dmiPluginRegistration) + + and: 'the module sync watchdog is triggered' + moduleSyncWatchdog.moduleSyncAdvisedCmHandles() + then: 'CM-handles go to LOCKED state' new PollingConditions().within(MODULE_SYNC_WAIT_TIME_IN_SECONDS, () -> { assert objectUnderTest.getCmHandleCompositeState('ch-1').cmHandleState == CmHandleState.LOCKED @@ -183,6 +208,9 @@ class CmHandleCreateSpec extends CpsIntegrationSpecBase { dmiDispatcher1.moduleNamesPerCmHandleId = ['ch-1': ['M1', 'M2'], 'ch-2': ['M1', 'M2']] dmiDispatcher1.isAvailable = true + and: 'the module sync watchdog is triggered TWICE' + 2.times { moduleSyncWatchdog.moduleSyncAdvisedCmHandles() } + then: 'Both CM-handles go to READY state' new PollingConditions().within(MODULE_SYNC_WAIT_TIME_IN_SECONDS, () -> { ['ch-1', 'ch-2'].each { cmHandleId -> @@ -198,4 +226,23 @@ class CmHandleCreateSpec extends CpsIntegrationSpecBase { cleanup: 'deregister CM handles' deregisterCmHandles(DMI1_URL, ['ch-1', 'ch-2']) } + + def subscribeAndClearPreviousMessages() { + kafkaConsumer = KafkaTestContainer.getConsumer('test-group', StringDeserializer.class) + kafkaConsumer.subscribe(['ncmp-events']) + kafkaConsumer.poll(Duration.ofMillis(500)) + } + + def getLatestConsumerRecords() { + def consumerRecords = [] + def retryAttempts = 10 + while (consumerRecords.size() < 2) { + retryAttempts-- + consumerRecords.addAll(kafkaConsumer.poll(Duration.ofMillis(100))) + if (retryAttempts == 0) + break + } + consumerRecords + } + } diff --git a/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleUpdateSpec.groovy b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleUpdateSpec.groovy index 2d1588ecf9..67011f811b 100644 --- a/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleUpdateSpec.groovy +++ b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleUpdateSpec.groovy @@ -44,7 +44,7 @@ class CmHandleUpdateSpec extends CpsIntegrationSpecBase { when: "CM-handle is registered for update with new alternate ID: $newAlternateId" def cmHandleToUpdate = new NcmpServiceCmHandle(cmHandleId: 'ch-1', alternateId: newAlternateId) def dmiPluginRegistrationResponse = - objectUnderTest.updateDmiRegistrationAndSyncModule(new DmiPluginRegistration(dmiPlugin: DMI1_URL, updatedCmHandles: [cmHandleToUpdate])) + objectUnderTest.updateDmiRegistration(new DmiPluginRegistration(dmiPlugin: DMI1_URL, updatedCmHandles: [cmHandleToUpdate])) then: 'registration gives successful response' assert dmiPluginRegistrationResponse.updatedCmHandles == [CmHandleRegistrationResponse.createSuccessResponse('ch-1')] @@ -74,7 +74,7 @@ class CmHandleUpdateSpec extends CpsIntegrationSpecBase { when: 'a CM-handle is registered for update with new alternate ID' def cmHandleToUpdate = new NcmpServiceCmHandle(cmHandleId: 'ch-1', alternateId: 'new') def dmiPluginRegistrationResponse = - objectUnderTest.updateDmiRegistrationAndSyncModule(new DmiPluginRegistration(dmiPlugin: DMI1_URL, updatedCmHandles: [cmHandleToUpdate])) + objectUnderTest.updateDmiRegistration(new DmiPluginRegistration(dmiPlugin: DMI1_URL, updatedCmHandles: [cmHandleToUpdate])) then: 'registration gives failure response, due to alternate ID being already associated' assert dmiPluginRegistrationResponse.updatedCmHandles == [CmHandleRegistrationResponse.createFailureResponse('ch-1', NcmpResponseStatus.ALTERNATE_ID_ALREADY_ASSOCIATED)] diff --git a/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleUpgradeSpec.groovy b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleUpgradeSpec.groovy index f93f58ce20..64449371fe 100644 --- a/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleUpgradeSpec.groovy +++ b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleUpgradeSpec.groovy @@ -48,7 +48,7 @@ class CmHandleUpgradeSpec extends CpsIntegrationSpecBase { when: "the CM-handle is upgraded with given moduleSetTag '${updatedModuleSetTag}'" def cmHandlesToUpgrade = new UpgradedCmHandles(cmHandles: [CM_HANDLE_ID], moduleSetTag: updatedModuleSetTag) - def dmiPluginRegistrationResponse = objectUnderTest.updateDmiRegistrationAndSyncModule( + def dmiPluginRegistrationResponse = objectUnderTest.updateDmiRegistration( new DmiPluginRegistration(dmiPlugin: DMI1_URL, upgradedCmHandles: cmHandlesToUpgrade)) then: 'registration gives successful response' @@ -63,6 +63,9 @@ class CmHandleUpgradeSpec extends CpsIntegrationSpecBase { when: 'DMI will return different modules for upgrade: M1 and M3' dmiDispatcher1.moduleNamesPerCmHandleId[CM_HANDLE_ID] = ['M1', 'M3'] + and: 'the module sync watchdog is triggered twice' + 2.times { moduleSyncWatchdog.moduleSyncAdvisedCmHandles() } + then: 'CM-handle goes to READY state' new PollingConditions().within(MODULE_SYNC_WAIT_TIME_IN_SECONDS, () -> { assert CmHandleState.READY == objectUnderTest.getCmHandleCompositeState(CM_HANDLE_ID).cmHandleState @@ -98,12 +101,15 @@ class CmHandleUpgradeSpec extends CpsIntegrationSpecBase { when: "CM-handle is upgraded to moduleSetTag '${updatedModuleSetTag}'" def cmHandlesToUpgrade = new UpgradedCmHandles(cmHandles: [CM_HANDLE_ID], moduleSetTag: updatedModuleSetTag) - def dmiPluginRegistrationResponse = objectUnderTest.updateDmiRegistrationAndSyncModule( + def dmiPluginRegistrationResponse = objectUnderTest.updateDmiRegistration( new DmiPluginRegistration(dmiPlugin: DMI1_URL, upgradedCmHandles: cmHandlesToUpgrade)) then: 'registration gives successful response' assert dmiPluginRegistrationResponse.upgradedCmHandles == [CmHandleRegistrationResponse.createSuccessResponse(CM_HANDLE_ID)] + and: 'the module sync watchdog is triggered twice' + 2.times { moduleSyncWatchdog.moduleSyncAdvisedCmHandles() } + and: 'CM-handle goes to READY state' new PollingConditions().within(MODULE_SYNC_WAIT_TIME_IN_SECONDS, () -> { assert CmHandleState.READY == objectUnderTest.getCmHandleCompositeState(CM_HANDLE_ID).cmHandleState @@ -132,7 +138,7 @@ class CmHandleUpgradeSpec extends CpsIntegrationSpecBase { when: 'CM-handle is upgraded with the same moduleSetTag' def cmHandlesToUpgrade = new UpgradedCmHandles(cmHandles: [CM_HANDLE_ID], moduleSetTag: 'same') - objectUnderTest.updateDmiRegistrationAndSyncModule( + objectUnderTest.updateDmiRegistration( new DmiPluginRegistration(dmiPlugin: DMI1_URL, upgradedCmHandles: cmHandlesToUpgrade)) then: 'CM-handle remains in READY state' @@ -157,9 +163,12 @@ class CmHandleUpgradeSpec extends CpsIntegrationSpecBase { when: 'the CM-handle is upgraded' def cmHandlesToUpgrade = new UpgradedCmHandles(cmHandles: [CM_HANDLE_ID], moduleSetTag: 'newTag') - objectUnderTest.updateDmiRegistrationAndSyncModule( + objectUnderTest.updateDmiRegistration( new DmiPluginRegistration(dmiPlugin: DMI1_URL, upgradedCmHandles: cmHandlesToUpgrade)) + and: 'the module sync watchdog is triggered twice' + 2.times { moduleSyncWatchdog.moduleSyncAdvisedCmHandles() } + then: 'CM-handle goes to LOCKED state with reason MODULE_UPGRADE_FAILED' new PollingConditions().within(MODULE_SYNC_WAIT_TIME_IN_SECONDS, () -> { def cmHandleCompositeState = objectUnderTest.getCmHandleCompositeState(CM_HANDLE_ID) diff --git a/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/ModuleSyncWatchdogIntegrationSpec.groovy b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/ModuleSyncWatchdogIntegrationSpec.groovy new file mode 100644 index 0000000000..e0bb437a7c --- /dev/null +++ b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/ModuleSyncWatchdogIntegrationSpec.groovy @@ -0,0 +1,99 @@ +/* + * ============LICENSE_START======================================================= + * Copyright (C) 2024 Nordix Foundation + * ================================================================================ + * 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 + +import org.onap.cps.integration.base.CpsIntegrationSpecBase +import org.onap.cps.ncmp.impl.inventory.sync.ModuleSyncWatchdog + +import java.util.concurrent.Executors + +class ModuleSyncWatchdogIntegrationSpec extends CpsIntegrationSpecBase { + + ModuleSyncWatchdog objectUnderTest + + def executorService = Executors.newFixedThreadPool(2) + def SYNC_SAMPLE_SIZE = 100 + + def setup() { + objectUnderTest = moduleSyncWatchdog + registerSequenceOfCmHandlesWithoutWaitForReady(DMI1_URL, NO_MODULE_SET_TAG, SYNC_SAMPLE_SIZE) + } + + def cleanup() { + try { + deregisterSequenceOfCmHandles(DMI1_URL, SYNC_SAMPLE_SIZE) + moduleSyncWorkQueue.clear() + } finally { + executorService.shutdownNow() + } + } + + def 'Watchdog is disabled for test.'() { + when: 'wait a while but less then the initial delay of 10 minutes' + Thread.sleep(3000) + then: 'the work queue remains empty' + assert moduleSyncWorkQueue.isEmpty() + } + + def 'Populate module sync work queue simultaneously on two parallel threads (CPS-2403).'() { + // This test failed before bug https://lf-onap.atlassian.net/browse/CPS-2403 was fixed + given: 'the queue is empty at the start' + assert moduleSyncWorkQueue.isEmpty() + when: 'attempt to populate the queue on the main (test) and another parallel thread at the same time' + objectUnderTest.populateWorkQueueIfNeeded() + executorService.execute(populateQueueWithoutDelay) + and: 'wait a little (to give all threads time to complete their task)' + Thread.sleep(50) + then: 'the queue size is exactly the sample size' + assert moduleSyncWorkQueue.size() == SYNC_SAMPLE_SIZE + } + + def 'Populate module sync work queue on two parallel threads with a slight difference in start time.'() { + // This test proved that the issue in CPS-2403 did not arise if the the queue was populated and given time to be distributed + given: 'the queue is empty at the start' + assert moduleSyncWorkQueue.isEmpty() + when: 'attempt to populate the queue on the main (test) and another parallel thread a little later' + objectUnderTest.populateWorkQueueIfNeeded() + executorService.execute(populateQueueWithDelay) + and: 'wait a little (to give all threads time to complete their task)' + Thread.sleep(50) + then: 'the queue size is exactly the sample size' + assert moduleSyncWorkQueue.size() == SYNC_SAMPLE_SIZE + } + + def populateQueueWithoutDelay = () -> { + try { + objectUnderTest.populateWorkQueueIfNeeded() + } catch (InterruptedException e) { + e.printStackTrace() + } + } + + def populateQueueWithDelay = () -> { + try { + Thread.sleep(10) + objectUnderTest.populateWorkQueueIfNeeded() + } catch (InterruptedException e) { + e.printStackTrace() + } + } + +} diff --git a/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/RestApiSpec.groovy b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/RestApiSpec.groovy index 1e1af556f1..7ce3cf5e17 100644 --- a/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/RestApiSpec.groovy +++ b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/RestApiSpec.groovy @@ -44,6 +44,8 @@ class RestApiSpec extends CpsIntegrationSpecBase { def requestBody = '{"dmiPlugin":"'+DMI1_URL+'","createdCmHandles":[{"cmHandle":"ch-1","alternateId":"alt-1"},{"cmHandle":"ch-2","alternateId":"alt-2"},{"cmHandle":"ch-3","alternateId":"alt-3"}]}' mvc.perform(post('/ncmpInventory/v1/ch').contentType(MediaType.APPLICATION_JSON).content(requestBody)) .andExpect(status().is2xxSuccessful()) + and: 'the module sync watchdog is triggered' + moduleSyncWatchdog.moduleSyncAdvisedCmHandles() then: 'CM-handles go to READY state' new PollingConditions().within(MODULE_SYNC_WAIT_TIME_IN_SECONDS, () -> { (1..3).each { @@ -65,15 +67,20 @@ class RestApiSpec extends CpsIntegrationSpecBase { ] }""".formatted(moduleName) expect: "a search for module ${moduleName} returns expected CM handles" - mvc.perform(post('/ncmp/v1/ch/id-searches').contentType(MediaType.APPLICATION_JSON).content(requestBodyWithModuleCondition)) + mvc.perform(post('/ncmp/v1/ch/id-searches'+outputAlternateId).contentType(MediaType.APPLICATION_JSON).content(requestBodyWithModuleCondition)) .andExpect(status().is2xxSuccessful()) - .andExpect(jsonPath('$[*]', containsInAnyOrder(expectedCmHandles.toArray()))) - .andExpect(jsonPath('$', hasSize(expectedCmHandles.size()))); + .andExpect(jsonPath('$[*]', containsInAnyOrder(expectedCmHandleReferences.toArray()))) + .andExpect(jsonPath('$', hasSize(expectedCmHandleReferences.size()))); where: - moduleName || expectedCmHandles - 'M1' || ['ch-1', 'ch-2', 'ch-3'] - 'M2' || ['ch-1', 'ch-2'] - 'M3' || ['ch-3'] + moduleName | outputAlternateId || expectedCmHandleReferences + 'M1' | '?outputAlternateId=false' || ['ch-1', 'ch-2', 'ch-3'] + 'M2' | '?outputAlternateId=false' || ['ch-1', 'ch-2'] + 'M3' | '?outputAlternateId=false' || ['ch-3'] + 'M1' | '?outputAlternateId=true' || ['alt-1', 'alt-2', 'alt-3'] + 'M2' | '?outputAlternateId=true' || ['alt-1', 'alt-2'] + 'M3' | '?outputAlternateId=true' || ['alt-3'] + 'M1' | '' || ['ch-1', 'ch-2', 'ch-3'] + } def 'Search for CM Handles using Cps Path Query.'() { diff --git a/integration-test/src/test/groovy/org/onap/cps/integration/performance/base/NcmpPerfTestBase.groovy b/integration-test/src/test/groovy/org/onap/cps/integration/performance/base/NcmpPerfTestBase.groovy index f6ae27d129..fb5a0c3eb1 100644 --- a/integration-test/src/test/groovy/org/onap/cps/integration/performance/base/NcmpPerfTestBase.groovy +++ b/integration-test/src/test/groovy/org/onap/cps/integration/performance/base/NcmpPerfTestBase.groovy @@ -27,11 +27,13 @@ import org.onap.cps.utils.ContentType import static org.onap.cps.ncmp.impl.inventory.NcmpPersistence.NCMP_DATASPACE_NAME import static org.onap.cps.ncmp.impl.inventory.NcmpPersistence.NCMP_DMI_REGISTRY_ANCHOR +import static org.onap.cps.ncmp.impl.inventory.NcmpPersistence.NCMP_DMI_REGISTRY_PARENT class NcmpPerfTestBase extends PerfTestBase { def static NCMP_PERFORMANCE_TEST_DATASPACE = 'ncmpPerformance' - def static REGISTRY_ANCHOR = 'ncmp-registry' + def static REGISTRY_ANCHOR = NCMP_DMI_REGISTRY_ANCHOR + def static REGISTRY_PARENT = NCMP_DMI_REGISTRY_PARENT def static REGISTRY_SCHEMA_SET = 'registrySchemaSet' def static TOTAL_CM_HANDLES = 20_000 def static CM_DATA_SUBSCRIPTIONS_ANCHOR = 'cm-data-subscriptions' @@ -70,30 +72,30 @@ class NcmpPerfTestBase extends PerfTestBase { } def createRegistrySchemaSet() { - def modelAsString = readResourceDataFile('ncmp-registry/dmi-registry@2024-02-23.yang') + def modelAsString = readResourceDataFile('inventory/dmi-registry@2024-02-23.yang') cpsModuleService.createSchemaSet(NCMP_PERFORMANCE_TEST_DATASPACE, REGISTRY_SCHEMA_SET, [registry: modelAsString]) } def addRegistryData() { cpsAnchorService.createAnchor(NCMP_PERFORMANCE_TEST_DATASPACE, REGISTRY_SCHEMA_SET, REGISTRY_ANCHOR) cpsDataService.saveData(NCMP_PERFORMANCE_TEST_DATASPACE, REGISTRY_ANCHOR, '{"dmi-registry": []}', now) - def innerNodeJsonTemplate = readResourceDataFile('ncmp-registry/innerNode.json') + def cmHandleJsonTemplate = readResourceDataFile('inventory/cmHandleTemplate.json') def batchSize = 100 for (def i = 0; i < TOTAL_CM_HANDLES; i += batchSize) { - def data = '{ "cm-handles": [' + (1..batchSize).collect { innerNodeJsonTemplate.replace('CMHANDLE_ID_HERE', (it + i).toString()) }.join(',') + ']}' - cpsDataService.saveListElements(NCMP_PERFORMANCE_TEST_DATASPACE, REGISTRY_ANCHOR, '/dmi-registry', data, now, ContentType.JSON) + def data = '{ "cm-handles": [' + (1..batchSize).collect { cmHandleJsonTemplate.replace('CM_HANDLE_ID_HERE', (it + i).toString()) }.join(',') + ']}' + cpsDataService.saveListElements(NCMP_PERFORMANCE_TEST_DATASPACE, REGISTRY_ANCHOR, REGISTRY_PARENT, data, now, ContentType.JSON) } } def addRegistryDataWithAlternateIdAsPath() { - def innerNodeJsonTemplate = readResourceDataFile('ncmp-registry/innerCmHandleNode.json') + def cmHandleWithAlternateIdTemplate = readResourceDataFile('inventory/cmHandleWithAlternateIdTemplate.json') def batchSize = 10 for (def i = 0; i < TOTAL_CM_HANDLES; i += batchSize) { def data = '{ "cm-handles": [' + (1..batchSize).collect { - innerNodeJsonTemplate.replace('CM_HANDLE_ID_HERE', (it + i).toString()) + cmHandleWithAlternateIdTemplate.replace('CM_HANDLE_ID_HERE', (it + i).toString()) .replace('ALTERNATE_ID_AS_PATH', (it + i).toString()) }.join(',') + ']}' - cpsDataService.saveListElements(NCMP_DATASPACE_NAME, NCMP_DMI_REGISTRY_ANCHOR, '/dmi-registry', data, now, ContentType.JSON) + cpsDataService.saveListElements(NCMP_DATASPACE_NAME, NCMP_DMI_REGISTRY_ANCHOR, REGISTRY_PARENT, data, now, ContentType.JSON) } } @@ -117,7 +119,7 @@ class NcmpPerfTestBase extends PerfTestBase { def result = cpsDataService.getDataNodes(NCMP_PERFORMANCE_TEST_DATASPACE, REGISTRY_ANCHOR, '/', FetchDescendantsOption.OMIT_DESCENDANTS) resourceMeter.stop() then: 'expected data exists' - assert result.xpath == ['/dmi-registry'] + assert result.xpath == [REGISTRY_PARENT] and: 'operation completes within expected time' recordAndAssertResourceUsage('NCMP pre-load test data', 15, resourceMeter.totalTimeInSeconds, diff --git a/integration-test/src/test/java/org/onap/cps/integration/KafkaTestContainer.java b/integration-test/src/test/java/org/onap/cps/integration/KafkaTestContainer.java index d41f752912..ff4aec4175 100644 --- a/integration-test/src/test/java/org/onap/cps/integration/KafkaTestContainer.java +++ b/integration-test/src/test/java/org/onap/cps/integration/KafkaTestContainer.java @@ -21,6 +21,7 @@ package org.onap.cps.integration; import java.util.HashMap; import java.util.Map; +import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.serialization.StringDeserializer; @@ -33,11 +34,12 @@ import org.testcontainers.utility.DockerImageName; * This ensures only one instance of Kafka container across the integration tests. * Avoid unnecessary resource and time consumption. */ +@Slf4j public class KafkaTestContainer extends KafkaContainer { private static final String IMAGE_NAME_AND_VERSION = "registry.nordix.org/onaptest/confluentinc/cp-kafka:6.2.1"; - private static KafkaTestContainer kafkaTestContainer; + private static volatile KafkaTestContainer kafkaTestContainer; private KafkaTestContainer() { super(DockerImageName.parse(IMAGE_NAME_AND_VERSION).asCompatibleSubstituteFor("confluentinc/cp-kafka")); @@ -51,8 +53,15 @@ public class KafkaTestContainer extends KafkaContainer { */ public static KafkaTestContainer getInstance() { if (kafkaTestContainer == null) { - kafkaTestContainer = new KafkaTestContainer(); - Runtime.getRuntime().addShutdownHook(new Thread(kafkaTestContainer::close)); + synchronized (KafkaTestContainer.class) { + if (kafkaTestContainer == null) { + kafkaTestContainer = new KafkaTestContainer(); + Runtime.getRuntime().addShutdownHook(new Thread(() -> { + log.info("Shutting down KafkaTestContainer..."); + kafkaTestContainer.stop(); + })); + } + } } return kafkaTestContainer; } @@ -63,8 +72,11 @@ public class KafkaTestContainer extends KafkaContainer { @Override public void start() { - super.start(); - System.setProperty("spring.kafka.properties.bootstrap.servers", kafkaTestContainer.getBootstrapServers()); + if (!isRunning()) { + super.start(); + System.setProperty("spring.kafka.properties.bootstrap.servers", getBootstrapServers()); + log.info("KafkaTestContainer started at {}", getBootstrapServers()); + } } @Override @@ -78,8 +90,9 @@ public class KafkaTestContainer extends KafkaContainer { configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaTestContainer.getBootstrapServers()); configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer); - configProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + configProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); configProps.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId); + configProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, Integer.MAX_VALUE); return configProps; } diff --git a/integration-test/src/test/resources/application-module-sync-delayed.yml b/integration-test/src/test/resources/application-module-sync-delayed.yml new file mode 100644 index 0000000000..7b9c6aea4f --- /dev/null +++ b/integration-test/src/test/resources/application-module-sync-delayed.yml @@ -0,0 +1,23 @@ +# ============LICENSE_START======================================================= +# Copyright (C) 2024 Nordix Foundation. +# ================================================================================ +# 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========================================================= + +test: + ncmp: + timers: + advised-modules-sync: + initial-delay-ms: 600000 + diff --git a/integration-test/src/test/resources/application.yml b/integration-test/src/test/resources/application.yml index 853797aac5..b786a3d4c5 100644 --- a/integration-test/src/test/resources/application.yml +++ b/integration-test/src/test/resources/application.yml @@ -179,7 +179,7 @@ ncmp: timers: advised-modules-sync: - sleep-time-ms: 1000 + sleep-time-ms: 1000000 cm-handle-data-sync: sleep-time-ms: 30000 subscription-forwarding: diff --git a/integration-test/src/test/resources/data/ncmp-registry/innerNode.json b/integration-test/src/test/resources/data/inventory/cmHandleTemplate.json index b6c65f3763..6577f4e560 100644 --- a/integration-test/src/test/resources/data/ncmp-registry/innerNode.json +++ b/integration-test/src/test/resources/data/inventory/cmHandleTemplate.json @@ -1,6 +1,6 @@ { - "id": "cm-CMHANDLE_ID_HERE", - "alternate-id": "alt-CMHANDLE_ID_HERE", + "id": "cm-CM_HANDLE_ID_HERE", + "alternate-id": "alt-CM_HANDLE_ID_HERE", "module-set-tag": "my-module-set-tag", "dmi-service-name": "http://ncmp-dmi-plugin-stub:8080", "dmi-data-service-name": "", @@ -21,4 +21,4 @@ } } } -}
\ No newline at end of file +} diff --git a/integration-test/src/test/resources/data/ncmp-registry/innerCmHandleNode.json b/integration-test/src/test/resources/data/inventory/cmHandleWithAlternateIdTemplate.json index 88446c4a0f..88446c4a0f 100644 --- a/integration-test/src/test/resources/data/ncmp-registry/innerCmHandleNode.json +++ b/integration-test/src/test/resources/data/inventory/cmHandleWithAlternateIdTemplate.json diff --git a/integration-test/src/test/resources/data/ncmp-registry/dmi-registry@2024-02-23.yang b/integration-test/src/test/resources/data/inventory/dmi-registry@2024-02-23.yang index d7b4ff7550..d7b4ff7550 100644 --- a/integration-test/src/test/resources/data/ncmp-registry/dmi-registry@2024-02-23.yang +++ b/integration-test/src/test/resources/data/inventory/dmi-registry@2024-02-23.yang |