CP-9575 Add support for additional mount in vsdk/dvp for dSources operations (#459)

SumoSourabh · web-flow · commit 8a9f174de266 · 2023-03-22T08:09:44.000+05:30
* CP-9575 Add support for additional mount in vsdk/dvp for dSources operations
diff --git a/common/setup.py b/common/setup.py
@@ -4,7 +4,7 @@
 PYTHON_SRC = 'src/main/python'
 
 install_requires = [
-    "dvp-api == 1.7.0",
+    "dvp-api == 1.7.1.dev0",
     "six >= 1.16, < 1.17",
 ]
 
diff --git a/libs/setup.py b/libs/setup.py
@@ -7,7 +7,7 @@
     version = version_file.read().strip()
 
 install_requires = [
-    "dvp-api == 1.7.0",
+    "dvp-api == 1.7.1.dev0",
     "dvp-common == {}".format(version),
     "six >= 1.16, < 1.17",
 ]
diff --git a/platform/setup.py b/platform/setup.py
@@ -7,7 +7,7 @@
     version = version_file.read().strip()
 
 install_requires = [
-    "dvp-api == 1.7.0",
+    "dvp-api == 1.7.1.dev0",
     "dvp-common == {}".format(version),
     "six >= 1.16, < 1.17",
 ]
diff --git a/platform/src/main/python/dlpx/virtualization/platform/_linked.py b/platform/src/main/python/dlpx/virtualization/platform/_linked.py
@@ -112,6 +112,30 @@ def source_size_decorator(source_size_impl):
 
         return source_size_decorator
 
+    @staticmethod
+    def _from_protobuf_remote_mount(remote_mount):
+        return Mount(remote_environment=RemoteEnvironment.from_proto(
+            remote_mount.remote_environment),
+            mount_path=remote_mount.mount_path,
+            shared_path=remote_mount.shared_path)
+
+    @staticmethod
+    def _get_mounts_from_request(request):
+        staged_mount = request.staged_source.staged_mount
+        mounts = request.staged_source.mounts
+        if mounts and len(mounts) > 0 and staged_mount and \
+                staged_mount.mount_path and len(staged_mount.mount_path) > 0 \
+                and staged_mount.remote_environment:
+            raise PluginRuntimeError(
+                'Either staged_mount or mounts can be present for staging source. '
+                'Found both staged_mount and mounts.')
+
+        if mounts and len(mounts) > 0:
+            return None, [
+                LinkedOperations._from_protobuf_remote_mount(m) for m in mounts]
+        else:
+            return LinkedOperations._from_protobuf_remote_mount(staged_mount), None
+
     def _internal_direct_pre_snapshot(self, request):
         """Pre Snapshot Wrapper for direct plugins.
 
@@ -331,19 +355,16 @@ def _internal_staged_pre_snapshot(self, request):
         linked_source = request.staged_source.linked_source
         staged_source_definition = (LinkedSourceDefinition.from_dict(
             json.loads(linked_source.parameters.json)))
-        staged_mount = request.staged_source.staged_mount
-        mount = Mount(remote_environment=RemoteEnvironment.from_proto(
-            staged_mount.remote_environment),
-            mount_path=staged_mount.mount_path,
-            shared_path=staged_mount.shared_path)
+        staged_mount, mounts = LinkedOperations._get_mounts_from_request(request)
         staged_source = StagedSource(
             guid=linked_source.guid,
             source_connection=RemoteConnection.from_proto(
                 request.staged_source.source_connection),
             parameters=staged_source_definition,
-            mount=mount,
+            mount=staged_mount,
             staged_connection=RemoteConnection.from_proto(
-                request.staged_source.staged_connection))
+                request.staged_source.staged_connection),
+            mounts=mounts)
 
         repository = RepositoryDefinition.from_dict(
             json.loads(request.repository.parameters.json))
@@ -406,19 +427,16 @@ def to_protobuf(snapshot):
 
         staged_source_definition = LinkedSourceDefinition.from_dict(
             json.loads(request.staged_source.linked_source.parameters.json))
-        mount = Mount(
-            remote_environment=RemoteEnvironment.from_proto(
-                request.staged_source.staged_mount.remote_environment),
-            mount_path=request.staged_source.staged_mount.mount_path,
-            shared_path=request.staged_source.staged_mount.shared_path)
+        staged_mount, mounts = LinkedOperations._get_mounts_from_request(request)
         staged_source = StagedSource(
             guid=request.staged_source.linked_source.guid,
             source_connection=RemoteConnection.from_proto(
                 request.staged_source.source_connection),
             parameters=staged_source_definition,
-            mount=mount,
+            mount=staged_mount,
             staged_connection=RemoteConnection.from_proto(
-                request.staged_source.staged_connection))
+                request.staged_source.staged_connection),
+            mounts=mounts)
 
         repository = RepositoryDefinition.from_dict(
             json.loads(request.repository.parameters.json))
@@ -480,19 +498,16 @@ def _internal_start_staging(self, request):
 
         staged_source_definition = LinkedSourceDefinition.from_dict(
             json.loads(request.staged_source.linked_source.parameters.json))
-        mount = Mount(
-            remote_environment=(RemoteEnvironment.from_proto(
-                request.staged_source.staged_mount.remote_environment)),
-            mount_path=request.staged_source.staged_mount.mount_path,
-            shared_path=request.staged_source.staged_mount.shared_path)
+        staged_mount, mounts = LinkedOperations._get_mounts_from_request(request)
         staged_source = StagedSource(
             guid=request.staged_source.linked_source.guid,
             source_connection=RemoteConnection.from_proto(
                 request.staged_source.source_connection),
             parameters=staged_source_definition,
-            mount=mount,
+            mount=staged_mount,
             staged_connection=RemoteConnection.from_proto(
-                request.staged_source.staged_connection))
+                request.staged_source.staged_connection),
+            mounts=mounts)
 
         repository = RepositoryDefinition.from_dict(
             json.loads(request.repository.parameters.json))
@@ -539,19 +554,16 @@ def _internal_stop_staging(self, request):
 
         staged_source_definition = LinkedSourceDefinition.from_dict(
             json.loads(request.staged_source.linked_source.parameters.json))
-        mount = Mount(
-            remote_environment=(RemoteEnvironment.from_proto(
-                request.staged_source.staged_mount.remote_environment)),
-            mount_path=request.staged_source.staged_mount.mount_path,
-            shared_path=request.staged_source.staged_mount.shared_path)
+        staged_mount, mounts = LinkedOperations._get_mounts_from_request(request)
         staged_source = StagedSource(
             guid=request.staged_source.linked_source.guid,
             source_connection=RemoteConnection.from_proto(
                 request.staged_source.source_connection),
             parameters=staged_source_definition,
-            mount=mount,
+            mount=staged_mount,
             staged_connection=RemoteConnection.from_proto(
-                request.staged_source.staged_connection))
+                request.staged_source.staged_connection),
+            mounts=mounts)
 
         repository = RepositoryDefinition.from_dict(
             json.loads(request.repository.parameters.json))
@@ -598,19 +610,16 @@ def _internal_status(self, request):
 
         staged_source_definition = LinkedSourceDefinition.from_dict(
             json.loads(request.staged_source.linked_source.parameters.json))
-        mount = Mount(
-            remote_environment=(RemoteEnvironment.from_proto(
-                request.staged_source.staged_mount.remote_environment)),
-            mount_path=request.staged_source.staged_mount.mount_path,
-            shared_path=request.staged_source.staged_mount.shared_path)
+        staged_mount, mounts = LinkedOperations._get_mounts_from_request(request)
         staged_source = StagedSource(
             guid=request.staged_source.linked_source.guid,
             source_connection=RemoteConnection.from_proto(
                 request.staged_source.source_connection),
             parameters=staged_source_definition,
-            mount=mount,
+            mount=staged_mount,
             staged_connection=RemoteConnection.from_proto(
-                request.staged_source.staged_connection))
+                request.staged_source.staged_connection),
+            mounts=mounts)
 
         repository = RepositoryDefinition.from_dict(
             json.loads(request.repository.parameters.json))
@@ -661,19 +670,16 @@ def _internal_worker(self, request):
 
         staged_source_definition = LinkedSourceDefinition.from_dict(
             json.loads(request.staged_source.linked_source.parameters.json))
-        mount = Mount(
-            remote_environment=(RemoteEnvironment.from_proto(
-                request.staged_source.staged_mount.remote_environment)),
-            mount_path=request.staged_source.staged_mount.mount_path,
-            shared_path=request.staged_source.staged_mount.shared_path)
+        staged_mount, mounts = LinkedOperations._get_mounts_from_request(request)
         staged_source = StagedSource(
             guid=request.staged_source.linked_source.guid,
             source_connection=RemoteConnection.from_proto(
                 request.staged_source.source_connection),
             parameters=staged_source_definition,
-            mount=mount,
+            mount=staged_mount,
             staged_connection=RemoteConnection.from_proto(
-                request.staged_source.staged_connection))
+                request.staged_source.staged_connection),
+            mounts=mounts)
 
         repository = RepositoryDefinition.from_dict(
             json.loads(request.repository.parameters.json))
@@ -712,6 +718,16 @@ def _internal_mount_specification(self, request):
         from generated.definitions import RepositoryDefinition
         from generated.definitions import LinkedSourceDefinition
 
+        def to_protobuf_subset_mount(subset_mount):
+            subset_mount_protobuf = common_pb2.SingleSubsetMount()
+            subset_mount_protobuf.mount_path = subset_mount.mount_path
+            subset_mount_protobuf.remote_environment.CopyFrom(
+                subset_mount.remote_environment.to_proto())
+            if subset_mount.shared_path:
+                subset_mount_protobuf.shared_path = subset_mount.shared_path
+
+            return subset_mount_protobuf
+
         def to_protobuf_single_mount(single_mount):
             if single_mount.shared_path:
                 raise PluginRuntimeError(
@@ -734,19 +750,16 @@ def to_protobuf_ownership_spec(ownership_spec):
 
         staged_source_definition = LinkedSourceDefinition.from_dict(
             json.loads(request.staged_source.linked_source.parameters.json))
-        mount = Mount(
-            remote_environment=(RemoteEnvironment.from_proto(
-                request.staged_source.staged_mount.remote_environment)),
-            mount_path=request.staged_source.staged_mount.mount_path,
-            shared_path=request.staged_source.staged_mount.shared_path)
+        staged_mount, mounts = LinkedOperations._get_mounts_from_request(request)
         staged_source = StagedSource(
             guid=request.staged_source.linked_source.guid,
             source_connection=RemoteConnection.from_proto(
                 request.staged_source.source_connection),
             parameters=staged_source_definition,
-            mount=mount,
+            mount=staged_mount,
             staged_connection=RemoteConnection.from_proto(
-                request.staged_source.staged_connection))
+                request.staged_source.staged_connection),
+            mounts=mounts)
 
         repository = RepositoryDefinition.from_dict(
             json.loads(request.repository.parameters.json))
@@ -760,18 +773,20 @@ def to_protobuf_ownership_spec(ownership_spec):
                                            type(mount_spec),
                                            MountSpecification)
 
-        # Only one mount is supported for linked sources.
+        staged_mount_spec_response = platform_pb2.StagedMountSpecResponse()
+
         mount_len = len(mount_spec.mounts)
-        if mount_len != 1:
+        if mount_len < 1:
             raise PluginRuntimeError(
-                'Exactly one mount must be provided for staging sources.'
-                ' Found {}'.format(mount_len))
-
-        staged_mount = to_protobuf_single_mount(mount_spec.mounts[0])
-
-        staged_mount_spec_response = platform_pb2.StagedMountSpecResponse()
-        staged_mount_spec_response.return_value.staged_mount.CopyFrom(
-            staged_mount)
+                'Mount must be provided for staging sources.'
+                ' Found {} mounts.'.format(mount_len))
+        elif mount_len > 1:
+            mounts = [to_protobuf_subset_mount(m) for m in mount_spec.mounts]
+            staged_mount_spec_response.return_value.mounts.extend(mounts)
+        else:
+            staged_mount = to_protobuf_single_mount(mount_spec.mounts[0])
+            staged_mount_spec_response.return_value.staged_mount.CopyFrom(
+                staged_mount)
 
         # Ownership spec is optional for linked sources.
         if mount_spec.ownership_specification:
@@ -812,19 +827,16 @@ def _internal_staged_source_size(self, request):
 
         staged_source_definition = LinkedSourceDefinition.from_dict(
             json.loads(request.staged_source.linked_source.parameters.json))
-        mount = Mount(
-            remote_environment=(RemoteEnvironment.from_proto(
-                request.staged_source.staged_mount.remote_environment)),
-            mount_path=request.staged_source.staged_mount.mount_path,
-            shared_path=request.staged_source.staged_mount.shared_path)
+        staged_mount, mounts = LinkedOperations._get_mounts_from_request(request)
         staged_source = StagedSource(
             guid=request.staged_source.linked_source.guid,
             source_connection=RemoteConnection.from_proto(
                 request.staged_source.source_connection),
             parameters=staged_source_definition,
-            mount=mount,
+            mount=staged_mount,
             staged_connection=RemoteConnection.from_proto(
-                request.staged_source.staged_connection))
+                request.staged_source.staged_connection),
+            mounts=mounts)
 
         repository = RepositoryDefinition.from_dict(
             json.loads(request.repository.parameters.json))
diff --git a/platform/src/main/python/dlpx/virtualization/platform/_plugin_classes.py b/platform/src/main/python/dlpx/virtualization/platform/_plugin_classes.py
@@ -59,7 +59,7 @@ def mounts(self):
 
 class StagedSource(object):
     def __init__(self, guid, source_connection, parameters, mount,
-                 staged_connection):
+                 staged_connection, mounts):
 
         self._guid = guid
         if not isinstance(source_connection, RemoteConnection):
@@ -72,6 +72,7 @@ def __init__(self, guid, source_connection, parameters, mount,
             raise IncorrectTypeError(RemoteConnection, 'staged_connection',
                                      type(staged_connection), RemoteConnection)
         self._staged_connection = staged_connection
+        self._mounts = mounts
 
     @property
     def guid(self):
@@ -93,7 +94,7 @@ def parameters(self):
 
     @property
     def mount(self):
-        """MountSpecification: The MountSpecification for this StagedSource."""
+        """MountSpecification: The SingleEntireMount for this StagedSource."""
         return self._mount
 
     @property
@@ -102,6 +103,11 @@ def staged_connection(self):
         staging environment for this StagedSource."""
         return self._staged_connection
 
+    @property
+    def mounts(self):
+        """list(Mount): The List of SingleSubsetMounts of this StagedSource."""
+        return self._mounts
+
 
 class DirectSource(object):
     def __init__(self, guid, connection, parameters):
diff --git a/platform/src/test/python/dlpx/virtualization/test_plugin.py b/platform/src/test/python/dlpx/virtualization/test_plugin.py
diff --git a/tools/src/test/python/dlpx/virtualization/_internal/test_package_util.py b/tools/src/test/python/dlpx/virtualization/_internal/test_package_util.py

Original file line number	Diff line number	Diff line change
`@@ -4,7 +4,7 @@`
`4`	`4`	`PYTHON_SRC = 'src/main/python'`
`5`	`5`
`6`	`6`	`install_requires = [`
`7`		`- "dvp-api == 1.7.0",`
	`7`	`+ "dvp-api == 1.7.1.dev0",`
`8`	`8`	`"six >= 1.16, < 1.17",`
`9`	`9`	`]`
`10`	`10`
Original file line number	Diff line number	Diff line change
`@@ -7,7 +7,7 @@`
`7`	`7`	`version = version_file.read().strip()`
`8`	`8`
`9`	`9`	`install_requires = [`
`10`		`- "dvp-api == 1.7.0",`
	`10`	`+ "dvp-api == 1.7.1.dev0",`
`11`	`11`	`"dvp-common == {}".format(version),`
`12`	`12`	`"six >= 1.16, < 1.17",`
`13`	`13`	`]`