Added legacy plugin tasks to graph only if graph is default

Also save graph_type in nailgun.task(AKA transaction)

Change-Id: I946a092df4d7c1bb143a47915cbc1455cc8c53c5
Closes-Bug: 1618758
Closes-Bug: 1618914
(cherry picked from commit fe87cfa835)

nailgun/nailgun/objects/cluster.py

@@ -1127,6 +1127,9 @@ class Cluster(NailgunObject):
         :return: deployment graph which includes metadata and tasks
         :rtype: dict
         """
+        if graph_type is None:
+            graph_type = consts.DEFAULT_DEPLOYMENT_GRAPH_TYPE
+
         release_deployment_graph = cls.get_release_deployment_graph(
             instance, graph_type=graph_type)
 

nailgun/nailgun/task/task.py

@@ -250,6 +250,7 @@ class DeploymentTask(BaseDeploymentTask):
         deployment_tasks = objects.Cluster.get_deployment_tasks(
             task.cluster, graph_type
         )
+        task.graph_type = graph_type or consts.DEFAULT_DEPLOYMENT_GRAPH_TYPE
         objects.Transaction.attach_tasks_snapshot(
             task,
             deployment_tasks
@@ -520,11 +521,13 @@ class ClusterTransaction(DeploymentTask):
 
         if objects.Cluster.is_propagate_task_deploy_enabled(cluster):
             logger.info("The legacy tasks adaptation is used.")
-            tasks = adapt_legacy_tasks(
-                tasks,
-                objects.Cluster.get_legacy_plugin_tasks(cluster),
-                role_resolver,
-            )
+            if transaction.graph_type == consts.DEFAULT_DEPLOYMENT_GRAPH_TYPE:
+                plugin_tasks = objects.Cluster.get_legacy_plugin_tasks(
+                    cluster
+                )
+            else:
+                plugin_tasks = None
+            tasks = adapt_legacy_tasks(tasks, plugin_tasks, role_resolver)
 
         directory, graph, metadata = lcm.TransactionSerializer.serialize(
             context,

nailgun/nailgun/test/integration/test_task_managers.py

@@ -83,9 +83,18 @@ class TestTaskManagers(BaseIntegrationTest):
         # deployment
         self.assertEqual(len(supertask.subtasks), 3)
         # provisioning task has less weight then deployment
-        provision_task = filter(
-            lambda t: t.name == TASK_NAMES.provision, supertask.subtasks)[0]
+        provision_task = next(
+            t for t in supertask.subtasks
+            if t.name == consts.TASK_NAMES.provision
+        )
         self.assertEqual(provision_task.weight, 0.4)
+        deployment_task = next(
+            t for t in supertask.subtasks
+            if t.name == consts.TASK_NAMES.deployment
+        )
+        self.assertEqual(
+            consts.DEFAULT_DEPLOYMENT_GRAPH_TYPE, deployment_task.graph_type
+        )
 
         cluster_name = cluster['name']
         self.assertIn(