Ken Rumer has proposed merging lp:~kenrumer/python-jenkins/build_info into lp:python-jenkins.
Requested reviews: Python Jenkins Developers (python-jenkins-developers) For more details, see: https://code.launchpad.net/~kenrumer/python-jenkins/build_info/+merge/94631 Gets build information from Jenkins. Can use to get the status of a build. Loop until the build is complete, return the artifact URL(s). Example usage: def start( **kwargs ): """ Start a build on the continuous integration server Arguments: service_name: Name of the service/application project_name: Name of the project/software branch vcs_server_type: type of version control server "svn" or "git" vcs_host: host name of the vcs server vcs_port: port number for the vcs server vcs_username: username to authenticate vcs server vcs_password: password to authenticate vcs server ci_host: host name of the continuous integreation server ci_port: port number of the continuous integreation server ci_username: username to authenticate continuous integreation server ci_password: password to authenticate continuous integreation server Outputs: version: the id defined for this build job. probably the timestamp of the build build_number: the build revision timestamp: the datestamp of the build duration: length of time the job took result: result of the build job artifacts: filename: the filename the continuous integration server built url: retrieval location on the continuous integration server """ import jenkins import time j = jenkins.Jenkins('http://'+kwargs['ci_host']+':'+kwargs['ci_port'], kwargs['ci_username'], kwargs['ci_password']) next_build_number = j.get_job_info('build_'+kwargs['vcs_server_type'])['nextBuildNumber'] params = { 'service_name': kwargs['service_name'], 'project_name': kwargs['project_name'], 'vcs_server_type': kwargs['vcs_server_type'], 'vcs_host': kwargs['vcs_host'], 'vcs_port': kwargs['vcs_port'], 'vcs_username': kwargs['vcs_username'], 'vcs_password': kwargs['vcs_password'] } output = j.build_job('build_'+kwargs['vcs_server_type'], params) last_completed_build_number = 0 last_successful_build_number = 0 while ( 1 ): if (j.get_job_info('build_'+kwargs['vcs_server_type'])['lastCompletedBuild'] == None): time.sleep(2) continue last_completed_build_number = j.get_job_info('build_'+kwargs['vcs_server_type'])['lastCompletedBuild']['number'] if (last_completed_build_number < next_build_number): time.sleep(2) continue build_info = j.get_build_info('build_'+kwargs['vcs_server_type'], next_build_number) break artifacts = [] for artifact in build_info['artifacts']: artifacts.append({ 'filename': artifact['fileName'], 'url': build_info['url'].encode('utf8')+'artifact/'+artifact['relativePath'].encode("utf8") }) return {'version': build_info['id'].encode('utf8'), 'build_number': build_info['number'], 'timestamp': build_info['timestamp'], 'duration': build_info['duration'], 'result': build_info['result'].encode("utf8"), 'artifacts': artifacts} -- https://code.launchpad.net/~kenrumer/python-jenkins/build_info/+merge/94631 Your team Python Jenkins Developers is requested to review the proposed merge of lp:~kenrumer/python-jenkins/build_info into lp:python-jenkins.
=== modified file 'jenkins/__init__.py' --- jenkins/__init__.py 2011-09-04 00:24:56 +0000 +++ jenkins/__init__.py 2012-02-24 23:20:22 +0000 @@ -1,429 +1,442 @@ -#!/usr/bin/env python -# Software License Agreement (BSD License) -# -# Copyright (c) 2010, Willow Garage, Inc. -# All rights reserved. -# -# Redistribution and use in source and binary forms, with or without -# modification, are permitted provided that the following conditions -# are met: -# -# * Redistributions of source code must retain the above copyright -# notice, this list of conditions and the following disclaimer. -# * Redistributions in binary form must reproduce the above -# copyright notice, this list of conditions and the following -# disclaimer in the documentation and/or other materials provided -# with the distribution. -# * Neither the name of Willow Garage, Inc. nor the names of its -# contributors may be used to endorse or promote products derived -# from this software without specific prior written permission. -# -# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS -# 'AS IS' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT -# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS -# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE -# COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, -# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, -# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; -# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER -# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT -# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN -# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE -# POSSIBILITY OF SUCH DAMAGE. -# -# Authors: -# Ken Conley <[email protected]> -# James Page <[email protected]> -# Tully Foote <[email protected]> -# Matthew Gertner <[email protected]> - -''' -Python API for Jenkins - -Examples:: - - j = jenkins.Jenkins('http://your_url_here', 'username', 'password') - j.get_jobs() - j.create_job('empty', jenkins.EMPTY_CONFIG_XML) - j.disable_job('empty') - j.copy_job('empty', 'empty_copy') - j.enable_job('empty_copy') - j.reconfig_job('empty_copy', jenkins.RECONFIG_XML) - - j.delete_job('empty') - j.delete_job('empty_copy') - - # build a parameterized job - j.build_job('api-test', {'param1': 'test value 1', 'param2': 'test value 2'}) -''' - -import sys -import urllib2 -import urllib -import base64 -import traceback -import json -import httplib - -INFO = 'api/json' -JOB_INFO = 'job/%(name)s/api/json?depth=0' -Q_INFO = 'queue/api/json?depth=0' -CREATE_JOB = 'createItem?name=%(name)s' #also post config.xml -CONFIG_JOB = 'job/%(name)s/config.xml' -DELETE_JOB = 'job/%(name)s/doDelete' -ENABLE_JOB = 'job/%(name)s/enable' -DISABLE_JOB = 'job/%(name)s/disable' -COPY_JOB = 'createItem?name=%(to_name)s&mode=copy&from=%(from_name)s' -BUILD_JOB = 'job/%(name)s/build' -BUILD_WITH_PARAMS_JOB = 'job/%(name)s/buildWithParameters' - - -CREATE_NODE = 'computer/doCreateItem?%s' -DELETE_NODE = 'computer/%(name)s/doDelete' -NODE_INFO = 'computer/%(name)s/api/json?depth=0' -NODE_TYPE = 'hudson.slaves.DumbSlave$DescriptorImpl' - - -#for testing only -EMPTY_CONFIG_XML = '''<?xml version='1.0' encoding='UTF-8'?> -<project> - <keepDependencies>false</keepDependencies> - <properties/> - <scm class='jenkins.scm.NullSCM'/> - <canRoam>true</canRoam> - <disabled>false</disabled> - <blockBuildWhenUpstreamBuilding>false</blockBuildWhenUpstreamBuilding> - <triggers class='vector'/> - <concurrentBuild>false</concurrentBuild> - <builders/> - <publishers/> - <buildWrappers/> -</project>''' - -#for testing only -RECONFIG_XML = '''<?xml version='1.0' encoding='UTF-8'?> -<project> - <keepDependencies>false</keepDependencies> - <properties/> - <scm class='jenkins.scm.NullSCM'/> - <canRoam>true</canRoam> - <disabled>false</disabled> - <blockBuildWhenUpstreamBuilding>false</blockBuildWhenUpstreamBuilding> - <triggers class='vector'/> - <concurrentBuild>false</concurrentBuild> -<builders> - <jenkins.tasks.Shell> - <command>export FOO=bar</command> - </jenkins.tasks.Shell> - </builders> - <publishers/> - <buildWrappers/> -</project>''' - -class JenkinsException(Exception): - ''' - General exception type for jenkins-API-related failures. - ''' - pass - -def auth_headers(username, password): - ''' - Simple implementation of HTTP Basic Authentication. Returns the 'Authentication' header value. - ''' - return 'Basic ' + base64.encodestring('%s:%s' % (username, password))[:-1] - -class Jenkins(object): - - def __init__(self, url, username=None, password=None): - ''' - Create handle to Jenkins instance. - - :param url: URL of Jenkins server, ``str`` - ''' - if url[-1] == '/': - self.server = url - else: - self.server = url + '/' - if username is not None and password is not None: - self.auth = auth_headers(username, password) - else: - self.auth = None - - def get_job_info(self, name): - ''' - Get job information dictionary. - - :param name: Job name, ``str`` - :returns: dictionary of job information - ''' - try: - response = self.jenkins_open(urllib2.Request(self.server + JOB_INFO%locals())) - if response: - return json.loads(response) - else: - raise JenkinsException('job[%s] does not exist'%name) - except urllib2.HTTPError: - raise JenkinsException('job[%s] does not exist'%name) - except ValueError: - raise JenkinsException("Could not parse JSON info for job[%s]"%name) - - def debug_job_info(self, job_name): - ''' - Print out job info in more readable format - ''' - for k, v in self.get_job_info(job_name).iteritems(): - print k, v - - def jenkins_open(self, req): - ''' - Utility routine for opening an HTTP request to a Jenkins server. This should only be used - to extends the :class:`Jenkins` API. - ''' - try: - if self.auth: - req.add_header('Authorization', self.auth) - return urllib2.urlopen(req).read() - except urllib2.HTTPError, e: - # Jenkins's funky authentication means its nigh impossible to distinguish errors. - if e.code in [401, 403, 500]: - raise JenkinsException('Error in request. Possibly authentication failed [%s]'%(e.code)) - # right now I'm getting 302 infinites on a successful delete - - def get_queue_info(self): - ''' - :returns: list of job dictionaries, ``[dict]`` - - Example:: - >>> queue_info = j.get_queue_info() - >>> print(queue_info[0]) - {u'task': {u'url': u'http://your_url/job/my_job/', u'color': u'aborted_anime', u'name': u'my_job'}, u'stuck': False, u'actions': [{u'causes': [{u'shortDescription': u'Started by timer'}]}], u'buildable': False, u'params': u'', u'buildableStartMilliseconds': 1315087293316, u'why': u'Build #2,532 is already in progress (ETA:10 min)', u'blocked': True} - ''' - return json.loads(self.jenkins_open(urllib2.Request(self.server + Q_INFO)))['items'] - - def get_info(self): - """ - Get information on this Master. This information - includes job list and view information. - - :returns: dictionary of information about Master, ``dict`` - - Example:: - - >>> info = j.get_info() - >>> jobs = info['jobs'] - >>> print(jobs[0]) - {u'url': u'http://your_url_here/job/my_job/', u'color': u'blue', u'name': u'my_job'} - - """ - try: - return json.loads(self.jenkins_open(urllib2.Request(self.server + INFO))) - except urllib2.HTTPError: - raise JenkinsException("Error communicating with server[%s]"%self.server) - except httplib.BadStatusLine: - raise JenkinsException("Error communicating with server[%s]"%self.server) - except ValueError: - raise JenkinsException("Could not parse JSON info for server[%s]"%self.server) - - def get_jobs(self): - """ - Get list of jobs running. Each job is a dictionary with - 'name', 'url', and 'color' keys. - - :returns: list of jobs, ``[ { str: str} ]`` - """ - return self.get_info()['jobs'] - - def copy_job(self, from_name, to_name): - ''' - Copy a Jenkins job - - :param from_name: Name of Jenkins job to copy from, ``str`` - :param to_name: Name of Jenkins job to copy to, ``str`` - ''' - self.get_job_info(from_name) - self.jenkins_open(urllib2.Request(self.server + COPY_JOB%locals(), '')) - if not self.job_exists(to_name): - raise JenkinsException('create[%s] failed'%(to_name)) - - def delete_job(self, name): - ''' - Delete Jenkins job permanently. - - :param name: Name of Jenkins job, ``str`` - ''' - self.get_job_info(name) - self.jenkins_open(urllib2.Request(self.server + DELETE_JOB%locals(), '')) - if self.job_exists(name): - raise JenkinsException('delete[%s] failed'%(name)) - - def enable_job(self, name): - ''' - Enable Jenkins job. - - :param name: Name of Jenkins job, ``str`` - ''' - self.get_job_info(name) - self.jenkins_open(urllib2.Request(self.server + ENABLE_JOB%locals(), '')) - - def disable_job(self, name): - ''' - Disable Jenkins job. To re-enable, call :meth:`Jenkins.enable_job`. - - :param name: Name of Jenkins job, ``str`` - ''' - self.get_job_info(name) - self.jenkins_open(urllib2.Request(self.server + DISABLE_JOB%locals(), '')) - - def job_exists(self, name): - ''' - :param name: Name of Jenkins job, ``str`` - :returns: ``True`` if Jenkins job exists - ''' - try: - self.get_job_info(name) - return True - except JenkinsException: - return False - - def create_job(self, name, config_xml): - ''' - Create a new Jenkins job - - :param name: Name of Jenkins job, ``str`` - :param config_xml: config file text, ``str`` - ''' - if self.job_exists(name): - raise JenkinsException('job[%s] already exists'%(name)) - - headers = {'Content-Type': 'text/xml'} - self.jenkins_open(urllib2.Request(self.server + CREATE_JOB%locals(), config_xml, headers)) - if not self.job_exists(name): - raise JenkinsException('create[%s] failed'%(name)) - - def get_job_config(self, name): - ''' - Get configuration of existing Jenkins job. - - :param name: Name of Jenkins job, ``str`` - :returns: job configuration (XML format) - ''' - get_config_url = self.server + CONFIG_JOB%locals() - return self.jenkins_open(urllib2.Request(get_config_url)) - - def reconfig_job(self, name, config_xml): - ''' - Change configuration of existing Jenkins job. To create a new job, see :meth:`Jenkins.create_job`. - - :param name: Name of Jenkins job, ``str`` - :param config_xml: New XML configuration, ``str`` - ''' - self.get_job_info(name) - headers = {'Content-Type': 'text/xml'} - reconfig_url = self.server + CONFIG_JOB%locals() - self.jenkins_open(urllib2.Request(reconfig_url, config_xml, headers)) - - def build_job_url(self, name, parameters=None, token=None): - ''' - Get URL to trigger build job. Authenticated setups may require configuring a token on the server side. - - :param parameters: parameters for job, or None., ``dict`` - :param token: (optional) token for building job, ``str`` - :returns: URL for building job - ''' - if parameters: - if token: - parameters['token'] = token - return self.server + BUILD_WITH_PARAMS_JOB%locals() + '?' + urllib.urlencode(parameters) - elif token: - return self.server + BUILD_JOB%locals() + '?' + urllib.urlencode({'token': token}) - else: - return self.server + BUILD_JOB%locals() - - def build_job(self, name, parameters=None, token=None): - ''' - Trigger build job. - - :param parameters: parameters for job, or ``None``, ``dict`` - ''' - if not self.job_exists(name): - raise JenkinsException('no such job[%s]'%(name)) - return self.jenkins_open(urllib2.Request(self.build_job_url(name, parameters, token))) - - def get_node_info(self, name): - ''' - Get node information dictionary - - :param name: Node name, ``str`` - :returns: Dictionary of node info, ``dict`` - ''' - try: - response = self.jenkins_open(urllib2.Request(self.server + NODE_INFO%locals())) - if response: - return json.loads(response) - else: - raise JenkinsException('node[%s] does not exist'%name) - except urllib2.HTTPError: - raise JenkinsException('node[%s] does not exist'%name) - except ValueError: - raise JenkinsException("Could not parse JSON info for node[%s]"%name) - - def node_exists(self, name): - ''' - :param name: Name of Jenkins node, ``str`` - :returns: ``True`` if Jenkins node exists - ''' - try: - self.get_node_info(name) - return True - except JenkinsException: - return False - - def delete_node(self, name): - ''' - Delete Jenkins node permanently. - - :param name: Name of Jenkins node, ``str`` - ''' - self.get_node_info(name) - self.jenkins_open(urllib2.Request(self.server + DELETE_NODE%locals(), '')) - if self.node_exists(name): - raise JenkinsException('delete[%s] failed'%(name)) - - - def create_node(self, name, numExecutors=2, nodeDescription=None, - remoteFS='/var/lib/jenkins', labels=None, exclusive=False): - ''' - :param name: name of node to create, ``str`` - :param numExecutors: number of executors for node, ``int`` - :param nodeDescription: Description of node, ``str`` - :param remoteFS: Remote filesystem location to use, ``str`` - :param labels: Labels to associate with node, ``str`` - :param exclusive: Use this node for tied jobs only, ``bool`` - ''' - if self.node_exists(name): - raise JenkinsException('node[%s] already exists'%(name)) - - mode = 'NORMAL' - if exclusive: - mode = 'EXCLUSIVE' - - params = { - 'name' : name, - 'type' : NODE_TYPE, - 'json' : json.dumps ({ - 'name' : name, - 'nodeDescription' : nodeDescription, - 'numExecutors' : numExecutors, - 'remoteFS' : remoteFS, - 'labelString' : labels, - 'mode' : mode, - 'type' : NODE_TYPE, - 'retentionStrategy' : { 'stapler-class' : 'hudson.slaves.RetentionStrategy$Always' }, - 'nodeProperties' : { 'stapler-class-bag' : 'true' }, - 'launcher' : { 'stapler-class' : 'hudson.slaves.JNLPLauncher' } - }) - } - - self.jenkins_open(urllib2.Request(self.server + CREATE_NODE%urllib.urlencode(params))) - if not self.node_exists(name): - raise JenkinsException('create[%s] failed'%(name)) +#!/usr/bin/env python +# Software License Agreement (BSD License) +# +# Copyright (c) 2010, Willow Garage, Inc. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions +# are met: +# +# * Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above +# copyright notice, this list of conditions and the following +# disclaimer in the documentation and/or other materials provided +# with the distribution. +# * Neither the name of Willow Garage, Inc. nor the names of its +# contributors may be used to endorse or promote products derived +# from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +# 'AS IS' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS +# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE +# COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, +# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, +# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; +# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER +# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT +# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN +# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# +# Authors: +# Ken Conley <[email protected]> +# James Page <[email protected]> +# Tully Foote <[email protected]> +# Matthew Gertner <[email protected]> + +''' +Python API for Jenkins + +Examples:: + + j = jenkins.Jenkins('http://your_url_here', 'username', 'password') + j.get_jobs() + j.create_job('empty', jenkins.EMPTY_CONFIG_XML) + j.disable_job('empty') + j.copy_job('empty', 'empty_copy') + j.enable_job('empty_copy') + j.reconfig_job('empty_copy', jenkins.RECONFIG_XML) + + j.delete_job('empty') + j.delete_job('empty_copy') + + # build a parameterized job + j.build_job('api-test', {'param1': 'test value 1', 'param2': 'test value 2'}) +''' + +import sys +import urllib2 +import urllib +import base64 +import traceback +import json +import httplib + +INFO = 'api/json' +JOB_INFO = 'job/%(name)s/api/json?depth=0' +Q_INFO = 'queue/api/json?depth=0' +CREATE_JOB = 'createItem?name=%(name)s' #also post config.xml +CONFIG_JOB = 'job/%(name)s/config.xml' +DELETE_JOB = 'job/%(name)s/doDelete' +ENABLE_JOB = 'job/%(name)s/enable' +DISABLE_JOB = 'job/%(name)s/disable' +COPY_JOB = 'createItem?name=%(to_name)s&mode=copy&from=%(from_name)s' +BUILD_JOB = 'job/%(name)s/build' +BUILD_WITH_PARAMS_JOB = 'job/%(name)s/buildWithParameters' +BUILD_INFO = 'job/%(name)s/%(number)d/api/json?depth=0' + + +CREATE_NODE = 'computer/doCreateItem?%s' +DELETE_NODE = 'computer/%(name)s/doDelete' +NODE_INFO = 'computer/%(name)s/api/json?depth=0' +NODE_TYPE = 'hudson.slaves.DumbSlave$DescriptorImpl' + + +#for testing only +EMPTY_CONFIG_XML = '''<?xml version='1.0' encoding='UTF-8'?> +<project> + <keepDependencies>false</keepDependencies> + <properties/> + <scm class='jenkins.scm.NullSCM'/> + <canRoam>true</canRoam> + <disabled>false</disabled> + <blockBuildWhenUpstreamBuilding>false</blockBuildWhenUpstreamBuilding> + <triggers class='vector'/> + <concurrentBuild>false</concurrentBuild> + <builders/> + <publishers/> + <buildWrappers/> +</project>''' + +#for testing only +RECONFIG_XML = '''<?xml version='1.0' encoding='UTF-8'?> +<project> + <keepDependencies>false</keepDependencies> + <properties/> + <scm class='jenkins.scm.NullSCM'/> + <canRoam>true</canRoam> + <disabled>false</disabled> + <blockBuildWhenUpstreamBuilding>false</blockBuildWhenUpstreamBuilding> + <triggers class='vector'/> + <concurrentBuild>false</concurrentBuild> +<builders> + <jenkins.tasks.Shell> + <command>export FOO=bar</command> + </jenkins.tasks.Shell> + </builders> + <publishers/> + <buildWrappers/> +</project>''' + +class JenkinsException(Exception): + ''' + General exception type for jenkins-API-related failures. + ''' + pass + +def auth_headers(username, password): + ''' + Simple implementation of HTTP Basic Authentication. Returns the 'Authentication' header value. + ''' + return 'Basic ' + base64.encodestring('%s:%s' % (username, password))[:-1] + +class Jenkins(object): + + def __init__(self, url, username=None, password=None): + ''' + Create handle to Jenkins instance. + + :param url: URL of Jenkins server, ``str`` + ''' + if url[-1] == '/': + self.server = url + else: + self.server = url + '/' + if username is not None and password is not None: + self.auth = auth_headers(username, password) + else: + self.auth = None + + def get_job_info(self, name): + ''' + Get job information dictionary. + + :param name: Job name, ``str`` + :returns: dictionary of job information + ''' + try: + response = self.jenkins_open(urllib2.Request(self.server + JOB_INFO%locals())) + if response: + return json.loads(response) + else: + raise JenkinsException('job[%s] does not exist'%name) + except urllib2.HTTPError: + raise JenkinsException('job[%s] does not exist'%name) + except ValueError: + raise JenkinsException("Could not parse JSON info for job[%s]"%name) + + def debug_job_info(self, job_name): + ''' + Print out job info in more readable format + ''' + for k, v in self.get_job_info(job_name).iteritems(): + print k, v + + def jenkins_open(self, req): + ''' + Utility routine for opening an HTTP request to a Jenkins server. This should only be used + to extends the :class:`Jenkins` API. + ''' + try: + if self.auth: + req.add_header('Authorization', self.auth) + return urllib2.urlopen(req).read() + except urllib2.HTTPError, e: + # Jenkins's funky authentication means its nigh impossible to distinguish errors. + if e.code in [401, 403, 500]: + raise JenkinsException('Error in request. Possibly authentication failed [%s]'%(e.code)) + # right now I'm getting 302 infinites on a successful delete + + def get_build_info(self, name, number): + try: + response = self.jenkins_open(urllib2.Request(self.server + BUILD_INFO%locals())) + if response: + return json.loads(response) + else: + raise JenkinsException('job[!s] number[!d] does not exist'.format(name, number)) + except urllib2.HTTPError: + raise JenkinsException('job[!s] number[!d] does not exist'.format(name, number)) + except ValueError: + raise JenkinsException("Could not parse JSON info for job[!s] number[!d]".format(name, number)) + + def get_queue_info(self): + ''' + :returns: list of job dictionaries, ``[dict]`` + + Example:: + >>> queue_info = j.get_queue_info() + >>> print(queue_info[0]) + {u'task': {u'url': u'http://your_url/job/my_job/', u'color': u'aborted_anime', u'name': u'my_job'}, u'stuck': False, u'actions': [{u'causes': [{u'shortDescription': u'Started by timer'}]}], u'buildable': False, u'params': u'', u'buildableStartMilliseconds': 1315087293316, u'why': u'Build #2,532 is already in progress (ETA:10 min)', u'blocked': True} + ''' + return json.loads(self.jenkins_open(urllib2.Request(self.server + Q_INFO)))['items'] + + def get_info(self): + """ + Get information on this Master. This information + includes job list and view information. + + :returns: dictionary of information about Master, ``dict`` + + Example:: + + >>> info = j.get_info() + >>> jobs = info['jobs'] + >>> print(jobs[0]) + {u'url': u'http://your_url_here/job/my_job/', u'color': u'blue', u'name': u'my_job'} + + """ + try: + return json.loads(self.jenkins_open(urllib2.Request(self.server + INFO))) + except urllib2.HTTPError: + raise JenkinsException("Error communicating with server[%s]"%self.server) + except httplib.BadStatusLine: + raise JenkinsException("Error communicating with server[%s]"%self.server) + except ValueError: + raise JenkinsException("Could not parse JSON info for server[%s]"%self.server) + + def get_jobs(self): + """ + Get list of jobs running. Each job is a dictionary with + 'name', 'url', and 'color' keys. + + :returns: list of jobs, ``[ { str: str} ]`` + """ + return self.get_info()['jobs'] + + def copy_job(self, from_name, to_name): + ''' + Copy a Jenkins job + + :param from_name: Name of Jenkins job to copy from, ``str`` + :param to_name: Name of Jenkins job to copy to, ``str`` + ''' + self.get_job_info(from_name) + self.jenkins_open(urllib2.Request(self.server + COPY_JOB%locals(), '')) + if not self.job_exists(to_name): + raise JenkinsException('create[%s] failed'%(to_name)) + + def delete_job(self, name): + ''' + Delete Jenkins job permanently. + + :param name: Name of Jenkins job, ``str`` + ''' + self.get_job_info(name) + self.jenkins_open(urllib2.Request(self.server + DELETE_JOB%locals(), '')) + if self.job_exists(name): + raise JenkinsException('delete[%s] failed'%(name)) + + def enable_job(self, name): + ''' + Enable Jenkins job. + + :param name: Name of Jenkins job, ``str`` + ''' + self.get_job_info(name) + self.jenkins_open(urllib2.Request(self.server + ENABLE_JOB%locals(), '')) + + def disable_job(self, name): + ''' + Disable Jenkins job. To re-enable, call :meth:`Jenkins.enable_job`. + + :param name: Name of Jenkins job, ``str`` + ''' + self.get_job_info(name) + self.jenkins_open(urllib2.Request(self.server + DISABLE_JOB%locals(), '')) + + def job_exists(self, name): + ''' + :param name: Name of Jenkins job, ``str`` + :returns: ``True`` if Jenkins job exists + ''' + try: + self.get_job_info(name) + return True + except JenkinsException: + return False + + def create_job(self, name, config_xml): + ''' + Create a new Jenkins job + + :param name: Name of Jenkins job, ``str`` + :param config_xml: config file text, ``str`` + ''' + if self.job_exists(name): + raise JenkinsException('job[%s] already exists'%(name)) + + headers = {'Content-Type': 'text/xml'} + self.jenkins_open(urllib2.Request(self.server + CREATE_JOB%locals(), config_xml, headers)) + if not self.job_exists(name): + raise JenkinsException('create[%s] failed'%(name)) + + def get_job_config(self, name): + ''' + Get configuration of existing Jenkins job. + + :param name: Name of Jenkins job, ``str`` + :returns: job configuration (XML format) + ''' + get_config_url = self.server + CONFIG_JOB%locals() + return self.jenkins_open(urllib2.Request(get_config_url)) + + def reconfig_job(self, name, config_xml): + ''' + Change configuration of existing Jenkins job. To create a new job, see :meth:`Jenkins.create_job`. + + :param name: Name of Jenkins job, ``str`` + :param config_xml: New XML configuration, ``str`` + ''' + self.get_job_info(name) + headers = {'Content-Type': 'text/xml'} + reconfig_url = self.server + CONFIG_JOB%locals() + self.jenkins_open(urllib2.Request(reconfig_url, config_xml, headers)) + + def build_job_url(self, name, parameters=None, token=None): + ''' + Get URL to trigger build job. Authenticated setups may require configuring a token on the server side. + + :param parameters: parameters for job, or None., ``dict`` + :param token: (optional) token for building job, ``str`` + :returns: URL for building job + ''' + if parameters: + if token: + parameters['token'] = token + return self.server + BUILD_WITH_PARAMS_JOB%locals() + '?' + urllib.urlencode(parameters) + elif token: + return self.server + BUILD_JOB%locals() + '?' + urllib.urlencode({'token': token}) + else: + return self.server + BUILD_JOB%locals() + + def build_job(self, name, parameters=None, token=None): + ''' + Trigger build job. + + :param parameters: parameters for job, or ``None``, ``dict`` + ''' + if not self.job_exists(name): + raise JenkinsException('no such job[%s]'%(name)) + return self.jenkins_open(urllib2.Request(self.build_job_url(name, parameters, token))) + + def get_node_info(self, name): + ''' + Get node information dictionary + + :param name: Node name, ``str`` + :returns: Dictionary of node info, ``dict`` + ''' + try: + response = self.jenkins_open(urllib2.Request(self.server + NODE_INFO%locals())) + if response: + return json.loads(response) + else: + raise JenkinsException('node[%s] does not exist'%name) + except urllib2.HTTPError: + raise JenkinsException('node[%s] does not exist'%name) + except ValueError: + raise JenkinsException("Could not parse JSON info for node[%s]"%name) + + def node_exists(self, name): + ''' + :param name: Name of Jenkins node, ``str`` + :returns: ``True`` if Jenkins node exists + ''' + try: + self.get_node_info(name) + return True + except JenkinsException: + return False + + def delete_node(self, name): + ''' + Delete Jenkins node permanently. + + :param name: Name of Jenkins node, ``str`` + ''' + self.get_node_info(name) + self.jenkins_open(urllib2.Request(self.server + DELETE_NODE%locals(), '')) + if self.node_exists(name): + raise JenkinsException('delete[%s] failed'%(name)) + + + def create_node(self, name, numExecutors=2, nodeDescription=None, + remoteFS='/var/lib/jenkins', labels=None, exclusive=False): + ''' + :param name: name of node to create, ``str`` + :param numExecutors: number of executors for node, ``int`` + :param nodeDescription: Description of node, ``str`` + :param remoteFS: Remote filesystem location to use, ``str`` + :param labels: Labels to associate with node, ``str`` + :param exclusive: Use this node for tied jobs only, ``bool`` + ''' + if self.node_exists(name): + raise JenkinsException('node[%s] already exists'%(name)) + + mode = 'NORMAL' + if exclusive: + mode = 'EXCLUSIVE' + + params = { + 'name' : name, + 'type' : NODE_TYPE, + 'json' : json.dumps ({ + 'name' : name, + 'nodeDescription' : nodeDescription, + 'numExecutors' : numExecutors, + 'remoteFS' : remoteFS, + 'labelString' : labels, + 'mode' : mode, + 'type' : NODE_TYPE, + 'retentionStrategy' : { 'stapler-class' : 'hudson.slaves.RetentionStrategy$Always' }, + 'nodeProperties' : { 'stapler-class-bag' : 'true' }, + 'launcher' : { 'stapler-class' : 'hudson.slaves.JNLPLauncher' } + }) + } + + self.jenkins_open(urllib2.Request(self.server + CREATE_NODE%urllib.urlencode(params))) + if not self.node_exists(name): + raise JenkinsException('create[%s] failed'%(name))
-- Mailing list: https://launchpad.net/~python-jenkins-developers Post to : [email protected] Unsubscribe : https://launchpad.net/~python-jenkins-developers More help : https://help.launchpad.net/ListHelp
