1012 lines
42 KiB
Python
1012 lines
42 KiB
Python
import json
|
|
|
|
import collections
|
|
from mock import patch, Mock
|
|
|
|
import jenkins
|
|
from tests.base import JenkinsTestBase
|
|
from tests.helper import build_response_mock
|
|
|
|
|
|
class JenkinsBuildConsoleTest(JenkinsTestBase):
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_simple(self, jenkins_mock):
|
|
jenkins_mock.return_value = "build console output..."
|
|
|
|
build_info = self.j.get_build_console_output(u'Test Job', number='52')
|
|
|
|
self.assertEqual(build_info, jenkins_mock.return_value)
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/Test%20Job/52/consoleText'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_in_folder(self, jenkins_mock):
|
|
jenkins_mock.return_value = "build console output..."
|
|
|
|
build_info = self.j.get_build_console_output(u'a Folder/Test Job', number='52')
|
|
|
|
self.assertEqual(build_info, jenkins_mock.return_value)
|
|
self.assertEqual(
|
|
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)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_matrix(self, jenkins_mock):
|
|
jenkins_mock.return_value = "build console output..."
|
|
|
|
build_info = self.j.get_build_console_output(u'a Folder/Test Job', number='52/label=matrix')
|
|
|
|
self.assertEqual(build_info, jenkins_mock.return_value)
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/a%20Folder/job/Test%20Job/52/label=matrix/consoleText'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_return_none(self, jenkins_mock):
|
|
jenkins_mock.return_value = None
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
# Test with a number for a printing case, to ensure it works
|
|
self.j.get_build_console_output(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_return_none(self, jenkins_mock):
|
|
jenkins_mock.return_value = None
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_console_output(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)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_matrix_return_none(self, jenkins_mock):
|
|
jenkins_mock.return_value = None
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_console_output(u'A Folder/TestJob', number='52/label=matrix')
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'job[A Folder/TestJob] number[52/label=matrix] does not exist')
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_return_invalid_json(self, jenkins_mock):
|
|
jenkins_mock.return_value = 'Invalid JSON'
|
|
|
|
console_output = self.j.get_build_console_output(u'TestJob', number='52')
|
|
self.assertEqual(console_output, jenkins_mock.return_value)
|
|
|
|
@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(
|
|
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')
|
|
|
|
@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:
|
|
# Test with a number for a printing case, to ensure it works
|
|
self.j.get_build_console_output(u'a Folder/TestJob', number=52)
|
|
self.assertEqual(
|
|
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')
|
|
|
|
|
|
class JenkinsBuildInfoTest(JenkinsTestBase):
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_simple(self, jenkins_mock):
|
|
build_info_to_return = {
|
|
u'building': False,
|
|
u'msg': u'test',
|
|
u'revision': 66,
|
|
u'user': u'unknown'
|
|
}
|
|
jenkins_mock.return_value = json.dumps(build_info_to_return)
|
|
|
|
build_info = self.j.get_build_info(u'Test Job', number='52')
|
|
|
|
self.assertEqual(build_info, build_info_to_return)
|
|
self.assertEqual(
|
|
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)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_in_folder(self, jenkins_mock):
|
|
build_info_to_return = {
|
|
u'building': False,
|
|
u'msg': u'test',
|
|
u'revision': 66,
|
|
u'user': u'unknown'
|
|
}
|
|
jenkins_mock.return_value = json.dumps(build_info_to_return)
|
|
|
|
build_info = self.j.get_build_info(u'a Folder/Test Job', number='52')
|
|
|
|
self.assertEqual(build_info, build_info_to_return)
|
|
self.assertEqual(
|
|
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)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_matrix(self, jenkins_mock):
|
|
build_info_to_return = {
|
|
u'building': False,
|
|
u'msg': u'test',
|
|
u'revision': 66,
|
|
u'user': u'unknown'
|
|
}
|
|
jenkins_mock.return_value = json.dumps(build_info_to_return)
|
|
|
|
build_info = self.j.get_build_info(u'a Folder/Test Job', number='52/label=matrix')
|
|
|
|
self.assertEqual(build_info, build_info_to_return)
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/a%20Folder/job/Test%20Job/52/label=matrix/api/json?depth=0'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_return_none(self, jenkins_mock):
|
|
jenkins_mock.return_value = None
|
|
|
|
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_return_invalid_json(self, jenkins_mock):
|
|
jenkins_mock.return_value = 'Invalid JSON'
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
# Test with a number for a printing case, to ensure it works
|
|
self.j.get_build_info(u'TestJob', number=52)
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Could not parse JSON info for job[TestJob] number[52]')
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch('jenkins.requests.Session.send', autospec=True)
|
|
def test_raise_HTTPError(self, session_send_mock):
|
|
session_send_mock.side_effect = iter([
|
|
build_response_mock(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')
|
|
|
|
@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')
|
|
|
|
|
|
class JenkinsStopBuildTest(JenkinsTestBase):
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_simple(self, jenkins_mock):
|
|
self.j.stop_build(u'Test Job', number='52')
|
|
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/Test%20Job/52/stop'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_in_folder(self, jenkins_mock):
|
|
|
|
self.j.stop_build(u'a Folder/Test Job', number='52')
|
|
|
|
self.assertEqual(
|
|
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)
|
|
|
|
|
|
class JenkinsDeleteBuildTest(JenkinsTestBase):
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_simple(self, jenkins_mock):
|
|
self.j.delete_build(u'Test Job', number='52')
|
|
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/Test%20Job/52/doDelete'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_in_folder(self, jenkins_mock):
|
|
|
|
self.j.delete_build(u'a Folder/Test Job', number='52')
|
|
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/a%20Folder/job/Test%20Job/52/doDelete'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
|
|
class JenkinsListRunningBuildsTest(JenkinsTestBase):
|
|
@patch.object(jenkins.Jenkins, 'get_node_info')
|
|
@patch.object(jenkins.Jenkins, 'get_nodes')
|
|
def test_with_builds_master(self, nodes_mock, node_info_mock):
|
|
nodes_to_return = [{
|
|
'name': "master", 'offline': False
|
|
}]
|
|
nodes_mock.return_value = nodes_to_return
|
|
build = {
|
|
"actions": [
|
|
{
|
|
"parameters": [
|
|
{
|
|
"name": "FOO",
|
|
"value": "foo"
|
|
},
|
|
{
|
|
"name": "BAR",
|
|
"value": "bar"
|
|
}
|
|
]
|
|
},
|
|
{
|
|
"causes": [
|
|
{
|
|
"shortDescription": "Started by user foo",
|
|
"userId": "foo",
|
|
"userName": "Foo Bar"
|
|
}
|
|
]
|
|
}
|
|
],
|
|
"artifacts": [],
|
|
"building": True,
|
|
"description": None,
|
|
"duration": 0,
|
|
"estimatedDuration": 20148,
|
|
"executor": {},
|
|
"fullDisplayName": "test #1",
|
|
"id": "2015-09-14_20-25-42",
|
|
"keepLog": False,
|
|
"number": 1,
|
|
"result": None,
|
|
"timestamp": 1442262342729,
|
|
"url": self.make_url('job/test/1/'),
|
|
"builtOn": "",
|
|
"changeSet": {
|
|
"items": [],
|
|
"kind": None
|
|
},
|
|
"culprits": []
|
|
}
|
|
node_info_to_return = {
|
|
"executors": [
|
|
{
|
|
"currentExecutable": None,
|
|
"currentWorkUnit": None,
|
|
"idle": True,
|
|
"likelyStuck": False,
|
|
"number": 0,
|
|
"progress": -1
|
|
},
|
|
{
|
|
"currentExecutable": build,
|
|
"currentWorkUnit": {},
|
|
"idle": False,
|
|
"likelyStuck": False,
|
|
"number": 1,
|
|
"progress": 14
|
|
}
|
|
],
|
|
}
|
|
node_info_mock.return_value = node_info_to_return
|
|
builds = self.j.get_running_builds()
|
|
self.assertEqual([{'name': 'test',
|
|
'number': 1,
|
|
'node': '(master)',
|
|
'executor': 1,
|
|
'url': self.make_url('job/test/1/')}], builds)
|
|
|
|
@patch.object(jenkins.Jenkins, 'get_node_info')
|
|
@patch.object(jenkins.Jenkins, 'get_nodes')
|
|
def test_with_builds_non_master(self, nodes_mock, node_info_mock):
|
|
nodes_to_return = [{
|
|
'name': "foo-slave", 'offline': False
|
|
}]
|
|
nodes_mock.return_value = nodes_to_return
|
|
build = {
|
|
"actions": [
|
|
{
|
|
"parameters": [
|
|
{
|
|
"name": "FOO",
|
|
"value": "foo"
|
|
},
|
|
{
|
|
"name": "BAR",
|
|
"value": "bar"
|
|
}
|
|
]
|
|
},
|
|
{
|
|
"causes": [
|
|
{
|
|
"shortDescription": "Started by user foo",
|
|
"userId": "foo",
|
|
"userName": "Foo Bar"
|
|
}
|
|
]
|
|
}
|
|
],
|
|
"artifacts": [],
|
|
"building": True,
|
|
"description": None,
|
|
"duration": 0,
|
|
"estimatedDuration": 20148,
|
|
"executor": {},
|
|
"fullDisplayName": "test #1",
|
|
"id": "2015-09-14_20-25-42",
|
|
"keepLog": False,
|
|
"number": 15,
|
|
"result": None,
|
|
"timestamp": 1442262342729,
|
|
"url": self.make_url("job/test/15/"),
|
|
"builtOn": "",
|
|
"changeSet": {
|
|
"items": [],
|
|
"kind": None
|
|
},
|
|
"culprits": []
|
|
}
|
|
node_info_to_return = {
|
|
"executors": [
|
|
{
|
|
"currentExecutable": None,
|
|
"currentWorkUnit": None,
|
|
"idle": True,
|
|
"likelyStuck": False,
|
|
"number": 1,
|
|
"progress": -1
|
|
},
|
|
{
|
|
"currentExecutable": build,
|
|
"currentWorkUnit": {},
|
|
"idle": False,
|
|
"likelyStuck": False,
|
|
"number": 0,
|
|
"progress": 14
|
|
}
|
|
],
|
|
}
|
|
node_info_mock.return_value = node_info_to_return
|
|
builds = self.j.get_running_builds()
|
|
self.assertEqual([{'name': 'test',
|
|
'number': 15,
|
|
'node': 'foo-slave',
|
|
'executor': 0,
|
|
'url': self.make_url('job/test/15/')}], builds)
|
|
|
|
@patch.object(jenkins.Jenkins, 'get_node_info')
|
|
@patch.object(jenkins.Jenkins, 'get_nodes')
|
|
def test_with_no_builds(self, nodes_mock, node_info_mock):
|
|
nodes_to_return = [{
|
|
'name': "master", 'offline': False
|
|
}]
|
|
nodes_mock.return_value = nodes_to_return
|
|
node_info_to_return = {
|
|
"executors": [
|
|
{
|
|
"currentExecutable": None,
|
|
"currentWorkUnit": None,
|
|
"idle": True,
|
|
"likelyStuck": False,
|
|
"number": 0,
|
|
"progress": -1
|
|
}
|
|
]
|
|
}
|
|
node_info_mock.return_value = node_info_to_return
|
|
builds = self.j.get_running_builds()
|
|
self.assertEqual([], builds)
|
|
|
|
@patch.object(jenkins.Jenkins, 'get_node_info')
|
|
@patch.object(jenkins.Jenkins, 'get_nodes')
|
|
def test_broken_slave(self, nodes_mock, node_info_mock):
|
|
nodes_to_return = [{
|
|
'name': "foo-slave", 'offline': False
|
|
}]
|
|
nodes_mock.return_value = nodes_to_return
|
|
|
|
def side_effect(*args, **kwargs):
|
|
if 'depth' in kwargs and kwargs['depth'] > 0:
|
|
raise jenkins.JenkinsException(
|
|
"Error in request. Possibly authentication failed"
|
|
"[500]: Server Error")
|
|
else:
|
|
return {"success": True}
|
|
|
|
node_info_mock.side_effect = side_effect
|
|
builds = self.j.get_running_builds()
|
|
# Should treat the slave as not running any builds
|
|
self.assertEqual([], builds)
|
|
|
|
@patch.object(jenkins.Jenkins, 'get_node_info')
|
|
@patch.object(jenkins.Jenkins, 'get_nodes')
|
|
def test_unnumbered_task_in_queue(self, nodes_mock, node_info_mock):
|
|
nodes_to_return = [{
|
|
'name': "foo-slave", 'offline': False
|
|
}]
|
|
nodes_mock.return_value = nodes_to_return
|
|
node_info_to_return = {
|
|
"executors": [
|
|
{
|
|
"currentExecutable": None,
|
|
"currentWorkUnit": None,
|
|
"idle": True,
|
|
"likelyStuck": False,
|
|
"number": 1,
|
|
"progress": -1
|
|
},
|
|
{
|
|
'currentExecutable': {},
|
|
'currentWorkUnit': {},
|
|
'idle': False,
|
|
'likelyStuck': False,
|
|
'number': 1,
|
|
'progress': 0
|
|
}
|
|
],
|
|
}
|
|
node_info_mock.return_value = node_info_to_return
|
|
builds = self.j.get_running_builds()
|
|
self.assertEqual([], builds)
|
|
|
|
|
|
class JenkinsBuildJobUrlTest(JenkinsTestBase):
|
|
|
|
def test_params_as_dict(self):
|
|
token = 'token123'
|
|
params = [
|
|
('m_select', 'value1'),
|
|
('s_select', 's_select2')
|
|
]
|
|
parameters = collections.OrderedDict(params)
|
|
self.assertEqual(
|
|
self.j.build_job_url('Test Job', parameters=parameters, token=token),
|
|
self.make_url(
|
|
'job/Test%20Job/buildWithParameters?m_select=value1'
|
|
'&s_select=s_select2&token=token123'))
|
|
|
|
def test_params_as_list(self):
|
|
token = 'token123'
|
|
params = [
|
|
('m_select', 'value1',),
|
|
('m_select', 'value3'),
|
|
('s_select', 's_select2')
|
|
]
|
|
self.assertEqual(
|
|
self.j.build_job_url('Test Job', parameters=params, token=token),
|
|
self.make_url(
|
|
'job/Test%20Job/buildWithParameters?m_select=value1'
|
|
'&m_select=value3&s_select=s_select2&token=token123'))
|
|
|
|
|
|
class JenkinsBuildEnvVarUrlTest(JenkinsTestBase):
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_simple(self, jenkins_mock):
|
|
jenkins_mock.return_value = '{}'
|
|
ret = self.j.get_build_env_vars(u'Test Job', number='52', depth=1)
|
|
self.assertEqual(ret, json.loads(jenkins_mock.return_value))
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/Test%20Job/52/injectedEnvVars/api/json?depth=1'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_in_folder(self, jenkins_mock):
|
|
jenkins_mock.return_value = '{}'
|
|
# Test with a number to ensure it works as well as string
|
|
ret = self.j.get_build_env_vars(u'a Folder/Test Job', number=52, depth=1)
|
|
self.assertEqual(ret, json.loads(jenkins_mock.return_value))
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/a%20Folder/job/Test%20Job/52/injectedEnvVars/api/json?depth=1'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_matrix(self, jenkins_mock):
|
|
jenkins_mock.return_value = '{}'
|
|
ret = self.j.get_build_env_vars(u'a Folder/Test Job', number='52/index=matrix', depth=1)
|
|
self.assertEqual(ret, json.loads(jenkins_mock.return_value))
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url(
|
|
'job/a%20Folder/job/Test%20Job/52/index=matrix/injectedEnvVars/api/json?depth=1'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch('jenkins.requests.Session.send', autospec=True)
|
|
def test_return_none(self, session_send_mock):
|
|
session_send_mock.side_effect = iter([
|
|
build_response_mock(404, reason="Not Found"), # crumb
|
|
build_response_mock(404, reason="Not Found"), # request
|
|
])
|
|
ret = self.j.get_build_env_vars(u'TestJob', number='52')
|
|
self.assertIsNone(ret)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_open_return_none(self, jenkins_mock):
|
|
jenkins_mock.return_value = None
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_env_vars(u'TestJob', number='52')
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'job[TestJob] number[52] does not exist')
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_return_invalid_json(self, jenkins_mock):
|
|
jenkins_mock.return_value = 'Invalid JSON'
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_env_vars(u'TestJob', number='52')
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Could not parse JSON info for job[TestJob] number[52]')
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch('jenkins.requests.Session.send', autospec=True)
|
|
def test_raise_HTTPError(self, session_send_mock):
|
|
session_send_mock.side_effect = iter([
|
|
build_response_mock(401, reason="Not Authorised"), # crumb
|
|
build_response_mock(401, reason="Not Authorised"), # request
|
|
])
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
# Test with a number for a printing case, to ensure it works
|
|
self.j.get_build_env_vars(u'TestJob', number=52)
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Error in request. Possibly authentication failed [401]: Not Authorised')
|
|
|
|
@patch('jenkins.requests.Session.send', autospec=True)
|
|
def test_in_folder_raise_HTTPError(self, session_send_mock):
|
|
session_send_mock.side_effect = iter([
|
|
build_response_mock(401, reason="Not Authorised"), # crumb
|
|
build_response_mock(401, reason="Not Authorised"), # request
|
|
])
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_env_vars(u'a Folder/TestJob', number='52')
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Error in request. Possibly authentication failed [401]: Not Authorised')
|
|
|
|
|
|
class JenkinsBuildTestReportUrlTest(JenkinsTestBase):
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_simple(self, jenkins_mock):
|
|
jenkins_mock.return_value = '{}'
|
|
ret = self.j.get_build_test_report(u'Test Job', number='52', depth=1)
|
|
self.assertEqual(ret, json.loads(jenkins_mock.return_value))
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/Test%20Job/52/testReport/api/json?depth=1'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_in_folder(self, jenkins_mock):
|
|
jenkins_mock.return_value = '{}'
|
|
# Test with a number to ensure it works as well as string
|
|
ret = self.j.get_build_test_report(u'a Folder/Test Job', number=52, depth=1)
|
|
self.assertEqual(ret, json.loads(jenkins_mock.return_value))
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/a%20Folder/job/Test%20Job/52/testReport/api/json?depth=1'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_matrix(self, jenkins_mock):
|
|
jenkins_mock.return_value = '{}'
|
|
ret = self.j.get_build_test_report(u'a Folder/Test Job', number='52/index=matrix', depth=1)
|
|
self.assertEqual(ret, json.loads(jenkins_mock.return_value))
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url(
|
|
'job/a%20Folder/job/Test%20Job/52/index=matrix/testReport/api/json?depth=1'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch('jenkins.requests.Session.send', autospec=True)
|
|
def test_return_none(self, session_send_mock):
|
|
session_send_mock.side_effect = iter([
|
|
build_response_mock(404, reason="Not Found"), # crumb
|
|
build_response_mock(404, reason="Not Found"), # request
|
|
])
|
|
ret = self.j.get_build_test_report(u'TestJob', number='52')
|
|
self.assertIsNone(ret)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_open_return_none(self, jenkins_mock):
|
|
jenkins_mock.return_value = None
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_test_report(u'TestJob', number='52')
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'job[TestJob] number[52] does not exist')
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_return_invalid_json(self, jenkins_mock):
|
|
jenkins_mock.return_value = 'Invalid JSON'
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_test_report(u'TestJob', number='52')
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Could not parse JSON info for job[TestJob] number[52]')
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch('jenkins.requests.Session.send', autospec=True)
|
|
def test_raise_HTTPError(self, session_send_mock):
|
|
session_send_mock.side_effect = iter([
|
|
build_response_mock(401, reason="Not Authorised"), # crumb
|
|
build_response_mock(401, reason="Not Authorised"), # request
|
|
])
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_test_report(u'TestJob', number='52')
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Error in request. Possibly authentication failed [401]: Not Authorised')
|
|
|
|
@patch('jenkins.requests.Session.send', autospec=True)
|
|
def test_in_folder_raise_HTTPError(self, session_send_mock):
|
|
session_send_mock.side_effect = iter([
|
|
build_response_mock(401, reason="Not Authorised"), # crumb
|
|
build_response_mock(401, reason="Not Authorised"), # request
|
|
])
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
# Test with a number for a printing case, to ensure it works
|
|
self.j.get_build_test_report(u'a Folder/TestJob', number=52)
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Error in request. Possibly authentication failed [401]: Not Authorised')
|
|
|
|
|
|
class JenkinsBuildArtifactUrlTest(JenkinsTestBase):
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_simple(self, jenkins_mock):
|
|
jenkins_mock.return_value = '{}'
|
|
ret = self.j.get_build_artifact(u'Test Job', number='52', artifact="filename")
|
|
self.assertEqual(ret, json.loads(jenkins_mock.return_value))
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/Test%20Job/52/artifact/filename'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_in_folder(self, jenkins_mock):
|
|
jenkins_mock.return_value = '{}'
|
|
ret = self.j.get_build_artifact(u'a Folder/Test Job', number='52', artifact="file name")
|
|
self.assertEqual(ret, json.loads(jenkins_mock.return_value))
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/a%20Folder/job/Test%20Job/52/artifact/file%20name'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_matrix(self, jenkins_mock):
|
|
jenkins_mock.return_value = '{}'
|
|
ret = self.j.get_build_artifact(u'a Folder/Test Job', number='52/index=matrix',
|
|
artifact="file name")
|
|
self.assertEqual(ret, json.loads(jenkins_mock.return_value))
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/a%20Folder/job/Test%20Job/52/index=matrix/artifact/file%20name'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch('jenkins.requests.Session.send', autospec=True)
|
|
def test_return_none(self, session_send_mock):
|
|
session_send_mock.side_effect = iter([
|
|
build_response_mock(404, reason="Not Found"), # crumb
|
|
build_response_mock(404, reason="Not Found"), # request
|
|
])
|
|
ret = self.j.get_build_artifact(u'TestJob', number='52', artifact="filename")
|
|
self.assertIsNone(ret)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_open_return_none(self, jenkins_mock):
|
|
jenkins_mock.return_value = None
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_artifact(u'TestJob', number='52', artifact="filename")
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'job[TestJob] number[52] does not exist')
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_return_invalid_json(self, jenkins_mock):
|
|
jenkins_mock.return_value = 'Invalid JSON'
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_artifact(u'TestJob', number='52', artifact="filename")
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Could not parse JSON info for job[TestJob] number[52]')
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch('jenkins.requests.Session.send', autospec=True)
|
|
def test_raise_HTTPError(self, session_send_mock):
|
|
session_send_mock.side_effect = iter([
|
|
build_response_mock(401, reason="Not Authorised"), # crumb
|
|
build_response_mock(401, reason="Not Authorised"), # request
|
|
])
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_artifact(u'TestJob', number='52', artifact="filename")
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Error in request. Possibly authentication failed [401]: Not Authorised')
|
|
|
|
@patch('jenkins.requests.Session.send', autospec=True)
|
|
def test_in_folder_raise_HTTPError(self, session_send_mock):
|
|
session_send_mock.side_effect = iter([
|
|
build_response_mock(401, reason="Not Authorised"), # crumb
|
|
build_response_mock(401, reason="Not Authorised"), # request
|
|
])
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_artifact(u'a Folder/TestJob', number='52', artifact="filename")
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Error in request. Possibly authentication failed [401]: Not Authorised')
|
|
|
|
|
|
class JenkinsBuildArtifactAsBytesUrlTest(JenkinsTestBase):
|
|
|
|
def streamMock(self, encoding=None, text=None, binary=None):
|
|
streamMock = Mock()
|
|
streamMock.__exit__ = Mock()
|
|
streamMock.__enter__ = Mock()
|
|
if encoding is None and text is None and binary is None:
|
|
streamMock.__enter__.return_value = None
|
|
return streamMock
|
|
streamMock.__enter__.return_value = Mock()
|
|
streamMock.__enter__.return_value.encoding = encoding
|
|
streamMock.__enter__.return_value.text = text
|
|
streamMock.__enter__.return_value.raw = Mock()
|
|
streamMock.__enter__.return_value.raw.read = Mock()
|
|
streamMock.__enter__.return_value.raw.read.return_value = binary
|
|
return streamMock
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open_stream')
|
|
def test_simple_ascii_artifact(self, jenkins_mock):
|
|
jenkins_mock.return_value = self.streamMock('utf-8', 'ascii')
|
|
ret = self.j.get_build_artifact_as_bytes(u'Test Job', number='52', artifact="filename")
|
|
self.assertEqual(ret, b'ascii')
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/Test%20Job/52/artifact/filename'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open_stream')
|
|
def test_simple_binary_artifact(self, jenkins_mock):
|
|
jenkins_mock.return_value = self.streamMock(binary=b'\0\1\2')
|
|
ret = self.j.get_build_artifact_as_bytes(u'Test Job', number='52', artifact="filename")
|
|
self.assertEqual(ret, b'\0\1\2')
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/Test%20Job/52/artifact/filename'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open_stream')
|
|
def test_in_folder(self, jenkins_mock):
|
|
jenkins_mock.return_value = self.streamMock('utf-8', 'ascii')
|
|
ret = self.j.get_build_artifact_as_bytes(u'a Folder/Test Job', number='52', artifact="file name")
|
|
self.assertEqual(ret, b'ascii')
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/a%20Folder/job/Test%20Job/52/artifact/file%20name'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open_stream')
|
|
def test_matrix(self, jenkins_mock):
|
|
jenkins_mock.return_value = self.streamMock('utf-8', 'ascii')
|
|
ret = self.j.get_build_artifact_as_bytes(u'a Folder/Test Job', number='52/index=matrix',
|
|
artifact="file name")
|
|
self.assertEqual(ret, b'ascii')
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/a%20Folder/job/Test%20Job/52/index=matrix/artifact/file%20name'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch('jenkins.requests.Session.send', autospec=True)
|
|
def test_404_item_not_found(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_artifact_as_bytes(u'TestJob', number='52', artifact="filename")
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Requested item could not be found')
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open_stream')
|
|
def test_open_return_none(self, jenkins_mock):
|
|
jenkins_mock.return_value = self.streamMock()
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_artifact_as_bytes(u'TestJob', number='52', artifact="filename")
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'job[TestJob] number[52] does not exist')
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch('jenkins.requests.Session.send', autospec=True)
|
|
def test_raise_HTTPError(self, session_send_mock):
|
|
session_send_mock.side_effect = iter([
|
|
build_response_mock(401, reason="Not Authorised"), # crumb
|
|
build_response_mock(401, reason="Not Authorised"), # request
|
|
])
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_artifact_as_bytes(u'TestJob', number='52', artifact="filename")
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Error in request. Possibly authentication failed [401]: Not Authorised')
|
|
|
|
@patch('jenkins.requests.Session.send', autospec=True)
|
|
def test_in_folder_raise_HTTPError(self, session_send_mock):
|
|
session_send_mock.side_effect = iter([
|
|
build_response_mock(401, reason="Not Authorised"), # crumb
|
|
build_response_mock(401, reason="Not Authorised"), # request
|
|
])
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_artifact_as_bytes(u'a Folder/TestJob', number='52', artifact="filename")
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Error in request. Possibly authentication failed [401]: Not Authorised')
|
|
|
|
|
|
class JenkinsBuildStagesUrlTest(JenkinsTestBase):
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_simple(self, jenkins_mock):
|
|
jenkins_mock.return_value = '{}'
|
|
ret = self.j.get_build_stages(u'Test Job', number='52')
|
|
self.assertEqual(ret, json.loads(jenkins_mock.return_value))
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/Test%20Job/52/wfapi/describe/'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_matrix(self, jenkins_mock):
|
|
jenkins_mock.return_value = '{}'
|
|
ret = self.j.get_build_stages(u'a Folder/Test Job', number='52/index=matrix')
|
|
self.assertEqual(ret, json.loads(jenkins_mock.return_value))
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/a%20Folder/job/Test%20Job/52/index=matrix/wfapi/describe/'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_in_folder(self, jenkins_mock):
|
|
jenkins_mock.return_value = '{}'
|
|
ret = self.j.get_build_stages(u'a Folder/Test Job', number='52')
|
|
self.assertEqual(ret, json.loads(jenkins_mock.return_value))
|
|
self.assertEqual(
|
|
jenkins_mock.call_args[0][0].url,
|
|
self.make_url('job/a%20Folder/job/Test%20Job/52/wfapi/describe/'))
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch('jenkins.requests.Session.send', autospec=True)
|
|
def test_return_none(self, session_send_mock):
|
|
session_send_mock.side_effect = iter([
|
|
build_response_mock(404, reason="Not Found"), # crumb
|
|
build_response_mock(404, reason="Not Found"), # request
|
|
])
|
|
ret = self.j.get_build_stages(u'TestJob', number='52')
|
|
self.assertIsNone(ret)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_open_return_none(self, jenkins_mock):
|
|
jenkins_mock.return_value = None
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
# Test with a number for a printing case, to ensure it works
|
|
self.j.get_build_stages(u'TestJob', number=52)
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'job[TestJob] number[52] does not exist')
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch.object(jenkins.Jenkins, 'jenkins_open')
|
|
def test_return_invalid_json(self, jenkins_mock):
|
|
jenkins_mock.return_value = 'Invalid JSON'
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_stages(u'TestJob', number='52')
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Could not parse JSON info for job[TestJob] number[52]')
|
|
self._check_requests(jenkins_mock.call_args_list)
|
|
|
|
@patch('jenkins.requests.Session.send', autospec=True)
|
|
def test_raise_HTTPError(self, session_send_mock):
|
|
session_send_mock.side_effect = iter([
|
|
build_response_mock(401, reason="Not Authorised"), # crumb
|
|
build_response_mock(401, reason="Not Authorised"), # request
|
|
])
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_stages(u'TestJob', number='52')
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Error in request. Possibly authentication failed [401]: Not Authorised')
|
|
|
|
@patch('jenkins.requests.Session.send', autospec=True)
|
|
def test_in_folder_raise_HTTPError(self, session_send_mock):
|
|
session_send_mock.side_effect = iter([
|
|
build_response_mock(401, reason="Not Authorised"), # crumb
|
|
build_response_mock(401, reason="Not Authorised"), # request
|
|
])
|
|
|
|
with self.assertRaises(jenkins.JenkinsException) as context_manager:
|
|
self.j.get_build_stages(u'a Folder/TestJob', number='52')
|
|
self.assertEqual(
|
|
str(context_manager.exception),
|
|
'Error in request. Possibly authentication failed [401]: Not Authorised')
|