Integration test for the writeDataJob

    - added Stub controller
    - created integration test for writeDataJob

Issue-ID: CPS-2236
Change-Id: I067a11dc5bfe629d50128cf303b2a81abc75a348
Signed-off-by: leventecsanyi <levente.csanyi@est.tech>

8 files changed, 165 insertions, 61 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 6855e49c54..5e46e95a0c 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
@@ -125,12 +125,17 @@ abstract class CpsIntegrationSpecBase extends Specification {
     @Autowired
     AlternateIdMatcher alternateIdMatcher
 
-    MockWebServer mockDmiServer = null
-    DmiDispatcher dmiDispatcher = new DmiDispatcher()
+    MockWebServer mockDmiServer1 = new MockWebServer()
+    MockWebServer mockDmiServer2 = new MockWebServer()
 
-    def DMI_URL = null
+    DmiDispatcher dmiDispatcher1 = new DmiDispatcher()
+    DmiDispatcher dmiDispatcher2 = new DmiDispatcher()
+
+    def DMI1_URL = null
+    def DMI2_URL = null
 
     static NO_MODULE_SET_TAG = ''
+    static NO_ALTERNATE_ID = ''
     static GENERAL_TEST_DATASPACE = 'generalTestDataspace'
     static BOOKSTORE_SCHEMA_SET = 'bookstoreSchemaSet'
     static MODULE_SYNC_WAIT_TIME_IN_SECONDS = 10
@@ -144,14 +149,19 @@ abstract class CpsIntegrationSpecBase extends Specification {
             createStandardBookStoreSchemaSet(GENERAL_TEST_DATASPACE)
             initialized = true
         }
-        mockDmiServer = new MockWebServer()
-        mockDmiServer.setDispatcher(dmiDispatcher)
-        mockDmiServer.start()
-        DMI_URL = String.format("http://%s:%s", mockDmiServer.getHostName(), mockDmiServer.getPort())
+        mockDmiServer1.setDispatcher(dmiDispatcher1)
+        mockDmiServer1.start()
+
+        mockDmiServer2.setDispatcher(dmiDispatcher2)
+        mockDmiServer2.start()
+
+        DMI1_URL = String.format("http://%s:%s", mockDmiServer1.getHostName(), mockDmiServer1.getPort())
+        DMI2_URL = String.format("http://%s:%s", mockDmiServer2.getHostName(), mockDmiServer2.getPort())
     }
 
     def cleanup() {
-        mockDmiServer.shutdown()
+        mockDmiServer1.shutdown()
+        mockDmiServer2.shutdown()
     }
 
     def static readResourceDataFile(filename) {
@@ -217,7 +227,11 @@ abstract class CpsIntegrationSpecBase extends Specification {
     // *** NCMP Integration Test Utilities ***
 
     def registerCmHandle(dmiPlugin, cmHandleId, moduleSetTag) {
-        def cmHandleToCreate = new NcmpServiceCmHandle(cmHandleId: cmHandleId, moduleSetTag: moduleSetTag)
+        registerCmHandle(dmiPlugin, cmHandleId, moduleSetTag, NO_ALTERNATE_ID)
+    }
+
+    def registerCmHandle(dmiPlugin, cmHandleId, moduleSetTag, alternateId) {
+        def cmHandleToCreate = new NcmpServiceCmHandle(cmHandleId: cmHandleId, moduleSetTag: moduleSetTag, alternateId: alternateId)
         networkCmProxyInventoryFacade.updateDmiRegistrationAndSyncModule(new DmiPluginRegistration(dmiPlugin: dmiPlugin, createdCmHandles: [cmHandleToCreate]))
         new PollingConditions().within(MODULE_SYNC_WAIT_TIME_IN_SECONDS, () -> {
             CmHandleState.READY == networkCmProxyInventoryFacade.getCmHandleCompositeState(cmHandleId).cmHandleState
diff --git a/integration-test/src/test/groovy/org/onap/cps/integration/base/DmiDispatcher.groovy b/integration-test/src/test/groovy/org/onap/cps/integration/base/DmiDispatcher.groovy
index 50ff93cebb..0b0e33c6f6 100644
--- a/integration-test/src/test/groovy/org/onap/cps/integration/base/DmiDispatcher.groovy
+++ b/integration-test/src/test/groovy/org/onap/cps/integration/base/DmiDispatcher.groovy
@@ -22,6 +22,7 @@ package org.onap.cps.integration.base
 
 import static org.onap.cps.integration.base.CpsIntegrationSpecBase.readResourceDataFile
 
+import groovy.json.JsonSlurper
 import java.util.regex.Matcher
 import okhttp3.mockwebserver.Dispatcher
 import okhttp3.mockwebserver.MockResponse
@@ -53,7 +54,10 @@ class DmiDispatcher extends Dispatcher {
     static final MODULE_RESOURCES_RESPONSE_TEMPLATE = readResourceDataFile('mock-dmi-responses/moduleResourcesTemplate.json')
 
     def isAvailable = true
-    Map<String, List<String>> moduleNamesPerCmHandleId = [:]
+
+    def jsonSlurper = new JsonSlurper()
+    def moduleNamesPerCmHandleId = [:]
+    def receivedSubJobs = [:]
     def lastAuthHeaderReceived
     def dmiResourceDataUrl
 
@@ -87,11 +91,23 @@ class DmiDispatcher extends Dispatcher {
             case ~'^/dmi/v1/data$':
                 return mockResponseWithBody(HttpStatus.ACCEPTED, '{}')
 
+            // get write sub job response
+            case ~'^/dmi/v1/writeJob/(.*)$':
+                return mockWriteJobResponse(request)
+
             default:
                 throw new IllegalArgumentException('Mock DMI does not implement endpoint ' + request.path)
         }
     }
 
+    def mockWriteJobResponse(request) {
+        def requestId = Matcher.lastMatcher[0][1]
+        def subJobWriteRequest = jsonSlurper.parseText(request.getBody().readUtf8())
+        this.receivedSubJobs.put(requestId, subJobWriteRequest)
+        def response = '{"subJobId":"some sub job id", "dmiServiceName":"some dmi service name", "dataProducerId":"some data producer id"}'
+        return mockResponseWithBody(HttpStatus.OK, response)
+    }
+
     private getModuleReferencesResponse(cmHandleId) {
         def moduleReferences = '{"schemas":[' + getModuleNamesForCmHandle(cmHandleId).collect {
             MODULE_REFERENCES_RESPONSE_TEMPLATE.replaceAll("<MODULE_NAME>", it)
diff --git a/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/BearerTokenPassthroughSpec.groovy b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/BearerTokenPassthroughSpec.groovy
index c91e750d6e..99e80323c2 100644
--- a/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/BearerTokenPassthroughSpec.groovy
+++ b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/BearerTokenPassthroughSpec.groovy
@@ -36,12 +36,12 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.
 class BearerTokenPassthroughSpec extends CpsIntegrationSpecBase {
 
     def setup() {
-        dmiDispatcher.moduleNamesPerCmHandleId['ch-1'] = ['M1', 'M2']
-        registerCmHandle(DMI_URL, 'ch-1', NO_MODULE_SET_TAG)
+        dmiDispatcher1.moduleNamesPerCmHandleId['ch-1'] = ['M1', 'M2']
+        registerCmHandle(DMI1_URL, 'ch-1', NO_MODULE_SET_TAG)
     }
 
     def cleanup() {
-        deregisterCmHandle(DMI_URL, 'ch-1')
+        deregisterCmHandle(DMI1_URL, 'ch-1')
     }
 
     def 'Bearer token is passed from NCMP to DMI in pass-through data operations.'() {
@@ -54,7 +54,7 @@ class BearerTokenPassthroughSpec extends CpsIntegrationSpecBase {
                     .andExpect(status().is2xxSuccessful())
 
         then: 'DMI has received request with bearer token'
-            assert dmiDispatcher.lastAuthHeaderReceived == 'Bearer some-bearer-token'
+            assert dmiDispatcher1.lastAuthHeaderReceived == 'Bearer some-bearer-token'
 
         where: 'all HTTP operations are applied'
             httpMethod << [GET, POST, PUT, PATCH, DELETE]
@@ -70,7 +70,7 @@ class BearerTokenPassthroughSpec extends CpsIntegrationSpecBase {
                     .andExpect(status().is2xxSuccessful())
 
         then: 'DMI has received request with no authorization header'
-            assert dmiDispatcher.lastAuthHeaderReceived == null
+            assert dmiDispatcher1.lastAuthHeaderReceived == null
 
         where: 'all HTTP operations are applied'
             httpMethod << [GET, POST, PUT, PATCH, DELETE]
@@ -94,7 +94,7 @@ class BearerTokenPassthroughSpec extends CpsIntegrationSpecBase {
 
         then: 'DMI will receive the async request with bearer token'
             new PollingConditions().within(3, () -> {
-                assert dmiDispatcher.lastAuthHeaderReceived == 'Bearer some-bearer-token'
+                assert dmiDispatcher1.lastAuthHeaderReceived == 'Bearer some-bearer-token'
             })
     }
 
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 c9a64e0ab8..3d526c6a42 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
@@ -48,14 +48,14 @@ class CmHandleCreateSpec extends CpsIntegrationSpecBase {
 
     def 'CM Handle registration is successful.'() {
         given: 'DMI will return modules when requested'
-            dmiDispatcher.moduleNamesPerCmHandleId['ch-1'] = ['M1', 'M2']
+            dmiDispatcher1.moduleNamesPerCmHandleId['ch-1'] = ['M1', 'M2']
 
         and: 'consumer subscribed to topic'
             kafkaConsumer.subscribe(['ncmp-events'])
 
         when: 'a CM-handle is registered for creation'
             def cmHandleToCreate = new NcmpServiceCmHandle(cmHandleId: 'ch-1')
-            def dmiPluginRegistration = new DmiPluginRegistration(dmiPlugin: DMI_URL, createdCmHandles: [cmHandleToCreate])
+            def dmiPluginRegistration = new DmiPluginRegistration(dmiPlugin: DMI1_URL, createdCmHandles: [cmHandleToCreate])
             def dmiPluginRegistrationResponse = objectUnderTest.updateDmiRegistrationAndSyncModule(dmiPluginRegistration)
 
         then: 'registration gives successful response'
@@ -81,16 +81,16 @@ class CmHandleCreateSpec extends CpsIntegrationSpecBase {
             assert ['M1', 'M2'] == objectUnderTest.getYangResourcesModuleReferences('ch-1').moduleName.sort()
 
         cleanup: 'deregister CM handle'
-            deregisterCmHandle(DMI_URL, 'ch-1')
+            deregisterCmHandle(DMI1_URL, 'ch-1')
     }
 
     def 'CM Handle goes to LOCKED state when DMI gives error during module sync.'() {
         given: 'DMI is not available to handle requests'
-            dmiDispatcher.isAvailable = false
+            dmiDispatcher1.isAvailable = false
 
         when: 'a CM-handle is registered for creation'
             def cmHandleToCreate = new NcmpServiceCmHandle(cmHandleId: 'ch-1')
-            def dmiPluginRegistration = new DmiPluginRegistration(dmiPlugin: DMI_URL, createdCmHandles: [cmHandleToCreate])
+            def dmiPluginRegistration = new DmiPluginRegistration(dmiPlugin: DMI1_URL, createdCmHandles: [cmHandleToCreate])
             objectUnderTest.updateDmiRegistrationAndSyncModule(dmiPluginRegistration)
 
         then: 'CM-handle goes to LOCKED state with reason MODULE_SYNC_FAILED'
@@ -104,19 +104,19 @@ class CmHandleCreateSpec extends CpsIntegrationSpecBase {
             assert objectUnderTest.getYangResourcesModuleReferences('ch-1').empty
 
         cleanup: 'deregister CM handle'
-            deregisterCmHandle(DMI_URL, 'ch-1')
+            deregisterCmHandle(DMI1_URL, 'ch-1')
     }
 
     def 'Create a CM-handle with existing moduleSetTag.'() {
         given: 'DMI will return modules when requested'
-            dmiDispatcher.moduleNamesPerCmHandleId = ['ch-1': ['M1', 'M2'], 'ch-2': ['M1', 'M3']]
+            dmiDispatcher1.moduleNamesPerCmHandleId = ['ch-1': ['M1', 'M2'], 'ch-2': ['M1', 'M3']]
         and: 'existing CM-handles cm-1 with moduleSetTag "A", and cm-2 with moduleSetTag "B"'
-            registerCmHandle(DMI_URL, 'ch-1', 'A')
-            registerCmHandle(DMI_URL, 'ch-2', 'B')
+            registerCmHandle(DMI1_URL, 'ch-1', 'A')
+            registerCmHandle(DMI1_URL, 'ch-2', 'B')
 
         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: DMI_URL, createdCmHandles: [cmHandleToCreate]))
+            objectUnderTest.updateDmiRegistrationAndSyncModule(new DmiPluginRegistration(dmiPlugin: DMI1_URL, createdCmHandles: [cmHandleToCreate]))
 
         then: 'the CM-handle goes to READY state'
             new PollingConditions().within(MODULE_SYNC_WAIT_TIME_IN_SECONDS, () -> {
@@ -130,16 +130,16 @@ class CmHandleCreateSpec extends CpsIntegrationSpecBase {
             assert ['M1', 'M3'] == objectUnderTest.getYangResourcesModuleReferences('ch-3').moduleName.sort()
 
         cleanup: 'deregister CM handles'
-            deregisterCmHandles(DMI_URL, ['ch-1', 'ch-2', 'ch-3'])
+            deregisterCmHandles(DMI1_URL, ['ch-1', 'ch-2', 'ch-3'])
     }
 
     def 'CM Handle retry after failed module sync.'() {
         given: 'DMI is not initially available to handle requests'
-            dmiDispatcher.isAvailable = false
+            dmiDispatcher1.isAvailable = false
 
         when: 'CM-handles are registered for creation'
-            def cmHandlesToCreate = [new NcmpServiceCmHandle(cmHandleId: 'ch-1')]
-            def dmiPluginRegistration = new DmiPluginRegistration(dmiPlugin: DMI_URL, createdCmHandles: cmHandlesToCreate)
+            def cmHandlesToCreate = [new NcmpServiceCmHandle(cmHandleId: 'ch-1'), new NcmpServiceCmHandle(cmHandleId: 'ch-2')]
+            def dmiPluginRegistration = new DmiPluginRegistration(dmiPlugin: DMI1_URL, createdCmHandles: cmHandlesToCreate)
             objectUnderTest.updateDmiRegistrationAndSyncModule(dmiPluginRegistration)
         then: 'CM-handles go to LOCKED state'
             new PollingConditions().within(MODULE_SYNC_WAIT_TIME_IN_SECONDS, () -> {
@@ -147,8 +147,8 @@ class CmHandleCreateSpec extends CpsIntegrationSpecBase {
             })
 
         when: 'DMI is available for retry'
-            dmiDispatcher.moduleNamesPerCmHandleId = ['ch-1': ['M1', 'M2']]
-            dmiDispatcher.isAvailable = true
+            dmiDispatcher1.moduleNamesPerCmHandleId = ['ch-1': ['M1', 'M2']]
+            dmiDispatcher1.isAvailable = true
         and: 'the LOCKED CM handle retry time elapses (actually just subtract 3 minutes from handles lastUpdateTime)'
             overrideCmHandleLastUpdateTime('ch-1', OffsetDateTime.now().minusMinutes(3))
 
@@ -159,7 +159,7 @@ class CmHandleCreateSpec extends CpsIntegrationSpecBase {
         and: 'CM-handle has expected modules'
             assert ['M1', 'M2'] == objectUnderTest.getYangResourcesModuleReferences('ch-1').moduleName.sort()
 
-        cleanup: 'deregister CM handle'
-            deregisterCmHandle(DMI_URL, 'ch-1')
+        cleanup: 'deregister CM handles'
+            deregisterCmHandles(DMI1_URL, ['ch-1', 'ch-2'])
     }
 }
diff --git a/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleResourceDataSpec.groovy b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleResourceDataSpec.groovy
index 8dfa118da6..418b3a4088 100644
--- a/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleResourceDataSpec.groovy
+++ b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleResourceDataSpec.groovy
@@ -32,13 +32,13 @@ class CmHandleResourceDataSpec extends CpsIntegrationSpecBase {
     NetworkCmProxyFacade objectUnderTest
 
     def setup() {
-        dmiDispatcher.moduleNamesPerCmHandleId['ch-1'] = ['M1', 'M2']
-        registerCmHandle(DMI_URL, 'ch-1', NO_MODULE_SET_TAG)
+        dmiDispatcher1.moduleNamesPerCmHandleId['ch-1'] = ['M1', 'M2']
+        registerCmHandle(DMI1_URL, 'ch-1', NO_MODULE_SET_TAG)
         objectUnderTest = networkCmProxyFacade
     }
 
     def cleanup() {
-        deregisterCmHandle(DMI_URL, 'ch-1')
+        deregisterCmHandle(DMI1_URL, 'ch-1')
     }
 
     def 'Get resource data having special chars into path & query param value.'() {
@@ -47,7 +47,7 @@ class CmHandleResourceDataSpec extends CpsIntegrationSpecBase {
             objectUnderTest.getResourceDataForCmHandle(cmResourceAddress, '(a=1,b=2)', 'my-client-topic', false, null)
         then: 'dmi resource data url is encoded correctly'
             new PollingConditions().within(5, () -> {
-                assert dmiDispatcher.dmiResourceDataUrl == '/dmi/v1/ch/ch-1/data/ds/ncmp-datastore%3Apassthrough-operational?resourceIdentifier=parent%2Fchild&options=%28a%3D1%2Cb%3D2%29&topic=my-client-topic'
+                assert dmiDispatcher1.dmiResourceDataUrl == '/dmi/v1/ch/ch-1/data/ds/ncmp-datastore%3Apassthrough-operational?resourceIdentifier=parent%2Fchild&options=%28a%3D1%2Cb%3D2%29&topic=my-client-topic'
             })
     }
 }
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 35ea0793ce..f93f58ce20 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
@@ -42,14 +42,14 @@ class CmHandleUpgradeSpec extends CpsIntegrationSpecBase {
 
     def 'Upgrade CM-handle with new moduleSetTag or no moduleSetTag.'() {
         given: 'a CM-handle is created with expected initial modules: M1 and M2'
-            dmiDispatcher.moduleNamesPerCmHandleId[CM_HANDLE_ID] = ['M1', 'M2']
-            registerCmHandle(DMI_URL, CM_HANDLE_ID, initialModuleSetTag)
+            dmiDispatcher1.moduleNamesPerCmHandleId[CM_HANDLE_ID] = ['M1', 'M2']
+            registerCmHandle(DMI1_URL, CM_HANDLE_ID, initialModuleSetTag)
             assert ['M1', 'M2'] == objectUnderTest.getYangResourcesModuleReferences(CM_HANDLE_ID).moduleName.sort()
 
         when: "the CM-handle is upgraded with given moduleSetTag '${updatedModuleSetTag}'"
             def cmHandlesToUpgrade = new UpgradedCmHandles(cmHandles: [CM_HANDLE_ID], moduleSetTag: updatedModuleSetTag)
             def dmiPluginRegistrationResponse = objectUnderTest.updateDmiRegistrationAndSyncModule(
-                    new DmiPluginRegistration(dmiPlugin: DMI_URL, upgradedCmHandles: cmHandlesToUpgrade))
+                    new DmiPluginRegistration(dmiPlugin: DMI1_URL, upgradedCmHandles: cmHandlesToUpgrade))
 
         then: 'registration gives successful response'
             assert dmiPluginRegistrationResponse.upgradedCmHandles == [CmHandleRegistrationResponse.createSuccessResponse(CM_HANDLE_ID)]
@@ -61,7 +61,7 @@ class CmHandleUpgradeSpec extends CpsIntegrationSpecBase {
             assert cmHandleCompositeState.lockReason.details == "Upgrade to ModuleSetTag: ${updatedModuleSetTag}"
 
         when: 'DMI will return different modules for upgrade: M1 and M3'
-            dmiDispatcher.moduleNamesPerCmHandleId[CM_HANDLE_ID] = ['M1', 'M3']
+            dmiDispatcher1.moduleNamesPerCmHandleId[CM_HANDLE_ID] = ['M1', 'M3']
 
         then: 'CM-handle goes to READY state'
             new PollingConditions().within(MODULE_SYNC_WAIT_TIME_IN_SECONDS, () -> {
@@ -75,7 +75,7 @@ class CmHandleUpgradeSpec extends CpsIntegrationSpecBase {
             assert ['M1', 'M3'] == objectUnderTest.getYangResourcesModuleReferences(CM_HANDLE_ID).moduleName.sort()
 
         cleanup: 'deregister CM-handle'
-            deregisterCmHandle(DMI_URL, CM_HANDLE_ID)
+            deregisterCmHandle(DMI1_URL, CM_HANDLE_ID)
 
         where:
             initialModuleSetTag | updatedModuleSetTag
@@ -87,19 +87,19 @@ class CmHandleUpgradeSpec extends CpsIntegrationSpecBase {
 
     def 'Upgrade CM-handle with existing moduleSetTag.'() {
         given: 'DMI will return modules for registration'
-            dmiDispatcher.moduleNamesPerCmHandleId[CM_HANDLE_ID] = ['M1', 'M2']
-            dmiDispatcher.moduleNamesPerCmHandleId[CM_HANDLE_ID_WITH_EXISTING_MODULE_SET_TAG] = ['M1', 'M3']
+            dmiDispatcher1.moduleNamesPerCmHandleId[CM_HANDLE_ID] = ['M1', 'M2']
+            dmiDispatcher1.moduleNamesPerCmHandleId[CM_HANDLE_ID_WITH_EXISTING_MODULE_SET_TAG] = ['M1', 'M3']
         and: "an existing CM-handle handle with moduleSetTag '${updatedModuleSetTag}'"
-            registerCmHandle(DMI_URL, CM_HANDLE_ID_WITH_EXISTING_MODULE_SET_TAG, updatedModuleSetTag)
+            registerCmHandle(DMI1_URL, CM_HANDLE_ID_WITH_EXISTING_MODULE_SET_TAG, updatedModuleSetTag)
             assert ['M1', 'M3'] == objectUnderTest.getYangResourcesModuleReferences(CM_HANDLE_ID_WITH_EXISTING_MODULE_SET_TAG).moduleName.sort()
         and: "a CM-handle with moduleSetTag '${initialModuleSetTag}' which will be upgraded"
-            registerCmHandle(DMI_URL, CM_HANDLE_ID, initialModuleSetTag)
+            registerCmHandle(DMI1_URL, CM_HANDLE_ID, initialModuleSetTag)
             assert ['M1', 'M2'] == objectUnderTest.getYangResourcesModuleReferences(CM_HANDLE_ID).moduleName.sort()
 
         when: "CM-handle is upgraded to moduleSetTag '${updatedModuleSetTag}'"
             def cmHandlesToUpgrade = new UpgradedCmHandles(cmHandles: [CM_HANDLE_ID], moduleSetTag: updatedModuleSetTag)
             def dmiPluginRegistrationResponse = objectUnderTest.updateDmiRegistrationAndSyncModule(
-                    new DmiPluginRegistration(dmiPlugin: DMI_URL, upgradedCmHandles: cmHandlesToUpgrade))
+                    new DmiPluginRegistration(dmiPlugin: DMI1_URL, upgradedCmHandles: cmHandlesToUpgrade))
 
         then: 'registration gives successful response'
             assert dmiPluginRegistrationResponse.upgradedCmHandles == [CmHandleRegistrationResponse.createSuccessResponse(CM_HANDLE_ID)]
@@ -116,7 +116,7 @@ class CmHandleUpgradeSpec extends CpsIntegrationSpecBase {
             assert ['M1', 'M3'] == objectUnderTest.getYangResourcesModuleReferences(CM_HANDLE_ID).moduleName.sort()
 
         cleanup: 'deregister CM-handle'
-            deregisterCmHandles(DMI_URL, [CM_HANDLE_ID, CM_HANDLE_ID_WITH_EXISTING_MODULE_SET_TAG])
+            deregisterCmHandles(DMI1_URL, [CM_HANDLE_ID, CM_HANDLE_ID_WITH_EXISTING_MODULE_SET_TAG])
 
         where:
             initialModuleSetTag | updatedModuleSetTag
@@ -126,14 +126,14 @@ class CmHandleUpgradeSpec extends CpsIntegrationSpecBase {
 
     def 'Skip upgrade of CM-handle with same moduleSetTag as before.'() {
         given: 'an existing CM-handle with expected initial modules: M1 and M2'
-            dmiDispatcher.moduleNamesPerCmHandleId[CM_HANDLE_ID] = ['M1', 'M2']
-            registerCmHandle(DMI_URL, CM_HANDLE_ID, 'same')
+            dmiDispatcher1.moduleNamesPerCmHandleId[CM_HANDLE_ID] = ['M1', 'M2']
+            registerCmHandle(DMI1_URL, CM_HANDLE_ID, 'same')
             assert ['M1', 'M2'] == objectUnderTest.getYangResourcesModuleReferences(CM_HANDLE_ID).moduleName.sort()
 
         when: 'CM-handle is upgraded with the same moduleSetTag'
             def cmHandlesToUpgrade = new UpgradedCmHandles(cmHandles: [CM_HANDLE_ID], moduleSetTag: 'same')
             objectUnderTest.updateDmiRegistrationAndSyncModule(
-                    new DmiPluginRegistration(dmiPlugin: DMI_URL, upgradedCmHandles: cmHandlesToUpgrade))
+                    new DmiPluginRegistration(dmiPlugin: DMI1_URL, upgradedCmHandles: cmHandlesToUpgrade))
 
         then: 'CM-handle remains in READY state'
             assert CmHandleState.READY == objectUnderTest.getCmHandleCompositeState(CM_HANDLE_ID).cmHandleState
@@ -145,20 +145,20 @@ class CmHandleUpgradeSpec extends CpsIntegrationSpecBase {
             assert ['M1', 'M2'] == objectUnderTest.getYangResourcesModuleReferences(CM_HANDLE_ID).moduleName.sort()
 
         cleanup: 'deregister CM-handle'
-            deregisterCmHandle(DMI_URL, CM_HANDLE_ID)
+            deregisterCmHandle(DMI1_URL, CM_HANDLE_ID)
     }
 
     def 'Upgrade of CM-handle fails due to DMI error.'() {
         given: 'a CM-handle exists'
-            dmiDispatcher.moduleNamesPerCmHandleId[CM_HANDLE_ID] = ['M1', 'M2']
-            registerCmHandle(DMI_URL, CM_HANDLE_ID, 'oldTag')
+            dmiDispatcher1.moduleNamesPerCmHandleId[CM_HANDLE_ID] = ['M1', 'M2']
+            registerCmHandle(DMI1_URL, CM_HANDLE_ID, 'oldTag')
         and: 'DMI is not available for upgrade'
-            dmiDispatcher.isAvailable = false
+            dmiDispatcher1.isAvailable = false
 
         when: 'the CM-handle is upgraded'
             def cmHandlesToUpgrade = new UpgradedCmHandles(cmHandles: [CM_HANDLE_ID], moduleSetTag: 'newTag')
             objectUnderTest.updateDmiRegistrationAndSyncModule(
-                    new DmiPluginRegistration(dmiPlugin: DMI_URL, upgradedCmHandles: cmHandlesToUpgrade))
+                    new DmiPluginRegistration(dmiPlugin: DMI1_URL, upgradedCmHandles: cmHandlesToUpgrade))
 
         then: 'CM-handle goes to LOCKED state with reason MODULE_UPGRADE_FAILED'
             new PollingConditions().within(MODULE_SYNC_WAIT_TIME_IN_SECONDS, () -> {
@@ -171,7 +171,7 @@ class CmHandleUpgradeSpec extends CpsIntegrationSpecBase {
             assert objectUnderTest.getNcmpServiceCmHandle(CM_HANDLE_ID).moduleSetTag == 'oldTag'
 
         cleanup: 'deregister CM-handle'
-            deregisterCmHandle(DMI_URL, CM_HANDLE_ID)
+            deregisterCmHandle(DMI1_URL, CM_HANDLE_ID)
     }
 
 }
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 ab189c2917..33973e547b 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
@@ -35,13 +35,13 @@ class RestApiSpec extends CpsIntegrationSpecBase {
 
     def 'Register CM Handles using REST API.'() {
         given: 'DMI will return modules'
-            dmiDispatcher.moduleNamesPerCmHandleId = [
+            dmiDispatcher1.moduleNamesPerCmHandleId = [
                 'ch-1': ['M1', 'M2'],
                 'ch-2': ['M1', 'M2'],
                 'ch-3': ['M1', 'M3']
             ]
         when: 'a POST request is made to register the CM Handles'
-            def requestBody = '{"dmiPlugin":"'+DMI_URL+'","createdCmHandles":[{"cmHandle":"ch-1"},{"cmHandle":"ch-2"},{"cmHandle":"ch-3"}]}'
+            def requestBody = '{"dmiPlugin":"'+DMI1_URL+'","createdCmHandles":[{"cmHandle":"ch-1"},{"cmHandle":"ch-2"},{"cmHandle":"ch-3"}]}'
             mvc.perform(post('/ncmpInventory/v1/ch').contentType(MediaType.APPLICATION_JSON).content(requestBody))
                     .andExpect(status().is2xxSuccessful())
         then: 'CM-handles go to READY state'
@@ -78,7 +78,7 @@ class RestApiSpec extends CpsIntegrationSpecBase {
 
     def 'De-register CM handles using REST API.'() {
         when: 'a POST request is made to deregister the CM Handle'
-            def requestBody = '{"dmiPlugin":"'+DMI_URL+'", "removedCmHandles": ["ch-1", "ch-2", "ch-3"]}'
+            def requestBody = '{"dmiPlugin":"'+DMI1_URL+'", "removedCmHandles": ["ch-1", "ch-2", "ch-3"]}'
             mvc.perform(post('/ncmpInventory/v1/ch').contentType(MediaType.APPLICATION_JSON).content(requestBody))
                     .andExpect(status().is2xxSuccessful())
         then: 'the CM handles are not found using GET'
diff --git a/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/WriteSubJobSpec.groovy b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/WriteSubJobSpec.groovy
new file mode 100644
index 0000000000..0999bda9f4
--- /dev/null
+++ b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/WriteSubJobSpec.groovy
@@ -0,0 +1,74 @@
+/*
+ *  ============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.api.datajobs.DataJobService
+import org.onap.cps.ncmp.api.datajobs.models.DataJobMetadata
+import org.onap.cps.ncmp.api.datajobs.models.DataJobWriteRequest
+import org.onap.cps.ncmp.api.datajobs.models.SubJobWriteResponse
+import org.onap.cps.ncmp.api.datajobs.models.WriteOperation
+import org.springframework.beans.factory.annotation.Autowired
+
+class WriteSubJobSpec extends CpsIntegrationSpecBase {
+
+    @Autowired
+    DataJobService dataJobService
+
+    def setup() {
+        dmiDispatcher1.moduleNamesPerCmHandleId['ch-1'] = ['M1']
+        dmiDispatcher1.moduleNamesPerCmHandleId['ch-2'] = ['M2']
+        dmiDispatcher2.moduleNamesPerCmHandleId['ch-3'] = ['M3']
+        registerCmHandle(DMI1_URL, 'ch-1', NO_MODULE_SET_TAG, 'p1')
+        registerCmHandle(DMI1_URL, 'ch-2', NO_MODULE_SET_TAG, 'p2')
+        registerCmHandle(DMI2_URL, 'ch-3', NO_MODULE_SET_TAG, 'p3')
+    }
+
+    def cleanup() {
+        deregisterCmHandle(DMI1_URL, 'ch-1')
+        deregisterCmHandle(DMI1_URL, 'ch-2')
+        deregisterCmHandle(DMI2_URL, 'ch-3')
+    }
+
+    def 'Create a sub-job write request.'() {
+        given: 'the required input data for the write job'
+            def dataJobWriteRequest = new DataJobWriteRequest([new WriteOperation('p1', '', '', null), new WriteOperation('p2', '', '', null), new WriteOperation('p3', '', '', null)])
+            def myDataJobMetadata = new DataJobMetadata('', '', '')
+            def dataJobId = 'my-data-job-id'
+        when: 'sending a write job to NCMP with 2 sub-jobs for DMI 1 and 1 sub-job for DMI 2'
+            def response = dataJobService.writeDataJob(dataJobId, myDataJobMetadata, dataJobWriteRequest)
+        then: 'each DMI received the expected sub-jobs and the response has the expected values'
+            assert response.size() == 2
+            assert response[0].class == SubJobWriteResponse.class
+            assert response[0].subJobId == "some sub job id"
+            assert response[0].dmiServiceName == "some dmi service name"
+            assert response[0].dataProducerId == "some data producer id"
+        and: 'dmi 1 received the correct job details'
+            def receivedSubJobsForDispatcher1 = dmiDispatcher1.receivedSubJobs['my-data-job-id']['data'].collect()
+            assert receivedSubJobsForDispatcher1.size() == 2
+            assert receivedSubJobsForDispatcher1[0]['path'] == 'p1'
+            assert receivedSubJobsForDispatcher1[1]['path'] == 'p2'
+        and: 'dmi 2 received the correct job details'
+            def receivedSubJobsForDispatcher2 = dmiDispatcher2.receivedSubJobs['my-data-job-id']['data'].collect()
+            assert receivedSubJobsForDispatcher2.size() == 1
+            assert receivedSubJobsForDispatcher2[0]['path'] == 'p3'
+    }
+}