jjb/python-jenkins
Code Issues Proposed changes

detect and respect http redirects

Browse Source 
Detects if API url is redirected and corrects it in order
to avoid doubling number of requests and even failing
POST ones because of having them silently converted by
requests from POST to GET.

Displays a warning to the user when redirection occurs
because almost always this means that they are using an
old endpoint, like a non secured one.

Change-Id: I7387bf150dad307342f9a6a91afbae32859bc82e
Signed-off-by: Sorin Sbarnea <ssbarnea@redhat.com>
This commit is contained in:
Sorin Sbarnea 2018-06-15 19:28:42 +01:00
parent 4e64f0f880
commit 4150a83d45
No known key found for this signature in database
GPG Key ID: B85725D917D27B8A
6 changed files with 45 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
jenkins/__init__.py
View File

@ -287,7 +287,7 @@ class Jenkins(object):
_timeout_warning_issued = False
def __init__(self, url, username=None, password=None,
timeout=socket._GLOBAL_DEFAULT_TIMEOUT):
timeout=socket._GLOBAL_DEFAULT_TIMEOUT, resolve=True):
'''Create handle to Jenkins instance.
All methods will raise :class:`JenkinsException` on failure.
@ -296,6 +296,7 @@ class Jenkins(object):
:param username: Server username, ``str``
:param password: Server password, ``str``
:param timeout: Server connection timeout in secs (default: not set), ``int``
:param resolve: Attempts to resolve and auto-correct API redirection. default: True ``bool``
'''
if url[-1] == '/':
self.server = url
@ -328,6 +329,25 @@ class Jenkins(object):
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
self._session.verify = False
if resolve:
self._resolve_api()
def _resolve_api(self):
'''Detects if Jenkins api frontend is redirect and corrects it if so
in order to avoid future redirects of each request or failures caused
by the fact that a redirected POST is transformed into a GET.
'''
r = self.jenkins_request(requests.Request('HEAD', self.server), add_crumb=False)
if r.url != self.server or r.status_code in [300, 301, 302, 303]:
if 'Location' in r.headers:
new_url = r.headers['Location']
else:
new_url = r.url
warnings.warn(
"Redirection from %s to %s detected, you may want to update your frontend url." % (
self.server, new_url))
self.server = new_url
def _get_encoded_params(self, params):
for k, v in params.items():
if k in ["name", "msg", "short_name", "from_short_name",

2
tests/base.py
View File

@ -30,7 +30,7 @@ class JenkinsTestBase(TestWithScenarios, unittest.TestCase):
'jenkins.requests_kerberos', None)
self.request_kerberos_module_patcher.start()
self.j = jenkins.Jenkins(self.base_url, 'test', 'test')
self.j = jenkins.Jenkins(self.base_url, 'test', 'test', resolve=False)
def tearDown(self):

23
tests/test_jenkins.py
View File

@ -30,7 +30,7 @@ class JenkinsConstructorTest(JenkinsTestBase):
self.assertEqual(self.j.crumb, None)
def test_url_without_trailing_slash(self):
j = jenkins.Jenkins(self.base_url, 'test', 'test')
j = jenkins.Jenkins(self.base_url, 'test', 'test', resolve=False)
j._maybe_add_auth()
self.assertEqual(j.server, self.make_url(''))
self.assertEqual(j.auth(self.req).headers['Authorization'],
@ -38,7 +38,7 @@ class JenkinsConstructorTest(JenkinsTestBase):
self.assertEqual(j.crumb, None)
def test_without_user_or_password(self):
j = jenkins.Jenkins('{0}'.format(self.base_url))
j = jenkins.Jenkins('{0}'.format(self.base_url), resolve=False)
j._maybe_add_auth()
self.assertEqual(j.server, self.make_url(''))
self.assertEqual(j.auth, None)
@ -47,7 +47,8 @@ class JenkinsConstructorTest(JenkinsTestBase):
def test_unicode_password(self):
j = jenkins.Jenkins('{0}'.format(self.base_url),
six.u('nonascii'),
six.u('\xe9\u20ac'))
six.u('\xe9\u20ac'),
resolve=False)
j._maybe_add_auth()
self.assertEqual(j.server, self.make_url(''))
self.assertEqual(j.auth(self.req).headers['Authorization'],
@ -58,7 +59,7 @@ class JenkinsConstructorTest(JenkinsTestBase):
long_str = 'a' * 60
long_str_b64 = 'YWFh' * 20
j = jenkins.Jenkins('{0}'.format(self.base_url), long_str, long_str)
j = jenkins.Jenkins('{0}'.format(self.base_url), long_str, long_str, resolve=False)
j._maybe_add_auth()
auth_header = j.auth(self.req).headers['Authorization']
@ -67,11 +68,11 @@ class JenkinsConstructorTest(JenkinsTestBase):
long_str_b64 + 'Om' + long_str_b64[2:] + 'YQ=='))
def test_default_timeout(self):
j = jenkins.Jenkins('{0}'.format(self.base_url))
j = jenkins.Jenkins('{0}'.format(self.base_url), resolve=False)
self.assertEqual(j.timeout, socket._GLOBAL_DEFAULT_TIMEOUT)
def test_custom_timeout(self):
j = jenkins.Jenkins('{0}'.format(self.base_url), timeout=300)
j = jenkins.Jenkins('{0}'.format(self.base_url), timeout=300, resolve=False)
self.assertEqual(j.timeout, 300)
@ -206,7 +207,9 @@ class JenkinsOpenTest(JenkinsTestBase):
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)
j = jenkins.Jenkins(self.make_url(''), 'test', 'test',
timeout=1,
resolve=False)
request = jenkins.requests.Request('GET', self.make_url('job/TestJob'))
with self.assertRaises(jenkins.JenkinsException) as context_manager:
@ -223,7 +226,7 @@ class JenkinsOpenTest(JenkinsTestBase):
@patch.object(jenkins.Jenkins, 'get_version',
return_value="Version42")
def test_wait_for_normal_op(self, version_mock, jenkins_mock):
j = jenkins.Jenkins('http://example.com', 'test', 'test')
j = jenkins.Jenkins('http://example.com', 'test', 'test', resolve=False)
self.assertTrue(j.wait_for_normal_op(0))
@patch.object(jenkins.Jenkins, 'jenkins_open',
@ -231,11 +234,11 @@ class JenkinsOpenTest(JenkinsTestBase):
@patch.object(jenkins.Jenkins, 'get_version',
side_effect=jenkins.EmptyResponseException())
def test_wait_for_normal_op__empty_response(self, version_mock, jenkins_mock):
j = jenkins.Jenkins('http://example.com', 'test', 'test')
j = jenkins.Jenkins('http://example.com', 'test', 'test', resolve=False)
self.assertFalse(j.wait_for_normal_op(0))
def test_wait_for_normal_op__negative_timeout(self):
j = jenkins.Jenkins('http://example.com', 'test', 'test')
j = jenkins.Jenkins('http://example.com', 'test', 'test', resolve=False)
with self.assertRaises(ValueError) as context_manager:
j.wait_for_normal_op(-1)
self.assertEqual(

4
tests/test_jenkins_sockets.py
View File

@ -22,7 +22,7 @@ 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)
None, None, timeout=0.1, resolve=False)
request = jenkins.requests.Request('GET', 'http://%s:%s/job/TestJob' %
self.server.server_address)
@ -32,7 +32,7 @@ class JenkinsRequestTimeoutTests(testtools.TestCase):
def test_jenkins_open_no_timeout(self):
j = jenkins.Jenkins("http://%s:%s" % self.server.server_address,
None, None)
None, None, resolve=False)
request = jenkins.requests.Request('GET', 'http://%s:%s/job/TestJob' %
self.server.server_address)

6
tests/test_plugins.py
View File

@ -175,7 +175,7 @@ class JenkinsPluginInfoTest(JenkinsPluginsBase):
json.dumps(self.plugin_info_json),
json.dumps(self.updated_plugin_info_json)
]
j = jenkins.Jenkins(self.make_url(''), 'test', 'test')
j = jenkins.Jenkins(self.make_url(''), 'test', 'test', resolve=False)
plugins_info = j.get_plugins()
self.assertEqual(plugins_info["mailer"]["version"],
@ -289,7 +289,7 @@ class PluginsTestScenarios(JenkinsPluginsBase):
equality operator defined for the scenario.
"""
plugin_name = "Jenkins Mailer Plugin"
j = jenkins.Jenkins(self.base_url, 'test', 'test')
j = jenkins.Jenkins(self.base_url, 'test', 'test', resolve=False)
plugin_info = j.get_plugins()[plugin_name]
v1 = plugin_info.get("version")
@ -307,7 +307,7 @@ class PluginsTestScenarios(JenkinsPluginsBase):
type of PluginVersion before comparing provides the same result.
"""
plugin_name = "Jenkins Mailer Plugin"
j = jenkins.Jenkins(self.base_url, 'test', 'test')
j = jenkins.Jenkins(self.base_url, 'test', 'test', resolve=False)
plugin_info = j.get_plugins()[plugin_name]
v1 = plugin_info.get("version")

10
tests/test_script.py
View File

@ -37,7 +37,7 @@ class JenkinsScriptTest(JenkinsTestBase):
def test_install_plugin(self, jenkins_mock):
'''Installation of plugins is done with the run_script method
'''
j = jenkins.Jenkins(self.make_url(''), 'test', 'test')
j = jenkins.Jenkins(self.make_url(''), 'test', 'test', resolve=False)
j.install_plugin("jabber")
self.assertEqual(
jenkins_mock.call_args[0][0].url,
@ -49,7 +49,7 @@ class JenkinsScriptTest(JenkinsTestBase):
def test_install_plugin_with_dependencies(self, run_script_mock, jenkins_mock):
'''Verify install plugins with dependencies
'''
j = jenkins.Jenkins(self.make_url(''), 'test', 'test')
j = jenkins.Jenkins(self.make_url(''), 'test', 'test', resolve=False)
j.install_plugin("jabber")
self.assertEqual(len(run_script_mock.call_args_list), 2)
self.assertEqual(run_script_mock.call_args_list[0][0][0],
@ -65,7 +65,7 @@ class JenkinsScriptTest(JenkinsTestBase):
def test_install_plugin_without_dependencies(self, run_script_mock, jenkins_mock):
'''Verify install plugins without dependencies
'''
j = jenkins.Jenkins(self.make_url(''), 'test', 'test')
j = jenkins.Jenkins(self.make_url(''), 'test', 'test', resolve=False)
j.install_plugin("jabber", include_dependencies=False)
self.assertEqual(len(run_script_mock.call_args_list), 2)
self.assertEqual(run_script_mock.call_args_list[0][0][0],
@ -81,7 +81,7 @@ class JenkinsScriptTest(JenkinsTestBase):
'''Verify install plugin does not need a restart
'''
run_script_mock.return_value = u'Result: false\n'
j = jenkins.Jenkins(self.make_url(''), 'test', 'test')
j = jenkins.Jenkins(self.make_url(''), 'test', 'test', resolve=False)
self.assertFalse(j.install_plugin("jabber"))
@patch.object(jenkins.Jenkins, 'jenkins_open')
@ -90,5 +90,5 @@ class JenkinsScriptTest(JenkinsTestBase):
'''Verify install plugin needs a restart
'''
run_script_mock.return_value = u'Result: true\n'
j = jenkins.Jenkins(self.make_url(''), 'test', 'test')
j = jenkins.Jenkins(self.make_url(''), 'test', 'test', resolve=False)
self.assertTrue(j.install_plugin("jabber"))