[Module Sync] Store CM-handle IDs in work queue

This fixes bug CPS-2474, handling various edge cases, such as
CM handles being deleted during module sync.

- Change moduleSyncWorkQueue to store CmHandleId instead of DataNode.
- Freshly fetch Cm Handles in module sync task, so latest CM-handle
  state is used, and only process ADVISED CM handles in module sync.

Issue-ID: CPS-2474
Signed-off-by: danielhanrahan <daniel.hanrahan@est.tech>
Change-Id: I53d5796c56014a2bfbe5b1c3f17d3991e4feef53

5 files changed, 62 insertions, 70 deletions

diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/CmHandleQueryServiceImplSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/CmHandleQueryServiceImplSpec.groovy
index be99479dc7..d19081cee5 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/CmHandleQueryServiceImplSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/CmHandleQueryServiceImplSpec.groovy
@@ -27,7 +27,6 @@ import org.onap.cps.impl.utils.CpsValidator
 import org.onap.cps.ncmp.api.inventory.models.TrustLevel
 import org.onap.cps.ncmp.impl.inventory.models.CmHandleState
 import org.onap.cps.api.model.DataNode
-import spock.lang.Shared
 import spock.lang.Specification
 
 import static org.onap.cps.ncmp.impl.inventory.NcmpPersistence.NCMP_DATASPACE_NAME
@@ -40,17 +39,14 @@ class CmHandleQueryServiceImplSpec extends Specification {
 
     def mockCpsQueryService = Mock(CpsQueryService)
     def mockCpsDataService = Mock(CpsDataService)
-
     def trustLevelPerDmiPlugin = [:]
-
     def trustLevelPerCmHandleId = [ 'PNFDemo': TrustLevel.COMPLETE, 'PNFDemo2': TrustLevel.NONE, 'PNFDemo4': TrustLevel.NONE ]
-
     def mockCpsValidator = Mock(CpsValidator)
 
     def objectUnderTest = new CmHandleQueryServiceImpl(mockCpsDataService, mockCpsQueryService, trustLevelPerDmiPlugin, trustLevelPerCmHandleId, mockCpsValidator)
 
-    @Shared
-    def static sampleDataNodes = [new DataNode()]
+    def static sampleDataNodes = [new DataNode(xpath: "/dmi-registry/cm-handles[@id='ch-1']"),
+                                  new DataNode(xpath: "/dmi-registry/cm-handles[@id='ch-2']")]
 
     def dataNodeWithPrivateField = '//additional-properties[@name=\"Contact3\" and @value=\"newemailforstore3@bookstore.com\"]/ancestor::cm-handles'
 
@@ -117,16 +113,16 @@ class CmHandleQueryServiceImplSpec extends Specification {
             result.size() == 1
     }
 
-    def 'Get CmHandles by it\'s state.'() {
+    def 'Get Ids of CmHandles by state.'() {
         given: 'a cm handle state to query'
             def cmHandleState = CmHandleState.ADVISED
         and: 'the persistence service returns a list of data nodes'
             mockCpsQueryService.queryDataNodes(NCMP_DATASPACE_NAME, NCMP_DMI_REGISTRY_ANCHOR,
-                '//state[@cm-handle-state="ADVISED"]/ancestor::cm-handles', INCLUDE_ALL_DESCENDANTS) >> sampleDataNodes
+                "//state[@cm-handle-state='ADVISED']", OMIT_DESCENDANTS) >> sampleDataNodes
         when: 'cm handles are fetched by state'
-            def result = objectUnderTest.queryCmHandlesByState(cmHandleState)
+            def result = objectUnderTest.queryCmHandleIdsByState(cmHandleState)
         then: 'the returned result matches the result from the persistence service'
-            assert result == sampleDataNodes
+            assert result.toSet() == ['ch-1', 'ch-2'].toSet()
     }
 
     def 'Check the state of a cmHandle when #scenario.'() {
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/sync/ModuleOperationsUtilsSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/sync/ModuleOperationsUtilsSpec.groovy
index de5f7e2fbe..f116e0e1dd 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/sync/ModuleOperationsUtilsSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/sync/ModuleOperationsUtilsSpec.groovy
@@ -79,15 +79,15 @@ class ModuleOperationsUtilsSpec extends Specification{
 
     def 'Get an advised Cm-Handle where ADVISED cm handle #scenario'() {
         given: 'the inventory persistence service returns a collection of data nodes'
-            mockCmHandleQueries.queryCmHandlesByState(CmHandleState.ADVISED) >> dataNodeCollection
-        when: 'get advised cm handles are fetched'
-            def yangModelCmHandles = objectUnderTest.getAdvisedCmHandles()
-        then: 'the returned data node collection is the correct size'
-            yangModelCmHandles.size() == expectedDataNodeSize
+            mockCmHandleQueries.queryCmHandleIdsByState(CmHandleState.ADVISED) >> cmHandleIds
+        when: 'advised cm handle ids are fetched'
+            def advisedCmHandleIds = objectUnderTest.getAdvisedCmHandleIds()
+        then: 'the expected cm handle ids are returned'
+            advisedCmHandleIds == cmHandleIds
         where: 'the following scenarios are used'
-            scenario         | dataNodeCollection || expectedCallsToGetYangModelCmHandle | expectedDataNodeSize
-            'exists'         | [dataNode]         || 1                                   | 1
-            'does not exist' | []                 || 0                                   | 0
+            scenario         | cmHandleIds
+            'exists'         | ['cm-handle-123']
+            'does not exist' | []
     }
 
     def 'Update Lock Reason, Details and Attempts where lock reason #scenario'() {
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/sync/ModuleSyncTasksSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/sync/ModuleSyncTasksSpec.groovy
index 5e2162ed0d..97c2488bc3 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/sync/ModuleSyncTasksSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/sync/ModuleSyncTasksSpec.groovy
@@ -36,9 +36,7 @@ import org.onap.cps.ncmp.impl.inventory.models.CmHandleState
 import org.onap.cps.ncmp.impl.inventory.models.YangModelCmHandle
 import org.onap.cps.ncmp.impl.inventory.sync.lcm.LcmEventsCmHandleStateHandler
 import org.onap.cps.api.exceptions.DataNodeNotFoundException
-import org.onap.cps.api.model.DataNode
 import org.slf4j.LoggerFactory
-import spock.lang.Ignore
 import spock.lang.Specification
 import java.util.concurrent.atomic.AtomicInteger
 
@@ -82,12 +80,13 @@ class ModuleSyncTasksSpec extends Specification {
 
     def 'Module Sync ADVISED cm handles.'() {
         given: 'cm handles in an ADVISED state'
-            def cmHandle1 = cmHandleAsDataNodeByIdAndState('cm-handle-1', CmHandleState.ADVISED)
-            def cmHandle2 = cmHandleAsDataNodeByIdAndState('cm-handle-2', CmHandleState.ADVISED)
-        and: 'the inventory persistence cm handle returns a ADVISED state for the any handle'
-            mockInventoryPersistence.getCmHandleState(_) >> new CompositeState(cmHandleState: CmHandleState.ADVISED)
+            def cmHandle1 = cmHandleByIdAndState('cm-handle-1', CmHandleState.ADVISED)
+            def cmHandle2 = cmHandleByIdAndState('cm-handle-2', CmHandleState.ADVISED)
+        and: 'the inventory persistence cm handle returns a ADVISED state for the handles'
+            mockInventoryPersistence.getYangModelCmHandle('cm-handle-1') >> cmHandle1
+            mockInventoryPersistence.getYangModelCmHandle('cm-handle-2') >> cmHandle2
         when: 'module sync poll is executed'
-            objectUnderTest.performModuleSync([cmHandle1, cmHandle2], batchCount)
+            objectUnderTest.performModuleSync(['cm-handle-1', 'cm-handle-2'], batchCount)
         then: 'module sync service deletes schemas set of each cm handle if it already exists'
             1 * mockModuleSyncService.deleteSchemaSetIfExists('cm-handle-1')
             1 * mockModuleSyncService.deleteSchemaSetIfExists('cm-handle-2')
@@ -103,18 +102,17 @@ class ModuleSyncTasksSpec extends Specification {
     }
 
     def 'Handle CM handle failure during #scenario and log MODULE_UPGRADE lock reason'() {
-        given: 'a CM handle in LOCKED state with a specific lock reason'
-            def cmHandle = cmHandleAsDataNodeByIdAndState('cm-handle', CmHandleState.LOCKED)
-            def expectedCmHandleState = new CompositeState(cmHandleState: CmHandleState.LOCKED, lockReason: CompositeState
-                    .LockReason.builder().lockReasonCategory(lockReasonCategory).details(lockReasonDetails).build())
-            1 * mockInventoryPersistence.getCmHandleState('cm-handle') >> expectedCmHandleState
+        given: 'a CM handle in ADVISED state with a specific lock reason'
+            def cmHandle = cmHandleByIdAndState('cm-handle', CmHandleState.ADVISED)
+            cmHandle.compositeState.lockReason = CompositeState.LockReason.builder().lockReasonCategory(lockReasonCategory).details(lockReasonDetails).build()
+            mockInventoryPersistence.getYangModelCmHandle('cm-handle') >> cmHandle
         and: 'module sync service attempts to sync/upgrade the CM handle and throws an exception'
             mockModuleSyncService.syncAndCreateSchemaSetAndAnchor(_) >> { throw new Exception('some exception') }
             mockModuleSyncService.syncAndUpgradeSchemaSet(_) >> { throw new Exception('some exception') }
         when: 'module sync is executed'
-            objectUnderTest.performModuleSync([cmHandle], batchCount)
+            objectUnderTest.performModuleSync(['cm-handle'], batchCount)
         then: 'lock reason is updated with number of attempts'
-            1 * mockSyncUtils.updateLockReasonWithAttempts(expectedCmHandleState, expectedLockReasonCategory, 'some exception')
+            1 * mockSyncUtils.updateLockReasonWithAttempts(_, expectedLockReasonCategory, 'some exception')
         and: 'the state handler is called to update the state to LOCKED'
             1 * mockLcmEventsCmHandleStateHandler.updateCmHandleStateBatch(_) >> { args ->
                 assertBatch(args, ['cm-handle'], CmHandleState.LOCKED)
@@ -128,25 +126,27 @@ class ModuleSyncTasksSpec extends Specification {
             'module upgrade' | MODULE_UPGRADE        | 'Upgrade in progress'                          || MODULE_UPGRADE_FAILED
     }
 
-    @Ignore  // TODO Enable this test once the bug CPS-2474 is fixed
     def 'Module sync succeeds even if a handle gets deleted during module sync.'() {
-        given: 'cm handles in an ADVISED state'
-            def cmHandle1 = cmHandleAsDataNodeByIdAndState('cm-handle-1', CmHandleState.ADVISED)
-            def cmHandle2 = cmHandleAsDataNodeByIdAndState('cm-handle-2', CmHandleState.ADVISED)
-        and: 'inventory persistence cannot find the first handle'
-            mockInventoryPersistence.getCmHandleState('cm-handle-1') >> { throw new DataNodeNotFoundException('dataspace', 'anchor', 'xpath') }
-        and: 'inventory persistence returns the second handle with ADVISED state'
-            mockInventoryPersistence.getCmHandleState('cm-handle-2') >> new CompositeState(cmHandleState: CmHandleState.ADVISED)
+        given: 'a cm handle which has been deleted'
+            mockInventoryPersistence.getYangModelCmHandle('cm-handle-1') >> { throw new DataNodeNotFoundException('dataspace', 'anchor', 'cm-handle-1') }
+        and: 'a cm handle which is being deleted'
+            mockInventoryPersistence.getYangModelCmHandle('cm-handle-2') >> cmHandleByIdAndState('cm-handle-2', CmHandleState.DELETING)
+        and: 'a cm handle in advised state'
+            mockInventoryPersistence.getYangModelCmHandle('cm-handle-3') >> cmHandleByIdAndState('cm-handle-3', CmHandleState.ADVISED)
         when: 'module sync poll is executed'
-            objectUnderTest.performModuleSync([cmHandle1, cmHandle2], batchCount)
+            objectUnderTest.performModuleSync(['cm-handle-1', 'cm-handle-2', 'cm-handle-3'], batchCount)
         then: 'no exception is thrown'
             noExceptionThrown()
         and: 'the deleted cm-handle did not sync'
             0 * mockModuleSyncService.syncAndCreateSchemaSetAndAnchor(_) >> { args -> assert args[0].id == 'cm-handle-1' }
-        and: 'the existing cm-handle synced'
-            1 * mockModuleSyncService.syncAndCreateSchemaSetAndAnchor(_) >> { args -> assert args[0].id == 'cm-handle-2' }
-        and: 'the state handler called'
-            1 * mockLcmEventsCmHandleStateHandler.updateCmHandleStateBatch(_)
+        and: 'the deleting cm-handle did not sync'
+            0 * mockModuleSyncService.syncAndCreateSchemaSetAndAnchor(_) >> { args -> assert args[0].id == 'cm-handle-2' }
+        and: 'the advised cm-handle synced'
+            1 * mockModuleSyncService.syncAndCreateSchemaSetAndAnchor(_) >> { args -> assert args[0].id == 'cm-handle-3' }
+        and: 'the state handler called for only the advised handle'
+            1 * mockLcmEventsCmHandleStateHandler.updateCmHandleStateBatch(_) >> { args ->
+                assertBatch(args, ['cm-handle-3'], CmHandleState.READY)
+            }
     }
 
     def 'Reset failed CM Handles #scenario.'() {
@@ -172,15 +172,15 @@ class ModuleSyncTasksSpec extends Specification {
 
     def 'Module Sync ADVISED cm handle without entry in progress map.'() {
         given: 'cm handles in an ADVISED state'
-            def cmHandle1 = cmHandleAsDataNodeByIdAndState('cm-handle-1', CmHandleState.ADVISED)
+            def cmHandle1 = cmHandleByIdAndState('cm-handle-1', CmHandleState.ADVISED)
         and: 'the inventory persistence cm handle returns a ADVISED state for the any handle'
-            mockInventoryPersistence.getCmHandleState(_) >> new CompositeState(cmHandleState: CmHandleState.ADVISED)
+            mockInventoryPersistence.getYangModelCmHandle('cm-handle-1') >> cmHandle1
         and: 'entry in progress map for other cm handle'
             moduleSyncStartedOnCmHandles.put('other-cm-handle', 'started')
         when: 'module sync poll is executed'
-            objectUnderTest.performModuleSync([cmHandle1], batchCount)
+            objectUnderTest.performModuleSync(['cm-handle-1'], batchCount)
         then: 'module sync service is invoked for cm handle'
-            1 * mockModuleSyncService.syncAndCreateSchemaSetAndAnchor(_)
+            1 * mockModuleSyncService.syncAndCreateSchemaSetAndAnchor(_) >> { args -> assert args[0].id == 'cm-handle-1' }
         and: 'the entry for other cm handle is still in the progress map'
             assert moduleSyncStartedOnCmHandles.get('other-cm-handle') != null
     }
@@ -199,11 +199,11 @@ class ModuleSyncTasksSpec extends Specification {
 
     def 'Sync and upgrade CM handle if in upgrade state for #scenario'() {
         given: 'a CM handle in an upgrade state'
-            def cmHandle = cmHandleAsDataNodeByIdAndState('cm-handle', CmHandleState.LOCKED)
-            def compositeState = new CompositeState(lockReason: CompositeState.LockReason.builder().lockReasonCategory(lockReasonCategory).build())
-            1 * mockInventoryPersistence.getCmHandleState('cm-handle') >> compositeState
+            def cmHandle = cmHandleByIdAndState('cm-handle', CmHandleState.ADVISED)
+            cmHandle.compositeState.setLockReason(CompositeState.LockReason.builder().lockReasonCategory(lockReasonCategory).build())
+            mockInventoryPersistence.getYangModelCmHandle('cm-handle') >> cmHandle
         when: 'module sync is executed'
-            objectUnderTest.performModuleSync([cmHandle], batchCount)
+            objectUnderTest.performModuleSync(['cm-handle'], batchCount)
         then: 'the module sync service should attempt to sync and upgrade the CM handle'
             1 * mockModuleSyncService.syncAndUpgradeSchemaSet(_) >> { args ->
                 assert args[0].id == 'cm-handle'
@@ -226,8 +226,8 @@ class ModuleSyncTasksSpec extends Specification {
             assert loggingEvent == null
     }
 
-    def cmHandleAsDataNodeByIdAndState(cmHandleId, cmHandleState) {
-        return new DataNode(anchorName: cmHandleId, leaves: ['id': cmHandleId, 'cm-handle-state': cmHandleState])
+    def cmHandleByIdAndState(cmHandleId, cmHandleState) {
+        return new YangModelCmHandle(id: cmHandleId, compositeState: new CompositeState(cmHandleState: cmHandleState))
     }
 
     def assertBatch(args, expectedCmHandleStatePerCmHandleIds, expectedCmHandleState) {
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/sync/ModuleSyncWatchdogSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/sync/ModuleSyncWatchdogSpec.groovy
index b8b3e45a8f..4cf07e4c24 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/sync/ModuleSyncWatchdogSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/sync/ModuleSyncWatchdogSpec.groovy
@@ -56,7 +56,7 @@ class ModuleSyncWatchdogSpec extends Specification {
 
     def 'Module sync advised cm handles with #scenario.'() {
         given: 'module sync utilities returns #numberOfAdvisedCmHandles advised cm handles'
-            mockModuleOperationsUtils.getAdvisedCmHandles() >> createDataNodes(numberOfAdvisedCmHandles)
+            mockModuleOperationsUtils.getAdvisedCmHandleIds() >> createCmHandleIds(numberOfAdvisedCmHandles)
         and: 'module sync utilities returns no failed (locked) cm handles'
             mockModuleOperationsUtils.getCmHandlesThatFailedModelSyncOrUpgrade() >> []
         and: 'the work queue is not locked'
@@ -79,7 +79,7 @@ class ModuleSyncWatchdogSpec extends Specification {
 
     def 'Module sync cm handles starts with no available threads.'() {
         given: 'module sync utilities returns a advise cm handles'
-            mockModuleOperationsUtils.getAdvisedCmHandles() >> createDataNodes(1)
+            mockModuleOperationsUtils.getAdvisedCmHandleIds() >> createCmHandleIds(1)
         and: 'the work queue is not locked'
             mockWorkQueueLock.tryLock() >> true
         and: 'the executor first has no threads but has one thread on the second attempt'
@@ -92,7 +92,7 @@ class ModuleSyncWatchdogSpec extends Specification {
 
     def 'Module sync advised cm handle already handled by other thread.'() {
         given: 'module sync utilities returns an advised cm handle'
-            mockModuleOperationsUtils.getAdvisedCmHandles() >> createDataNodes(1)
+            mockModuleOperationsUtils.getAdvisedCmHandleIds() >> createCmHandleIds(1)
         and: 'the work queue is not locked'
             mockWorkQueueLock.tryLock() >> true
         and: 'the executor has a thread available'
@@ -107,9 +107,9 @@ class ModuleSyncWatchdogSpec extends Specification {
 
     def 'Module sync with previous cm handle(s) left in work queue.'() {
         given: 'there is still a cm handle in the queue'
-            moduleSyncWorkQueue.offer(new DataNode())
+            moduleSyncWorkQueue.offer('ch-1')
         and: 'sync utilities returns many advise cm handles'
-            mockModuleOperationsUtils.getAdvisedCmHandles() >> createDataNodes(500)
+            mockModuleOperationsUtils.getAdvisedCmHandleIds() >> createCmHandleIds(500)
         and: 'the executor has plenty threads available'
             spiedAsyncTaskExecutor.getAsyncTaskParallelismLevel() >> 10
         when: ' module sync is started'
@@ -130,7 +130,7 @@ class ModuleSyncWatchdogSpec extends Specification {
 
     def 'Module Sync Locking.'() {
         given: 'module sync utilities returns an advised cm handle'
-            mockModuleOperationsUtils.getAdvisedCmHandles() >> createDataNodes(1)
+            mockModuleOperationsUtils.getAdvisedCmHandleIds() >> createCmHandleIds(1)
         and: 'can lock is : #canLock'
             mockWorkQueueLock.tryLock() >> canLock
         when: 'attempt to populate the work queue'
@@ -152,9 +152,7 @@ class ModuleSyncWatchdogSpec extends Specification {
             noExceptionThrown()
     }
 
-    def createDataNodes(numberOfDataNodes) {
-        def dataNodes = []
-        numberOfDataNodes.times { dataNodes.add(new DataNode()) }
-        return dataNodes
+    def createCmHandleIds(numberOfCmHandles) {
+        return (numberOfCmHandles > 0) ? (1..numberOfCmHandles).collect { 'ch-'+it } : []
     }
 }
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/sync/SynchronizationCacheConfigSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/sync/SynchronizationCacheConfigSpec.groovy
index 6914273209..3213e5d442 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/sync/SynchronizationCacheConfigSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/inventory/sync/SynchronizationCacheConfigSpec.groovy
@@ -24,22 +24,20 @@ import com.hazelcast.collection.ISet
 import com.hazelcast.config.Config
 import com.hazelcast.core.Hazelcast
 import com.hazelcast.map.IMap
-import org.onap.cps.api.model.DataNode
+import java.util.concurrent.BlockingQueue
+import java.util.concurrent.TimeUnit
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.boot.test.context.SpringBootTest
 import org.springframework.test.context.ContextConfiguration
 import spock.lang.Specification
 import spock.util.concurrent.PollingConditions
 
-import java.util.concurrent.BlockingQueue
-import java.util.concurrent.TimeUnit
-
 @SpringBootTest
 @ContextConfiguration(classes = [SynchronizationCacheConfig])
 class SynchronizationCacheConfigSpec extends Specification {
 
     @Autowired
-    BlockingQueue<DataNode> moduleSyncWorkQueue
+    BlockingQueue<String> moduleSyncWorkQueue
 
     @Autowired
     IMap<String, Object> moduleSyncStartedOnCmHandles