Add support for handling promotions

Promotions can be attached to jobs via the REST API described at:

https://issues.jenkins-ci.org/browse/JENKINS-8963

Co-Authored-By: Guido Günther <agx@sigxcpu.org>
Co-Authored-By: Joao Vale <jpvale@gmail.com>

Change-Id: I756a35f53f96ba6e46e71f36ea99f62d32b604d2
This commit is contained in:
Guido Günther 2016-01-12 18:42:17 +01:00
parent dccae5fd10
commit 0c503b724c
3 changed files with 487 additions and 0 deletions

View File

@ -165,3 +165,38 @@ Jenkins job.
next_bn = server.get_job_info('job_name')['nextBuildNumber']
server.set_next_build_number('job_name', next_bn + 50)
Example 9: Working with Build Promotions
----------------------------------------
Requires the `Promoted Builds Plugin
<https://wiki.jenkins-ci.org/display/JENKINS/Promoted+Builds+Plugin>`_
for Jenkins.
This is an example showing how to create, configure and delete a
promotion process for an existing job.
The job in this example is named *prom_job* and it needs to have this
config xml snippet before creating the promotion:
::
<properties>
<hudson.plugins.promoted__builds.JobPropertyImpl>
<activeProcessNames>
<string>prom_name</string>
</activeProcessNames>
</hudson.plugins.promoted__builds.JobPropertyImpl>
</properties>
where *prom_name* is the name of the promotion that will get added to the job.
::
server.create_promotion('prom_name', 'prom_job', jenkins.EMPTY_PROMO_CONFIG_XML)
server.promotion_exists('prom_name', 'prom_job')
print server.get_promotions('prom_job')
server.reconfig_promotion('prom_name', 'prom_job', jenkins.PROMO_RECONFIG_XML)
print server.get_promotion_config('prom_name', 'prom_job')
server.delete_promotion('prom_name', 'prom_job')

View File

@ -110,6 +110,11 @@ CREATE_VIEW = 'createView?name=%(name)s'
CONFIG_VIEW = 'view/%(name)s/config.xml'
DELETE_VIEW = 'view/%(name)s/doDelete'
SCRIPT_TEXT = 'scriptText'
PROMOTION_NAME = '%(folder_url)sjob/%(short_name)s/promotion/process/%(name)s/api/json?tree=name'
PROMOTION_INFO = '%(folder_url)sjob/%(short_name)s/promotion/api/json?depth=%(depth)s'
DELETE_PROMOTION = '%(folder_url)sjob/%(short_name)s/promotion/process/%(name)s/doDelete'
CREATE_PROMOTION = '%(folder_url)sjob/%(short_name)s/promotion/createProcess?name=%(name)s'
CONFIG_PROMOTION = '%(folder_url)sjob/%(short_name)s/promotion/process/%(name)s/config.xml'
QUIET_DOWN = 'quietDown'
# for testing only
@ -170,6 +175,34 @@ EMPTY_VIEW_CONFIG_XML = '''<?xml version="1.0" encoding="UTF-8"?>
</columns>
</hudson.model.ListView>'''
# for testing only
EMPTY_PROMO_CONFIG_XML = '''<?xml version='1.0' encoding='UTF-8'?>
<hudson.plugins.promoted__builds.PromotionProcess>
<properties/>
<scm class="hudson.scm.NullSCM"/>
<canRoam>false</canRoam>
<triggers/>
<conditions/>
<icon>star-gold</icon>
<buildSteps/>
</hudson.plugins.promoted__builds.PromotionProcess>'''
# for testing only
PROMO_RECONFIG_XML = '''<?xml version='1.0' encoding='UTF-8'?>
<hudson.plugins.promoted__builds.PromotionProcess>
<keepDependencies>false</keepDependencies>
<properties/>
<scm class="hudson.scm.NullSCM"/>
<triggers/>
<icon>star-green</icon>
<buildSteps>
<hudson.tasks.Shell>
<command>ls -l</command>
</hudson.tasks.Shell>
</buildSteps>
</hudson.plugins.promoted__builds.PromotionProcess>
'''
class JenkinsException(Exception):
'''General exception type for jenkins-API-related failures.'''
@ -1324,6 +1357,143 @@ class Jenkins(object):
request = Request(self._build_url(CONFIG_VIEW, locals()))
return self.jenkins_open(request)
def get_promotion_name(self, name, job_name):
'''Return the name of a promotion using the API.
That is roughly an identity method which can be used to
quickly verify a promotion exists for a job or is accessible
without causing too much stress on the server side.
:param job_name: Job name, ``str``
:param name: Promotion name, ``str``
:returns: Name of promotion or None
'''
folder_url, short_name = self._get_job_folder(job_name)
try:
response = self.jenkins_open(Request(
self._build_url(PROMOTION_NAME, locals())))
except NotFoundException:
return None
else:
actual = json.loads(response)['name']
if actual != name:
raise JenkinsException(
'Jenkins returned an unexpected promotion name %s '
'(expected: %s)' % (actual, name))
return actual
def assert_promotion_exists(self, name, job_name,
exception_message='promotion[%s] does not '
'exist for job[%s]'):
'''Raise an exception if a job lacks a promotion
:param job_name: Job name, ``str``
:param name: Name of Jenkins promotion, ``str``
:param exception_message: Message to use for the exception. Formatted
with ``name`` and ``job_name``
:throws: :class:`JenkinsException` whenever the promotion
does not exist on a job
'''
if not self.promotion_exists(name, job_name):
raise JenkinsException(exception_message % (name, job_name))
def promotion_exists(self, name, job_name):
'''Check whether a job has a certain promotion
:param job_name: Job name, ``str``
:param name: Name of Jenkins promotion, ``str``
:returns: ``True`` if Jenkins promotion exists
'''
return self.get_promotion_name(name, job_name) == name
def get_promotions_info(self, job_name, depth=0):
'''Get promotion information dictionary of a job
:param name: job_name, ``str``
:param depth: JSON depth, ``int``
:returns: Dictionary of promotion info, ``dict``
'''
folder_url, short_name = self._get_job_folder(job_name)
try:
response = self.jenkins_open(Request(
self._build_url(PROMOTION_INFO, locals())))
if response:
return json.loads(response)
else:
raise JenkinsException('job[%s] does not exist' % job_name)
except HTTPError:
raise JenkinsException('job[%s] does not exist' % job_name)
except ValueError:
raise JenkinsException("Could not parse JSON info for "
"promotions of job[%s]" % job_name)
def get_promotions(self, job_name):
"""Get list of promotions running.
Each promotion is a dictionary with 'name' and 'url' keys.
:param job_name: Job name, ``str``
:returns: list of promotions, ``[ { str: str} ]``
"""
return self.get_promotions_info(job_name)['processes']
def delete_promotion(self, name, job_name):
'''Delete Jenkins promotion permanently.
:param job_name: Job name, ``str``
:param name: Name of Jenkins promotion, ``str``
'''
folder_url, short_name = self._get_job_folder(job_name)
self.jenkins_open(Request(
self._build_url(DELETE_PROMOTION, locals()), b''
))
if self.promotion_exists(name, job_name):
raise JenkinsException('delete[%s] from job[%s] failed' %
(name, job_name))
def create_promotion(self, name, job_name, config_xml):
'''Create a new Jenkins promotion
:param name: Name of Jenkins promotion, ``str``
:param job_name: Job name, ``str``
:param config_xml: config file text, ``str``
'''
if self.promotion_exists(name, job_name):
raise JenkinsException('promotion[%s] already exists at job[%s]'
% (name, job_name))
folder_url, short_name = self._get_job_folder(job_name)
self.jenkins_open(Request(
self._build_url(CREATE_PROMOTION, locals()),
config_xml.encode('utf-8'), DEFAULT_HEADERS))
self.assert_promotion_exists(name, job_name, 'create[%s] at '
'job[%s] failed')
def reconfig_promotion(self, name, job_name, config_xml):
'''Change configuration of existing Jenkins promotion.
To create a new promotion, see :meth:`Jenkins.create_promotion`.
:param name: Name of Jenkins promotion, ``str``
:param job_name: Job name, ``str``
:param config_xml: New XML configuration, ``str``
'''
folder_url, short_name = self._get_job_folder(job_name)
reconfig_url = self._build_url(CONFIG_PROMOTION, locals())
self.jenkins_open(Request(reconfig_url, config_xml.encode('utf-8'),
DEFAULT_HEADERS))
def get_promotion_config(self, name, job_name):
'''Get configuration of existing Jenkins promotion.
:param name: Name of Jenkins promotion, ``str``
:param job_name: Job name, ``str``
:returns: promotion configuration (XML format)
'''
folder_url, short_name = self._get_job_folder(job_name)
request = Request(self._build_url(CONFIG_PROMOTION, locals()))
return self.jenkins_open(request)
def quiet_down(self):
'''Prepare Jenkins for shutdown.

282
tests/test_promotion.py Normal file
View File

@ -0,0 +1,282 @@
import json
from mock import patch
import jenkins
from tests.base import JenkinsTestBase
from six.moves.urllib.error import HTTPError
class JenkinsPromotionsTestBase(JenkinsTestBase):
config_xml = """<hudson.plugins.promoted__builds.PromotionProcess>
</hudson.plugins.promoted__builds.PromotionProcess>"""
class JenkinsGetPromotionNameTest(JenkinsPromotionsTestBase):
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_simple(self, jenkins_mock):
promotion_name_to_return = {u'name': 'Test Promotion'}
jenkins_mock.return_value = json.dumps(promotion_name_to_return)
promotion_name = self.j.get_promotion_name(u'Test Promotion',
u'Test Job')
self.assertEqual(promotion_name, 'Test Promotion')
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
self.make_url('job/Test%20Job/promotion/process/'
'Test%20Promotion/api/json?tree=name'))
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_return_none(self, jenkins_mock):
jenkins_mock.side_effect = jenkins.NotFoundException()
promotion_name = self.j.get_promotion_name(u'TestPromotion',
u'Test Job')
self.assertEqual(promotion_name, None)
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
self.make_url('job/Test%20Job/promotion/process/'
'TestPromotion/api/json?tree=name'))
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_unexpected_promotion_name(self, jenkins_mock):
promotion_name_to_return = {u'name': 'not the right name'}
jenkins_mock.return_value = json.dumps(promotion_name_to_return)
with self.assertRaises(jenkins.JenkinsException) as context_manager:
self.j.get_promotion_name(u'TestPromotion', u'TestJob')
self.assertEqual(
jenkins_mock.call_args_list[0][0][0].get_full_url(),
self.make_url('job/TestJob/promotion/process/TestPromotion'
'/api/json?tree=name'))
self.assertEqual(
str(context_manager.exception),
'Jenkins returned an unexpected promotion name {0} '
'(expected: {1})'.format(promotion_name_to_return['name'],
'TestPromotion'))
self._check_requests(jenkins_mock.call_args_list)
class JenkinsAssertPromotionTest(JenkinsPromotionsTestBase):
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_promotion_missing(self, jenkins_mock):
jenkins_mock.side_effect = jenkins.NotFoundException()
with self.assertRaises(jenkins.JenkinsException) as context_manager:
self.j.assert_promotion_exists('NonExistent', 'TestJob')
self.assertEqual(
str(context_manager.exception),
'promotion[NonExistent] does not exist for job[TestJob]')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_promotion_exists(self, jenkins_mock):
jenkins_mock.side_effect = [
json.dumps({'name': 'ExistingPromotion'}),
]
self.j.assert_promotion_exists('ExistingPromotion', 'TestJob')
self._check_requests(jenkins_mock.call_args_list)
class JenkinsPromotionExistsTest(JenkinsPromotionsTestBase):
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_promotion_missing(self, jenkins_mock):
jenkins_mock.side_effect = jenkins.NotFoundException()
self.assertEqual(self.j.promotion_exists('NonExistent', 'TestJob'),
False)
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_promotion_exists(self, jenkins_mock):
jenkins_mock.side_effect = [
json.dumps({'name': 'ExistingPromotion'}),
]
self.assertEqual(self.j.promotion_exists('ExistingPromotion',
'TestJob'),
True)
self._check_requests(jenkins_mock.call_args_list)
class JenkinsGetPromotionsTest(JenkinsPromotionsTestBase):
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_simple(self, jenkins_mock):
promotions = {
u'url': (u'http://your_url_here/jobs/TestJob/promotions'
u'/my_promotion/'),
u'name': u'my_promotion',
}
promotion_info_to_return = {u'processes': promotions}
jenkins_mock.return_value = json.dumps(promotion_info_to_return)
promotion_info = self.j.get_promotions('TestJob')
self.assertEqual(promotion_info, promotions)
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
self.make_url('job/TestJob/promotion/api/json?depth=0'))
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_nonexistent(self, jenkins_mock):
jenkins_mock.side_effect = [
None,
HTTPError,
]
with self.assertRaises(jenkins.JenkinsException) as context_manager:
self.j.get_promotions('TestJob')
self.assertEqual(
str(context_manager.exception),
'job[TestJob] does not exist')
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_invalid_json(self, jenkins_mock):
jenkins_mock.return_value = '{invalid_json}'
with self.assertRaises(jenkins.JenkinsException) as context_manager:
self.j.get_promotions('TestJob')
self.assertEqual(
str(context_manager.exception),
"Could not parse JSON info for promotions of job[TestJob]")
class JenkinsDeletePromotionTest(JenkinsPromotionsTestBase):
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_simple(self, jenkins_mock):
jenkins_mock.side_effect = [
None,
jenkins.NotFoundException(),
]
self.j.delete_promotion(u'Test Promotion', 'TestJob')
self.assertEqual(
jenkins_mock.call_args_list[0][0][0].get_full_url(),
self.make_url('job/TestJob/promotion/process/'
'Test%20Promotion/doDelete'))
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_failed(self, jenkins_mock):
jenkins_mock.side_effect = [
json.dumps({'name': 'TestPromotion'}),
json.dumps({'name': 'TestPromotion'}),
json.dumps({'name': 'TestPromotion'}),
]
with self.assertRaises(jenkins.JenkinsException) as context_manager:
self.j.delete_promotion(u'TestPromotion', 'TestJob')
self.assertEqual(
jenkins_mock.call_args_list[0][0][0].get_full_url(),
self.make_url('job/TestJob/promotion/process/'
'TestPromotion/doDelete'))
self.assertEqual(
str(context_manager.exception),
'delete[TestPromotion] from job[TestJob] failed')
self._check_requests(jenkins_mock.call_args_list)
class JenkinsCreatePromotionTest(JenkinsPromotionsTestBase):
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_simple(self, jenkins_mock):
jenkins_mock.side_effect = [
jenkins.NotFoundException(),
None,
json.dumps({'name': 'Test Promotion'}),
]
self.j.create_promotion(u'Test Promotion', 'Test Job', self.config_xml)
self.assertEqual(
jenkins_mock.call_args_list[1][0][0].get_full_url(),
self.make_url('job/Test%20Job/promotion/'
'createProcess?name=Test%20Promotion'))
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_already_exists(self, jenkins_mock):
jenkins_mock.side_effect = [
json.dumps({'name': 'TestPromotion'}),
None,
]
with self.assertRaises(jenkins.JenkinsException) as context_manager:
self.j.create_promotion(u'TestPromotion', 'TestJob',
self.config_xml)
self.assertEqual(
jenkins_mock.call_args_list[0][0][0].get_full_url(),
self.make_url('job/TestJob/promotion/process/'
'TestPromotion/api/json?tree=name'))
self.assertEqual(
str(context_manager.exception),
'promotion[TestPromotion] already exists at job[TestJob]')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_failed(self, jenkins_mock):
jenkins_mock.side_effect = [
jenkins.NotFoundException(),
None,
jenkins.NotFoundException(),
]
with self.assertRaises(jenkins.JenkinsException) as context_manager:
self.j.create_promotion(u'TestPromotion', 'TestJob',
self.config_xml)
self.assertEqual(
jenkins_mock.call_args_list[0][0][0].get_full_url(),
self.make_url('job/TestJob/promotion/process/'
'TestPromotion/api/json?tree=name'))
self.assertEqual(
jenkins_mock.call_args_list[1][0][0].get_full_url(),
self.make_url('job/TestJob/promotion/'
'createProcess?name=TestPromotion'))
self.assertEqual(
str(context_manager.exception),
'create[TestPromotion] at job[TestJob] failed')
self._check_requests(jenkins_mock.call_args_list)
class JenkinsReconfigPromotionTest(JenkinsPromotionsTestBase):
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_simple(self, jenkins_mock):
jenkins_mock.side_effect = [
json.dumps({'name': 'Test Promotion'}),
None,
]
self.j.reconfig_promotion(u'Test Promotion', u'Test Job',
self.config_xml)
self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(),
self.make_url('job/Test%20Job/promotion/process/'
'Test%20Promotion/config.xml'))
self._check_requests(jenkins_mock.call_args_list)
class JenkinsGetPromotionConfigTest(JenkinsPromotionsTestBase):
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_encodes_promotion_name(self, jenkins_mock):
self.j.get_promotion_config(u'Test Promotion', u'Test Job')
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
self.make_url('job/Test%20Job/promotion/process/'
'Test%20Promotion/config.xml'))
self._check_requests(jenkins_mock.call_args_list)