Send ssh and ssl data removal task with reset task

The additional task is sent by manager to 'naily' rpc queue in similar
way that provision and deployment task are casted to ensure the order of
execution.

Change-Id: I5204a258ac0f00cf9184bd4903ff82c13e68de6a
Closes-bug: #1507361
(cherry picked from commit 0446b99f10)

nailgun/nailgun/task/manager.py

@@ -707,21 +707,30 @@ class ResetEnvironmentTaskManager(TaskManager):
                 consts.TASK_NAMES.stop_deployment
             ])
         )
+
         for task in obsolete_tasks:
             db().delete(task)
         db().commit()
 
-        task = Task(
+        supertask = Task(
             name=consts.TASK_NAMES.reset_environment,
             cluster=self.cluster
         )
-        db().add(task)
-        db.commit()
-        self._call_silently(
-            task,
-            tasks.ResetEnvironmentTask
+        db().add(supertask)
+        al = TaskHelper.create_action_log(supertask)
+
+        remove_keys_task = supertask.create_subtask(
+            consts.TASK_NAMES.reset_environment
         )
-        return task
+
+        db.commit()
+
+        rpc.cast('naily', [
+            tasks.ResetEnvironmentTask.message(supertask),
+            tasks.RemoveClusterKeys.message(remove_keys_task)
+        ])
+        TaskHelper.update_action_log(supertask, al)
+        return supertask
 
 
 class UpdateEnvironmentTaskManager(TaskManager):

nailgun/nailgun/task/task.py

@@ -641,9 +641,36 @@ class ResetEnvironmentTask(object):
         db().commit()
         return rpc_message
 
+
+class RemoveClusterKeys(object):
+    """Task that deletes all ssh and ssl data for deployed environment
+
+    Meant to be run after environment reset to make sure that new keys will be
+    generated.
+    """
+
     @classmethod
-    def execute(cls, task):
-        rpc.cast('naily', cls.message(task))
+    def message(cls, task):
+        rpc_message = make_astute_message(
+            task,
+            "execute_tasks",
+            "reset_environment_resp",
+            {
+                "tasks": [
+                    tasks_templates.make_shell_task(
+                        [consts.MASTER_ROLE],
+                        {
+                            "parameters": {
+                                "cmd": "rm -rf /var/lib/fuel/keys/{0}".format(
+                                    task.cluster.id),
+                                "timeout": 30
+                            }
+                        }
+                    )
+                ]
+            }
+        )
+        return rpc_message
 
 
 class ClusterDeletionTask(object):

nailgun/nailgun/test/integration/test_reset_environment.py

@@ -14,6 +14,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import mock
 from oslo_serialization import jsonutils
 
 from nailgun.db.sqlalchemy.models import Notification
@@ -121,3 +122,35 @@ class TestResetEnvironment(BaseIntegrationTest):
         self.env.refresh_nodes()
         self.assertEqual(node_db.pending_addition, True)
         self.assertEqual(node_db.pending_deletion, False)
+
+    @fake_tasks(
+        override_state={"progress": 100, "status": "ready"},
+        recover_nodes=False,
+        ia_nodes_count=1
+    )
+    def test_reset_environment_tasks(self):
+        self.env.create(
+            cluster_kwargs={},
+            nodes_kwargs=[
+                {"name": "First",
+                 "pending_addition": True},
+                {"name": "Second",
+                 "roles": ["compute"],
+                 "pending_addition": True}
+            ]
+        )
+        cluster_db = self.env.clusters[0]
+        supertask = self.env.launch_deployment()
+        self.env.wait_ready(supertask, 60)
+
+        for n in cluster_db.nodes:
+            self.assertEqual(n.status, "ready")
+            self.assertEqual(n.pending_addition, False)
+
+        with mock.patch('nailgun.task.task.rpc.cast') as cast_mock:
+            self.env.reset_environment()
+            casted_tasks = cast_mock.call_args[0][1]
+            self.assertEqual(len(casted_tasks), 2)
+
+            self.assertEqual(casted_tasks[0]['method'], 'reset_environment')
+            self.assertEqual(casted_tasks[1]['method'], 'execute_tasks')