Fix deployment input validation

Adds logic to validate manually launched deployments before creating
the deployment, while also validating auto deployments' inputs
before creating the deployment execution.

Author: Dany9966

coriolis/conductor/rpc/server.py

@@ -1399,13 +1399,7 @@ def _get_provider_types(self, ctxt, endpoint):
                 "No provider found for: %s" % endpoint.type)
         return provider_types["types"]
 
-    def _execute_deployment(self, ctxt, deployment, force):
-        transfer = self._get_transfer(
-            ctxt, deployment.transfer_id, include_task_info=True)
-        skip_os_morphing = deployment.skip_os_morphing
-        clone_disks = deployment.clone_disks
-        user_scripts = deployment.user_scripts
-
+    def _validate_deployment_inputs(self, ctxt, deployment, transfer, force):
         self._check_transfer_running_executions(ctxt, transfer)
         self._check_valid_transfer_tasks_execution(transfer, force)
         for instance, info in transfer.info.items():
@@ -1415,10 +1409,19 @@ def _execute_deployment(self, ctxt, deployment, force):
                     f"for instance: {instance}. If transferred disks are "
                     "deleted, the transfer needs to be executed anew "
                     "before a deployment can occur")
-        deployment.info = transfer.info
         self._check_minion_pools_for_action(ctxt, deployment)
         self._check_reservation_for_transfer(transfer)
 
+    def _execute_deployment(self, ctxt, deployment, force):
+        transfer = self._get_transfer(
+            ctxt, deployment.transfer_id, include_task_info=True)
+        skip_os_morphing = deployment.skip_os_morphing
+        clone_disks = deployment.clone_disks
+        user_scripts = deployment.user_scripts
+
+        if deployment.deployer_id:
+            self._validate_deployment_inputs(ctxt, deployment, transfer, force)
+        deployment.info = transfer.info
         destination_endpoint = self.get_endpoint(
             ctxt, transfer.destination_endpoint_id)
         destination_provider_types = self._get_provider_types(
@@ -1709,6 +1712,8 @@ def deploy_transfer_instances(
         if instance_osmorphing_minion_pool_mappings:
             deployment.instance_osmorphing_minion_pool_mappings.update(
                 instance_osmorphing_minion_pool_mappings)
+        if not wait_for_execution:
+            self._validate_deployment_inputs(ctxt, deployment, transfer, force)
 
         db_api.add_deployment(ctxt, deployment)
         LOG.info("Deployment created: %s", deployment.id)

coriolis/tests/conductor/rpc/test_server.py

@@ -2216,14 +2216,17 @@ def test_get_provider_types(self, mock_get_available_providers):
         mock_get_available_providers.assert_called_once_with(
             mock.sentinel.context)
 
+    @mock.patch.object(server.ConductorServerEndpoint,
+                       '_validate_deployment_inputs')
     @mock.patch.object(models, 'Deployment')
     @mock.patch.object(db_api, 'add_deployment')
     @mock.patch.object(server.ConductorServerEndpoint, '_get_transfer')
     @mock.patch.object(server.ConductorServerEndpoint, '_execute_deployment')
     @mock.patch.object(server.ConductorServerEndpoint, 'get_deployment')
     def test_deploy_transfer_instances(
             self, mock_get_deployment, mock_execute_deployment,
-            mock_get_transfer, mock_add_deployment, mock_deployment_model):
+            mock_get_transfer, mock_add_deployment, mock_deployment_model,
+            mock_validate_deployment_inputs):
         transfer_mock = mock.MagicMock()
         transfer_mock.instance_osmorphing_minion_pool_mappings = {
             mock.sentinel.instance1: mock.sentinel.pool1}
@@ -2234,13 +2237,14 @@ def test_deploy_transfer_instances(
         mock_get_transfer.return_value = transfer_mock
         osm_pool_mappings = {mock.sentinel.instance1: mock.sentinel.pool2}
         deployment = mock_deployment_model.return_value
+        force = False
 
         result = testutils.get_wrapped_function(
             self.server.deploy_transfer_instances)(
                 self.server,
                 mock.sentinel.ctxt,
                 mock.sentinel.transfer_id,
-                force=False,
+                force=force,
                 clone_disks=True,
                 instance_osmorphing_minion_pool_mappings=osm_pool_mappings,
                 skip_os_morphing=False,
@@ -2268,6 +2272,8 @@ def test_deploy_transfer_instances(
             mock.sentinel.ctxt, deployment, False)
         mock_get_deployment.assert_called_once_with(
             mock.sentinel.ctxt, deployment.id)
+        mock_validate_deployment_inputs.assert_called_once_with(
+            mock.sentinel.ctxt, deployment, transfer_mock, force)
 
     @mock.patch.object(models, 'Deployment')
     @mock.patch.object(db_api, 'add_deployment')