225 lines
8.3 KiB
Python
225 lines
8.3 KiB
Python
import json
|
|
import socket
|
|
|
|
from mock import patch, Mock
|
|
import six
|
|
from six.moves.urllib.error import HTTPError
|
|
|
|
import jenkins
|
|
from tests.base import JenkinsTestBase
|
|
|
|
|
|
def get_mock_urlopen_return_value(a_dict=None):
|
|
if a_dict is None:
|
|
a_dict = {}
|
|
return six.BytesIO(json.dumps(a_dict).encode('utf-8'))
|
|
|
|
|
|
class JenkinsConstructorTest(JenkinsTestBase):
|
|
|
|
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.crumb, None)
|
|
|
|
def test_url_without_trailing_slash(self):
|
|
j = jenkins.Jenkins(self.base_url, 'test', 'test')
|
|
self.assertEqual(j.server, self.make_url(''))
|
|
self.assertEqual(j.auth, b'Basic dGVzdDp0ZXN0')
|
|
self.assertEqual(j.crumb, None)
|
|
|
|
def test_without_user_or_password(self):
|
|
j = jenkins.Jenkins('{0}'.format(self.base_url))
|
|
self.assertEqual(j.server, self.make_url(''))
|
|
self.assertEqual(j.auth, None)
|
|
self.assertEqual(j.crumb, None)
|
|
|
|
def test_unicode_password(self):
|
|
j = jenkins.Jenkins('{0}'.format(self.base_url),
|
|
six.u('nonascii'),
|
|
six.u('\xe9\u20ac'))
|
|
self.assertEqual(j.server, self.make_url(''))
|
|
self.assertEqual(j.auth, b'Basic bm9uYXNjaWk6w6nigqw=')
|
|
self.assertEqual(j.crumb, None)
|
|
|
|
def test_long_user_or_password(self):
|
|
long_str = 'a' * 60
|
|
long_str_b64 = 'YWFh' * 20
|
|
|
|
j = jenkins.Jenkins('{0}'.format(self.base_url), long_str, long_str)
|
|
|
|
self.assertNotIn(b"\n", j.auth)
|
|
self.assertEqual(j.auth.decode('utf-8'), 'Basic %s' % (
|
|
long_str_b64 + 'Om' + long_str_b64[2:] + 'YQ=='))
|
|
|
|
def test_default_timeout(self):
|
|
j = jenkins.Jenkins('{0}'.format(self.base_url))
|
|
self.assertEqual(j.timeout, socket._GLOBAL_DEFAULT_TIMEOUT)
|
|
|
|
def test_custom_timeout(self):
|
|
j = jenkins.Jenkins('{0}'.format(self.base_url), timeout=300)
|
|
self.assertEqual(j.timeout, 300)
|
|
|
|
|
|
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'))
|
|
|
|
self.j.maybe_add_crumb(request)
|
|
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].get_full_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'))
|
|
|
|
self.j.maybe_add_crumb(request)
|
|
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].get_full_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'))
|
|
|
|
self.j.maybe_add_crumb(request)
|
|
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].get_full_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):
|
|
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'))
|
|
|
|
response = self.j.jenkins_open(request)
|
|
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].get_full_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'))
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.jenkins_open(request, add_crumb=False)
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Error in request. Possibly authentication failed [401]: '
|
|
'basic auth failed')
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].get_full_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'))
|
|
|
|
with self.assertRaises(jenkins.NotFoundException) as context_manager:
|
|
self.j.jenkins_open(request, add_crumb=False)
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Requested item could not be found')
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].get_full_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'))
|
|
|
|
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))
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].get_full_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'))
|
|
|
|
with self.assertRaises(HTTPError) as context_manager:
|
|
self.j.jenkins_open(request, add_crumb=False)
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'HTTP Error 501: Not implemented')
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].get_full_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(
|
|
reason="timed out")
|
|
j = jenkins.Jenkins(self.make_url(''), 'test', 'test', timeout=1)
|
|
request = jenkins.Request(self.make_url('job/TestJob'))
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
j.jenkins_open(request, add_crumb=False)
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Error in request: timed out')
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].get_full_url(),
|
|
self.make_url('job/TestJob'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|