More refactoring of jobs

Move more of db_upgrade into a generic shell script executor.

Change-Id: I71d4e4ac800d1b7dd8f66d00b2ef757d5c39d0ab

2 files changed, 133 insertions, 136 deletions

diff --git a/turbo_hipster/lib/models.py b/turbo_hipster/lib/models.py
index eb2f92f..66f26c8 100644
--- a/turbo_hipster/lib/models.py
+++ b/turbo_hipster/lib/models.py
@@ -30,15 +30,43 @@ class Task(object):
         self.global_config = global_config
         self.plugin_config = plugin_config
         self.job_name = job_name
+        self._reset()
 
+        # Define the number of steps we will do to determine our progress.
+        self.total_steps = 0
+
+    def _reset(self):
         self.job = None
         self.job_arguments = None
         self.work_data = None
         self.cancelled = False
-
-        # Define the number of steps we will do to determine our progress.
+        self.success = True
+        self.messages = []
         self.current_step = 0
-        self.total_steps = 0
+
+    def start_job(self, job):
+        self._reset()
+        self.job = job
+
+        if self.job is not None:
+            try:
+                self.job_arguments = \
+                    json.loads(self.job.arguments.decode('utf-8'))
+                self.log.debug("Got job from ZUUL %s" % self.job_arguments)
+
+                # Send an initial WORK_DATA and WORK_STATUS packets
+                self._send_work_data()
+
+                # Execute the job_steps
+                self.do_job_steps()
+
+                # Finally, send updated work data and completed packets
+                self._send_final_results()
+
+            except Exception as e:
+                self.log.exception('Exception handling log event.')
+                if not self.cancelled:
+                    self.job.sendWorkException(str(e).encode('utf-8'))
 
     def stop_worker(self, number):
         # Check the number is for this job instance
@@ -49,27 +77,6 @@ class Task(object):
             self.cancelled = True
             # TODO: Work out how to kill current step
 
-    @common.task_step
-    def _grab_patchset(self, job_args, job_log_file_path):
-        """ Checkout the reference into config['git_working_dir'] """
-
-        self.log.debug("Grab the patchset we want to test against")
-        local_path = os.path.join(self.global_config['git_working_dir'],
-                                  self.job_name, job_args['ZUUL_PROJECT'])
-        if not os.path.exists(local_path):
-            os.makedirs(local_path)
-
-        git_args = copy.deepcopy(job_args)
-        git_args['GIT_ORIGIN'] = 'git://git.openstack.org/'
-
-        cmd = os.path.join(os.path.join(os.path.dirname(__file__),
-                                        'gerrit-git-prep.sh'))
-        cmd += ' https://review.openstack.org'
-        cmd += ' http://zuul.rcbops.com'
-        utils.execute_to_log(cmd, job_log_file_path, env=git_args,
-                             cwd=local_path)
-        return local_path
-
     def _get_work_data(self):
         if self.work_data is None:
             hostname = os.uname()[1]
@@ -87,6 +94,15 @@ class Task(object):
                        json.dumps(self._get_work_data()))
         self.job.sendWorkData(json.dumps(self._get_work_data()))
 
+    def _send_final_results(self):
+        self._send_work_data()
+
+        if self.work_data['result'] is 'SUCCESS':
+            self.job.sendWorkComplete(
+                json.dumps(self._get_work_data()))
+        else:
+            self.job.sendWorkFail()
+
     def _do_next_step(self):
         """ Send a WORK_STATUS command to the gearman server.
         This can provide a progress bar. """
@@ -110,43 +126,58 @@ class ShellTask(Task):
         # Define the number of steps we will do to determine our progress.
         self.total_steps = 4
 
-    def start_job(self, job):
-        self.job = job
-        self.success = True
-        self.messages = []
+    def _reset(self):
+        super(ShellTask, self)._reset()
+        self.git_path = None
 
-        if self.job is not None:
-            try:
-                self.job_arguments = \
-                    json.loads(self.job.arguments.decode('utf-8'))
-                self.log.debug("Got job from ZUUL %s" % self.job_arguments)
+    def do_job_steps(self, job):
+        # Step 1: Checkout updates from git
+        self._grab_patchset(self.job_arguments,
+                            self.job_datasets[0]['job_log_file_path'])
 
-                # Send an initial WORK_DATA and WORK_STATUS packets
-                self._send_work_data()
+        # Step 2: Run shell script
+        self._execute_script()
+
+        # Step 3: Analyse logs for errors
+        self._parse_and_check_results()
+
+        # Step 4: handle the results (and upload etc)
+        self._handle_results()
+
+    @common.task_step
+    def _grab_patchset(self, job_args, job_log_file_path):
+        """ Checkout the reference into config['git_working_dir'] """
 
-                # Step 1: Checkout updates from git!
-                self.git_path = self._grab_patchset(
-                    self.job_arguments,
-                    self.job_datasets[0]['job_log_file_path'])
+        self.log.debug("Grab the patchset we want to test against")
+        local_path = os.path.join(self.global_config['git_working_dir'],
+                                  self.job_name, job_args['ZUUL_PROJECT'])
+        if not os.path.exists(local_path):
+            os.makedirs(local_path)
 
-                # Step 3: execute shell script
-                # TODO
+        git_args = copy.deepcopy(job_args)
+        git_args['GIT_ORIGIN'] = 'git://git.openstack.org/'
 
-                # Step 4: Analyse logs for errors
-                # TODO
+        cmd = os.path.join(os.path.join(os.path.dirname(__file__),
+                                        'gerrit-git-prep.sh'))
+        cmd += ' https://review.openstack.org'
+        cmd += ' http://zuul.rcbops.com'
+        utils.execute_to_log(cmd, job_log_file_path, env=git_args,
+                             cwd=local_path)
+        self.git_path = local_path
+        return local_path
 
-                # Step 5: handle the results (and upload etc)
-                # TODO
+    @common.task_step
+    def _execute_script(self):
+        # Run script
+        self.script_return_code = 0
 
-                # Finally, send updated work data and completed packets
-                self._send_work_data()
+    @common.task_step
+    def _parse_and_check_results(self):
+        if self.script_return_code > 0:
+            self.success = False
+            self.messages.append('Return code from test script was non-zero '
+                                 '(%d)' % self.script_return_code)
 
-                if self.work_data['result'] is 'SUCCESS':
-                    self.job.sendWorkComplete(
-                        json.dumps(self._get_work_data()))
-                else:
-                    self.job.sendWorkFail()
-            except Exception as e:
-                self.log.exception('Exception handling log event.')
-                if not self.cancelled:
-                    self.job.sendWorkException(str(e).encode('utf-8'))
+    @common.task_step
+    def _handle_results(self):
+        pass
diff --git a/turbo_hipster/task_plugins/gate_real_db_upgrade/task.py b/turbo_hipster/task_plugins/gate_real_db_upgrade/task.py
index 3dae2f3..3d271c0 100644
--- a/turbo_hipster/task_plugins/gate_real_db_upgrade/task.py
+++ b/turbo_hipster/task_plugins/gate_real_db_upgrade/task.py
@@ -32,7 +32,7 @@ MIGRATION_START_RE = re.compile('([0-9]+) -> ([0-9]+)\.\.\.$')
 MIGRATION_END_RE = re.compile('^done$')
 
 
-class Runner(models.Task):
+class Runner(models.ShellTask):
 
     """ This thread handles the actual sql-migration tests.
         It pulls in a gearman job from the  build:gate-real-db-upgrade
@@ -50,52 +50,51 @@ class Runner(models.Task):
         # Define the number of steps we will do to determine our progress.
         self.total_steps = 5
 
-    def start_job(self, job):
-        self.job = job
-        self.success = True
-        self.messages = []
-
-        if self.job is not None:
-            try:
-                self.job_arguments = \
-                    json.loads(self.job.arguments.decode('utf-8'))
-                self.log.debug("Got job from ZUUL %s" % self.job_arguments)
-
-                # Send an initial WORK_DATA and WORK_STATUS packets
-                self._send_work_data()
-
-                # Step 1: Figure out which datasets to run
-                self.job_datasets = self._get_job_datasets()
-
-                # Step 2: Checkout updates from git!
-                self.git_path = self._grab_patchset(
-                    self.job_arguments,
-                    self.job_datasets[0]['job_log_file_path'])
-
-                # Step 3: Run migrations on datasets
-                if self._execute_migrations() > 0:
-                    self.success = False
-                    self.messages.append('Return code from test script was '
-                                         'non-zero')
-
-                # Step 4: Analyse logs for errors
-                self._check_all_dataset_logs_for_errors()
-
-                # Step 5: handle the results (and upload etc)
-                self._handle_results()
-
-                # Finally, send updated work data and completed packets
-                self._send_work_data()
-
-                if self.work_data['result'] is 'SUCCESS':
-                    self.job.sendWorkComplete(
-                        json.dumps(self._get_work_data()))
-                else:
-                    self.job.sendWorkFail()
-            except Exception as e:
-                self.log.exception('Exception handling log event.')
-                if not self.cancelled:
-                    self.job.sendWorkException(str(e).encode('utf-8'))
+    def do_job_steps(self):
+        # Step 1: Figure out which datasets to run
+        self.job_datasets = self._get_job_datasets()
+
+        # all other steps are common to running a shell script
+        super(Runner, self).job_steps()
+
+    @common.task_step
+    def _get_job_datasets(self):
+        """ Take the applicable datasets for this job and set them up in
+        self.job_datasets """
+
+        job_datasets = []
+        for dataset in self._get_datasets():
+            # Only load a dataset if it is the right project and we
+            # know how to process the upgrade
+            if (self.job_arguments['ZUUL_PROJECT'] ==
+                    dataset['config']['project'] and
+                    self._get_project_command(dataset['config']['type'])):
+                dataset['determined_path'] = utils.determine_job_identifier(
+                    self.job_arguments, self.plugin_config['function'],
+                    self.job.unique
+                )
+                dataset['job_log_file_path'] = os.path.join(
+                    self.global_config['jobs_working_dir'],
+                    dataset['determined_path'],
+                    dataset['name'] + '.log'
+                )
+                dataset['result'] = 'UNTESTED'
+                dataset['command'] = \
+                    self._get_project_command(dataset['config']['type'])
+
+                job_datasets.append(dataset)
+
+        return job_datasets
+
+    @common.task_step
+    def _execute_script(self):
+        # Run script
+        self.script_return_code = self._execute_migrations()
+
+    @common.task_step
+    def _parse_and_check_results(self):
+        super(Runner, self)._parse_and_check_results()
+        self._check_all_dataset_logs_for_errors()
 
     @common.task_step
     def _handle_results(self):
@@ -108,11 +107,8 @@ class Runner(models.Task):
         self.log.debug("Index URL found at %s" % index_url)
         self.work_data['url'] = index_url
 
-    @common.task_step
     def _check_all_dataset_logs_for_errors(self):
         self.log.debug('Check logs for errors')
-        self.success = True
-        self.messages = []
 
         for i, dataset in enumerate(self.job_datasets):
             success, messages = handle_results.check_log_file(
@@ -156,35 +152,6 @@ class Runner(models.Task):
 
         return self.datasets
 
-    @common.task_step
-    def _get_job_datasets(self):
-        """ Take the applicable datasets for this job and set them up in
-        self.job_datasets """
-
-        job_datasets = []
-        for dataset in self._get_datasets():
-            # Only load a dataset if it is the right project and we
-            # know how to process the upgrade
-            if (self.job_arguments['ZUUL_PROJECT'] ==
-                    dataset['config']['project'] and
-                    self._get_project_command(dataset['config']['type'])):
-                dataset['determined_path'] = utils.determine_job_identifier(
-                    self.job_arguments, self.plugin_config['function'],
-                    self.job.unique
-                )
-                dataset['job_log_file_path'] = os.path.join(
-                    self.global_config['jobs_working_dir'],
-                    dataset['determined_path'],
-                    dataset['name'] + '.log'
-                )
-                dataset['result'] = 'UNTESTED'
-                dataset['command'] = \
-                    self._get_project_command(dataset['config']['type'])
-
-                job_datasets.append(dataset)
-
-        return job_datasets
-
     def _get_project_command(self, db_type):
         command = (self.job_arguments['ZUUL_PROJECT'].split('/')[-1] + '_' +
                    db_type + '_migrations.sh')
@@ -193,7 +160,6 @@ class Runner(models.Task):
             return command
         return False
 
-    @common.task_step
     def _execute_migrations(self):
         """ Execute the migration on each dataset in datasets """