diff --git a/jenkins/__init__.py b/jenkins/__init__.py index dbb8e1c..50a11ee 100755 --- a/jenkins/__init__.py +++ b/jenkins/__init__.py @@ -46,7 +46,6 @@ See examples at :doc:`examples` ''' -import base64 import json import re import socket @@ -55,24 +54,18 @@ import time import warnings import multi_key_dict -import six +import requests +import requests.exceptions as req_exc from six.moves.http_client import BadStatusLine -from six.moves.urllib.error import HTTPError from six.moves.urllib.error import URLError from six.moves.urllib.parse import quote, urlencode, urljoin, urlparse -from six.moves.urllib.request import Request, install_opener, build_opener, urlopen from jenkins import plugins try: - import kerberos - assert kerberos # pyflakes - from jenkins import urllib_kerb - opener = build_opener() - opener.add_handler(urllib_kerb.HTTPNegotiateHandler()) - install_opener(opener) + import requests_kerberos except ImportError: - pass + requests_kerberos = None if sys.version_info < (2, 7, 0): @@ -240,17 +233,6 @@ class TimeoutException(JenkinsException): '''A special exception to call out in the case of a socket timeout.''' -def auth_headers(username, password): - '''Simple implementation of HTTP Basic Authentication. - - Returns the 'Authentication' header value. - ''' - auth = '%s:%s' % (username, password) - if isinstance(auth, six.text_type): - auth = auth.encode('utf-8') - return b'Basic ' + base64.b64encode(auth) - - class Jenkins(object): _timeout_warning_issued = False @@ -269,12 +251,25 @@ class Jenkins(object): self.server = url else: self.server = url + '/' + + self._auths = [('anonymous', None)] + self._auth_resolved = False if username is not None and password is not None: - self.auth = auth_headers(username, password) - else: - self.auth = None + self._auths[0] = ( + 'basic', + requests.auth.HTTPBasicAuth( + username.encode('utf-8'), password.encode('utf-8')) + ) + + if requests_kerberos is not None: + self._auths.append( + ('kerberos', requests_kerberos.HTTPKerberosAuth()) + ) + + self.auth = None self.crumb = None self.timeout = timeout + self._session = requests.Session() def _get_encoded_params(self, params): for k, v in params.items(): @@ -296,14 +291,48 @@ class Jenkins(object): # We don't know yet whether we need a crumb if self.crumb is None: try: - response = self.jenkins_open(Request( - self._build_url(CRUMB_URL)), add_crumb=False) + response = self.jenkins_open(requests.Request( + 'GET', self._build_url(CRUMB_URL)), add_crumb=False) except (NotFoundException, EmptyResponseException): self.crumb = False else: self.crumb = json.loads(response) if self.crumb: - req.add_header(self.crumb['crumbRequestField'], self.crumb['crumb']) + req.headers[self.crumb['crumbRequestField']] = self.crumb['crumb'] + + def _maybe_add_auth(self): + + if self._auth_resolved: + return + + if len(self._auths) == 1: + # If we only have one auth mechanism specified, just require it + self._session.auth = self._auths[0][1] + else: + # Attempt the list of auth mechanisms and keep the first that works + # otherwise default to the first one in the list (last popped). + # This is a hack to allow the transparent use of kerberos to work + # in future, we should require explicit request to use kerberos + failures = [] + for name, auth in reversed(self._auths): + try: + self.jenkins_open( + requests.Request('GET', self._build_url(INFO), + auth=auth), + add_crumb=False, resolve_auth=False) + self._session.auth = auth + break + except Exception as exc: + # assume authentication failure + failures.append("auth(%s) %s" % (name, exc)) + continue + else: + raise JenkinsException( + 'Unable to authenticate with any scheme:\n%s' + % '\n'.join(failures)) + + self._auth_resolved = True + self.auth = self._session.auth def _add_missing_builds(self, data): """Query Jenkins to get all builds of a job. @@ -327,8 +356,9 @@ class Jenkins(object): if all_builds_loaded: return data folder_url, short_name = self._get_job_folder(data["name"]) - response = self.jenkins_open(Request(self._build_url(ALL_BUILDS, - locals()))) + response = self.jenkins_open(requests.Request( + 'GET', self._build_url(ALL_BUILDS, locals()) + )) if response: data["builds"] = json.loads(response)["allBuilds"] else: @@ -352,8 +382,8 @@ class Jenkins(object): ''' folder_url, short_name = self._get_job_folder(name) try: - response = self.jenkins_open(Request( - self._build_url(JOB_INFO, locals()) + response = self.jenkins_open(requests.Request( + 'GET', self._build_url(JOB_INFO, locals()) )) if response: if fetch_all_builds: @@ -362,7 +392,7 @@ class Jenkins(object): return json.loads(response) else: raise JenkinsException('job[%s] does not exist' % name) - except HTTPError: + except (req_exc.HTTPError, NotFoundException): raise JenkinsException('job[%s] does not exist' % name) except ValueError: raise JenkinsException( @@ -397,8 +427,8 @@ class Jenkins(object): ''' folder_url, short_name = self._get_job_folder(name) try: - response = self.jenkins_open(Request( - self._build_url(JOB_NAME, locals()) + response = self.jenkins_open(requests.Request( + 'GET', self._build_url(JOB_NAME, locals()) )) except NotFoundException: return None @@ -415,40 +445,57 @@ class Jenkins(object): for k, v in self.get_job_info(job_name).items(): print(k, v) - def jenkins_open(self, req, add_crumb=True): + def _response_handler(self, response): + '''Handle response objects''' + + # raise exceptions if occurred + response.raise_for_status() + + headers = response.headers + if (headers.get('content-length') is None and + headers.get('transfer-encoding') is None): + # response body should only exist if one of these is provided + raise EmptyResponseException( + "Error communicating with server[%s]: " + "empty response" % self.server) + + # Reponse objects will automatically return unicode encoded + # when accessing .text property + return response + + def _request(self, req): + + r = self._session.prepare_request(req) + return self._session.send(r, timeout=self.timeout) + + def jenkins_open(self, req, add_crumb=True, resolve_auth=True): '''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) + if resolve_auth: + self._maybe_add_auth() if add_crumb: self.maybe_add_crumb(req) - response = urlopen(req, timeout=self.timeout).read() - if response is None: - raise EmptyResponseException( - "Error communicating with server[%s]: " - "empty response" % self.server) - return response.decode('utf-8') - except HTTPError as e: + + return self._response_handler( + self._request(req)).text + + except req_exc.HTTPError as e: # Jenkins's funky authentication means its nigh impossible to # distinguish errors. - if e.code in [401, 403, 500]: - # six.moves.urllib.error.HTTPError provides a 'reason' - # attribute for all python version except for ver 2.6 - # Falling back to HTTPError.msg since it contains the - # same info as reason + if e.response.status_code in [401, 403, 500]: raise JenkinsException( 'Error in request. ' + 'Possibly authentication failed [%s]: %s' % ( - e.code, e.msg) + e.response.status_code, e.response.reason) ) - elif e.code == 404: + elif e.response.status_code == 404: raise NotFoundException('Requested item could not be found') else: raise - except socket.timeout as e: + except req_exc.Timeout as e: raise TimeoutException('Error in request: %s' % (e)) except URLError as e: # python 2.6 compatibility to ensure same exception raised @@ -476,15 +523,15 @@ class Jenkins(object): ''' folder_url, short_name = self._get_job_folder(name) try: - response = self.jenkins_open(Request( - self._build_url(BUILD_INFO, locals()) + response = self.jenkins_open(requests.Request( + 'GET', self._build_url(BUILD_INFO, locals()) )) if response: return json.loads(response) else: raise JenkinsException('job[%s] number[%d] does not exist' % (name, number)) - except HTTPError: + except (req_exc.HTTPError, NotFoundException): raise JenkinsException('job[%s] number[%d] does not exist' % (name, number)) except ValueError: @@ -502,7 +549,7 @@ class Jenkins(object): {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( - Request(self._build_url(Q_INFO)) + requests.Request('GET', self._build_url(Q_INFO)) ))['items'] def cancel_queue(self, id): @@ -514,8 +561,9 @@ class Jenkins(object): # https://issues.jenkins-ci.org/browse/JENKINS-21311 try: self.jenkins_open( - Request(self._build_url(CANCEL_QUEUE, locals()), b'', - headers={'Referer': self.server})) + requests.Request( + 'POST', self._build_url(CANCEL_QUEUE, locals()), + headers={'Referer': self.server})) except NotFoundException: # Exception is expected; cancel_queue() is a best-effort # mechanism, so ignore it @@ -546,9 +594,9 @@ class Jenkins(object): url += query try: return json.loads(self.jenkins_open( - Request(self._build_url(url)) + requests.Request('GET', self._build_url(INFO)) )) - except (HTTPError, BadStatusLine): + except (req_exc.HTTPError, BadStatusLine): raise BadHTTPException("Error communicating with server[%s]" % self.server) except ValueError: @@ -570,7 +618,9 @@ class Jenkins(object): """ try: - response = self.jenkins_open(Request(self._build_url(WHOAMI_URL))) + response = self.jenkins_open(requests.Request( + 'GET', self._build_url(WHOAMI_URL) + )) if response is None: raise EmptyResponseException( "Error communicating with server[%s]: " @@ -578,7 +628,7 @@ class Jenkins(object): return json.loads(response) - except (HTTPError, BadStatusLine): + except (req_exc.HTTPError, BadStatusLine): raise BadHTTPException("Error communicating with server[%s]" % self.server) @@ -595,21 +645,13 @@ class Jenkins(object): """ try: - request = Request(self._build_url('')) - request.add_header('X-Jenkins', '0.0') - response = urlopen(request, timeout=self.timeout) - if response is None: - raise EmptyResponseException( - "Error communicating with server[%s]: " - "empty response" % self.server) + request = requests.Request('GET', self._build_url('')) + request.headers['X-Jenkins'] = '0.0' + response = self._response_handler(self._request(request)) - if six.PY2: - return response.info().getheader('X-Jenkins') + return response.headers['X-Jenkins'] - if six.PY3: - return response.getheader('X-Jenkins') - - except (HTTPError, BadStatusLine): + except (req_exc.HTTPError, BadStatusLine): raise BadHTTPException("Error communicating with server[%s]" % self.server) @@ -711,8 +753,8 @@ class Jenkins(object): try: plugins_info_json = json.loads(self.jenkins_open( - Request(self._build_url(PLUGIN_INFO, locals())))) - except (HTTPError, BadStatusLine): + requests.Request('GET', self._build_url(PLUGIN_INFO, locals())))) + except (req_exc.HTTPError, BadStatusLine): raise BadHTTPException("Error communicating with server[%s]" % self.server) except ValueError: @@ -758,7 +800,7 @@ class Jenkins(object): """ if view_name: - return self._get_view_jobs(view_name=view_name) + return self._get_view_jobs(name=view_name) else: return self.get_all_jobs(folder_depth=folder_depth) @@ -849,8 +891,9 @@ class Jenkins(object): raise JenkinsException('copy[%s to %s] failed, source and destination ' 'folder must be the same' % (from_name, to_name)) - self.jenkins_open(Request( - self._build_url(COPY_JOB, locals()), b'')) + self.jenkins_open(requests.Request( + 'POST', self._build_url(COPY_JOB, locals()) + )) self.assert_job_exists(to_name, 'create[%s] failed') def rename_job(self, from_name, to_name): @@ -869,8 +912,9 @@ class Jenkins(object): if from_folder_url != to_folder_url: raise JenkinsException('rename[%s to %s] failed, source and destination folder ' 'must be the same' % (from_name, to_name)) - self.jenkins_open(Request( - self._build_url(RENAME_JOB, locals()), b'')) + self.jenkins_open(requests.Request( + 'POST', self._build_url(RENAME_JOB, locals()) + )) self.assert_job_exists(to_name, 'rename[%s] failed') def delete_job(self, name): @@ -879,8 +923,9 @@ class Jenkins(object): :param name: Name of Jenkins job, ``str`` ''' folder_url, short_name = self._get_job_folder(name) - self.jenkins_open(Request( - self._build_url(DELETE_JOB, locals()), b'')) + self.jenkins_open(requests.Request( + 'POST', self._build_url(DELETE_JOB, locals()) + )) if self.job_exists(name): raise JenkinsException('delete[%s] failed' % (name)) @@ -890,8 +935,9 @@ class Jenkins(object): :param name: Name of Jenkins job, ``str`` ''' folder_url, short_name = self._get_job_folder(name) - self.jenkins_open(Request( - self._build_url(ENABLE_JOB, locals()), b'')) + self.jenkins_open(requests.Request( + 'POST', self._build_url(ENABLE_JOB, locals()) + )) def disable_job(self, name): '''Disable Jenkins job. @@ -901,8 +947,9 @@ class Jenkins(object): :param name: Name of Jenkins job, ``str`` ''' folder_url, short_name = self._get_job_folder(name) - self.jenkins_open(Request( - self._build_url(DISABLE_JOB, locals()), b'')) + self.jenkins_open(requests.Request( + 'POST', self._build_url(DISABLE_JOB, locals()) + )) def set_next_build_number(self, name, number): '''Set a job's next build number. @@ -925,9 +972,9 @@ class Jenkins(object): >>> server.set_next_build_number('job_name', next_bn + 50) ''' folder_url, short_name = self._get_job_folder(name) - self.jenkins_open( - Request(self._build_url(SET_JOB_BUILD_NUMBER, locals()), - ("nextBuildNumber=%d" % number).encode('utf-8'))) + self.jenkins_open(requests.Request( + 'POST', self._build_url(SET_JOB_BUILD_NUMBER, locals()), + data=("nextBuildNumber=%d" % number).encode('utf-8'))) def job_exists(self, name): '''Check whether a job exists @@ -985,9 +1032,11 @@ class Jenkins(object): raise JenkinsException('job[%s] already exists' % (name)) try: - self.jenkins_open(Request( - self._build_url(CREATE_JOB, locals()), - config_xml.encode('utf-8'), DEFAULT_HEADERS)) + self.jenkins_open(requests.Request( + 'POST', self._build_url(CREATE_JOB, locals()), + data=config_xml.encode('utf-8'), + headers=DEFAULT_HEADERS + )) except NotFoundException: raise JenkinsException('Cannot create job[%s] because folder ' 'for the job does not exist' % (name)) @@ -1000,7 +1049,7 @@ class Jenkins(object): :returns: job configuration (XML format) ''' folder_url, short_name = self._get_job_folder(name) - request = Request(self._build_url(CONFIG_JOB, locals())) + request = requests.Request('GET', self._build_url(CONFIG_JOB, locals())) return self.jenkins_open(request) def reconfig_job(self, name, config_xml): @@ -1013,8 +1062,11 @@ class Jenkins(object): ''' folder_url, short_name = self._get_job_folder(name) reconfig_url = self._build_url(CONFIG_JOB, locals()) - self.jenkins_open(Request(reconfig_url, config_xml.encode('utf-8'), - DEFAULT_HEADERS)) + self.jenkins_open(requests.Request( + 'POST', reconfig_url, + data=config_xml.encode('utf-8'), + headers=DEFAULT_HEADERS + )) def build_job_url(self, name, parameters=None, token=None): '''Get URL to trigger build job. @@ -1045,8 +1097,8 @@ class Jenkins(object): :param parameters: parameters for job, or ``None``, ``dict`` :param token: Jenkins API token ''' - return self.jenkins_open(Request( - self.build_job_url(name, parameters, token), b'')) + return self.jenkins_open(requests.Request( + 'POST', self.build_job_url(name, parameters, token))) def run_script(self, script): '''Execute a groovy script on the jenkins master. @@ -1062,8 +1114,10 @@ class Jenkins(object): Plugin:mailer, Plugin:jquery, Plugin:antisamy-markup-formatter, Plugin:maven-plugin, Plugin:pam-auth]' ''' - return self.jenkins_open(Request(self._build_url(SCRIPT_TEXT), - "script=".encode('utf-8') + quote(script).encode('utf-8'))) + return self.jenkins_open( + requests.Request( + 'POST', self._build_url(SCRIPT_TEXT), + data="script=".encode('utf-8') + quote(script).encode('utf-8'))) def install_plugin(self, name, include_dependencies=True): '''Install a plugin and its dependencies from the Jenkins public @@ -1105,8 +1159,9 @@ class Jenkins(object): :param number: Jenkins build number for the job, ``int`` ''' folder_url, short_name = self._get_job_folder(name) - self.jenkins_open(Request( - self._build_url(STOP_BUILD, locals()), b'')) + self.jenkins_open(requests.Request( + 'POST', self._build_url(STOP_BUILD, locals()) + )) def get_running_builds(self): '''Return list of running builds. @@ -1165,10 +1220,11 @@ class Jenkins(object): :returns: List of nodes, ``[ { str: str, str: bool} ]`` ''' try: - nodes_data = json.loads(self.jenkins_open(Request(self._build_url(NODE_LIST)))) + nodes_data = json.loads(self.jenkins_open( + requests.Request('GET', self._build_url(NODE_LIST)))) return [{'name': c["displayName"], 'offline': c["offline"]} for c in nodes_data["computer"]] - except (HTTPError, BadStatusLine): + except (req_exc.HTTPError, BadStatusLine): raise BadHTTPException("Error communicating with server[%s]" % self.server) except ValueError: @@ -1183,13 +1239,14 @@ class Jenkins(object): :returns: Dictionary of node info, ``dict`` ''' try: - response = self.jenkins_open(Request( - self._build_url(NODE_INFO, locals()))) + response = self.jenkins_open(requests.Request( + 'GET', self._build_url(NODE_INFO, locals()) + )) if response: return json.loads(response) else: raise JenkinsException('node[%s] does not exist' % name) - except HTTPError: + except (req_exc.HTTPError, NotFoundException): raise JenkinsException('node[%s] does not exist' % name) except ValueError: raise JenkinsException("Could not parse JSON info for node[%s]" @@ -1225,8 +1282,9 @@ class Jenkins(object): :param name: Name of Jenkins node, ``str`` ''' self.get_node_info(name) - self.jenkins_open(Request( - self._build_url(DELETE_NODE, locals()), b'')) + self.jenkins_open(requests.Request( + 'POST', self._build_url(DELETE_NODE, locals()) + )) if self.node_exists(name): raise JenkinsException('delete[%s] failed' % (name)) @@ -1239,8 +1297,9 @@ class Jenkins(object): info = self.get_node_info(name) if info['offline']: return - self.jenkins_open(Request( - self._build_url(TOGGLE_OFFLINE, locals()), b'')) + self.jenkins_open(requests.Request( + 'POST', self._build_url(TOGGLE_OFFLINE, locals()) + )) def enable_node(self, name): '''Enable a node @@ -1251,8 +1310,9 @@ class Jenkins(object): if not info['offline']: return msg = '' - self.jenkins_open(Request( - self._build_url(TOGGLE_OFFLINE, locals()), b'')) + self.jenkins_open(requests.Request( + 'POST', self._build_url(TOGGLE_OFFLINE, locals()) + )) def create_node(self, name, numExecutors=2, nodeDescription=None, remoteFS='/var/lib/jenkins', labels=None, exclusive=False, @@ -1297,9 +1357,9 @@ class Jenkins(object): 'json': json.dumps(inner_params) } - self.jenkins_open(Request( - self._build_url(CREATE_NODE, locals()), - urlencode(params).encode('utf-8'))) + self.jenkins_open(requests.Request( + 'POST', self._build_url(CREATE_NODE, locals()), data=params) + ) self.assert_node_exists(name, 'create[%s] failed') @@ -1309,7 +1369,7 @@ class Jenkins(object): :param name: Jenkins node name, ``str`` ''' get_config_url = self._build_url(CONFIG_NODE, locals()) - return self.jenkins_open(Request(get_config_url)) + return self.jenkins_open(requests.Request('GET', get_config_url)) def reconfig_node(self, name, config_xml): '''Change the configuration for an existing node. @@ -1318,7 +1378,11 @@ class Jenkins(object): :param config_xml: New XML configuration, ``str`` ''' reconfig_url = self._build_url(CONFIG_NODE, locals()) - self.jenkins_open(Request(reconfig_url, config_xml.encode('utf-8'), DEFAULT_HEADERS)) + self.jenkins_open(requests.Request( + 'POST', reconfig_url, + data=config_xml.encode('utf-8'), + headers=DEFAULT_HEADERS + )) def get_build_console_output(self, name, number): '''Get build console text. @@ -1329,15 +1393,15 @@ class Jenkins(object): ''' folder_url, short_name = self._get_job_folder(name) try: - response = self.jenkins_open(Request( - self._build_url(BUILD_CONSOLE_OUTPUT, locals()) + response = self.jenkins_open(requests.Request( + 'GET', self._build_url(BUILD_CONSOLE_OUTPUT, locals()) )) if response: return response else: raise JenkinsException('job[%s] number[%d] does not exist' % (name, number)) - except HTTPError: + except (req_exc.HTTPError, NotFoundException): raise JenkinsException('job[%s] number[%d] does not exist' % (name, number)) @@ -1359,7 +1423,7 @@ class Jenkins(object): return folder_url, short_name - def _get_view_jobs(self, view_name): + def _get_view_jobs(self, name): '''Get list of jobs on the view specified. Each job is a dictionary with 'name', 'url', 'color' and 'fullname' @@ -1375,18 +1439,18 @@ class Jenkins(object): ''' try: - response = self.jenkins_open(Request( - self._build_url(VIEW_JOBS, {u'name': view_name}) + response = self.jenkins_open(requests.Request( + 'GET', self._build_url(VIEW_JOBS, locals()) )) if response: jobs = json.loads(response)['jobs'] else: - raise JenkinsException('view[%s] does not exist' % view_name) - except HTTPError: - raise JenkinsException('view[%s] does not exist' % view_name) + raise JenkinsException('view[%s] does not exist' % name) + except NotFoundException: + raise JenkinsException('view[%s] does not exist' % name) except ValueError: raise JenkinsException( - 'Could not parse JSON info for view[%s]' % view_name) + 'Could not parse JSON info for view[%s]' % name) for job_dict in jobs: job_dict.update({u'fullname': job_dict[u'name']}) @@ -1404,8 +1468,8 @@ class Jenkins(object): :returns: Name of view or None ''' try: - response = self.jenkins_open(Request( - self._build_url(VIEW_NAME, locals()))) + response = self.jenkins_open(requests.Request( + 'GET', self._build_url(VIEW_NAME, locals()))) except NotFoundException: return None else: @@ -1426,7 +1490,7 @@ class Jenkins(object): :throws: :class:`JenkinsException` whenever the view does not exist ''' if not self.view_exists(name): - raise JenkinsException(exception_message % name) + raise NotFoundException(exception_message % name) def view_exists(self, name): '''Check whether a view exists @@ -1451,8 +1515,8 @@ class Jenkins(object): :param name: Name of Jenkins view, ``str`` ''' - self.jenkins_open(Request( - self._build_url(DELETE_VIEW, locals()), b'' + self.jenkins_open(requests.Request( + 'POST', self._build_url(DELETE_VIEW, locals()) )) if self.view_exists(name): raise JenkinsException('delete[%s] failed' % (name)) @@ -1466,9 +1530,11 @@ class Jenkins(object): if self.view_exists(name): raise JenkinsException('view[%s] already exists' % (name)) - self.jenkins_open(Request( - self._build_url(CREATE_VIEW, locals()), - config_xml.encode('utf-8'), DEFAULT_HEADERS)) + self.jenkins_open(requests.Request( + 'POST', self._build_url(CREATE_VIEW, locals()), + data=config_xml.encode('utf-8'), + headers=DEFAULT_HEADERS + )) self.assert_view_exists(name, 'create[%s] failed') def reconfig_view(self, name, config_xml): @@ -1480,8 +1546,11 @@ class Jenkins(object): :param config_xml: New XML configuration, ``str`` ''' reconfig_url = self._build_url(CONFIG_VIEW, locals()) - self.jenkins_open(Request(reconfig_url, config_xml.encode('utf-8'), - DEFAULT_HEADERS)) + self.jenkins_open(requests.Request( + 'POST', reconfig_url, + data=config_xml.encode('utf-8'), + headers=DEFAULT_HEADERS + )) def get_view_config(self, name): '''Get configuration of existing Jenkins view. @@ -1489,7 +1558,7 @@ class Jenkins(object): :param name: Name of Jenkins view, ``str`` :returns: view configuration (XML format) ''' - request = Request(self._build_url(CONFIG_VIEW, locals())) + request = requests.Request('GET', self._build_url(CONFIG_VIEW, locals())) return self.jenkins_open(request) def get_promotion_name(self, name, job_name): @@ -1505,8 +1574,8 @@ class Jenkins(object): ''' folder_url, short_name = self._get_job_folder(job_name) try: - response = self.jenkins_open(Request( - self._build_url(PROMOTION_NAME, locals()))) + response = self.jenkins_open(requests.Request( + 'GET', self._build_url(PROMOTION_NAME, locals()))) except NotFoundException: return None else: @@ -1550,13 +1619,13 @@ class Jenkins(object): ''' folder_url, short_name = self._get_job_folder(job_name) try: - response = self.jenkins_open(Request( - self._build_url(PROMOTION_INFO, locals()))) + response = self.jenkins_open(requests.Request( + 'GET', self._build_url(PROMOTION_INFO, locals()))) if response: return json.loads(response) else: raise JenkinsException('job[%s] does not exist' % job_name) - except HTTPError: + except req_exc.HTTPError: raise JenkinsException('job[%s] does not exist' % job_name) except ValueError: raise JenkinsException("Could not parse JSON info for " @@ -1579,8 +1648,8 @@ class Jenkins(object): :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'' + self.jenkins_open(requests.Request( + 'POST', self._build_url(DELETE_PROMOTION, locals()) )) if self.promotion_exists(name, job_name): raise JenkinsException('delete[%s] from job[%s] failed' % @@ -1598,9 +1667,9 @@ class Jenkins(object): % (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.jenkins_open(requests.Request( + 'POST', self._build_url(CREATE_PROMOTION, locals()), + data=config_xml.encode('utf-8'), headers=DEFAULT_HEADERS)) self.assert_promotion_exists(name, job_name, 'create[%s] at ' 'job[%s] failed') @@ -1615,8 +1684,11 @@ class Jenkins(object): ''' 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)) + self.jenkins_open(requests.Request( + 'POST', reconfig_url, + data=config_xml.encode('utf-8'), + headers=DEFAULT_HEADERS + )) def get_promotion_config(self, name, job_name): '''Get configuration of existing Jenkins promotion. @@ -1626,7 +1698,8 @@ class Jenkins(object): :returns: promotion configuration (XML format) ''' folder_url, short_name = self._get_job_folder(job_name) - request = Request(self._build_url(CONFIG_PROMOTION, locals())) + request = requests.Request( + 'GET', self._build_url(CONFIG_PROMOTION, locals())) return self.jenkins_open(request) def quiet_down(self): @@ -1635,7 +1708,7 @@ class Jenkins(object): No new builds will be started allowing running builds to complete prior to shutdown of the server. ''' - request = Request(self._build_url(QUIET_DOWN)) + request = requests.Request('POST', self._build_url(QUIET_DOWN)) self.jenkins_open(request) info = self.get_info() if not info['quietingDown']: diff --git a/jenkins/urllib_kerb.py b/jenkins/urllib_kerb.py deleted file mode 100644 index 490dc77..0000000 --- a/jenkins/urllib_kerb.py +++ /dev/null @@ -1,121 +0,0 @@ -# Copyright (C) 2015 OpenStack Foundation -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. - -import logging -import re - -import kerberos -from six.moves.urllib import error, request - -logger = logging.getLogger(__name__) - - -class HTTPNegotiateHandler(request.BaseHandler): - handler_order = 490 # before Digest auth - - def __init__(self, max_tries=5): - self.krb_context = None - self.tries = 0 - self.max_tries = max_tries - self.re_extract_auth = re.compile('.*?Negotiate\s*([^,]*)', re.I) - - def http_error_401(self, req, fp, code, msg, headers): - logger.debug("INSIDE http_error_401") - try: - try: - krb_req = self._extract_krb_value(headers) - except ValueError: - # Negotiate header not found or a similar error - # we can't handle this, let the next handler have a go - return None - - if not krb_req: - # First reply from server (no neg value) - self.tries = 0 - krb_req = "" - else: - if self.tries > self.max_tries: - raise error.HTTPError( - req.get_full_url(), 401, "Negotiate auth failed", - headers, None) - - self.tries += 1 - try: - krb_resp = self._krb_response(req.host, krb_req) - - req.add_unredirected_header('Authorization', - "Negotiate %s" % krb_resp) - - resp = self.parent.open(req, timeout=req.timeout) - self._authenticate_server(resp.headers) - return resp - - except kerberos.GSSError as err: - try: - msg = err.args[1][0] - except Exception: - msg = "Negotiate auth failed" - logger.debug(msg) - return None # let the next handler (if any) have a go - - finally: - if self.krb_context is not None: - kerberos.authGSSClientClean(self.krb_context) - self.krb_context = None - - def _krb_response(self, host, krb_val): - logger.debug("INSIDE _krb_response") - - _dummy, self.krb_context = kerberos.authGSSClientInit("HTTP@%s" % host) - kerberos.authGSSClientStep(self.krb_context, krb_val) - response = kerberos.authGSSClientResponse(self.krb_context) - - logger.debug("kerb auth successful") - - return response - - def _authenticate_server(self, headers): - logger.debug("INSIDE _authenticate_server") - try: - val = self._extract_krb_value(headers) - except ValueError: - logger.critical("Server authentication failed." - "Auth value couldn't be extracted from headers.") - return None - if not val: - logger.critical("Server authentication failed." - "Empty 'Negotiate' value.") - return None - - kerberos.authGSSClientStep(self.krb_context, val) - - def _extract_krb_value(self, headers): - logger.debug("INSIDE _extract_krb_value") - header = headers.get('www-authenticate', None) - - if header is None: - msg = "www-authenticate header not found" - logger.debug(msg) - raise ValueError(msg) - - if "negotiate" in header.lower(): - matches = self.re_extract_auth.search(header) - if matches: - return matches.group(1) - else: - return "" - else: - msg = "Negotiate not in www-authenticate header (%s)" % header - logger.debug(msg) - raise ValueError(msg) diff --git a/requirements.txt b/requirements.txt index eb72f18..5c014cc 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,4 @@ six>=1.3.0 pbr>=0.8.2 multi_key_dict +requests diff --git a/test-requirements.txt b/test-requirements.txt index 4d51894..06dd44d 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -1,9 +1,10 @@ coverage>=3.6 hacking>=0.5.6,<0.11 -kerberos>=1.2.4 mock<1.1 unittest2 python-subunit +requests-mock>=1.4.0 +requests-kerberos sphinx>=1.2,<1.3.0 testrepository testscenarios diff --git a/tests/base.py b/tests/base.py index 86a8180..4105bbd 100644 --- a/tests/base.py +++ b/tests/base.py @@ -1,6 +1,5 @@ import sys -from six.moves.urllib.request import build_opener from testscenarios import TestWithScenarios import jenkins @@ -25,7 +24,8 @@ class JenkinsTestBase(TestWithScenarios, unittest.TestCase): def setUp(self): super(JenkinsTestBase, self).setUp() - self.opener = build_opener() + # TODO(darragh) would be useful if this could be mocked + jenkins.requests_kerberos = None self.j = jenkins.Jenkins(self.base_url, 'test', 'test') @@ -35,17 +35,4 @@ class JenkinsTestBase(TestWithScenarios, unittest.TestCase): 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) + req[0][0].prepare() diff --git a/tests/helper.py b/tests/helper.py index 7c00b84..82514db 100644 --- a/tests/helper.py +++ b/tests/helper.py @@ -1,8 +1,11 @@ import functools +import json from multiprocessing import Process from multiprocessing import Queue import traceback +from mock import Mock +import requests from six.moves import socketserver @@ -73,3 +76,31 @@ class NullServer(socketserver.TCPServer): socketserver.TCPServer.__init__( self, server_address, socketserver.BaseRequestHandler, *args, **kwargs) + + +def build_response_mock(status_code, json_body=None, headers=None, **kwargs): + real_response = requests.Response() + real_response.status_code = status_code + + text = None + if json_body is not None: + text = json.dumps(json_body) + if headers is not {}: + real_response.headers['content-length'] = len(text) + + if headers is not None: + for k, v in headers.items(): + real_response.headers[k] = v + + for k, v in kwargs.items(): + setattr(real_response, k, v) + + response = Mock(wraps=real_response, autospec=True) + if text: + response.text = text + + # for some reason, wraps cannot handle attributes which are dicts + # and accessed by key + response.headers = real_response.headers + + return response diff --git a/tests/jobs/test_build.py b/tests/jobs/test_build.py index b437fd7..07e17c4 100644 --- a/tests/jobs/test_build.py +++ b/tests/jobs/test_build.py @@ -14,7 +14,7 @@ class JenkinsBuildJobTest(JenkinsJobsTestBase): build_info = self.j.build_job(u'Test Job') - self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(), + self.assertEqual(jenkins_mock.call_args[0][0].url, self.make_url('job/Test%20Job/build')) self.assertEqual(build_info, {'foo': 'bar'}) self._check_requests(jenkins_mock.call_args_list) @@ -27,7 +27,7 @@ class JenkinsBuildJobTest(JenkinsJobsTestBase): build_info = self.j.build_job(u'a Folder/Test Job') - self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(), + self.assertEqual(jenkins_mock.call_args[0][0].url, self.make_url('job/a%20Folder/job/Test%20Job/build')) self.assertEqual(build_info, {'foo': 'bar'}) self._check_requests(jenkins_mock.call_args_list) @@ -40,7 +40,7 @@ class JenkinsBuildJobTest(JenkinsJobsTestBase): build_info = self.j.build_job(u'TestJob', token='some_token') - self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(), + self.assertEqual(jenkins_mock.call_args[0][0].url, self.make_url('job/TestJob/build?token=some_token')) self.assertEqual(build_info, {'foo': 'bar'}) self._check_requests(jenkins_mock.call_args_list) @@ -53,7 +53,7 @@ class JenkinsBuildJobTest(JenkinsJobsTestBase): build_info = self.j.build_job(u'a Folder/TestJob', token='some_token') - self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(), + self.assertEqual(jenkins_mock.call_args[0][0].url, self.make_url('job/a%20Folder/job/TestJob/build?token=some_token')) self.assertEqual(build_info, {'foo': 'bar'}) self._check_requests(jenkins_mock.call_args_list) @@ -69,8 +69,8 @@ class JenkinsBuildJobTest(JenkinsJobsTestBase): parameters={'when': 'now', 'why': 'because I felt like it'}, token='some_token') - self.assertTrue('token=some_token' in jenkins_mock.call_args[0][0].get_full_url()) - 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.assertTrue('token=some_token' in jenkins_mock.call_args[0][0].url) + self.assertTrue('when=now' in jenkins_mock.call_args[0][0].url) + self.assertTrue('why=because+I+felt+like+it' in jenkins_mock.call_args[0][0].url) self.assertEqual(build_info, {'foo': 'bar'}) self._check_requests(jenkins_mock.call_args_list) diff --git a/tests/jobs/test_config.py b/tests/jobs/test_config.py index 692d6b6..da07ebb 100644 --- a/tests/jobs/test_config.py +++ b/tests/jobs/test_config.py @@ -11,7 +11,7 @@ class JenkinsGetJobConfigTest(JenkinsJobsTestBase): self.j.get_job_config(u'Test Job') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/Test%20Job/config.xml')) self._check_requests(jenkins_mock.call_args_list) @@ -20,6 +20,6 @@ class JenkinsGetJobConfigTest(JenkinsJobsTestBase): self.j.get_job_config(u'a folder/Test Job') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/a%20folder/job/Test%20Job/config.xml')) self._check_requests(jenkins_mock.call_args_list) diff --git a/tests/jobs/test_copy.py b/tests/jobs/test_copy.py index 766339e..6d0c449 100644 --- a/tests/jobs/test_copy.py +++ b/tests/jobs/test_copy.py @@ -18,7 +18,7 @@ class JenkinsCopyJobTest(JenkinsJobsTestBase): self.j.copy_job(u'Test Job', u'Test Job_2') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('createItem?name=Test%20Job_2&mode=copy&from=Test%20Job')) self.assertTrue(self.j.job_exists('Test Job_2')) self._check_requests(jenkins_mock.call_args_list) @@ -34,7 +34,7 @@ class JenkinsCopyJobTest(JenkinsJobsTestBase): self.j.copy_job(u'a Folder/Test Job', u'a Folder/Test Job_2') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/a%20Folder/createItem?name=Test%20Job_2' '&mode=copy&from=Test%20Job')) self.assertTrue(self.j.job_exists('a Folder/Test Job_2')) @@ -50,7 +50,7 @@ class JenkinsCopyJobTest(JenkinsJobsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.copy_job(u'TestJob', u'TestJob_2') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('createItem?name=TestJob_2&mode=copy&from=TestJob')) self.assertEqual( str(context_manager.exception), @@ -67,7 +67,7 @@ class JenkinsCopyJobTest(JenkinsJobsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.copy_job(u'a Folder/TestJob', u'a Folder/TestJob_2') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/a%20Folder/createItem?name=TestJob_2&mode=copy' '&from=TestJob')) self.assertEqual( diff --git a/tests/jobs/test_create.py b/tests/jobs/test_create.py index ea8c2e4..a26ca71 100644 --- a/tests/jobs/test_create.py +++ b/tests/jobs/test_create.py @@ -18,7 +18,7 @@ class JenkinsCreateJobTest(JenkinsJobsTestBase): self.j.create_job(u'Test Job', self.config_xml) self.assertEqual( - jenkins_mock.call_args_list[1][0][0].get_full_url(), + jenkins_mock.call_args_list[1][0][0].url, self.make_url('createItem?name=Test%20Job')) self._check_requests(jenkins_mock.call_args_list) @@ -33,7 +33,7 @@ class JenkinsCreateJobTest(JenkinsJobsTestBase): self.j.create_job(u'a Folder/Test Job', self.config_xml) self.assertEqual( - jenkins_mock.call_args_list[1][0][0].get_full_url(), + jenkins_mock.call_args_list[1][0][0].url, self.make_url('job/a%20Folder/createItem?name=Test%20Job')) self._check_requests(jenkins_mock.call_args_list) @@ -47,7 +47,7 @@ class JenkinsCreateJobTest(JenkinsJobsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.create_job(u'TestJob', self.config_xml) self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/TestJob/api/json?tree=name')) self.assertEqual( str(context_manager.exception), @@ -64,7 +64,7 @@ class JenkinsCreateJobTest(JenkinsJobsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.create_job(u'a Folder/TestJob', self.config_xml) self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/a%20Folder/job/TestJob/api/json?tree=name')) self.assertEqual( str(context_manager.exception), @@ -82,10 +82,10 @@ class JenkinsCreateJobTest(JenkinsJobsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.create_job(u'TestJob', self.config_xml) self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/TestJob/api/json?tree=name')) self.assertEqual( - jenkins_mock.call_args_list[1][0][0].get_full_url(), + jenkins_mock.call_args_list[1][0][0].url, self.make_url('createItem?name=TestJob')) self.assertEqual( str(context_manager.exception), @@ -103,10 +103,10 @@ class JenkinsCreateJobTest(JenkinsJobsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.create_job(u'a Folder/TestJob', self.config_xml) self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/a%20Folder/job/TestJob/api/json?tree=name')) self.assertEqual( - jenkins_mock.call_args_list[1][0][0].get_full_url(), + jenkins_mock.call_args_list[1][0][0].url, self.make_url('job/a%20Folder/createItem?name=TestJob')) self.assertEqual( str(context_manager.exception), diff --git a/tests/jobs/test_debug.py b/tests/jobs/test_debug.py index 397c38b..b081110 100644 --- a/tests/jobs/test_debug.py +++ b/tests/jobs/test_debug.py @@ -20,7 +20,7 @@ class JenkinsDebugJobInfoTest(JenkinsJobsTestBase): self.j.debug_job_info(u'Test Job') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/Test%20Job/api/json?depth=0')) self._check_requests(jenkins_mock.call_args_list) @@ -37,6 +37,6 @@ class JenkinsDebugJobInfoTest(JenkinsJobsTestBase): self.j.debug_job_info(u'a Folder/Test Job') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/a%20Folder/job/Test%20Job/api/json?depth=0')) self._check_requests(jenkins_mock.call_args_list) diff --git a/tests/jobs/test_delete.py b/tests/jobs/test_delete.py index a8a2421..d7fa442 100644 --- a/tests/jobs/test_delete.py +++ b/tests/jobs/test_delete.py @@ -17,7 +17,7 @@ class JenkinsDeleteJobTest(JenkinsJobsTestBase): self.j.delete_job(u'Test Job') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/Test%20Job/doDelete')) self._check_requests(jenkins_mock.call_args_list) @@ -31,7 +31,7 @@ class JenkinsDeleteJobTest(JenkinsJobsTestBase): self.j.delete_job(u'a Folder/Test Job') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/a%20Folder/job/Test%20Job/doDelete')) self._check_requests(jenkins_mock.call_args_list) @@ -46,7 +46,7 @@ class JenkinsDeleteJobTest(JenkinsJobsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.delete_job(u'TestJob') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/TestJob/doDelete')) self.assertEqual( str(context_manager.exception), @@ -64,7 +64,7 @@ class JenkinsDeleteJobTest(JenkinsJobsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.delete_job(u'a Folder/TestJob') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/a%20Folder/job/TestJob/doDelete')) self.assertEqual( str(context_manager.exception), diff --git a/tests/jobs/test_disable.py b/tests/jobs/test_disable.py index c309d39..5170bf8 100644 --- a/tests/jobs/test_disable.py +++ b/tests/jobs/test_disable.py @@ -17,7 +17,7 @@ class JenkinsDisableJobTest(JenkinsJobsTestBase): self.j.disable_job(u'Test Job') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/Test%20Job/disable')) self.assertTrue(self.j.job_exists('Test Job')) self._check_requests(jenkins_mock.call_args_list) @@ -32,7 +32,7 @@ class JenkinsDisableJobTest(JenkinsJobsTestBase): self.j.disable_job(u'a Folder/Test Job') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/a%20Folder/job/Test%20Job/disable')) self.assertTrue(self.j.job_exists('a Folder/Test Job')) self._check_requests(jenkins_mock.call_args_list) diff --git a/tests/jobs/test_enable.py b/tests/jobs/test_enable.py index b04b3cb..a1f6467 100644 --- a/tests/jobs/test_enable.py +++ b/tests/jobs/test_enable.py @@ -17,7 +17,7 @@ class JenkinsEnableJobTest(JenkinsJobsTestBase): self.j.enable_job(u'TestJob') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/TestJob/enable')) self.assertTrue(self.j.job_exists('TestJob')) self._check_requests(jenkins_mock.call_args_list) @@ -32,7 +32,7 @@ class JenkinsEnableJobTest(JenkinsJobsTestBase): self.j.enable_job(u'a Folder/TestJob') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/a%20Folder/job/TestJob/enable')) self.assertTrue(self.j.job_exists('a Folder/TestJob')) self._check_requests(jenkins_mock.call_args_list) diff --git a/tests/jobs/test_get.py b/tests/jobs/test_get.py index 483fd85..4a625c5 100644 --- a/tests/jobs/test_get.py +++ b/tests/jobs/test_get.py @@ -2,6 +2,7 @@ import json from mock import patch import jenkins +from tests.helper import build_response_mock from tests.jobs.base import build_jobs_list_responses from tests.jobs.base import JenkinsGetJobsTestBase @@ -23,8 +24,8 @@ class JenkinsGetJobsTest(JenkinsGetJobsTestBase): jobs[u'fullname'] = jobs[u'name'] self.assertEqual(job_info, [jobs]) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), - self.make_url('api/json?tree=jobs[url,color,name,jobs]')) + jenkins_mock.call_args[0][0].url, + self.make_url('api/json')) self._check_requests(jenkins_mock.call_args_list) @patch.object(jenkins.Jenkins, 'jenkins_open') @@ -79,7 +80,7 @@ class JenkinsGetJobsTest(JenkinsGetJobsTestBase): self.assertEqual(view_jobs[1][u'name'], u'community.first') self.assertEqual(view_jobs[1][u'name'], view_jobs[1][u'fullname']) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url( 'view/Test%20View/api/json?tree=jobs[url,color,name]' )) @@ -93,7 +94,7 @@ class JenkinsGetJobsTest(JenkinsGetJobsTestBase): self.j.get_jobs(view_name=u'Test View') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url( 'view/Test%20View/api/json?tree=jobs[url,color,name]' )) @@ -110,7 +111,7 @@ class JenkinsGetJobsTest(JenkinsGetJobsTestBase): self.j.get_jobs(view_name=u'Test View') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url( 'view/Test%20View/api/json?tree=jobs[url,color,name]' )) @@ -119,25 +120,21 @@ class JenkinsGetJobsTest(JenkinsGetJobsTestBase): 'Could not parse JSON info for view[Test View]') self._check_requests(jenkins_mock.call_args_list) - @patch.object(jenkins.Jenkins, 'jenkins_open') - def test_get_view_jobs_raise_HTTPError(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.HTTPError( - self.make_url( - 'view/Test%20View/api/json?tree=jobs[url,color,name]'), - code=401, - msg="basic auth failed", - hdrs=[], - fp=None) + @patch('jenkins.requests.Session.send', autospec=True) + def test_get_view_jobs_raise_HTTPError(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 + ]) with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_jobs(view_name=u'Test View') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + session_send_mock.call_args_list[1][0][1].url, self.make_url( 'view/Test%20View/api/json?tree=jobs[url,color,name]' )) self.assertEqual( str(context_manager.exception), 'view[Test View] does not exist') - self._check_requests(jenkins_mock.call_args_list) diff --git a/tests/jobs/test_info.py b/tests/jobs/test_info.py index a272758..685f374 100644 --- a/tests/jobs/test_info.py +++ b/tests/jobs/test_info.py @@ -2,6 +2,7 @@ import json from mock import patch import jenkins +from tests.helper import build_response_mock from tests.jobs.base import JenkinsJobsTestBase @@ -21,7 +22,7 @@ class JenkinsGetJobInfoTest(JenkinsJobsTestBase): self.assertEqual(job_info, job_info_to_return) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/Test%20Job/api/json?depth=0')) self._check_requests(jenkins_mock.call_args_list) @@ -48,10 +49,10 @@ class JenkinsGetJobInfoTest(JenkinsJobsTestBase): self.assertEqual(job_info, expected) self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/Test%20Job/api/json?depth=0')) self.assertEqual( - jenkins_mock.call_args_list[1][0][0].get_full_url(), + jenkins_mock.call_args_list[1][0][0].url, self.make_url( 'job/Test%20Job/api/json?tree=allBuilds[number,url]')) self._check_requests(jenkins_mock.call_args_list) @@ -70,7 +71,7 @@ class JenkinsGetJobInfoTest(JenkinsJobsTestBase): self.assertEqual(job_info, job_info_to_return) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/a%20Folder/job/Test%20Job/api/json?depth=0')) self._check_requests(jenkins_mock.call_args_list) @@ -101,7 +102,7 @@ class JenkinsGetJobInfoTest(JenkinsJobsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_job_info(u'TestJob') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/TestJob/api/json?depth=0')) self.assertEqual( str(context_manager.exception), @@ -115,47 +116,41 @@ class JenkinsGetJobInfoTest(JenkinsJobsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_job_info(u'TestJob') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/TestJob/api/json?depth=0')) 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_raise_HTTPError(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.HTTPError( - self.make_url('job/TestJob/api/json?depth=0'), - code=401, - msg="basic auth failed", - hdrs=[], - fp=None) + @patch('jenkins.requests.Session.send', autospec=True) + def test_raise_HTTPError(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 + ]) with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_job_info(u'TestJob') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + session_send_mock.call_args_list[1][0][1].url, self.make_url('job/TestJob/api/json?depth=0')) 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_in_folder_raise_HTTPError(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.HTTPError( - self.make_url('job/a%20Folder/job/TestJob/api/json?depth=0'), - code=401, - msg="basic auth failed", - hdrs=[], - fp=None) + @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(404, reason="Not Found"), # crumb + build_response_mock(404, reason="Not Found"), # request + ]) with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_job_info(u'a Folder/TestJob') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + session_send_mock.call_args_list[1][0][1].url, self.make_url('job/a%20Folder/job/TestJob/api/json?depth=0')) self.assertEqual( str(context_manager.exception), 'job[a Folder/TestJob] does not exist') - self._check_requests(jenkins_mock.call_args_list) diff --git a/tests/jobs/test_name.py b/tests/jobs/test_name.py index 39f5263..652cd2e 100644 --- a/tests/jobs/test_name.py +++ b/tests/jobs/test_name.py @@ -16,7 +16,7 @@ class JenkinsGetJobNameTest(JenkinsJobsTestBase): self.assertEqual(job_name, 'Test Job') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/Test%20Job/api/json?tree=name')) self._check_requests(jenkins_mock.call_args_list) @@ -29,7 +29,7 @@ class JenkinsGetJobNameTest(JenkinsJobsTestBase): self.assertEqual(job_name, 'Test Job') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/a%20Folder/job/Test%20Job/api/json?tree=name')) self._check_requests(jenkins_mock.call_args_list) @@ -41,7 +41,7 @@ class JenkinsGetJobNameTest(JenkinsJobsTestBase): self.assertEqual(job_name, None) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/TestJob/api/json?tree=name')) self._check_requests(jenkins_mock.call_args_list) @@ -53,7 +53,7 @@ class JenkinsGetJobNameTest(JenkinsJobsTestBase): self.assertEqual(job_name, None) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/a%20Folder/job/TestJob/api/json?tree=name')) self._check_requests(jenkins_mock.call_args_list) @@ -65,7 +65,7 @@ class JenkinsGetJobNameTest(JenkinsJobsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_job_name(u'TestJob') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/TestJob/api/json?tree=name')) self.assertEqual( str(context_manager.exception), @@ -81,7 +81,7 @@ class JenkinsGetJobNameTest(JenkinsJobsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_job_name(u'a Folder/TestJob') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/a%20Folder/job/TestJob/api/json?tree=name')) self.assertEqual( str(context_manager.exception), diff --git a/tests/jobs/test_reconfig.py b/tests/jobs/test_reconfig.py index 1bef54f..a7598de 100644 --- a/tests/jobs/test_reconfig.py +++ b/tests/jobs/test_reconfig.py @@ -16,7 +16,7 @@ class JenkinsReconfigJobTest(JenkinsJobsTestBase): self.j.reconfig_job(u'Test Job', self.config_xml) - self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(), + self.assertEqual(jenkins_mock.call_args[0][0].url, self.make_url('job/Test%20Job/config.xml')) self._check_requests(jenkins_mock.call_args_list) @@ -29,6 +29,6 @@ class JenkinsReconfigJobTest(JenkinsJobsTestBase): self.j.reconfig_job(u'a Folder/Test Job', self.config_xml) - self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(), + self.assertEqual(jenkins_mock.call_args[0][0].url, self.make_url('job/a%20Folder/job/Test%20Job/config.xml')) self._check_requests(jenkins_mock.call_args_list) diff --git a/tests/jobs/test_rename.py b/tests/jobs/test_rename.py index 15ac9ae..69a9c02 100644 --- a/tests/jobs/test_rename.py +++ b/tests/jobs/test_rename.py @@ -18,7 +18,7 @@ class JenkinsRenameJobTest(JenkinsJobsTestBase): self.j.rename_job(u'Test Job', u'Test Job_2') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/Test%20Job/doRename?newName=Test%20Job_2')) self.assertTrue(self.j.job_exists('Test Job_2')) self._check_requests(jenkins_mock.call_args_list) @@ -34,7 +34,7 @@ class JenkinsRenameJobTest(JenkinsJobsTestBase): self.j.rename_job(u'a Folder/Test Job', u'a Folder/Test Job_2') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/a%20Folder/job/Test%20Job/doRename?newName=Test%20Job_2')) self.assertTrue(self.j.job_exists('Test Job_2')) self._check_requests(jenkins_mock.call_args_list) @@ -49,7 +49,7 @@ class JenkinsRenameJobTest(JenkinsJobsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.rename_job(u'TestJob', u'TestJob_2') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/TestJob/doRename?newName=TestJob_2')) self.assertEqual( str(context_manager.exception), @@ -66,7 +66,7 @@ class JenkinsRenameJobTest(JenkinsJobsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.rename_job(u'a Folder/TestJob', u'a Folder/TestJob_2') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/a%20Folder/job/TestJob/doRename?newName=TestJob_2')) self.assertEqual( str(context_manager.exception), diff --git a/tests/jobs/test_set_next_build_number.py b/tests/jobs/test_set_next_build_number.py index c87ff86..6bb7920 100644 --- a/tests/jobs/test_set_next_build_number.py +++ b/tests/jobs/test_set_next_build_number.py @@ -11,7 +11,7 @@ class JenkinsSetNextBuildNumberTest(JenkinsJobsTestBase): self.j.set_next_build_number('TestJob', 1234) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/TestJob/nextbuildnumber/submit')) self.assertEqual( jenkins_mock.call_args[0][0].data, diff --git a/tests/test_build.py b/tests/test_build.py index c6d3728..af0ad47 100644 --- a/tests/test_build.py +++ b/tests/test_build.py @@ -3,6 +3,7 @@ from mock import patch import jenkins from tests.base import JenkinsTestBase +from tests.helper import build_response_mock class JenkinsBuildConsoleTest(JenkinsTestBase): @@ -15,7 +16,7 @@ class JenkinsBuildConsoleTest(JenkinsTestBase): self.assertEqual(build_info, jenkins_mock.return_value) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/Test%20Job/52/consoleText')) self._check_requests(jenkins_mock.call_args_list) @@ -27,7 +28,7 @@ class JenkinsBuildConsoleTest(JenkinsTestBase): self.assertEqual(build_info, jenkins_mock.return_value) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/a%20Folder/job/Test%20Job/52/consoleText')) self._check_requests(jenkins_mock.call_args_list) @@ -59,45 +60,38 @@ class JenkinsBuildConsoleTest(JenkinsTestBase): console_output = self.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_raise_HTTPError(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.HTTPError( - self.make_url('job/TestJob/52/consoleText'), - code=401, - msg="basic auth failed", - hdrs=[], - fp=None) + @patch('jenkins.requests.Session.send') + def test_raise_HTTPError(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 + ]) with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_build_console_output(u'TestJob', number=52) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + session_send_mock.call_args_list[1][0][0].url, self.make_url('job/TestJob/52/consoleText')) 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_in_folder_raise_HTTPError(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.HTTPError( - self.make_url('job/a%20Folder/job/TestJob/52/consoleText'), - code=401, - msg="basic auth failed", - hdrs=[], - fp=None) + @patch('jenkins.requests.Session.send') + def test_in_folder_raise_HTTPError(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 + ]) with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_build_console_output(u'a Folder/TestJob', number=52) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + session_send_mock.call_args_list[1][0][0].url, self.make_url('job/a%20Folder/job/TestJob/52/consoleText')) self.assertEqual( str(context_manager.exception), 'job[a Folder/TestJob] number[52] does not exist') - self._check_requests(jenkins_mock.call_args_list) class JenkinsBuildInfoTest(JenkinsTestBase): @@ -116,7 +110,7 @@ class JenkinsBuildInfoTest(JenkinsTestBase): self.assertEqual(build_info, build_info_to_return) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/Test%20Job/52/api/json?depth=0')) self._check_requests(jenkins_mock.call_args_list) @@ -134,7 +128,7 @@ class JenkinsBuildInfoTest(JenkinsTestBase): self.assertEqual(build_info, build_info_to_return) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/a%20Folder/job/Test%20Job/52/api/json?depth=0')) self._check_requests(jenkins_mock.call_args_list) @@ -160,37 +154,31 @@ class JenkinsBuildInfoTest(JenkinsTestBase): '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_raise_HTTPError(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.HTTPError( - self.make_url('job/TestJob/api/json?depth=0'), - code=401, - msg="basic auth failed", - hdrs=[], - fp=None) + @patch('jenkins.requests.Session.send', autospec=True) + def test_raise_HTTPError(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 + ]) with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_build_info(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_in_folder_raise_HTTPError(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.HTTPError( - self.make_url('job/a%20Folder/job/TestJob/api/json?depth=0'), - code=401, - msg="basic auth failed", - hdrs=[], - fp=None) + @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(404, reason="Not Found"), # crumb + build_response_mock(404, reason="Not Found"), # request + ]) with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_build_info(u'a Folder/TestJob', number=52) self.assertEqual( str(context_manager.exception), 'job[a Folder/TestJob] number[52] does not exist') - self._check_requests(jenkins_mock.call_args_list) class JenkinsStopBuildTest(JenkinsTestBase): @@ -200,7 +188,7 @@ class JenkinsStopBuildTest(JenkinsTestBase): self.j.stop_build(u'Test Job', number=52) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/Test%20Job/52/stop')) self._check_requests(jenkins_mock.call_args_list) @@ -210,7 +198,7 @@ class JenkinsStopBuildTest(JenkinsTestBase): self.j.stop_build(u'a Folder/Test Job', number=52) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/a%20Folder/job/Test%20Job/52/stop')) self._check_requests(jenkins_mock.call_args_list) diff --git a/tests/test_info.py b/tests/test_info.py index 5bab926..404e2d2 100644 --- a/tests/test_info.py +++ b/tests/test_info.py @@ -3,6 +3,7 @@ from mock import patch import jenkins from tests.base import JenkinsTestBase +from tests.helper import build_response_mock class JenkinsInfoTest(JenkinsTestBase): @@ -22,28 +23,25 @@ class JenkinsInfoTest(JenkinsTestBase): self.assertEqual(job_info, job_info_to_return) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('api/json')) self._check_requests(jenkins_mock.call_args_list) - @patch.object(jenkins.Jenkins, 'jenkins_open') - def test_raise_HTTPError(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.HTTPError( - self.make_url('job/TestJob/api/json?depth=0'), - code=401, - msg="basic auth failed", - hdrs=[], - fp=None) + @patch('jenkins.requests.Session.send', autospec=True) + def test_raise_HTTPError(self, session_send_mock): + session_send_mock.side_effect = iter([ + build_response_mock(404, reason="Not Found"), # crumb + build_response_mock(499, reason="Unhandled Error"), # request + ]) with self.assertRaises(jenkins.BadHTTPException) as context_manager: self.j.get_info() self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + session_send_mock.call_args_list[1][0][1].url, self.make_url('api/json')) self.assertEqual( str(context_manager.exception), - 'Error communicating with server[{0}/]'.format(self.base_url)) - self._check_requests(jenkins_mock.call_args_list) + 'Error communicating with server[{0}]'.format(self.make_url(''))) @patch.object(jenkins.Jenkins, 'jenkins_open') def test_raise_BadStatusLine(self, jenkins_mock): @@ -52,11 +50,11 @@ class JenkinsInfoTest(JenkinsTestBase): with self.assertRaises(jenkins.BadHTTPException) as context_manager: self.j.get_info() self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('api/json')) self.assertEqual( str(context_manager.exception), - 'Error communicating with server[{0}/]'.format(self.base_url)) + 'Error communicating with server[{0}]'.format(self.make_url(''))) self._check_requests(jenkins_mock.call_args_list) @patch.object(jenkins.Jenkins, 'jenkins_open') @@ -66,26 +64,26 @@ class JenkinsInfoTest(JenkinsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_info() self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('api/json')) self.assertEqual( str(context_manager.exception), - 'Could not parse JSON info for server[{0}/]'.format(self.base_url)) + 'Could not parse JSON info for server[{0}]'.format(self.make_url(''))) self._check_requests(jenkins_mock.call_args_list) @patch.object(jenkins.Jenkins, 'jenkins_open') def test_return_empty_response(self, jenkins_mock): jenkins_mock.side_effect = jenkins.JenkinsException( - "Error communicating with server[{0}/]: empty response". - format(self.base_url)) + "Error communicating with server[{0}]: empty response". + format(self.make_url(''))) with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_info() self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('api/json')) self.assertEqual( str(context_manager.exception), - 'Error communicating with server[{0}/]: ' - 'empty response'.format(self.base_url)) + 'Error communicating with server[{0}]: ' + 'empty response'.format(self.make_url(''))) self._check_requests(jenkins_mock.call_args_list) diff --git a/tests/test_jenkins.py b/tests/test_jenkins.py index 1f972f2..073158a 100644 --- a/tests/test_jenkins.py +++ b/tests/test_jenkins.py @@ -1,12 +1,13 @@ import json import socket -from mock import patch, Mock +from mock import patch import six -from six.moves.urllib.error import HTTPError + +from tests.base import JenkinsTestBase +from tests.helper import build_response_mock import jenkins -from tests.base import JenkinsTestBase def get_mock_urlopen_return_value(a_dict=None): @@ -17,19 +18,28 @@ def get_mock_urlopen_return_value(a_dict=None): class JenkinsConstructorTest(JenkinsTestBase): + def setUp(self): + super(JenkinsConstructorTest, self).setUp() + self.req = jenkins.requests.Request('GET', self.base_url) + self.j._maybe_add_auth() + def test_url_with_trailing_slash(self): self.assertEqual(self.j.server, self.make_url('')) - self.assertEqual(self.j.auth, b'Basic dGVzdDp0ZXN0') + self.assertEqual(self.j.auth(self.req).headers['Authorization'], + 'Basic dGVzdDp0ZXN0') self.assertEqual(self.j.crumb, None) def test_url_without_trailing_slash(self): j = jenkins.Jenkins(self.base_url, 'test', 'test') + j._maybe_add_auth() self.assertEqual(j.server, self.make_url('')) - self.assertEqual(j.auth, b'Basic dGVzdDp0ZXN0') + self.assertEqual(j.auth(self.req).headers['Authorization'], + 'Basic dGVzdDp0ZXN0') self.assertEqual(j.crumb, None) def test_without_user_or_password(self): j = jenkins.Jenkins('{0}'.format(self.base_url)) + j._maybe_add_auth() self.assertEqual(j.server, self.make_url('')) self.assertEqual(j.auth, None) self.assertEqual(j.crumb, None) @@ -38,8 +48,10 @@ class JenkinsConstructorTest(JenkinsTestBase): j = jenkins.Jenkins('{0}'.format(self.base_url), six.u('nonascii'), six.u('\xe9\u20ac')) + j._maybe_add_auth() self.assertEqual(j.server, self.make_url('')) - self.assertEqual(j.auth, b'Basic bm9uYXNjaWk6w6nigqw=') + self.assertEqual(j.auth(self.req).headers['Authorization'], + 'Basic bm9uYXNjaWk6w6nigqw=') self.assertEqual(j.crumb, None) def test_long_user_or_password(self): @@ -47,9 +59,11 @@ class JenkinsConstructorTest(JenkinsTestBase): long_str_b64 = 'YWFh' * 20 j = jenkins.Jenkins('{0}'.format(self.base_url), long_str, long_str) + j._maybe_add_auth() - self.assertNotIn(b"\n", j.auth) - self.assertEqual(j.auth.decode('utf-8'), 'Basic %s' % ( + auth_header = j.auth(self.req).headers['Authorization'] + self.assertNotIn("\n", auth_header) + self.assertEqual(auth_header, 'Basic %s' % ( long_str_b64 + 'Om' + long_str_b64[2:] + 'YQ==')) def test_default_timeout(self): @@ -63,80 +77,74 @@ class JenkinsConstructorTest(JenkinsTestBase): class JenkinsMaybeAddCrumbTest(JenkinsTestBase): - @patch('jenkins.urlopen') - def test_simple(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.NotFoundException() - request = jenkins.Request(self.make_url('job/TestJob')) + @patch('jenkins.requests.Session.send', autospec=True) + def test_simple(self, session_send_mock): + session_send_mock.return_value = build_response_mock( + 404, reason="Not Found") + request = jenkins.requests.Request('http://example.com/job/TestJob') self.j.maybe_add_crumb(request) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + session_send_mock.call_args[0][1].url, self.make_url('crumbIssuer/api/json')) self.assertFalse(self.j.crumb) self.assertFalse('.crumb' in request.headers) - self._check_requests(jenkins_mock.call_args_list) - @patch('jenkins.urlopen') - def test_with_data(self, jenkins_mock): - jenkins_mock.return_value = get_mock_urlopen_return_value(self.crumb_data) - request = jenkins.Request(self.make_url('job/TestJob')) + @patch('jenkins.requests.Session.send', autospec=True) + def test_with_data(self, session_send_mock): + session_send_mock.return_value = build_response_mock( + 200, self.crumb_data) + request = jenkins.requests.Request('GET', 'http://example.com/job/TestJob') self.j.maybe_add_crumb(request) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + session_send_mock.call_args[0][1].url, self.make_url('crumbIssuer/api/json')) self.assertEqual(self.j.crumb, self.crumb_data) self.assertEqual(request.headers['.crumb'], self.crumb_data['crumb']) - self._check_requests(jenkins_mock.call_args_list) @patch.object(jenkins.Jenkins, 'jenkins_open') def test_return_empty_response(self, jenkins_mock): "Don't try to create crumb header from an empty response" jenkins_mock.side_effect = jenkins.EmptyResponseException("empty response") - request = jenkins.Request(self.make_url('job/TestJob')) + request = jenkins.requests.Request('GET', 'http://example.com/job/TestJob') self.j.maybe_add_crumb(request) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('crumbIssuer/api/json')) self.assertFalse(self.j.crumb) self.assertFalse('.crumb' in request.headers) - self._check_requests(jenkins_mock.call_args_list) class JenkinsOpenTest(JenkinsTestBase): - @patch('jenkins.urlopen') - def test_simple(self, jenkins_mock): + @patch('jenkins.requests.Session.send', autospec=True) + def test_simple(self, session_send_mock): data = {'foo': 'bar'} - jenkins_mock.side_effect = [ - get_mock_urlopen_return_value(self.crumb_data), - get_mock_urlopen_return_value(data), - ] - request = jenkins.Request(self.make_url('job/TestJob')) + session_send_mock.side_effect = iter([ + build_response_mock(200, self.crumb_data), + build_response_mock(200, data), + ]) + request = jenkins.requests.Request('GET', self.make_url('job/TestJob')) response = self.j.jenkins_open(request) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + session_send_mock.call_args[0][1].url, self.make_url('job/TestJob')) self.assertEqual(response, json.dumps(data)) self.assertEqual(self.j.crumb, self.crumb_data) self.assertEqual(request.headers['.crumb'], self.crumb_data['crumb']) - self._check_requests(jenkins_mock.call_args_list) - @patch('jenkins.urlopen') - def test_response_403(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.HTTPError( - self.make_url('job/TestJob'), - code=401, - msg="basic auth failed", - hdrs=[], - fp=None) - request = jenkins.Request(self.make_url('job/TestJob')) + @patch('jenkins.requests.Session.send', autospec=True) + def test_response_403(self, session_send_mock): + request = jenkins.requests.Request('GET', self.make_url('job/TestJob')) + session_send_mock.return_value = build_response_mock( + 401, reason="basic auth failed") with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.jenkins_open(request, add_crumb=False) @@ -145,19 +153,14 @@ class JenkinsOpenTest(JenkinsTestBase): 'Error in request. Possibly authentication failed [401]: ' 'basic auth failed') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + session_send_mock.call_args[0][1].url, self.make_url('job/TestJob')) - self._check_requests(jenkins_mock.call_args_list) - @patch('jenkins.urlopen') - def test_response_404(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.HTTPError( - self.make_url('job/TestJob'), - code=404, - msg="basic auth failed", - hdrs=[], - fp=None) - request = jenkins.Request(self.make_url('job/TestJob')) + @patch('jenkins.requests.Session.send', autospec=True) + def test_response_404(self, session_send_mock): + request = jenkins.requests.Request('GET', self.make_url('job/TestJob')) + session_send_mock.return_value = build_response_mock( + 404, reason="basic auth failed") with self.assertRaises(jenkins.NotFoundException) as context_manager: self.j.jenkins_open(request, add_crumb=False) @@ -165,53 +168,46 @@ class JenkinsOpenTest(JenkinsTestBase): str(context_manager.exception), 'Requested item could not be found') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + session_send_mock.call_args[0][1].url, self.make_url('job/TestJob')) - self._check_requests(jenkins_mock.call_args_list) - @patch('jenkins.urlopen') - def test_empty_response(self, jenkins_mock): - jenkins_mock.return_value = Mock(**{'read.return_value': None}) - - request = jenkins.Request(self.make_url('job/TestJob')) + @patch('jenkins.requests.Session.send', autospec=True) + def test_empty_response(self, session_send_mock): + request = jenkins.requests.Request('GET', self.make_url('job/TestJob')) + session_send_mock.return_value = build_response_mock( + 401, reason="basic auth failed") with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.jenkins_open(request, False) self.assertEqual( str(context_manager.exception), - 'Error communicating with server[{0}/]: ' - 'empty response'.format(self.base_url)) + 'Error in request. Possibly authentication failed [401]: ' + 'basic auth failed') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + session_send_mock.call_args[0][1].url, self.make_url('job/TestJob')) - self._check_requests(jenkins_mock.call_args_list) - @patch('jenkins.urlopen') - def test_response_501(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.HTTPError( - self.make_url('job/TestJob'), - code=501, - msg="Not implemented", - hdrs=[], - fp=None) - request = jenkins.Request(self.make_url('job/TestJob')) + @patch('jenkins.requests.Session.send', autospec=True) + def test_response_501(self, session_send_mock): + request = jenkins.requests.Request('GET', self.make_url('job/TestJob')) + session_send_mock.return_value = build_response_mock( + 501, reason="Not implemented") - with self.assertRaises(HTTPError) as context_manager: + with self.assertRaises(jenkins.req_exc.HTTPError) as context_manager: self.j.jenkins_open(request, add_crumb=False) self.assertEqual( str(context_manager.exception), - 'HTTP Error 501: Not implemented') + '501 Server Error: Not implemented for url: None') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + session_send_mock.call_args[0][1].url, self.make_url('job/TestJob')) - self._check_requests(jenkins_mock.call_args_list) - @patch('jenkins.urlopen') - def test_timeout(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.URLError( + @patch('jenkins.requests.Session.send', autospec=True) + def test_timeout(self, session_send_mock): + session_send_mock.side_effect = jenkins.URLError( reason="timed out") j = jenkins.Jenkins(self.make_url(''), 'test', 'test', timeout=1) - request = jenkins.Request(self.make_url('job/TestJob')) + request = jenkins.requests.Request('GET', self.make_url('job/TestJob')) with self.assertRaises(jenkins.JenkinsException) as context_manager: j.jenkins_open(request, add_crumb=False) @@ -219,9 +215,8 @@ class JenkinsOpenTest(JenkinsTestBase): str(context_manager.exception), 'Error in request: timed out') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + session_send_mock.call_args[0][1].url, self.make_url('job/TestJob')) - self._check_requests(jenkins_mock.call_args_list) @patch.object(jenkins.Jenkins, 'jenkins_open', return_value=json.dumps({'mode': 'NORMAL'})) diff --git a/tests/test_jenkins_sockets.py b/tests/test_jenkins_sockets.py index 5377fac..0b30085 100644 --- a/tests/test_jenkins_sockets.py +++ b/tests/test_jenkins_sockets.py @@ -23,8 +23,8 @@ class JenkinsRequestTimeoutTests(testtools.TestCase): def test_jenkins_open_timeout(self): j = jenkins.Jenkins("http://%s:%s" % self.server.server_address, None, None, timeout=0.1) - request = jenkins.Request('http://%s:%s/job/TestJob' % - self.server.server_address) + request = jenkins.requests.Request('GET', 'http://%s:%s/job/TestJob' % + self.server.server_address) # assert our request times out when no response with testtools.ExpectedException(jenkins.TimeoutException): @@ -33,8 +33,8 @@ class JenkinsRequestTimeoutTests(testtools.TestCase): def test_jenkins_open_no_timeout(self): j = jenkins.Jenkins("http://%s:%s" % self.server.server_address, None, None) - request = jenkins.Request('http://%s:%s/job/TestJob' % - self.server.server_address) + request = jenkins.requests.Request('GET', 'http://%s:%s/job/TestJob' % + self.server.server_address) # assert we don't timeout quickly like previous test when # no timeout defined. diff --git a/tests/test_kerberos.py b/tests/test_kerberos.py deleted file mode 100644 index dc73a06..0000000 --- a/tests/test_kerberos.py +++ /dev/null @@ -1,124 +0,0 @@ -import kerberos -assert kerberos # pyflakes -from mock import patch, Mock -from six.moves.urllib.request import Request -import testtools - -from jenkins import urllib_kerb - - -class KerberosTests(testtools.TestCase): - - @patch('kerberos.authGSSClientResponse') - @patch('kerberos.authGSSClientStep') - @patch('kerberos.authGSSClientInit') - @patch('kerberos.authGSSClientClean') - def test_http_error_401_simple(self, clean_mock, init_mock, step_mock, response_mock): - headers_from_server = {'www-authenticate': 'Negotiate xxx'} - - init_mock.side_effect = lambda x: (x, "context") - response_mock.return_value = "foo" - - parent_mock = Mock() - parent_return_mock = Mock() - parent_return_mock.headers = {'www-authenticate': "Negotiate bar"} - parent_mock.open.return_value = parent_return_mock - - request_mock = Mock(spec=self._get_dummy_request()) - h = urllib_kerb.HTTPNegotiateHandler() - h.add_parent(parent_mock) - rv = h.http_error_401(request_mock, "", "", "", headers_from_server) - - init_mock.assert_called() - step_mock.assert_any_call("context", "xxx") - # verify authGSSClientStep was called for response as well - step_mock.assert_any_call("context", "bar") - response_mock.assert_called_with("context") - request_mock.add_unredirected_header.assert_called_with( - 'Authorization', 'Negotiate %s' % "foo") - self.assertEqual(rv, parent_return_mock) - clean_mock.assert_called_with("context") - - @patch('kerberos.authGSSClientResponse') - @patch('kerberos.authGSSClientStep') - @patch('kerberos.authGSSClientInit') - @patch('kerberos.authGSSClientClean') - def test_http_error_401_gsserror(self, clean_mock, init_mock, step_mock, response_mock): - headers_from_server = {'www-authenticate': 'Negotiate xxx'} - - init_mock.side_effect = kerberos.GSSError - - h = urllib_kerb.HTTPNegotiateHandler() - rv = h.http_error_401(Mock(spec=self._get_dummy_request()), "", "", "", - headers_from_server) - self.assertEqual(rv, None) - - @patch('kerberos.authGSSClientResponse') - @patch('kerberos.authGSSClientStep') - @patch('kerberos.authGSSClientInit') - @patch('kerberos.authGSSClientClean') - def test_http_error_401_empty(self, clean_mock, init_mock, step_mock, response_mock): - headers_from_server = {} - - h = urllib_kerb.HTTPNegotiateHandler() - rv = h.http_error_401(Mock(spec=self._get_dummy_request()), "", "", "", - headers_from_server) - self.assertEqual(rv, None) - - @patch('kerberos.authGSSClientResponse') - @patch('kerberos.authGSSClientStep') - @patch('kerberos.authGSSClientInit') - def test_krb_response_simple(self, init_mock, step_mock, response_mock): - response_mock.return_value = "foo" - init_mock.return_value = ("bar", "context") - h = urllib_kerb.HTTPNegotiateHandler() - rv = h._krb_response("host", "xxx") - self.assertEqual(rv, "foo") - - @patch('kerberos.authGSSClientResponse') - @patch('kerberos.authGSSClientStep') - @patch('kerberos.authGSSClientInit') - def test_krb_response_gsserror(self, init_mock, step_mock, response_mock): - response_mock.side_effect = kerberos.GSSError - init_mock.return_value = ("bar", "context") - h = urllib_kerb.HTTPNegotiateHandler() - with testtools.ExpectedException(kerberos.GSSError): - h._krb_response("host", "xxx") - - @patch('kerberos.authGSSClientStep') - def test_authenticate_server_simple(self, step_mock): - headers_from_server = {'www-authenticate': 'Negotiate xxx'} - h = urllib_kerb.HTTPNegotiateHandler() - h.krb_context = "foo" - h._authenticate_server(headers_from_server) - step_mock.assert_called_with("foo", "xxx") - - @patch('kerberos.authGSSClientStep') - def test_authenticate_server_empty(self, step_mock): - headers_from_server = {'www-authenticate': 'Negotiate'} - h = urllib_kerb.HTTPNegotiateHandler() - rv = h._authenticate_server(headers_from_server) - self.assertEqual(rv, None) - - def test_extract_krb_value_simple(self): - headers_from_server = {'www-authenticate': 'Negotiate xxx'} - h = urllib_kerb.HTTPNegotiateHandler() - rv = h._extract_krb_value(headers_from_server) - self.assertEqual(rv, "xxx") - - def test_extract_krb_value_empty(self): - headers_from_server = {} - h = urllib_kerb.HTTPNegotiateHandler() - with testtools.ExpectedException(ValueError): - h._extract_krb_value(headers_from_server) - - def test_extract_krb_value_invalid(self): - headers_from_server = {'www-authenticate': 'Foo-&#@^%:; bar'} - h = urllib_kerb.HTTPNegotiateHandler() - with testtools.ExpectedException(ValueError): - h._extract_krb_value(headers_from_server) - - def _get_dummy_request(self): - r = Request('http://example.com') - r.timeout = 10 - return r diff --git a/tests/test_node.py b/tests/test_node.py index 4e67b90..cf2fb00 100644 --- a/tests/test_node.py +++ b/tests/test_node.py @@ -2,7 +2,9 @@ import json from mock import patch import jenkins +import requests_mock from tests.base import JenkinsTestBase +from tests.helper import build_response_mock class JenkinsNodesTestBase(JenkinsTestBase): @@ -42,41 +44,41 @@ class JenkinsGetNodesTest(JenkinsNodesTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_nodes() self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('computer/api/json')) self.assertEqual( str(context_manager.exception), - 'Could not parse JSON info for server[{0}/]'.format(self.base_url)) + 'Could not parse JSON info for server[{0}]'.format( + self.make_url(''))) self._check_requests(jenkins_mock.call_args_list) - @patch('jenkins.urlopen') - def test_raise_BadStatusLine(self, urlopen_mock): - urlopen_mock.side_effect = jenkins.BadStatusLine('not a valid status line') + @patch('jenkins.requests.Session.send', autospec=True) + def test_raise_BadStatusLine(self, session_send_mock): + session_send_mock.side_effect = jenkins.BadStatusLine( + 'not a valid status line') with self.assertRaises(jenkins.BadHTTPException) as context_manager: self.j.get_nodes() self.assertEqual( str(context_manager.exception), - 'Error communicating with server[{0}/]'.format(self.base_url)) - self._check_requests(urlopen_mock.call_args_list) + 'Error communicating with server[{0}]'.format( + self.make_url(''))) - @patch.object(jenkins.Jenkins, 'jenkins_open') - def test_raise_HTTPError(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.HTTPError( - self.make_url('job/TestJob'), - code=401, - msg="basic auth failed", - hdrs=[], - fp=None) + @patch('jenkins.requests.Session.send', autospec=True) + def test_raise_HTTPError(self, session_send_mock): + session_send_mock.side_effect = iter([ + build_response_mock(404, reason="Not Found"), # crumb + build_response_mock(499, reason="Unhandled Error"), # request + ]) with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_nodes() self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + session_send_mock.call_args_list[1][0][1].url, self.make_url('computer/api/json')) self.assertEqual( str(context_manager.exception), - 'Error communicating with server[{0}/]'.format(self.base_url)) - self._check_requests(jenkins_mock.call_args_list) + 'Error communicating with server[{0}]'.format( + self.make_url(''))) class JenkinsGetNodeInfoTest(JenkinsNodesTestBase): @@ -87,11 +89,11 @@ class JenkinsGetNodeInfoTest(JenkinsNodesTestBase): json.dumps(self.node_info), ] + self._check_requests(jenkins_mock.call_args_list) self.assertEqual(self.j.get_node_info('test node'), self.node_info) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('computer/test%20node/api/json?depth=0')) - self._check_requests(jenkins_mock.call_args_list) @patch.object(jenkins.Jenkins, 'jenkins_open') def test_return_invalid_json(self, jenkins_mock): @@ -101,32 +103,30 @@ class JenkinsGetNodeInfoTest(JenkinsNodesTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_node_info('test_node') + + self._check_requests(jenkins_mock.call_args_list) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('computer/test_node/api/json?depth=0')) 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_raise_HTTPError(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.HTTPError( - self.make_url('job/TestJob'), - code=401, - msg="basic auth failed", - hdrs=[], - fp=None) + @patch('jenkins.requests.Session.send', autospec=True) + def test_raise_HTTPError(self, session_send_mock): + session_send_mock.side_effect = iter([ + build_response_mock(404, reason="Not Found"), # crumb + build_response_mock(499, reason="Unhandled Error"), # request + ]) with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_node_info('test_node') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + session_send_mock.call_args_list[1][0][1].url, self.make_url('computer/test_node/api/json?depth=0')) self.assertEqual( str(context_manager.exception), 'node[test_node] does not exist') - self._check_requests(jenkins_mock.call_args_list) class JenkinsAssertNodeExistsTest(JenkinsNodesTestBase): @@ -137,10 +137,11 @@ class JenkinsAssertNodeExistsTest(JenkinsNodesTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.assert_node_exists('NonExistentNode') + + self._check_requests(jenkins_mock.call_args_list) 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_node_exists(self, jenkins_mock): @@ -164,11 +165,11 @@ class JenkinsDeleteNodeTest(JenkinsNodesTestBase): self.j.delete_node('test node') + self._check_requests(jenkins_mock.call_args_list) self.assertEqual( - jenkins_mock.call_args_list[1][0][0].get_full_url(), + jenkins_mock.call_args_list[1][0][0].url, self.make_url('computer/test%20node/doDelete')) self.assertFalse(self.j.node_exists('test node')) - self._check_requests(jenkins_mock.call_args_list) @patch.object(jenkins.Jenkins, 'jenkins_open') def test_failed(self, jenkins_mock): @@ -180,42 +181,55 @@ class JenkinsDeleteNodeTest(JenkinsNodesTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.delete_node('test_node') + + self._check_requests(jenkins_mock.call_args_list) self.assertEqual( - jenkins_mock.call_args_list[1][0][0].get_full_url(), + jenkins_mock.call_args_list[1][0][0].url, self.make_url('computer/test_node/doDelete')) self.assertEqual( str(context_manager.exception), 'delete[test_node] failed') - self._check_requests(jenkins_mock.call_args_list) class JenkinsCreateNodeTest(JenkinsNodesTestBase): - @patch.object(jenkins.Jenkins, 'jenkins_open') - def test_simple(self, jenkins_mock): - jenkins_mock.side_effect = [ - None, - None, - json.dumps(self.node_info), - json.dumps(self.node_info), - ] + @requests_mock.Mocker() + def test_simple(self, req_mock): + req_mock.get(self.make_url(jenkins.CRUMB_URL)) + req_mock.post(self.make_url(jenkins.CREATE_NODE), status_code=200, + text='success', headers={'content-length': '7'}) + req_mock.get( + self.make_url('computer/test%20node/api/json?depth=0'), + [{'status_code': 404, 'headers': {'content-length': '9'}, + 'text': 'NOT FOUND'}, + {'status_code': 200, 'json': {'displayName': 'test%20node'}, + 'headers': {'content-length': '20'}} + ]) self.j.create_node('test node', exclusive=True) - self.assertEqual( - jenkins_mock.call_args_list[1][0][0].get_full_url().split('?')[0], - self.make_url('computer/doCreateItem')) + actual = req_mock.request_history[2] + self.assertEqual(actual.url, self.make_url('computer/doCreateItem')) + self.assertIn('name=test+node', actual.body) self.assertTrue(self.j.node_exists('test node')) - self._check_requests(jenkins_mock.call_args_list) - @patch.object(jenkins.Jenkins, 'jenkins_open') - def test_urlencode(self, jenkins_mock): - jenkins_mock.side_effect = [ - None, - None, - json.dumps(self.node_info), - json.dumps(self.node_info), - ] + @requests_mock.Mocker() + def test_urlencode(self, req_mock): + # resp 0 (don't care about this succeeding) + req_mock.get(self.make_url(jenkins.CRUMB_URL)) + # resp 2 + req_mock.post(self.make_url(jenkins.CREATE_NODE), status_code=200, + text='success', headers={'content-length': '7'}) + # resp 1 & 3 + req_mock.get( + self.make_url('computer/10.0.0.1%2Btest-node/api/json?depth=0'), + [{'status_code': 404, 'headers': {'content-length': '9'}, + 'text': 'NOT FOUND'}, + {'status_code': 200, + 'json': {'displayName': '10.0.0.1+test-node'}, + 'headers': {'content-length': '20'}} + ]) + params = { 'port': '22', 'username': 'juser', @@ -232,11 +246,10 @@ class JenkinsCreateNodeTest(JenkinsNodesTestBase): launcher=jenkins.LAUNCHER_SSH, launcher_params=params) - actual = jenkins_mock.call_args_list[1][0][0].data.decode('utf-8') + actual = req_mock.request_history[2].body # As python dicts do not guarantee order so the parameters get - # re-ordered when it gets processed by _get_encoded_params(), - # verify sections of the URL with self.assertIn() instead of - # the entire URL + # re-ordered when it gets processed by requests, verify sections + # of the URL with self.assertIn() instead of the entire URL self.assertIn(u'name=10.0.0.1%2Btest-node', actual) self.assertIn(u'type=hudson.slaves.DumbSlave%24DescriptorImpl', actual) self.assertIn(u'username%22%3A+%22juser', actual) @@ -250,20 +263,21 @@ class JenkinsCreateNodeTest(JenkinsNodesTestBase): self.assertIn(u'port%22%3A+%2222', actual) self.assertIn(u'remoteFS%22%3A+%22%2Fhome%2Fjuser', actual) self.assertIn(u'labelString%22%3A+%22precise', actual) - 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(self.node_info), - ] + @requests_mock.Mocker() + def test_already_exists(self, req_mock): + req_mock.get(self.make_url(jenkins.CRUMB_URL)) + req_mock.get( + self.make_url('computer/test_node/api/json?depth=0'), + status_code=200, json=self.node_info, + headers={'content-length': '20'} + ) with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.create_node('test_node') 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_failed(self, jenkins_mock): @@ -277,7 +291,7 @@ class JenkinsCreateNodeTest(JenkinsNodesTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.create_node('test_node') self.assertEqual( - jenkins_mock.call_args_list[1][0][0].get_full_url().split('?')[0], + jenkins_mock.call_args_list[1][0][0].url, self.make_url('computer/doCreateItem')) self.assertEqual( str(context_manager.exception), @@ -297,9 +311,9 @@ class JenkinsEnableNodeTest(JenkinsNodesTestBase): self.j.enable_node('test node') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), - '{0}/computer/test%20node/' - 'toggleOffline?offlineMessage='.format(self.base_url)) + jenkins_mock.call_args[0][0].url, + self.make_url('computer/test%20node/' + 'toggleOffline?offlineMessage=')) jenkins_mock.side_effect = [json.dumps(self.online_node_info)] node_info = self.j.get_node_info('test node') @@ -318,7 +332,7 @@ class JenkinsEnableNodeTest(JenkinsNodesTestBase): # Node was not offline; so enable_node skips toggle # Last call to jenkins was to check status self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('computer/test_node/api/json?depth=0')) jenkins_mock.side_effect = [json.dumps(self.online_node_info)] @@ -339,9 +353,9 @@ class JenkinsDisableNodeTest(JenkinsNodesTestBase): self.j.disable_node('test node') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), - '{0}/computer/test%20node/' - 'toggleOffline?offlineMessage='.format(self.base_url)) + jenkins_mock.call_args[0][0].url, + self.make_url('computer/test%20node/' + 'toggleOffline?offlineMessage=')) jenkins_mock.side_effect = [json.dumps(self.offline_node_info)] node_info = self.j.get_node_info('test node') @@ -360,7 +374,7 @@ class JenkinsDisableNodeTest(JenkinsNodesTestBase): # Node was already offline; so disable_node skips toggle # Last call to jenkins was to check status self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('computer/test_node/api/json?depth=0')) jenkins_mock.side_effect = [json.dumps(self.offline_node_info)] diff --git a/tests/test_plugins.py b/tests/test_plugins.py index de2ae7d..0834687 100644 --- a/tests/test_plugins.py +++ b/tests/test_plugins.py @@ -38,6 +38,7 @@ from testscenarios.scenarios import multiply_scenarios import jenkins from jenkins import plugins from tests.base import JenkinsTestBase +from tests.helper import build_response_mock class JenkinsPluginsBase(JenkinsTestBase): @@ -83,7 +84,7 @@ class JenkinsPluginsInfoTest(JenkinsPluginsBase): plugins_info = self.j.get_plugins_info() self.assertEqual(plugins_info, self.plugin_info_json['plugins']) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('pluginManager/api/json?depth=2')) self._check_requests(jenkins_mock.call_args_list) @@ -103,7 +104,7 @@ class JenkinsPluginsInfoTest(JenkinsPluginsBase): self.j.get_plugins_info(depth=1) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('pluginManager/api/json?depth=1')) self._check_requests(jenkins_mock.call_args_list) @@ -114,7 +115,7 @@ class JenkinsPluginsInfoTest(JenkinsPluginsBase): with self.assertRaises(jenkins.BadHTTPException) as context_manager: self.j.get_plugins_info() self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('pluginManager/api/json?depth=2')) self.assertEqual( str(context_manager.exception), @@ -128,28 +129,23 @@ class JenkinsPluginsInfoTest(JenkinsPluginsBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_plugins_info() self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('pluginManager/api/json?depth=2')) self.assertEqual( str(context_manager.exception), 'Could not parse JSON info for server[{0}/]'.format(self.base_url)) self._check_requests(jenkins_mock.call_args_list) - @patch.object(jenkins.Jenkins, 'jenkins_open') - def test_raise_HTTPError(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.HTTPError( - self.make_url('job/pluginManager/api/json?depth=2'), - code=401, - msg="basic auth failed", - hdrs=[], - fp=None) + @patch('jenkins.requests.Session.send', autospec=True) + def test_raise_HTTPError(self, session_send_mock): + session_send_mock.return_value = build_response_mock( + 499, reason="Unhandled Error") with self.assertRaises(jenkins.BadHTTPException) as context_manager: self.j.get_plugins_info(depth=52) self.assertEqual( str(context_manager.exception), 'Error communicating with server[{0}/]'.format(self.base_url)) - self._check_requests(jenkins_mock.call_args_list) class JenkinsPluginInfoTest(JenkinsPluginsBase): @@ -209,7 +205,7 @@ class JenkinsPluginInfoTest(JenkinsPluginsBase): self.j.get_plugin_info('test', depth=1) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('pluginManager/api/json?depth=1')) self._check_requests(jenkins_mock.call_args_list) @@ -220,7 +216,7 @@ class JenkinsPluginInfoTest(JenkinsPluginsBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_plugin_info('test') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('pluginManager/api/json?depth=2')) self.assertEqual( str(context_manager.exception), @@ -234,28 +230,25 @@ class JenkinsPluginInfoTest(JenkinsPluginsBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_plugin_info('test') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('pluginManager/api/json?depth=2')) self.assertEqual( str(context_manager.exception), 'Could not parse JSON info for server[{0}/]'.format(self.base_url)) self._check_requests(jenkins_mock.call_args_list) - @patch.object(jenkins.Jenkins, 'jenkins_open') - def test_raise_HTTPError(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.HTTPError( - self.make_url('job/pluginManager/api/json?depth=2'), - code=401, - msg="basic auth failed", - hdrs=[], - fp=None) + @patch('jenkins.requests.Session.send', autospec=True) + def test_raise_HTTPError(self, session_send_mock): + session_send_mock.side_effect = iter([ + build_response_mock(404, reason="Not Found"), # crumb + build_response_mock(499, reason="Unhandled Error"), # request + ]) with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_plugin_info(u'TestPlugin', depth=52) self.assertEqual( str(context_manager.exception), 'Error communicating with server[{0}/]'.format(self.base_url)) - self._check_requests(jenkins_mock.call_args_list) class PluginsTestScenarios(JenkinsPluginsBase): diff --git a/tests/test_promotion.py b/tests/test_promotion.py index a9fc425..70b2413 100644 --- a/tests/test_promotion.py +++ b/tests/test_promotion.py @@ -24,7 +24,7 @@ class JenkinsGetPromotionNameTest(JenkinsPromotionsTestBase): self.assertEqual(promotion_name, 'Test Promotion') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/Test%20Job/promotion/process/' 'Test%20Promotion/api/json?tree=name')) self._check_requests(jenkins_mock.call_args_list) @@ -38,7 +38,7 @@ class JenkinsGetPromotionNameTest(JenkinsPromotionsTestBase): self.assertEqual(promotion_name, None) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/Test%20Job/promotion/process/' 'TestPromotion/api/json?tree=name')) self._check_requests(jenkins_mock.call_args_list) @@ -51,7 +51,7 @@ class JenkinsGetPromotionNameTest(JenkinsPromotionsTestBase): 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(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/TestJob/promotion/process/TestPromotion' '/api/json?tree=name')) self.assertEqual( @@ -122,7 +122,7 @@ class JenkinsGetPromotionsTest(JenkinsPromotionsTestBase): self.assertEqual(promotion_info, promotions) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/TestJob/promotion/api/json?depth=0')) self._check_requests(jenkins_mock.call_args_list) @@ -164,7 +164,7 @@ class JenkinsDeletePromotionTest(JenkinsPromotionsTestBase): self.j.delete_promotion(u'Test Promotion', 'TestJob') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/TestJob/promotion/process/' 'Test%20Promotion/doDelete')) self._check_requests(jenkins_mock.call_args_list) @@ -180,7 +180,7 @@ class JenkinsDeletePromotionTest(JenkinsPromotionsTestBase): 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(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/TestJob/promotion/process/' 'TestPromotion/doDelete')) self.assertEqual( @@ -202,7 +202,7 @@ class JenkinsCreatePromotionTest(JenkinsPromotionsTestBase): 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(), + jenkins_mock.call_args_list[1][0][0].url, self.make_url('job/Test%20Job/promotion/' 'createProcess?name=Test%20Promotion')) self._check_requests(jenkins_mock.call_args_list) @@ -218,7 +218,7 @@ class JenkinsCreatePromotionTest(JenkinsPromotionsTestBase): self.j.create_promotion(u'TestPromotion', 'TestJob', self.config_xml) self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('job/TestJob/promotion/process/' 'TestPromotion/api/json?tree=name')) self.assertEqual( @@ -238,11 +238,11 @@ class JenkinsCreatePromotionTest(JenkinsPromotionsTestBase): self.j.create_promotion(u'TestPromotion', 'TestJob', self.config_xml) self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].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(), + jenkins_mock.call_args_list[1][0][0].url, self.make_url('job/TestJob/promotion/' 'createProcess?name=TestPromotion')) self.assertEqual( @@ -263,7 +263,7 @@ class JenkinsReconfigPromotionTest(JenkinsPromotionsTestBase): 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.assertEqual(jenkins_mock.call_args[0][0].url, self.make_url('job/Test%20Job/promotion/process/' 'Test%20Promotion/config.xml')) self._check_requests(jenkins_mock.call_args_list) @@ -276,7 +276,7 @@ class JenkinsGetPromotionConfigTest(JenkinsPromotionsTestBase): self.j.get_promotion_config(u'Test Promotion', u'Test Job') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('job/Test%20Job/promotion/process/' 'Test%20Promotion/config.xml')) self._check_requests(jenkins_mock.call_args_list) diff --git a/tests/test_queue.py b/tests/test_queue.py index cd7157f..a40d20c 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -15,7 +15,7 @@ class JenkinsCancelQueueTest(JenkinsTestBase): self.j.cancel_queue(52) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('queue/cancelItem?id=52')) self._check_requests(jenkins_mock.call_args_list) @@ -28,7 +28,7 @@ class JenkinsCancelQueueTest(JenkinsTestBase): self.j.cancel_queue(52) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('queue/cancelItem?id=52')) self._check_requests(jenkins_mock.call_args_list) @@ -67,6 +67,6 @@ class JenkinsQueueInfoTest(JenkinsTestBase): self.assertEqual(queue_info, queue_info_to_return['items']) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('queue/api/json?depth=0')) self._check_requests(jenkins_mock.call_args_list) diff --git a/tests/test_quiet_down.py b/tests/test_quiet_down.py index f4e2818..e5158e9 100644 --- a/tests/test_quiet_down.py +++ b/tests/test_quiet_down.py @@ -16,10 +16,10 @@ class JenkinsQuietDownTest(JenkinsTestBase): self.j.quiet_down() self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('quietDown')) self.assertEqual( - jenkins_mock.call_args_list[1][0][0].get_full_url(), + jenkins_mock.call_args_list[1][0][0].url, self.make_url('api/json')) self._check_requests(jenkins_mock.call_args_list) @@ -34,10 +34,10 @@ class JenkinsQuietDownTest(JenkinsTestBase): self.j.quiet_down() self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('quietDown')) self.assertEqual( - jenkins_mock.call_args_list[1][0][0].get_full_url(), + jenkins_mock.call_args_list[1][0][0].url, self.make_url('api/json')) self.assertEqual( str(context_manager.exception), @@ -54,6 +54,6 @@ class JenkinsQuietDownTest(JenkinsTestBase): self.j.quiet_down() self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('quietDown')) self._check_requests(jenkins_mock.call_args_list) diff --git a/tests/test_script.py b/tests/test_script.py index 616c5ff..c73bcd6 100644 --- a/tests/test_script.py +++ b/tests/test_script.py @@ -12,7 +12,7 @@ class JenkinsScriptTest(JenkinsTestBase): self.j.run_script(u'println(\"Hello World!\")') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('scriptText')) self._check_requests(jenkins_mock.call_args_list) @@ -21,7 +21,7 @@ class JenkinsScriptTest(JenkinsTestBase): self.j.run_script(u'if (a == b && c ==d) { println(\"Yes\")}') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('scriptText')) self.assertIn(quote('&&'), jenkins_mock.call_args[0][0].data.decode('utf8')) self._check_requests(jenkins_mock.call_args_list) @@ -33,7 +33,7 @@ class JenkinsScriptTest(JenkinsTestBase): j = jenkins.Jenkins(self.make_url(''), 'test', 'test') j.install_plugin("jabber") self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('scriptText')) self._check_requests(jenkins_mock.call_args_list) diff --git a/tests/test_version.py b/tests/test_version.py index 13e35ee..67efe92 100644 --- a/tests/test_version.py +++ b/tests/test_version.py @@ -1,57 +1,46 @@ -from mock import patch, Mock -import six +from mock import patch import jenkins from tests.base import JenkinsTestBase +from tests.helper import build_response_mock class JenkinsVersionTest(JenkinsTestBase): - @patch('jenkins.urlopen') - def test_some_version(self, urlopen_mock): - mock_response = Mock() - if six.PY2: - config = {'info.return_value.getheader.return_value': 'Version42'} - - if six.PY3: - config = {'getheader.return_value': 'Version42'} - - mock_response.configure_mock(**config) - urlopen_mock.side_effect = [mock_response] + @patch('jenkins.requests.Session.send', autospec=True) + def test_some_version(self, session_send_mock): + session_send_mock.return_value = build_response_mock( + 200, headers={'X-Jenkins': 'Version42', 'Content-Length': 0}) self.assertEqual(self.j.get_version(), 'Version42') - self._check_requests(urlopen_mock.call_args_list) - @patch('jenkins.urlopen') - def test_raise_HTTPError(self, urlopen_mock): - urlopen_mock.side_effect = jenkins.HTTPError( - self.make_url(''), - code=503, - msg="internal server error", - hdrs=[], - fp=None) + @patch('jenkins.requests.Session.send', autospec=True) + def test_raise_HTTPError(self, session_send_mock): + session_send_mock.side_effect = iter([ + build_response_mock(404, reason="Not Found"), # crumb + build_response_mock(499, reason="Unhandled Error"), # request + ]) + with self.assertRaises(jenkins.BadHTTPException) as context_manager: self.j.get_version() self.assertEqual( str(context_manager.exception), 'Error communicating with server[{0}/]'.format(self.base_url)) - self._check_requests(urlopen_mock.call_args_list) - @patch('jenkins.urlopen') - def test_raise_BadStatusLine(self, urlopen_mock): - urlopen_mock.side_effect = jenkins.BadStatusLine('not a valid status line') + @patch('jenkins.requests.Session.send', autospec=True) + def test_raise_BadStatusLine(self, session_send_mock): + session_send_mock.side_effect = jenkins.BadStatusLine('not a valid status line') with self.assertRaises(jenkins.BadHTTPException) as context_manager: self.j.get_version() self.assertEqual( str(context_manager.exception), 'Error communicating with server[{0}/]'.format(self.base_url)) - self._check_requests(urlopen_mock.call_args_list) - @patch('jenkins.urlopen', return_value=None) - def test_return_empty_response(self, urlopen_mock): + @patch('jenkins.requests.Session.send', autospec=True) + def test_return_empty_response(self, session_send_mock): + session_send_mock.return_value = build_response_mock(0) with self.assertRaises(jenkins.EmptyResponseException) as context_manager: self.j.get_version() self.assertEqual( str(context_manager.exception), 'Error communicating with server[{0}/]:' ' empty response'.format(self.base_url)) - self._check_requests(urlopen_mock.call_args_list) diff --git a/tests/test_view.py b/tests/test_view.py index 45b2587..183c129 100644 --- a/tests/test_view.py +++ b/tests/test_view.py @@ -24,7 +24,7 @@ class JenkinsGetViewNameTest(JenkinsViewsTestBase): self.assertEqual(view_name, 'Test View') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('view/Test%20View/api/json?tree=name')) self._check_requests(jenkins_mock.call_args_list) @@ -36,7 +36,7 @@ class JenkinsGetViewNameTest(JenkinsViewsTestBase): self.assertEqual(view_name, None) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('view/TestView/api/json?tree=name')) self._check_requests(jenkins_mock.call_args_list) @@ -48,7 +48,7 @@ class JenkinsGetViewNameTest(JenkinsViewsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.get_view_name(u'TestView') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('view/TestView/api/json?tree=name')) self.assertEqual( str(context_manager.exception), @@ -94,7 +94,7 @@ class JenkinsGetViewsTest(JenkinsViewsTestBase): self.assertEqual(view_info, views) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('api/json')) self._check_requests(jenkins_mock.call_args_list) @@ -111,7 +111,7 @@ class JenkinsDeleteViewTest(JenkinsViewsTestBase): self.j.delete_view(u'Test View') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('view/Test%20View/doDelete')) self._check_requests(jenkins_mock.call_args_list) @@ -126,7 +126,7 @@ class JenkinsDeleteViewTest(JenkinsViewsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.delete_view(u'TestView') self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('view/TestView/doDelete')) self.assertEqual( str(context_manager.exception), @@ -147,7 +147,7 @@ class JenkinsCreateViewTest(JenkinsViewsTestBase): self.j.create_view(u'Test View', self.config_xml) self.assertEqual( - jenkins_mock.call_args_list[1][0][0].get_full_url(), + jenkins_mock.call_args_list[1][0][0].url, self.make_url('createView?name=Test%20View')) self._check_requests(jenkins_mock.call_args_list) @@ -161,7 +161,7 @@ class JenkinsCreateViewTest(JenkinsViewsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.create_view(u'TestView', self.config_xml) self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('view/TestView/api/json?tree=name')) self.assertEqual( str(context_manager.exception), @@ -179,10 +179,10 @@ class JenkinsCreateViewTest(JenkinsViewsTestBase): with self.assertRaises(jenkins.JenkinsException) as context_manager: self.j.create_view(u'TestView', self.config_xml) self.assertEqual( - jenkins_mock.call_args_list[0][0][0].get_full_url(), + jenkins_mock.call_args_list[0][0][0].url, self.make_url('view/TestView/api/json?tree=name')) self.assertEqual( - jenkins_mock.call_args_list[1][0][0].get_full_url(), + jenkins_mock.call_args_list[1][0][0].url, self.make_url('createView?name=TestView')) self.assertEqual( str(context_manager.exception), @@ -201,7 +201,7 @@ class JenkinsReconfigViewTest(JenkinsViewsTestBase): self.j.reconfig_view(u'Test View', self.config_xml) - self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(), + self.assertEqual(jenkins_mock.call_args[0][0].url, self.make_url('view/Test%20View/config.xml')) self._check_requests(jenkins_mock.call_args_list) @@ -213,6 +213,6 @@ class JenkinsGetViewConfigTest(JenkinsViewsTestBase): self.j.get_view_config(u'Test View') self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('view/Test%20View/config.xml')) self._check_requests(jenkins_mock.call_args_list) diff --git a/tests/test_whoami.py b/tests/test_whoami.py index e449011..16eb782 100644 --- a/tests/test_whoami.py +++ b/tests/test_whoami.py @@ -3,6 +3,7 @@ from mock import patch import jenkins from tests.base import JenkinsTestBase +from tests.helper import build_response_mock class JenkinsWhoamiTest(JenkinsTestBase): @@ -29,22 +30,19 @@ class JenkinsWhoamiTest(JenkinsTestBase): self.assertEqual(user, user_to_return) self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + jenkins_mock.call_args[0][0].url, self.make_url('me/api/json')) self._check_requests(jenkins_mock.call_args_list) - @patch.object(jenkins.Jenkins, 'jenkins_open') - def test_raise_HTTPError(self, jenkins_mock): - jenkins_mock.side_effect = jenkins.HTTPError( - self.make_url('me/api/json'), - code=401, - msg='basic auth failed', - hdrs=[], - fp=None) + @patch('jenkins.requests.Session.send', autospec=True) + def test_raise_HTTPError(self, session_send_mock): + session_send_mock.side_effect = iter([ + build_response_mock(404, reason="Not Found"), # crumb + build_response_mock(401, reason="Basic Auth Failed"), # request + ]) with self.assertRaises(jenkins.JenkinsException): self.j.get_whoami() self.assertEqual( - jenkins_mock.call_args[0][0].get_full_url(), + session_send_mock.call_args_list[1][0][1].url, self.make_url('me/api/json')) - self._check_requests(jenkins_mock.call_args_list)