python-jenkinsapi 0.2.30-1 / usr / lib / python2.7 / dist-packages / jenkinsapi / job.py

"""
Module for jenkinsapi Job
"""

from collections import defaultdict
from jenkinsapi.build import Build
from jenkinsapi.custom_exceptions import (
    NoBuildData,
    NotConfiguredSCM,
    NotFound,
    NotInQueue,
    NotSupportSCM,
    UnknownQueueItem,
    BadParams,
)
from jenkinsapi.jenkinsbase import JenkinsBase
from jenkinsapi.mutable_jenkins_thing import MutableJenkinsThing
from jenkinsapi.queue import QueueItem
import json
import logging

import xml.etree.ElementTree as ET


try:
    import urlparse
except ImportError:
    # Python3
    import urllib.parse as urlparse


SVN_URL = './scm/locations/hudson.scm.SubversionSCM_-ModuleLocation/remote'
GIT_URL = './scm/userRemoteConfigs/hudson.plugins.git.UserRemoteConfig/url'
HG_URL = './scm/source'
GIT_BRANCH = './scm/branches/hudson.plugins.git.BranchSpec/name'
HG_BRANCH = './scm/branch'
DEFAULT_HG_BRANCH_NAME = 'default'

log = logging.getLogger(__name__)


class Job(JenkinsBase, MutableJenkinsThing):

    """
    Represents a jenkins job
    A job can hold N builds which are the actual execution environments
    """

    def __init__(self, url, name, jenkins_obj):
        self.name = name
        self.jenkins = jenkins_obj
        self._revmap = None
        self._config = None
        self._element_tree = None
        self._scm_map = {
            'hudson.scm.SubversionSCM': 'svn',
            'hudson.plugins.git.GitSCM': 'git',
            'hudson.plugins.mercurial.MercurialSCM': 'hg',
            'hudson.scm.NullSCM': 'NullSCM'
        }
        self._scmurlmap = {
            'svn': lambda element_tree: list(element_tree.findall(SVN_URL)),
            'git': lambda element_tree: list(element_tree.findall(GIT_URL)),
            'hg': lambda element_tree: list(element_tree.findall(HG_URL)),
            None: lambda element_tree: []
        }
        self._scmbranchmap = {
            'svn': lambda element_tree: [],
            'git': lambda element_tree: list(element_tree.findall(GIT_BRANCH)),
            'hg': self._get_hg_branch,
            None: lambda element_tree: []
        }
        JenkinsBase.__init__(self, url)

    def __str__(self):
        return self.name

    def get_description(self):
        return self._data["description"]

    def get_jenkins_obj(self):
        return self.jenkins

    # When the name of the hg branch used in the job is default hg branch (i.e.
    # default), Mercurial plugin doesn't store default branch name in
    # config XML file of the job. Create XML node corresponding to
    # default branch
    def _get_hg_branch(self, element_tree):
        branches = element_tree.findall(HG_BRANCH)
        if not branches:
            hg_default_branch = ET.Element('branch')
            hg_default_branch.text = DEFAULT_HG_BRANCH_NAME
            branches.append(hg_default_branch)
        return branches

    def poll(self, tree=None):
        data = super(Job, self).poll(tree=tree)
        if not tree:
            self._data = self._add_missing_builds(self._data)
        else:
            return data

    # pylint: disable=E1123
    # Unexpected keyword arg 'params'
    def _add_missing_builds(self, data):
        """
        Query Jenkins to get all builds of the job in the data object.

        Jenkins API loads the first 100 builds and thus may not contain
        all builds information. This method checks if all builds are loaded
        in the data object and updates it with the missing builds if needed.
        """
        if not data.get("builds"):
            return data
        # do not call _buildid_for_type here: it would poll and do an infinite
        # loop
        oldest_loaded_build_number = data["builds"][-1]["number"]
        if not self._data['firstBuild']:
            first_build_number = oldest_loaded_build_number
        else:
            first_build_number = self._data["firstBuild"]["number"]
        all_builds_loaded = (oldest_loaded_build_number == first_build_number)
        if all_builds_loaded:
            return data
        response = self.poll(tree='allBuilds[number,url]')
        data['builds'] = response['allBuilds']
        return data

    def _get_config_element_tree(self):
        """
        The ElementTree objects creation is unnecessary, it can be
        a singleton per job
        """
        if self._config is None:
            self.load_config()

        if self._element_tree is None:
            self._element_tree = ET.fromstring(self._config)
        return self._element_tree

    def get_build_triggerurl(self, files, build_params=None):
        if (files and build_params) or (not self.has_params()):
            # If job has file parameters and non-file parameters - it must be
            # triggered using "/build", not by "/buildWithParameters"
            # "/buildWithParameters" will ignore non-file parameters
            return "%s/build" % self.baseurl
        return "%s/buildWithParameters" % self.baseurl

    @staticmethod
    def _mk_json_from_build_parameters(build_params, file_params=None):
        """
        Build parameters must be submitted in a particular format
        Key-Value pairs would be far too simple, no no!
        Watch and read on and behold!
        """
        assert isinstance(
            build_params, dict), 'Build parameters must be a dict'

        build_p = [{'name': k, 'value': str(v)}
                   for k, v in sorted(build_params.items())]
        out = {'parameter': build_p}
        if file_params:
            file_p = [{'name': k, 'file': k}
                      for k in file_params.keys()]
            out['parameter'].extend(file_p)

        if len(out['parameter']) == 1:
            out['parameter'] = out['parameter'][0]

        return out

    @staticmethod
    def mk_json_from_build_parameters(build_params, file_params=None):
        json_structure = Job._mk_json_from_build_parameters(
            build_params,
            file_params
        )
        json_structure['statusCode'] = "303"
        json_structure['redirectTo'] = "."
        return json.dumps(json_structure)

    def invoke(self, securitytoken=None, block=False,
               build_params=None, cause=None, files=None, delay=5):
        assert isinstance(block, bool)
        if build_params and (not self.has_params()):
            raise BadParams("This job does not support parameters")

        params = {}  # Via Get string

        if securitytoken:
            params['token'] = securitytoken

        # Either copy the params dict or make a new one.
        build_params = build_params and dict(
            build_params.items()) or {}  # Via POSTed JSON

        url = self.get_build_triggerurl(files, build_params)
        if cause:
            build_params['cause'] = cause

        # Build require params as form fields
        # and as Json.
        data = {
            'json': self.mk_json_from_build_parameters(
                build_params,
                files)
        }
        data.update(build_params)

        response = self.jenkins.requester.post_and_confirm_status(
            url,
            data=data,
            params=params,
            files=files,
            valid=[200, 201, 303],
            allow_redirects=False
        )

        redirect_url = response.headers['location']

        if not redirect_url.startswith("%s/queue/item" % self.jenkins.baseurl):
            if files:
                raise ValueError('Builds with file parameters are not '
                                 'supported by this jenkinsapi version. '
                                 'Please use previous version.')
            else:
                raise ValueError("Not a Queue URL: %s" % redirect_url)

        qi = QueueItem(redirect_url, self.jenkins)
        if block:
            qi.block_until_complete(delay=delay)
        return qi

    def _buildid_for_type(self, buildtype):
        """Gets a buildid for a given type of build"""
        KNOWNBUILDTYPES = [
            "lastStableBuild",
            "lastSuccessfulBuild",
            "lastBuild",
            "lastCompletedBuild",
            "firstBuild",
            "lastFailedBuild"]
        assert buildtype in KNOWNBUILDTYPES, ('Unknown build info type: %s'
                                              % buildtype)

        data = self.poll(tree='%s[number]' % buildtype)

        if not data.get(buildtype):
            raise NoBuildData(buildtype)
        return data[buildtype]["number"]

    def get_first_buildnumber(self):
        """
        Get the numerical ID of the first build.
        """
        return self._buildid_for_type("firstBuild")

    def get_last_stable_buildnumber(self):
        """
        Get the numerical ID of the last stable build.
        """
        return self._buildid_for_type("lastStableBuild")

    def get_last_good_buildnumber(self):
        """
        Get the numerical ID of the last good build.
        """
        return self._buildid_for_type("lastSuccessfulBuild")

    def get_last_failed_buildnumber(self):
        """
        Get the numerical ID of the last good build.
        """
        return self._buildid_for_type(buildtype="lastFailedBuild")

    def get_last_buildnumber(self):
        """
        Get the numerical ID of the last build.
        """
        return self._buildid_for_type("lastBuild")

    def get_last_completed_buildnumber(self):
        """
        Get the numerical ID of the last complete build.
        """
        return self._buildid_for_type("lastCompletedBuild")

    def get_build_dict(self):
        builds = self.poll(tree='builds[number,url]')
        if not builds:
            raise NoBuildData(repr(self))
        builds = self._add_missing_builds(builds)
        builds = builds['builds']
        last_build = self.poll(tree='lastBuild[number,url]')['lastBuild']
        if builds and last_build and \
                builds[0]['number'] != last_build['number']:
            builds = [last_build] + builds
        # FIXME SO how is this supposed to work if build is false-y?
        # I don't think that builds *can* be false here, so I don't
        # understand the test above.
        return dict((build["number"], build["url"]) for build in builds)

    def get_revision_dict(self):
        """
        Get dictionary of all revisions with a list of buildnumbers (int)
        that used that particular revision
        """
        revs = defaultdict(list)
        if 'builds' not in self._data:
            raise NoBuildData(repr(self))
        for buildnumber in self.get_build_ids():
            revs[self.get_build(buildnumber)
                 .get_revision()].append(buildnumber)
        return revs

    def get_build_ids(self):
        """
        Return a sorted list of all good builds as ints.
        """
        return reversed(sorted(self.get_build_dict().keys()))

    def get_next_build_number(self):
        """
        Return the next build number that Jenkins will assign.
        """
        return self._data.get('nextBuildNumber', 0)

    def get_last_stable_build(self):
        """
        Get the last stable build
        """
        bn = self.get_last_stable_buildnumber()
        return self.get_build(bn)

    def get_last_good_build(self):
        """
        Get the last good build
        """
        bn = self.get_last_good_buildnumber()
        return self.get_build(bn)

    def get_last_build(self):
        """
        Get the last build
        """
        bn = self.get_last_buildnumber()
        return self.get_build(bn)

    def get_first_build(self):
        bn = self.get_first_buildnumber()
        return self.get_build(bn)

    def get_last_build_or_none(self):
        """
        Get the last build or None if there is no builds
        """
        try:
            return self.get_last_build()
        except NoBuildData:
            return None

    def get_last_completed_build(self):
        """
        Get the last build regardless of status
        """
        bn = self.get_last_completed_buildnumber()
        return self.get_build(bn)

    def get_buildnumber_for_revision(self, revision, refresh=False):
        """

        :param revision: subversion revision to look for, int
        :param refresh: boolean, whether or not to refresh the
            revision -> buildnumber map
        :return: list of buildnumbers, [int]
        """
        if self.get_scm_type() == 'svn' and not isinstance(revision, int):
            revision = int(revision)
        if self._revmap is None or refresh:
            self._revmap = self.get_revision_dict()
        try:
            return self._revmap[revision]
        except KeyError:
            raise NotFound("Couldn't find a build with that revision")

    def get_build(self, buildnumber):
        assert isinstance(buildnumber, int)
        url = self.get_build_dict()[buildnumber]
        return Build(url, buildnumber, job=self)

    def get_build_metadata(self, buildnumber):
        """
        Get the build metadata for a given build number. For large builds with
        tons of tests, this method is faster than get_build by returning less
        data.
        """
        assert isinstance(buildnumber, int)
        url = self.get_build_dict()[buildnumber]
        return Build(url, buildnumber, job=self, depth=0)

    def __getitem__(self, buildnumber):
        return self.get_build(buildnumber)

    def __len__(self):
        return len(self.get_build_dict())

    def is_queued_or_running(self):
        return self.is_queued() or self.is_running()

    def is_queued(self):
        data = self.poll(tree='inQueue')
        return data.get('inQueue', False)

    def get_queue_item(self):
        """
        Return a QueueItem if this object is in a queue, otherwise raise
        an exception
        """
        if not self.is_queued():
            raise UnknownQueueItem()
        return QueueItem(self.jenkins, **self._data['queueItem'])

    def is_running(self):
        # self.poll()
        try:
            build = self.get_last_build_or_none()
            if build is not None:
                return build.is_running()
        except NoBuildData:
            log.info(
                "No build info available for %s, assuming not running.",
                str(self))
        return False

    def get_config(self):
        '''Returns the config.xml from the job'''
        response = self.jenkins.requester.get_and_confirm_status(
            "%(baseurl)s/config.xml" % self.__dict__)
        return response.text

    def load_config(self):
        self._config = self.get_config()

    def get_scm_type(self):
        element_tree = self._get_config_element_tree()
        scm_class = element_tree.find('scm').get('class')
        scm = self._scm_map.get(scm_class)
        if not scm:
            raise NotSupportSCM(
                'SCM class "%s" not supported by API for job "%s"'
                % (scm_class, self.name))
        if scm == 'NullSCM':
            raise NotConfiguredSCM(
                'SCM is not configured for job "%s"' % self.name)
        return scm

    def get_scm_url(self):
        """
        Get list of project SCM urls
        For some SCM's jenkins allow to configure and use number of SCM url's
        : return: list of SCM urls
        """
        element_tree = self._get_config_element_tree()
        scm = self.get_scm_type()
        scm_url_list = [scm_url.text for scm_url in self._scmurlmap[
            scm](element_tree)]
        return scm_url_list

    def get_scm_branch(self):
        """
        Get list of SCM branches
        : return: list of SCM branches
        """
        element_tree = self._get_config_element_tree()
        scm = self.get_scm_type()
        return [scm_branch.text
                for scm_branch in self._scmbranchmap[scm](element_tree)]

    def modify_scm_branch(self, new_branch, old_branch=None):
        """
        Modify SCM ("Source Code Management") branch name for configured job.
        :param new_branch : new repository branch name to set.
            If job has multiple branches configured and "old_branch"
            not provided - method will allways modify first url.
        :param old_branch (optional): exact value of branch name
            to be replaced.
            For some SCM's jenkins allow set multiple branches per job
            this parameter intended to indicate which branch need to be
            modified
        """
        element_tree = self._get_config_element_tree()
        scm = self.get_scm_type()
        scm_branch_list = self._scmbranchmap[scm](element_tree)
        if scm_branch_list and not old_branch:
            scm_branch_list[0].text = new_branch
            self.update_config(ET.tostring(element_tree))
        else:
            for scm_branch in scm_branch_list:
                if scm_branch.text == old_branch:
                    scm_branch.text = new_branch
                    self.update_config(ET.tostring(element_tree))

    def modify_scm_url(self, new_source_url, old_source_url=None):
        """
        Modify SCM ("Source Code Management") url for configured job.
        :param new_source_url : new repository url to set.
            If job has multiple repositories configured and "old_source_url"
            not provided - method will allways modify first url.
        :param old_source_url (optional): for some SCM's jenkins allows
            settting multiple repositories per job
            this parameter intended to indicate which repository need
            to be modified
        """
        element_tree = self._get_config_element_tree()
        scm = self.get_scm_type()
        scm_url_list = self._scmurlmap[scm](element_tree)
        if scm_url_list and not old_source_url:
            scm_url_list[0].text = new_source_url
            self.update_config(ET.tostring(element_tree))
        else:
            for scm_url in scm_url_list:
                if scm_url.text == old_source_url:
                    scm_url.text = new_source_url
                    self.update_config(ET.tostring(element_tree))

    def get_config_xml_url(self):
        return '%s/config.xml' % self.baseurl

    def update_config(self, config, full_response=False):
        """
        Update the config.xml to the job
        Also refresh the ElementTree object since the config has changed
        :param full_response (optional): if True, it will return the full
            response object instead of just the response text.
            Useful for debugging and validation workflows.
        """
        url = self.get_config_xml_url()
        try:
            if isinstance(
                    config, unicode):  # pylint: disable=undefined-variable
                config = str(config)
        except NameError:
            # Python3 already a str
            pass

        response = self.jenkins.requester.post_url(url, params={}, data=config)
        self._element_tree = ET.fromstring(config)

        if full_response:
            return response

        return response.text

    def get_downstream_jobs(self):
        """
        Get all the possible downstream jobs
        :return List of Job
        """
        downstream_jobs = []
        try:
            for j in self._data['downstreamProjects']:
                downstream_jobs.append(
                    self.get_jenkins_obj()[j['name']])
        except KeyError:
            return []
        return downstream_jobs

    def get_downstream_job_names(self):
        """
        Get all the possible downstream job names
        :return List of String
        """
        downstream_jobs = []
        try:
            for j in self._data['downstreamProjects']:
                downstream_jobs.append(j['name'])
        except KeyError:
            return []
        return downstream_jobs

    def get_upstream_job_names(self):
        """
        Get all the possible upstream job names
        :return List of String
        """
        upstream_jobs = []
        try:
            for j in self._data['upstreamProjects']:
                upstream_jobs.append(j['name'])
        except KeyError:
            return []
        return upstream_jobs

    def get_upstream_jobs(self):
        """
        Get all the possible upstream jobs
        :return List of Job
        """
        upstream_jobs = []
        try:
            for j in self._data['upstreamProjects']:
                upstream_jobs.append(self.get_jenkins_obj().get_job(j['name']))
        except KeyError:
            return []
        return upstream_jobs

    def is_enabled(self):
        data = self.poll(tree='color')
        return data.get('color', None) != 'disabled'

    def disable(self):
        '''Disable job'''
        url = "%s/disable" % self.baseurl
        return self.get_jenkins_obj().requester.post_url(url, data='')

    def enable(self):
        '''Enable job'''
        url = "%s/enable" % self.baseurl
        return self.get_jenkins_obj().requester.post_url(url, data='')

    def delete_from_queue(self):
        """
        Delete a job from the queue only if it's enqueued
        :raise NotInQueue if the job is not in the queue
        """
        if not self.is_queued():
            raise NotInQueue()
        queue_id = self._data['queueItem']['id']
        url = urlparse.urljoin(self.get_jenkins_obj().get_queue().baseurl,
                               'cancelItem?id=%s' % queue_id)
        self.get_jenkins_obj().requester.post_and_confirm_status(url, data='')
        return True

    def get_params(self):
        """
        Get the parameters for this job. Format varies by parameter type. Here
        is an example string parameter:
            {
                'type': 'StringParameterDefinition',
                'description': 'Parameter description',
                'defaultParameterValue': {'value': 'default value'},
                'name': 'FOO_BAR'
            }
        """
        places = ['actions', 'property']
        found_definitions = False

        for place in places:
            if found_definitions:
                return
            actions = (x for x in self._data[place] if x is not None)
            for action in actions:
                try:
                    for param in action['parameterDefinitions']:
                        found_definitions = True
                        yield param
                except KeyError:
                    continue

    def get_params_list(self):
        """
        Gets the list of parameter names for this job.
        """
        return [param['name'] for param in self.get_params()]

    def has_params(self):
        """
        If job has parameters, returns True, else False
        """
        return any("parameterDefinitions" in a for a in (
            self._data["actions"] or self._data["property"]) if a)

    def has_queued_build(self, build_params):
        """Returns True if a build with build_params is currently queued."""
        queue = self.jenkins.get_queue()
        queued_builds = queue.get_queue_items_for_job(self.name)
        for build in queued_builds:
            if build.get_parameters() == build_params:
                return True
        return False