Add support of selected_task_ids to ClusterTransaction

Now when you run only one task from CLI or API this not starts the whole
deployment

Change-Id: Ibf339eaacee4f5596d3c5338323f79605604ffc1
Related-Bug: #1567112

nailgun/nailgun/task/task.py

@@ -372,7 +372,24 @@ class ClusterTransaction(DeploymentTask):
         return ['task_deploy']
 
     @classmethod
-    def task_deploy(cls, transaction, tasks, nodes, force=False, **kwargs):
+    def mark_skipped(cls, tasks, ids_not_to_skip):
+        """Change tasks type which ids not present in ids_not_to_skip to skipped
+
+        :param tasks: the list of deployment tasks to execute
+        :param ids_not_to_skip: the list of task ids that will be not skipped
+        """
+        task_ids = set(ids_not_to_skip)
+
+        for task in tasks:
+            if task['id'] not in task_ids:
+                task = task.copy()
+                task['type'] = consts.ORCHESTRATOR_TASK_TYPES.skipped
+
+            yield task
+
+    @classmethod
+    def task_deploy(cls, transaction, tasks, nodes, force=False,
+                    selected_task_ids=None, **kwargs):
         logger.info("The cluster transaction is initiated.")
         logger.info("cluster serialization is started.")
         # we should update information for all nodes except deleted
@@ -397,9 +414,14 @@ class ClusterTransaction(DeploymentTask):
         # TODO(bgaifullin) Primary roles applied in deployment_serializers
         # need to move this code from deployment serializer
         # also role resolver should be created after serialization completed
+        if selected_task_ids:
+            tasks = cls.mark_skipped(tasks, selected_task_ids)
+
         role_resolver = RoleResolver(nodes)
         directory, graph = lcm.TransactionSerializer.serialize(
-            context, tasks, role_resolver
+            context,
+            tasks,
+            role_resolver,
         )
         logger.info("tasks serialization is finished.")
         return {

nailgun/nailgun/test/base.py

@@ -950,7 +950,11 @@ class EnvironmentManager(object):
         raise Exception(
             'Cluster with ID "{0}" was not found.'.format(cluster_id))
 
-    def _launch_for_selected_nodes(self, handler, nodes_uids, cluster_id):
+    def _launch_for_selected_nodes(self, handler, nodes_uids, cluster_id,
+                                   body=None):
+        if body is None:
+            body = {}
+
         if self.clusters:
             cluster = self._get_cluster_by_id(cluster_id)
             if not nodes_uids:
@@ -961,7 +965,7 @@ class EnvironmentManager(object):
             ) + '?nodes={0}'.format(','.join(nodes_uids))
             resp = self.app.put(
                 action_url,
-                '{}',
+                jsonutils.dumps(body),
                 headers=self.default_headers,
                 expect_errors=True
             )
@@ -988,6 +992,13 @@ class EnvironmentManager(object):
             'DeploySelectedNodes', nodes_uids, cluster_id
         )
 
+    def launch_deployment_selected_tasks(self,
+                                         nodes_uids, cluster_id, task_ids):
+        return self._launch_for_selected_nodes(
+            'DeploySelectedNodesWithTasks', nodes_uids, cluster_id,
+            task_ids or [],
+        )
+
     def _launch_for_cluster(self, handler, cluster_id):
         if self.clusters:
             cluster_id = self._get_cluster_by_id(cluster_id).id

nailgun/nailgun/test/integration/test_task_managers.py

@@ -1250,6 +1250,48 @@ class TestTaskManagers(BaseIntegrationTest):
             [consts.MASTER_NODE_UID, None] + nodes_uids, tasks_graph
         )
 
+    @mock.patch('nailgun.task.task.rpc.cast')
+    @mock.patch('objects.Cluster.get_deployment_tasks')
+    def test_only_certain_tasks_run_in_deploy(self, tasks_mock, rpc_mock):
+        task = {
+            'id': 'test', 'parameters': {}, 'type': 'puppet',
+            'roles': ['master'], 'version': '2.1.0',
+        }
+
+        tasks = []
+        for i in range(5):
+            task_copy = task.copy()
+            task_copy['id'] = 'test' + str(i)
+            tasks.append(task_copy)
+
+        tasks_mock.return_value = tasks
+
+        self.env.create(
+            nodes_kwargs=[
+                {'status': NODE_STATUSES.provisioned, 'roles': ['controller']},
+                {'status': NODE_STATUSES.provisioned, 'roles': ['compute']},
+                {'status': NODE_STATUSES.provisioned, 'roles': ['cinder']},
+            ],
+            release_kwargs={
+                'operating_system': consts.RELEASE_OS.ubuntu,
+                'version': 'mitaka-9.0',
+            },
+        )
+        cluster = self.env.clusters[-1]
+        task_ids = ['test0', 'test3']
+        task = self.env.launch_deployment_selected_tasks(
+            [n.uid for n in cluster.nodes],
+            cluster.id, task_ids)
+
+        self.assertNotEqual(consts.TASK_STATUSES.error, task.status)
+
+        tasks_graph = rpc_mock.call_args[0][1]['args']['tasks_graph']
+        for task in tasks_graph['master']:
+            if task['id'] in task_ids:
+                self.assertEqual(task['type'], 'puppet')
+            else:
+                self.assertEqual(task['type'], 'skipped')
+
 
 class TestUpdateDnsmasqTaskManagers(BaseIntegrationTest):