From 0c503b724c96dd6f2b1dae923383b86d2f9bbb45 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Guido=20G=C3=BCnther?= Date: Tue, 12 Jan 2016 18:42:17 +0100 Subject: [PATCH] Add support for handling promotions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Co-Authored-By: Joao Vale Change-Id: I756a35f53f96ba6e46e71f36ea99f62d32b604d2 --- doc/source/examples.rst | 35 +++++ jenkins/__init__.py | 170 ++++++++++++++++++++++++ tests/test_promotion.py | 282 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 487 insertions(+) create mode 100644 tests/test_promotion.py diff --git a/doc/source/examples.rst b/doc/source/examples.rst index 812343e..350afd2 100644 --- a/doc/source/examples.rst +++ b/doc/source/examples.rst @@ -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 +`_ +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: + +:: + + + + prom_name + + + + +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') diff --git a/jenkins/__init__.py b/jenkins/__init__.py index 34cae71..9f6075a 100644 --- a/jenkins/__init__.py +++ b/jenkins/__init__.py @@ -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 = ''' ''' +# for testing only +EMPTY_PROMO_CONFIG_XML = ''' + + + + false + + + star-gold + +''' + +# for testing only +PROMO_RECONFIG_XML = ''' + + false + + + + star-green + + + ls -l + + + +''' + 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. diff --git a/tests/test_promotion.py b/tests/test_promotion.py new file mode 100644 index 0000000..a9fc425 --- /dev/null +++ b/tests/test_promotion.py @@ -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 = """ + """ + + +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)