def _trigger_job()

in services/lambda-mxnet-ci-bot/CIBot.py [0:0]

• 21 lines of code
• 7 McCabe index (conditional complexity)

    def _trigger_job(self, jenkins_obj, job, branch):
        """
        This method triggers a particular jenkins job
        :param jenkins_obj Jenkins Object
        :param job name of the job to be triggered
        :param branch e.g. master/PR
        """
        try:
            name = "mxnet-validation/" + job + "/" + branch
            job = jenkins_obj[name]
            # check if the job is already in queue or running and kill that pending job first in that case
            self._pending_build_cleanup(job, name)
            logging.info(f'invoking {name}')
            response = job.invoke(block=False)
            logging.info(response)
            return True
        except jenkinsapi.custom_exceptions.UnknownJob:
            if not self.auto_trigger:
                # Jenkins Automatic Trigger is disabled
                # branch isn't discovered yet
                # trigger a scan of multi-branch pipeline
                url = self.jenkins_url + "multibranch-webhook-trigger/invoke"
                job_trigger_token = self._get_job_trigger_token(job)
                headers = {"token": job_trigger_token}
                r = requests.post(url, headers=headers)
                logging.info(r.text)
                return True
            else:
                raise Exception("Unable to invoke job due to unknownJob error")
        except Exception as e:
            raise Exception("Unable to invoke job due to %s", exc_info=e)