jjb
/
python-jenkins
Code Issues Proposed changes

Modify tests to exercise Request on python3 

Need to exercise the Request object on python3 to ensure that the
parameters passed to it are in the correct format. Use a simple
combination of the opener and required handler to be able to have the
url handling code parse the Request object created for performing API
calls.

Change-Id: Ia6ce75958bb18a743b7fd88d9512fcba3128a527
This commit is contained in:
Darragh Bailey 2015-05-14 18:07:36 +01:00
parent 5b29a91ded
commit 76410adc19
2 changed files with 135 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
jenkins/__init__.py
View File

@ -360,7 +360,7 @@ class Jenkins(object):
# https://issues.jenkins-ci.org/browse/JENKINS-21311
try:
self.jenkins_open(
Request(self.server + CANCEL_QUEUE % locals(), '',
Request(self.server + CANCEL_QUEUE % locals(), b'',
headers={'Referer': self.server}))
except NotFoundException:
# Exception is expected; cancel_queue() is a best-effort
@ -513,7 +513,8 @@ class Jenkins(object):
:param to_name: Name of Jenkins job to copy to, ``str``
'''
self.jenkins_open(Request(
self.server + COPY_JOB % self._get_encoded_params(locals()), ''))
self.server + COPY_JOB % self._get_encoded_params(locals()),
b''))
self.assert_job_exists(to_name, 'create[%s] failed')
def rename_job(self, from_name, to_name):
@ -523,7 +524,8 @@ class Jenkins(object):
:param to_name: New Jenkins job name, ``str``
'''
self.jenkins_open(Request(
self.server + RENAME_JOB % self._get_encoded_params(locals()), ''))
self.server + RENAME_JOB % self._get_encoded_params(locals()),
b''))
self.assert_job_exists(to_name, 'rename[%s] failed')
def delete_job(self, name):
@ -532,7 +534,8 @@ class Jenkins(object):
:param name: Name of Jenkins job, ``str``
'''
self.jenkins_open(Request(
self.server + DELETE_JOB % self._get_encoded_params(locals()), ''))
self.server + DELETE_JOB % self._get_encoded_params(locals()),
b''))
if self.job_exists(name):
raise JenkinsException('delete[%s] failed' % (name))
@ -542,7 +545,8 @@ class Jenkins(object):
:param name: Name of Jenkins job, ``str``
'''
self.jenkins_open(Request(
self.server + ENABLE_JOB % self._get_encoded_params(locals()), ''))
self.server + ENABLE_JOB % self._get_encoded_params(locals()),
b''))
def disable_job(self, name):
'''Disable Jenkins job.
@ -552,7 +556,8 @@ class Jenkins(object):
:param name: Name of Jenkins job, ``str``
'''
self.jenkins_open(Request(
self.server + DISABLE_JOB % self._get_encoded_params(locals()), ''))
self.server + DISABLE_JOB % self._get_encoded_params(locals()),
b''))
def job_exists(self, name):
'''Check whether a job exists
@ -592,7 +597,8 @@ class Jenkins(object):
raise JenkinsException('job[%s] already exists' % (name))
self.jenkins_open(Request(
self.server + CREATE_JOB % self._get_encoded_params(locals()), config_xml, DEFAULT_HEADERS))
self.server + CREATE_JOB % self._get_encoded_params(locals()),
config_xml.encode('utf-8'), DEFAULT_HEADERS))
self.assert_job_exists(name, 'create[%s] failed')
def get_job_config(self, name):
@ -613,7 +619,8 @@ class Jenkins(object):
:param config_xml: New XML configuration, ``str``
'''
reconfig_url = self.server + CONFIG_JOB % self._get_encoded_params(locals())
self.jenkins_open(Request(reconfig_url, config_xml, DEFAULT_HEADERS))
self.jenkins_open(Request(reconfig_url, config_xml.encode('utf-8'),
DEFAULT_HEADERS))
def build_job_url(self, name, parameters=None, token=None):
'''Get URL to trigger build job.
@ -644,7 +651,7 @@ class Jenkins(object):
:param token: Jenkins API token
'''
return self.jenkins_open(Request(
self.build_job_url(name, parameters, token), ""))
self.build_job_url(name, parameters, token), b''))
def stop_build(self, name, number):
'''Stop a running Jenkins build.
@ -723,7 +730,8 @@ class Jenkins(object):
'''
self.get_node_info(name)
self.jenkins_open(Request(
self.server + DELETE_NODE % self._get_encoded_params(locals()), ''))
self.server + DELETE_NODE % self._get_encoded_params(locals()),
b''))
if self.node_exists(name):
raise JenkinsException('delete[%s] failed' % (name))
@ -737,7 +745,8 @@ class Jenkins(object):
if info['offline']:
return
self.jenkins_open(Request(
self.server + TOGGLE_OFFLINE % self._get_encoded_params(locals()), ''))
self.server + TOGGLE_OFFLINE % self._get_encoded_params(locals()),
b''))
def enable_node(self, name):
'''Enable a node
@ -749,7 +758,8 @@ class Jenkins(object):
return
msg = ''
self.jenkins_open(Request(
self.server + TOGGLE_OFFLINE % self._get_encoded_params(locals()), ''))
self.server + TOGGLE_OFFLINE % self._get_encoded_params(locals()),
b''))
def create_node(self, name, numExecutors=2, nodeDescription=None,
remoteFS='/var/lib/jenkins', labels=None, exclusive=False,

113
tests/test_jenkins.py
View File

@ -9,6 +9,7 @@ else:
from mock import patch, Mock
import six
from six.moves.urllib.error import HTTPError
from six.moves.urllib.request import build_opener
from tests.helper import jenkins
@ -43,6 +44,28 @@ class JenkinsTest(unittest.TestCase):
]
}
def setUp(self):
super(JenkinsTest, self).setUp()
self.opener = build_opener()
def _check_requests(self, requests):
for req in requests:
self._check_request(req[0][0])
def _check_request(self, request):
# taken from opener.open() in request
# attribute request.type is only set automatically for python 3
# requests, must use request.get_type() for python 2.7
protocol = request.type or request.get_type()
# check that building the request doesn't throw any exception
meth_name = protocol + "_request"
for processor in self.opener.process_request.get(protocol, []):
meth = getattr(processor, meth_name)
request = meth(request)
def test_constructor_url_with_trailing_slash(self):
j = jenkins.Jenkins('http://example.com/', 'test', 'test')
self.assertEqual(j.server, 'http://example.com/')
@ -95,6 +118,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/job/Test%20Job/config.xml')
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.urlopen')
def test_maybe_add_crumb(self, jenkins_mock):
@ -109,6 +133,7 @@ class JenkinsTest(unittest.TestCase):
'http://example.com/crumbIssuer/api/json')
self.assertFalse(j.crumb)
self.assertFalse('.crumb' in request.headers)
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.urlopen')
def test_maybe_add_crumb__with_data(self, jenkins_mock):
@ -127,6 +152,7 @@ class JenkinsTest(unittest.TestCase):
'http://example.com/crumbIssuer/api/json')
self.assertEqual(j.crumb, crumb_data)
self.assertEqual(request.headers['.crumb'], crumb_data['crumb'])
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_maybe_add_crumb__empty_response(self, jenkins_mock):
@ -142,6 +168,7 @@ class JenkinsTest(unittest.TestCase):
'http://example.com/crumbIssuer/api/json')
self.assertFalse(j.crumb)
self.assertFalse('.crumb' in request.headers)
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.urlopen')
def test_jenkins_open(self, jenkins_mock):
@ -165,6 +192,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(response, json.dumps(data))
self.assertEqual(j.crumb, crumb_data)
self.assertEqual(request.headers['.crumb'], crumb_data['crumb'])
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.urlopen')
def test_jenkins_open__403(self, jenkins_mock):
@ -186,6 +214,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
'http://example.com/job/TestJob')
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.urlopen')
def test_jenkins_open__404(self, jenkins_mock):
@ -206,6 +235,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
'http://example.com/job/TestJob')
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.urlopen')
def test_jenkins_open__empty_response(self, jenkins_mock):
@ -223,6 +253,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
'http://example.com/job/TestJob')
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.urlopen')
def test_jenkins_open__501(self, jenkins_mock):
@ -243,6 +274,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
'http://example.com/job/TestJob')
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.urlopen')
def test_jenkins_open__timeout(self, jenkins_mock):
@ -259,6 +291,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
'http://example.com/job/TestJob')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_assert_job_exists__job_missing(self, jenkins_mock):
@ -270,6 +303,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'job[NonExistent] does not exist')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_assert_job_exists__job_exists(self, jenkins_mock):
@ -278,6 +312,7 @@ class JenkinsTest(unittest.TestCase):
]
j = jenkins.Jenkins('http://example.com/', 'test', 'test')
j.assert_job_exists('ExistingJob')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_create_job(self, jenkins_mock):
@ -298,6 +333,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args_list[1][0][0].get_full_url(),
'http://example.com/createItem?name=Test%20Job')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_create_job__already_exists(self, jenkins_mock):
@ -320,6 +356,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'job[TestJob] already exists')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_create_job__create_failed(self, jenkins_mock):
@ -346,6 +383,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'create[TestJob] failed')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_reconfig_job(self, jenkins_mock):
@ -364,6 +402,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/job/Test%20Job/config.xml')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_build_job(self, jenkins_mock):
@ -377,6 +416,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/job/Test%20Job/build')
self.assertEqual(build_info, {'foo': 'bar'})
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_build_job__with_token(self, jenkins_mock):
@ -390,6 +430,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/job/TestJob/build?token=some_token')
self.assertEqual(build_info, {'foo': 'bar'})
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_build_job__with_parameters_and_token(self, jenkins_mock):
@ -407,6 +448,7 @@ class JenkinsTest(unittest.TestCase):
self.assertTrue('when=now' in jenkins_mock.call_args[0][0].get_full_url())
self.assertTrue('why=because+I+felt+like+it' in jenkins_mock.call_args[0][0].get_full_url())
self.assertEqual(build_info, {'foo': 'bar'})
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_stop_build(self, jenkins_mock):
@ -417,6 +459,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/job/Test%20Job/52/stop')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_build_console_output(self, jenkins_mock):
@ -429,6 +472,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/job/Test%20Job/52/consoleText')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_build_console_output__None(self, jenkins_mock):
@ -440,6 +484,7 @@ class JenkinsTest(unittest.TestCase):
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_get_build_console_output__invalid_json(self, jenkins_mock):
@ -448,6 +493,7 @@ class JenkinsTest(unittest.TestCase):
console_output = j.get_build_console_output(u'TestJob', number=52)
self.assertEqual(console_output, jenkins_mock.return_value)
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_build_console_output__HTTPError(self, jenkins_mock):
@ -467,6 +513,7 @@ class JenkinsTest(unittest.TestCase):
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_get_build_info(self, jenkins_mock):
@ -485,6 +532,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/job/Test%20Job/52/api/json?depth=0')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_build_info__None(self, jenkins_mock):
@ -496,6 +544,7 @@ class JenkinsTest(unittest.TestCase):
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_get_build_info__invalid_json(self, jenkins_mock):
@ -507,6 +556,7 @@ class JenkinsTest(unittest.TestCase):
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.object(jenkins.Jenkins, 'jenkins_open')
def test_get_build_info__HTTPError(self, jenkins_mock):
@ -523,6 +573,7 @@ class JenkinsTest(unittest.TestCase):
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_get_job_info(self, jenkins_mock):
@ -541,6 +592,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/job/Test%20Job/api/json?depth=0')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_job_info_regex(self, jenkins_mock):
@ -561,6 +613,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(len(j.get_job_info_regex('job')), 3)
self.assertEqual(len(j.get_job_info_regex('project')), 1)
self.assertEqual(len(j.get_job_info_regex('[Yy]our-[Jj]ob-1')), 2)
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_job_info__None(self, jenkins_mock):
@ -575,6 +628,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'job[TestJob] does not exist')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_job_info__invalid_json(self, jenkins_mock):
@ -589,6 +643,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'Could not parse JSON info for job[TestJob]')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_job_info__HTTPError(self, jenkins_mock):
@ -608,6 +663,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'job[TestJob] does not exist')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_debug_job_info(self, jenkins_mock):
@ -625,6 +681,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/job/Test%20Job/api/json?depth=0')
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.urlopen')
def test_get_version__some_version(self, urlopen_mock):
@ -640,6 +697,7 @@ class JenkinsTest(unittest.TestCase):
mock_response.configure_mock(**config)
urlopen_mock.side_effect = [mock_response]
self.assertEqual(j.get_version(), 'Version42')
self._check_requests(urlopen_mock.call_args_list)
@patch('jenkins.urlopen')
def test_get_version__HTTPError(self, urlopen_mock):
@ -655,6 +713,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'Error communicating with server[http://example.com/]')
self._check_requests(urlopen_mock.call_args_list)
@patch('jenkins.urlopen')
def test_get_version__BadStatusLine(self, urlopen_mock):
@ -665,6 +724,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'Error communicating with server[http://example.com/]')
self._check_requests(urlopen_mock.call_args_list)
@patch('jenkins.urlopen', return_value=None)
def test_get_version__empty_response(self, urlopen_mock):
@ -675,6 +735,7 @@ class JenkinsTest(unittest.TestCase):
str(context_manager.exception),
'Error communicating with server[http://example.com/]:'
' empty response')
self._check_requests(urlopen_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_jobs_count(self, jenkins_mock):
@ -687,6 +748,7 @@ class JenkinsTest(unittest.TestCase):
jenkins_mock.return_value = json.dumps(job_info_to_return)
j = jenkins.Jenkins('http://example.com/', 'test', 'test')
self.assertEqual(j.jobs_count(), 3)
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_jobs(self, jenkins_mock):
@ -705,6 +767,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/api/json')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_info(self, jenkins_mock):
@ -724,6 +787,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/api/json')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_plugins_info(self, jenkins_mock):
@ -737,6 +801,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/pluginManager/api/json?depth=2')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_plugins_info_none(self, jenkins_mock):
@ -748,6 +813,7 @@ class JenkinsTest(unittest.TestCase):
plugins_info = j.get_plugins_info()
self.assertEqual(plugins_info, [])
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_plugins_info_depth(self, jenkins_mock):
@ -759,6 +825,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/pluginManager/api/json?depth=1')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_plugins_info__BadStatusLine(self, jenkins_mock):
@ -773,6 +840,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'Error communicating with server[http://example.com/]')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_plugins_info__ValueError(self, jenkins_mock):
@ -787,6 +855,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'Could not parse JSON info for server[http://example.com/]')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_plugins_info__HTTPError(self, jenkins_mock):
@ -803,6 +872,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'Error communicating with server[http://example.com/]')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_plugin_info_shortname(self, jenkins_mock):
@ -813,6 +883,7 @@ class JenkinsTest(unittest.TestCase):
# expected to return info on a single plugin
plugin_info = j.get_plugin_info("mailer")
self.assertEqual(plugin_info, self.plugin_info_json['plugins'][0])
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_plugin_info_longname(self, jenkins_mock):
@ -823,6 +894,7 @@ class JenkinsTest(unittest.TestCase):
# expected to return info on a single plugin
plugin_info = j.get_plugin_info("Jenkins Mailer Plugin")
self.assertEqual(plugin_info, self.plugin_info_json['plugins'][0])
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_plugin_info_none(self, jenkins_mock):
@ -833,6 +905,7 @@ class JenkinsTest(unittest.TestCase):
# expected not to find bogus so should return None
plugin_info = j.get_plugin_info("bogus")
self.assertEqual(plugin_info, None)
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_plugin_info_depth(self, jenkins_mock):
@ -844,6 +917,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/pluginManager/api/json?depth=1')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_plugin_info__BadStatusLine(self, jenkins_mock):
@ -858,6 +932,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'Error communicating with server[http://example.com/]')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_plugin_info__ValueError(self, jenkins_mock):
@ -872,6 +947,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'Could not parse JSON info for server[http://example.com/]')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_plugin_info__HTTPError(self, jenkins_mock):
@ -888,6 +964,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'Error communicating with server[http://example.com/]')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_info__HTTPError(self, jenkins_mock):
@ -907,6 +984,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'Error communicating with server[http://example.com/]')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_info__BadStatusLine(self, jenkins_mock):
@ -921,6 +999,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'Error communicating with server[http://example.com/]')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_info__ValueError(self, jenkins_mock):
@ -935,6 +1014,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'Could not parse JSON info for server[http://example.com/]')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_info__empty_response(self, jenkins_mock):
@ -952,6 +1032,7 @@ class JenkinsTest(unittest.TestCase):
str(context_manager.exception),
'Error communicating with server[http://example.com/]: '
'empty response')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_copy_job(self, jenkins_mock):
@ -969,6 +1050,7 @@ class JenkinsTest(unittest.TestCase):
'http://example.com/createItem'
'?name=Test%20Job_2&mode=copy&from=Test%20Job')
self.assertTrue(j.job_exists('Test Job_2'))
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_copy_job__create_failed(self, jenkins_mock):
@ -987,6 +1069,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'create[TestJob_2] failed')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_rename_job(self, jenkins_mock):
@ -1003,6 +1086,7 @@ class JenkinsTest(unittest.TestCase):
jenkins_mock.call_args_list[0][0][0].get_full_url(),
'http://example.com/job/Test%20Job/doRename?newName=Test%20Job_2')
self.assertTrue(j.job_exists('Test Job_2'))
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_rename_job__rename_failed(self, jenkins_mock):
@ -1020,6 +1104,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'rename[TestJob_2] failed')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_delete_job(self, jenkins_mock):
@ -1034,6 +1119,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args_list[0][0][0].get_full_url(),
'http://example.com/job/Test%20Job/doDelete')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_delete_job__delete_failed(self, jenkins_mock):
@ -1052,6 +1138,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'delete[TestJob] failed')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_enable_job(self, jenkins_mock):
@ -1067,6 +1154,7 @@ class JenkinsTest(unittest.TestCase):
jenkins_mock.call_args_list[0][0][0].get_full_url(),
'http://example.com/job/TestJob/enable')
self.assertTrue(j.job_exists('TestJob'))
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_disable_job(self, jenkins_mock):
@ -1082,6 +1170,7 @@ class JenkinsTest(unittest.TestCase):
jenkins_mock.call_args_list[0][0][0].get_full_url(),
'http://example.com/job/Test%20Job/disable')
self.assertTrue(j.job_exists('Test Job'))
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_job_name(self, jenkins_mock):
@ -1095,6 +1184,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/job/Test%20Job/api/json?tree=name')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_job_name__None(self, jenkins_mock):
@ -1107,6 +1197,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/job/TestJob/api/json?tree=name')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_job_name__unexpected_job_name(self, jenkins_mock):
@ -1123,6 +1214,7 @@ class JenkinsTest(unittest.TestCase):
str(context_manager.exception),
'Jenkins returned an unexpected job name {0} '
'(expected: {1})'.format(job_name_to_return['name'], 'TestJob'))
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_cancel_queue(self, jenkins_mock):
@ -1135,6 +1227,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/queue/cancelItem?id=52')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open',
side_effect=jenkins.NotFoundException('not found'))
@ -1148,6 +1241,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/queue/cancelItem?id=52')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_nodes(self, jenkins_mock):
@ -1160,6 +1254,7 @@ class JenkinsTest(unittest.TestCase):
j = jenkins.Jenkins('http://example.com/', 'test', 'test')
self.assertEqual(j.get_nodes(),
[{'name': 'master', 'offline': False}])
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_nodes__invalid_json(self, jenkins_mock):
@ -1176,6 +1271,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'Could not parse JSON info for server[http://example.com/]')
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.urlopen')
def test_get_nodes__BadStatusLine(self, urlopen_mock):
@ -1186,6 +1282,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'Error communicating with server[http://example.com/]')
self._check_requests(urlopen_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_nodes__HTTPError(self, jenkins_mock):
@ -1205,6 +1302,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'Error communicating with server[http://example.com/]')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_node_info(self, jenkins_mock):
@ -1221,6 +1319,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
'http://example.com/computer/test%20node/api/json?depth=0')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_node_info__invalid_json(self, jenkins_mock):
@ -1237,6 +1336,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'Could not parse JSON info for node[test_node]')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_node_info__HTTPError(self, jenkins_mock):
@ -1256,6 +1356,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'node[test_node] does not exist')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_assert_node_exists__node_missing(self, jenkins_mock):
@ -1267,6 +1368,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'node[NonExistentNode] does not exist')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_assert_node_exists__node_exists(self, jenkins_mock):
@ -1275,6 +1377,7 @@ class JenkinsTest(unittest.TestCase):
]
j = jenkins.Jenkins('http://example.com/', 'test', 'test')
j.assert_node_exists('ExistingNode')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_delete_node(self, jenkins_mock):
@ -1296,6 +1399,7 @@ class JenkinsTest(unittest.TestCase):
jenkins_mock.call_args_list[1][0][0].get_full_url(),
'http://example.com/computer/test%20node/doDelete')
self.assertFalse(j.node_exists('test node'))
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_delete_node__delete_failed(self, jenkins_mock):
@ -1318,6 +1422,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'delete[test_node] failed')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_create_node(self, jenkins_mock):
@ -1339,6 +1444,7 @@ class JenkinsTest(unittest.TestCase):
jenkins_mock.call_args_list[1][0][0].get_full_url().split('?')[0],
'http://example.com/computer/doCreateItem')
self.assertTrue(j.node_exists('test node'))
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_create_node__node_already_exists(self, jenkins_mock):
@ -1356,6 +1462,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'node[test_node] already exists')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_create_node__create_failed(self, jenkins_mock):
@ -1375,6 +1482,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
str(context_manager.exception),
'create[test_node] failed')
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_enable_node(self, jenkins_mock):
@ -1404,6 +1512,7 @@ class JenkinsTest(unittest.TestCase):
jenkins_mock.side_effect = [json.dumps(expected_node_info)]
node_info = j.get_node_info('test node')
self.assertEqual(node_info, expected_node_info)
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_enable_node__offline_False(self, jenkins_mock):
@ -1434,6 +1543,7 @@ class JenkinsTest(unittest.TestCase):
jenkins_mock.side_effect = [json.dumps(expected_node_info)]
node_info = j.get_node_info('test_node')
self.assertEqual(node_info, expected_node_info)
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_disable_node(self, jenkins_mock):
@ -1463,6 +1573,7 @@ class JenkinsTest(unittest.TestCase):
jenkins_mock.side_effect = [json.dumps(expected_node_info)]
node_info = j.get_node_info('test node')
self.assertEqual(node_info, expected_node_info)
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_disable_node__offline_True(self, jenkins_mock):
@ -1493,6 +1604,7 @@ class JenkinsTest(unittest.TestCase):
jenkins_mock.side_effect = [json.dumps(expected_node_info)]
node_info = j.get_node_info('test_node')
self.assertEqual(node_info, expected_node_info)
self._check_requests(jenkins_mock.call_args_list)
@patch.object(jenkins.Jenkins, 'jenkins_open')
def test_get_queue_info(self, jenkins_mock):
@ -1529,3 +1641,4 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
u'http://example.com/queue/api/json?depth=0')
self._check_requests(jenkins_mock.call_args_list)