jjb/python-jenkins
Code Issues Proposed changes

Add functions to retrieve more data from Jenkins

Browse Source 
If tests results are added to builds, or environment variables are injected
into a job, there are separate API calls to retrieve that data.  Add functions
to do so.

Change-Id: Ie0a51839d527db6d440719eeb44cd1cc89644d06
This commit is contained in:
JP Sullivan 2018-04-20 19:44:41 +01:00
parent 5f5d54b897
commit 827f28a223
2 changed files with 214 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
jenkins/__init__.py
View File

@ -116,6 +116,10 @@ STOP_BUILD = '%(folder_url)sjob/%(short_name)s/%(number)s/stop'
BUILD_WITH_PARAMS_JOB = '%(folder_url)sjob/%(short_name)s/buildWithParameters'
BUILD_INFO = '%(folder_url)sjob/%(short_name)s/%(number)d/api/json?depth=%(depth)s'
BUILD_CONSOLE_OUTPUT = '%(folder_url)sjob/%(short_name)s/%(number)d/consoleText'
BUILD_ENV_VARS = '%(folder_url)sjob/%(short_name)s/%(number)d/injectedEnvVars/api/json' + \
'?depth=%(depth)s'
BUILD_TEST_REPORT = '%(folder_url)sjob/%(short_name)s/%(number)d/testReport/api/json' + \
'?depth=%(depth)s'
DELETE_BUILD = '%(folder_url)sjob/%(short_name)s/%(number)s/doDelete'
WIPEOUT_JOB_WORKSPACE = '%(folder_url)sjob/%(short_name)s/doWipeOutWorkspace'
NODE_LIST = 'computer/api/json'
@ -637,6 +641,56 @@ class Jenkins(object):
% (name, number)
)
def get_build_env_vars(self, name, number, depth=0):
'''Get build environment variables.
:param name: Job name, ``str``
:param number: Build number, ``int``
:param depth: JSON depth, ``int``
:returns: dictionary of build env vars, ``dict`` or None for workflow jobs,
or if InjectEnvVars plugin not installed
'''
folder_url, short_name = self._get_job_folder(name)
try:
response = self.jenkins_open(requests.Request(
'GET', self._build_url(BUILD_ENV_VARS, locals())))
if response:
return json.loads(response)
else:
raise JenkinsException('job[%s] number[%d] does not exist' % (name, number))
except req_exc.HTTPError:
raise JenkinsException('job[%s] number[%d] does not exist' % (name, number))
except ValueError:
raise JenkinsException(
'Could not parse JSON info for job[%s] number[%d]' % (name, number))
except NotFoundException:
# This can happen on workflow jobs, or if InjectEnvVars plugin not installed
return None
def get_build_test_report(self, name, number, depth=0):
'''Get test results report.
:param name: Job name, ``str``
:param number: Build number, ``int``
:returns: dictionary of test report results, ``dict`` or None if there is no Test Report
'''
folder_url, short_name = self._get_job_folder(name)
try:
response = self.jenkins_open(requests.Request(
'GET', self._build_url(BUILD_TEST_REPORT, locals())))
if response:
return json.loads(response)
else:
raise JenkinsException('job[%s] number[%d] does not exist' % (name, number))
except req_exc.HTTPError:
raise JenkinsException('job[%s] number[%d] does not exist' % (name, number))
except ValueError:
raise JenkinsException(
'Could not parse JSON info for job[%s] number[%d]' % (name, number))
except NotFoundException:
# This can happen if the test report wasn't generated for any reason
return None
def get_queue_info(self):
''':returns: list of job dictionaries, ``[dict]``

160
tests/test_build.py
View File

@ -495,3 +495,163 @@ class JenkinsBuildJobUrlTest(JenkinsTestBase):
self.make_url(
'job/Test%20Job/buildWithParameters?m_select=value1'
'&m_select=value3&s_select=s_select2&token=token123'))
class JenkinsBuildEnvVarUrlTest(JenkinsTestBase):
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_simple(self, jenkins_mock):
jenkins_mock.return_value = '{}'
ret = self.j.get_build_env_vars(u'Test Job', number=52, depth=1)
self.assertEqual(ret, json.loads(jenkins_mock.return_value))
self.assertEqual(
jenkins_mock.call_args[0][0].url,
self.make_url('job/Test%20Job/52/injectedEnvVars/api/json?depth=1'))
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_in_folder(self, jenkins_mock):
jenkins_mock.return_value = '{}'
ret = self.j.get_build_env_vars(u'a Folder/Test Job', number=52, depth=1)
self.assertEqual(ret, json.loads(jenkins_mock.return_value))
self.assertEqual(
jenkins_mock.call_args[0][0].url,
self.make_url('job/a%20Folder/job/Test%20Job/52/injectedEnvVars/api/json?depth=1'))
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.requests.Session.send', autospec=True)
def test_return_none(self, session_send_mock):
session_send_mock.side_effect = iter([
build_response_mock(404, reason="Not Found"), # crumb
build_response_mock(404, reason="Not Found"), # request
])
ret = self.j.get_build_env_vars(u'TestJob', number=52)
self.assertIsNone(ret)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_open_return_none(self, jenkins_mock):
jenkins_mock.return_value = None
with self.assertRaises(jenkins.JenkinsException) as context_manager:
self.j.get_build_env_vars(u'TestJob', number=52)
self.assertEqual(
str(context_manager.exception),
'job[TestJob] number[52] does not exist')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_return_invalid_json(self, jenkins_mock):
jenkins_mock.return_value = 'Invalid JSON'
with self.assertRaises(jenkins.JenkinsException) as context_manager:
self.j.get_build_env_vars(u'TestJob', number=52)
self.assertEqual(
str(context_manager.exception),
'Could not parse JSON info for job[TestJob] number[52]')
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.requests.Session.send', autospec=True)
def test_raise_HTTPError(self, session_send_mock):
session_send_mock.side_effect = iter([
build_response_mock(401, reason="Not Authorised"), # crumb
build_response_mock(401, reason="Not Authorised"), # request
])
with self.assertRaises(jenkins.JenkinsException) as context_manager:
self.j.get_build_env_vars(u'TestJob', number=52)
self.assertEqual(
str(context_manager.exception),
'Error in request. Possibly authentication failed [401]: Not Authorised')
@patch('jenkins.requests.Session.send', autospec=True)
def test_in_folder_raise_HTTPError(self, session_send_mock):
session_send_mock.side_effect = iter([
build_response_mock(401, reason="Not Authorised"), # crumb
build_response_mock(401, reason="Not Authorised"), # request
])
with self.assertRaises(jenkins.JenkinsException) as context_manager:
self.j.get_build_env_vars(u'a Folder/TestJob', number=52)
self.assertEqual(
str(context_manager.exception),
'Error in request. Possibly authentication failed [401]: Not Authorised')
class JenkinsBuildTestReportUrlTest(JenkinsTestBase):
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_simple(self, jenkins_mock):
jenkins_mock.return_value = '{}'
ret = self.j.get_build_test_report(u'Test Job', number=52, depth=1)
self.assertEqual(ret, json.loads(jenkins_mock.return_value))
self.assertEqual(
jenkins_mock.call_args[0][0].url,
self.make_url('job/Test%20Job/52/testReport/api/json?depth=1'))
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_in_folder(self, jenkins_mock):
jenkins_mock.return_value = '{}'
ret = self.j.get_build_test_report(u'a Folder/Test Job', number=52, depth=1)
self.assertEqual(ret, json.loads(jenkins_mock.return_value))
self.assertEqual(
jenkins_mock.call_args[0][0].url,
self.make_url('job/a%20Folder/job/Test%20Job/52/testReport/api/json?depth=1'))
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.requests.Session.send', autospec=True)
def test_return_none(self, session_send_mock):
session_send_mock.side_effect = iter([
build_response_mock(404, reason="Not Found"), # crumb
build_response_mock(404, reason="Not Found"), # request
])
ret = self.j.get_build_test_report(u'TestJob', number=52)
self.assertIsNone(ret)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_open_return_none(self, jenkins_mock):
jenkins_mock.return_value = None
with self.assertRaises(jenkins.JenkinsException) as context_manager:
self.j.get_build_test_report(u'TestJob', number=52)
self.assertEqual(
str(context_manager.exception),
'job[TestJob] number[52] does not exist')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_return_invalid_json(self, jenkins_mock):
jenkins_mock.return_value = 'Invalid JSON'
with self.assertRaises(jenkins.JenkinsException) as context_manager:
self.j.get_build_test_report(u'TestJob', number=52)
self.assertEqual(
str(context_manager.exception),
'Could not parse JSON info for job[TestJob] number[52]')
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.requests.Session.send', autospec=True)
def test_raise_HTTPError(self, session_send_mock):
session_send_mock.side_effect = iter([
build_response_mock(401, reason="Not Authorised"), # crumb
build_response_mock(401, reason="Not Authorised"), # request
])
with self.assertRaises(jenkins.JenkinsException) as context_manager:
self.j.get_build_test_report(u'TestJob', number=52)
self.assertEqual(
str(context_manager.exception),
'Error in request. Possibly authentication failed [401]: Not Authorised')
@patch('jenkins.requests.Session.send', autospec=True)
def test_in_folder_raise_HTTPError(self, session_send_mock):
session_send_mock.side_effect = iter([
build_response_mock(401, reason="Not Authorised"), # crumb
build_response_mock(401, reason="Not Authorised"), # request
])
with self.assertRaises(jenkins.JenkinsException) as context_manager:
self.j.get_build_test_report(u'a Folder/TestJob', number=52)
self.assertEqual(
str(context_manager.exception),
'Error in request. Possibly authentication failed [401]: Not Authorised')