diff --git a/doc/source/install.rst b/doc/source/install.rst
index 2d89828..9e5430c 100644
--- a/doc/source/install.rst
+++ b/doc/source/install.rst
@@ -50,6 +50,15 @@ every module. To run the unit tests, execute the command::
* Note: View ``tox.ini`` to run tests on other versions of Python.
+Due to how the tests are split up into a dedicated class per API method, it is
+possible to execute tests against a single API at a time. To execute the tests
+for the :py:meth:`.Jenkins.get_version` API execute the command::
+
+ tox -e py27 -- tests.test_version.JenkinsVersionTest
+
+For further details on how to list tests available and different ways to
+execute them, see https://wiki.openstack.org/wiki/Testr.
+
Test Coverage
-------------
diff --git a/tests/base.py b/tests/base.py
new file mode 100644
index 0000000..b5583ca
--- /dev/null
+++ b/tests/base.py
@@ -0,0 +1,42 @@
+import sys
+
+from six.moves.urllib.request import build_opener
+
+import jenkins
+
+if sys.version_info < (2, 7):
+ import unittest2 as unittest
+else:
+ import unittest
+
+
+class JenkinsTestBase(unittest.TestCase):
+
+ crumb_data = {
+ "crumb": "dab177f483b3dd93483ef6716d8e792d",
+ "crumbRequestField": ".crumb",
+ }
+
+ def setUp(self):
+ super(JenkinsTestBase, self).setUp()
+ self.opener = build_opener()
+
+ self.j = jenkins.Jenkins('http://example.com/', 'test', 'test')
+
+ 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)
diff --git a/tests/jobs/__init__.py b/tests/jobs/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/jobs/base.py b/tests/jobs/base.py
new file mode 100644
index 0000000..de1827d
--- /dev/null
+++ b/tests/jobs/base.py
@@ -0,0 +1,10 @@
+from tests.base import JenkinsTestBase
+
+
+class JenkinsJobsTestBase(JenkinsTestBase):
+
+ config_xml = """
+ <matrix-project>
+ <actions/>
+ <description>Foo</description>
+ </matrix-project>"""
diff --git a/tests/jobs/test_assert.py b/tests/jobs/test_assert.py
new file mode 100644
index 0000000..83e34f6
--- /dev/null
+++ b/tests/jobs/test_assert.py
@@ -0,0 +1,27 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.jobs.base import JenkinsJobsTestBase
+
+
+class JenkinsAssertJobExistsTest(JenkinsJobsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_job_missing(self, jenkins_mock):
+ jenkins_mock.side_effect = jenkins.NotFoundException()
+
+ with self.assertRaises(jenkins.JenkinsException) as context_manager:
+ self.j.assert_job_exists('NonExistent')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'job[NonExistent] does not exist')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_job_exists(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ json.dumps({'name': 'ExistingJob'}),
+ ]
+ self.j.assert_job_exists('ExistingJob')
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/jobs/test_build.py b/tests/jobs/test_build.py
new file mode 100644
index 0000000..2ab6025
--- /dev/null
+++ b/tests/jobs/test_build.py
@@ -0,0 +1,50 @@
+from mock import patch
+
+import jenkins
+from tests.jobs.base import JenkinsJobsTestBase
+
+
+class JenkinsBuildJobTest(JenkinsJobsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ {'foo': 'bar'},
+ ]
+
+ build_info = self.j.build_job(u'Test Job')
+
+ self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/job/Test%20Job/build')
+ self.assertEqual(build_info, {'foo': 'bar'})
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_with_token(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ {'foo': 'bar'},
+ ]
+
+ build_info = self.j.build_job(u'TestJob', token='some_token')
+
+ self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/job/TestJob/build?token=some_token')
+ self.assertEqual(build_info, {'foo': 'bar'})
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_with_parameters_and_token(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ {'foo': 'bar'},
+ ]
+
+ build_info = self.j.build_job(
+ u'TestJob',
+ 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.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
new file mode 100644
index 0000000..7299258
--- /dev/null
+++ b/tests/jobs/test_config.py
@@ -0,0 +1,16 @@
+from mock import patch
+
+import jenkins
+from tests.jobs.base import JenkinsJobsTestBase
+
+
+class JenkinsGetJobConfigTest(JenkinsJobsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_encodes_job_name(self, jenkins_mock):
+ self.j.get_job_config(u'Test Job')
+
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/job/Test%20Job/config.xml')
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/jobs/test_copy.py b/tests/jobs/test_copy.py
new file mode 100644
index 0000000..71c4736
--- /dev/null
+++ b/tests/jobs/test_copy.py
@@ -0,0 +1,43 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.jobs.base import JenkinsJobsTestBase
+
+
+class JenkinsCopyJobTest(JenkinsJobsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ json.dumps({'name': 'Test Job_2'}),
+ json.dumps({'name': 'Test Job_2'}),
+ json.dumps({'name': 'Test Job_2'}),
+ ]
+
+ self.j.copy_job(u'Test Job', u'Test Job_2')
+
+ self.assertEqual(
+ jenkins_mock.call_args_list[0][0][0].get_full_url(),
+ 'http://example.com/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)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_failed(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ None,
+ jenkins.NotFoundException(),
+ ]
+
+ 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(),
+ 'http://example.com/createItem'
+ '?name=TestJob_2&mode=copy&from=TestJob')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'create[TestJob_2] failed')
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/jobs/test_count.py b/tests/jobs/test_count.py
new file mode 100644
index 0000000..6426ade
--- /dev/null
+++ b/tests/jobs/test_count.py
@@ -0,0 +1,20 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.jobs.base import JenkinsJobsTestBase
+
+
+class JenkinsJobsCountTest(JenkinsJobsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jobs = [
+ {u'url': u'http://localhost:8080/job/guava/', u'color': u'notbuilt', u'name': u'guava'},
+ {u'url': u'http://localhost:8080/job/kiwi/', u'color': u'blue', u'name': u'kiwi'},
+ {u'url': u'http://localhost:8080/job/lemon/', u'color': u'red', u'name': u'lemon'}
+ ]
+ job_info_to_return = {u'jobs': jobs}
+ jenkins_mock.return_value = json.dumps(job_info_to_return)
+ self.assertEqual(self.j.jobs_count(), 3)
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/jobs/test_create.py b/tests/jobs/test_create.py
new file mode 100644
index 0000000..37ce3ae
--- /dev/null
+++ b/tests/jobs/test_create.py
@@ -0,0 +1,61 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.jobs.base import JenkinsJobsTestBase
+
+
+class JenkinsCreateJobTest(JenkinsJobsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ jenkins.NotFoundException(),
+ None,
+ json.dumps({'name': 'Test Job'}),
+ ]
+
+ self.j.create_job(u'Test Job', self.config_xml)
+
+ self.assertEqual(
+ jenkins_mock.call_args_list[1][0][0].get_full_url(),
+ 'http://example.com/createItem?name=Test%20Job')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_already_exists(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ json.dumps({'name': 'TestJob'}),
+ None,
+ ]
+
+ 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(),
+ 'http://example.com/job/TestJob/api/json?tree=name')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'job[TestJob] already exists')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_failed(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ jenkins.NotFoundException(),
+ None,
+ jenkins.NotFoundException(),
+ ]
+
+ 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(),
+ 'http://example.com/job/TestJob/api/json?tree=name')
+ self.assertEqual(
+ jenkins_mock.call_args_list[1][0][0].get_full_url(),
+ 'http://example.com/createItem?name=TestJob')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'create[TestJob] failed')
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/jobs/test_debug.py b/tests/jobs/test_debug.py
new file mode 100644
index 0000000..6d6379a
--- /dev/null
+++ b/tests/jobs/test_debug.py
@@ -0,0 +1,25 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.jobs.base import JenkinsJobsTestBase
+
+
+class JenkinsDebugJobInfoTest(JenkinsJobsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_debug_job_info(self, jenkins_mock):
+ job_info_to_return = {
+ u'building': False,
+ u'msg': u'test',
+ u'revision': 66,
+ u'user': u'unknown'
+ }
+ jenkins_mock.return_value = json.dumps(job_info_to_return)
+
+ self.j.debug_job_info(u'Test Job')
+
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/job/Test%20Job/api/json?depth=0')
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/jobs/test_delete.py b/tests/jobs/test_delete.py
new file mode 100644
index 0000000..ca96b60
--- /dev/null
+++ b/tests/jobs/test_delete.py
@@ -0,0 +1,40 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.jobs.base import JenkinsJobsTestBase
+
+
+class JenkinsDeleteJobTest(JenkinsJobsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ None,
+ jenkins.NotFoundException(),
+ ]
+
+ self.j.delete_job(u'Test Job')
+
+ self.assertEqual(
+ jenkins_mock.call_args_list[0][0][0].get_full_url(),
+ 'http://example.com/job/Test%20Job/doDelete')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_failed(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ json.dumps({'name': 'TestJob'}),
+ json.dumps({'name': 'TestJob'}),
+ json.dumps({'name': 'TestJob'}),
+ ]
+
+ 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(),
+ 'http://example.com/job/TestJob/doDelete')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'delete[TestJob] failed')
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/jobs/test_disable.py b/tests/jobs/test_disable.py
new file mode 100644
index 0000000..06b0ed0
--- /dev/null
+++ b/tests/jobs/test_disable.py
@@ -0,0 +1,23 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.jobs.base import JenkinsJobsTestBase
+
+
+class JenkinsDisableJobTest(JenkinsJobsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ json.dumps({'name': 'Test Job'}),
+ json.dumps({'name': 'Test Job'}),
+ ]
+
+ self.j.disable_job(u'Test Job')
+
+ self.assertEqual(
+ jenkins_mock.call_args_list[0][0][0].get_full_url(),
+ 'http://example.com/job/Test%20Job/disable')
+ self.assertTrue(self.j.job_exists('Test Job'))
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/jobs/test_enable.py b/tests/jobs/test_enable.py
new file mode 100644
index 0000000..aeffb64
--- /dev/null
+++ b/tests/jobs/test_enable.py
@@ -0,0 +1,23 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.jobs.base import JenkinsJobsTestBase
+
+
+class JenkinsEnableJobTest(JenkinsJobsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ json.dumps({'name': 'TestJob'}),
+ json.dumps({'name': 'TestJob'}),
+ ]
+
+ self.j.enable_job(u'TestJob')
+
+ self.assertEqual(
+ jenkins_mock.call_args_list[0][0][0].get_full_url(),
+ 'http://example.com/job/TestJob/enable')
+ self.assertTrue(self.j.job_exists('TestJob'))
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/jobs/test_get.py b/tests/jobs/test_get.py
new file mode 100644
index 0000000..b57f8ae
--- /dev/null
+++ b/tests/jobs/test_get.py
@@ -0,0 +1,26 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.jobs.base import JenkinsJobsTestBase
+
+
+class JenkinsGetJobsTest(JenkinsJobsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jobs = {
+ u'url': u'http://your_url_here/job/my_job/',
+ u'color': u'blue',
+ u'name': u'my_job',
+ }
+ job_info_to_return = {u'jobs': jobs}
+ jenkins_mock.return_value = json.dumps(job_info_to_return)
+
+ job_info = self.j.get_jobs()
+
+ self.assertEqual(job_info, jobs)
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/api/json')
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/jobs/test_info.py b/tests/jobs/test_info.py
new file mode 100644
index 0000000..3df80ff
--- /dev/null
+++ b/tests/jobs/test_info.py
@@ -0,0 +1,93 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.jobs.base import JenkinsJobsTestBase
+
+
+class JenkinsGetJobInfoTest(JenkinsJobsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ job_info_to_return = {
+ u'building': False,
+ u'msg': u'test',
+ u'revision': 66,
+ u'user': u'unknown'
+ }
+ jenkins_mock.return_value = json.dumps(job_info_to_return)
+
+ job_info = self.j.get_job_info(u'Test Job')
+
+ self.assertEqual(job_info, job_info_to_return)
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/job/Test%20Job/api/json?depth=0')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_regex(self, jenkins_mock):
+ jobs = [
+ {u'name': u'my-job-1'},
+ {u'name': u'my-job-2'},
+ {u'name': u'your-job-1'},
+ {u'name': u'Your-Job-1'},
+ {u'name': u'my-project-1'},
+ ]
+ job_info_to_return = {u'jobs': jobs}
+ jenkins_mock.return_value = json.dumps(job_info_to_return)
+
+ self.assertEqual(len(self.j.get_job_info_regex('her-job')), 0)
+ self.assertEqual(len(self.j.get_job_info_regex('my-job-1')), 1)
+ self.assertEqual(len(self.j.get_job_info_regex('my-job')), 2)
+ self.assertEqual(len(self.j.get_job_info_regex('job')), 3)
+ self.assertEqual(len(self.j.get_job_info_regex('project')), 1)
+ self.assertEqual(len(self.j.get_job_info_regex('[Yy]our-[Jj]ob-1')), 2)
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_return_none(self, jenkins_mock):
+ jenkins_mock.return_value = None
+
+ 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(),
+ u'http://example.com/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_return_invalid_json(self, jenkins_mock):
+ jenkins_mock.return_value = 'Invalid JSON'
+
+ 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(),
+ u'http://example.com/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(
+ 'http://example.com/job/TestJob/api/json?depth=0',
+ code=401,
+ msg="basic auth failed",
+ hdrs=[],
+ fp=None)
+
+ 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(),
+ u'http://example.com/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)
diff --git a/tests/jobs/test_name.py b/tests/jobs/test_name.py
new file mode 100644
index 0000000..bbcc8f8
--- /dev/null
+++ b/tests/jobs/test_name.py
@@ -0,0 +1,49 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.jobs.base import JenkinsJobsTestBase
+
+
+class JenkinsGetJobNameTest(JenkinsJobsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ job_name_to_return = {u'name': 'Test Job'}
+ jenkins_mock.return_value = json.dumps(job_name_to_return)
+
+ job_name = self.j.get_job_name(u'Test Job')
+
+ self.assertEqual(job_name, 'Test Job')
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/job/Test%20Job/api/json?tree=name')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_return_none(self, jenkins_mock):
+ jenkins_mock.side_effect = jenkins.NotFoundException()
+
+ job_name = self.j.get_job_name(u'TestJob')
+
+ self.assertEqual(job_name, None)
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/job/TestJob/api/json?tree=name')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_unexpected_job_name(self, jenkins_mock):
+ job_name_to_return = {u'name': 'not the right name'}
+ jenkins_mock.return_value = json.dumps(job_name_to_return)
+
+ 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(),
+ 'http://example.com/job/TestJob/api/json?tree=name')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'Jenkins returned an unexpected job name {0} '
+ '(expected: {1})'.format(job_name_to_return['name'], 'TestJob'))
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/jobs/test_reconfig.py b/tests/jobs/test_reconfig.py
new file mode 100644
index 0000000..09eef5d
--- /dev/null
+++ b/tests/jobs/test_reconfig.py
@@ -0,0 +1,21 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.jobs.base import JenkinsJobsTestBase
+
+
+class JenkinsReconfigJobTest(JenkinsJobsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ json.dumps({'name': 'Test Job'}),
+ None,
+ ]
+
+ self.j.reconfig_job(u'Test Job', self.config_xml)
+
+ self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/job/Test%20Job/config.xml')
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/jobs/test_rename.py b/tests/jobs/test_rename.py
new file mode 100644
index 0000000..27d9267
--- /dev/null
+++ b/tests/jobs/test_rename.py
@@ -0,0 +1,41 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.jobs.base import JenkinsJobsTestBase
+
+
+class JenkinsRenameJobTest(JenkinsJobsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ json.dumps({'name': 'Test Job_2'}),
+ json.dumps({'name': 'Test Job_2'}),
+ json.dumps({'name': 'Test Job_2'}),
+ ]
+
+ self.j.rename_job(u'Test Job', u'Test Job_2')
+
+ self.assertEqual(
+ jenkins_mock.call_args_list[0][0][0].get_full_url(),
+ 'http://example.com/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)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_failed(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ None,
+ jenkins.NotFoundException(),
+ ]
+
+ 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(),
+ 'http://example.com/job/TestJob/doRename?newName=TestJob_2')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'rename[TestJob_2] failed')
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/test_build.py b/tests/test_build.py
new file mode 100644
index 0000000..75ed0e2
--- /dev/null
+++ b/tests/test_build.py
@@ -0,0 +1,129 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.base import JenkinsTestBase
+
+
+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].get_full_url(),
+ u'http://example.com/job/Test%20Job/52/consoleText')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_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'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'
+
+ 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(
+ 'http://example.com/job/TestJob/52/consoleText',
+ code=401,
+ msg="basic auth failed",
+ hdrs=[],
+ fp=None)
+
+ 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(),
+ u'http://example.com/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)
+
+
+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].get_full_url(),
+ u'http://example.com/job/Test%20Job/52/api/json?depth=0')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_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:
+ 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.object(jenkins.Jenkins, 'jenkins_open')
+ def test_raise_HTTPError(self, jenkins_mock):
+ jenkins_mock.side_effect = jenkins.HTTPError(
+ 'http://example.com/job/TestJob/api/json?depth=0',
+ code=401,
+ msg="basic auth failed",
+ hdrs=[],
+ fp=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)
+
+
+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].get_full_url(),
+ u'http://example.com/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
new file mode 100644
index 0000000..b519be9
--- /dev/null
+++ b/tests/test_info.py
@@ -0,0 +1,91 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.base import JenkinsTestBase
+
+
+class JenkinsInfoTest(JenkinsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ job_info_to_return = {
+ u'jobs': {
+ u'url': u'http://your_url_here/job/my_job/',
+ u'color': u'blue',
+ u'name': u'my_job',
+ }
+ }
+ jenkins_mock.return_value = json.dumps(job_info_to_return)
+
+ job_info = self.j.get_info()
+
+ self.assertEqual(job_info, job_info_to_return)
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/api/json')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_raise_HTTPError(self, jenkins_mock):
+ jenkins_mock.side_effect = jenkins.HTTPError(
+ 'http://example.com/job/TestJob/api/json?depth=0',
+ code=401,
+ msg="basic auth failed",
+ hdrs=[],
+ fp=None)
+
+ with self.assertRaises(jenkins.BadHTTPException) as context_manager:
+ self.j.get_info()
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/api/json')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'Error communicating with server[http://example.com/]')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_raise_BadStatusLine(self, jenkins_mock):
+ jenkins_mock.side_effect = jenkins.BadStatusLine('not a valid status line')
+
+ with self.assertRaises(jenkins.BadHTTPException) as context_manager:
+ self.j.get_info()
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/api/json')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'Error communicating with server[http://example.com/]')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_return_invalid_json(self, jenkins_mock):
+ jenkins_mock.return_value = 'not valid JSON'
+
+ with self.assertRaises(jenkins.JenkinsException) as context_manager:
+ self.j.get_info()
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/api/json')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'Could not parse JSON info for server[http://example.com/]')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_return_empty_response(self, jenkins_mock):
+ jenkins_mock.side_effect = jenkins.JenkinsException(
+ "Error communicating with server[http://example.com/]: "
+ "empty response")
+
+ with self.assertRaises(jenkins.JenkinsException) as context_manager:
+ self.j.get_info()
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/api/json')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'Error communicating with server[http://example.com/]: '
+ 'empty response')
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/test_jenkins.py b/tests/test_jenkins.py
index c817702..5d3de72 100644
--- a/tests/test_jenkins.py
+++ b/tests/test_jenkins.py
@@ -1,17 +1,12 @@
import json
import socket
-import sys
-if sys.version_info < (2, 7):
- import unittest2 as unittest
-else:
- import unittest
from mock import patch, Mock
import six
from six.moves.urllib.error import HTTPError
-from six.moves.urllib.request import build_opener
import jenkins
+from tests.base import JenkinsTestBase
def get_mock_urlopen_return_value(a_dict=None):
@@ -20,71 +15,26 @@ def get_mock_urlopen_return_value(a_dict=None):
return six.BytesIO(json.dumps(a_dict).encode('utf-8'))
-class JenkinsTest(unittest.TestCase):
+class JenkinsConstructorTest(JenkinsTestBase):
- plugin_info_json = {
- u"plugins":
- [
- {
- u"active": u'true',
- u"backupVersion": u'null',
- u"bundled": u'true',
- u"deleted": u'false',
- u"dependencies": [],
- u"downgradable": u'false',
- u"enabled": u'true',
- u"hasUpdate": u'true',
- u"longName": u"Jenkins Mailer Plugin",
- u"pinned": u'false',
- u"shortName": u"mailer",
- u"supportsDynamicLoad": u"MAYBE",
- u"url": u"http://wiki.jenkins-ci.org/display/JENKINS/Mailer",
- u"version": u"1.5"
- }
- ]
- }
+ def test_url_with_trailing_slash(self):
+ self.assertEqual(self.j.server, 'http://example.com/')
+ self.assertEqual(self.j.auth, b'Basic dGVzdDp0ZXN0')
+ self.assertEqual(self.j.crumb, None)
- def setUp(self):
- super(JenkinsTest, self).setUp()
- self.opener = build_opener()
-
- def _check_requests(self, requests):
-
- for req in requests:
- self._check_request(req[0][0])
-
- def _check_request(self, request):
-
- # taken from opener.open() in request
- # attribute request.type is only set automatically for python 3
- # requests, must use request.get_type() for python 2.7
- protocol = request.type or request.get_type()
-
- # check that building the request doesn't throw any exception
- meth_name = protocol + "_request"
- for processor in self.opener.process_request.get(protocol, []):
- meth = getattr(processor, meth_name)
- request = meth(request)
-
- def test_constructor_url_with_trailing_slash(self):
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
- self.assertEqual(j.server, 'http://example.com/')
- self.assertEqual(j.auth, b'Basic dGVzdDp0ZXN0')
- self.assertEqual(j.crumb, None)
-
- def test_constructor_url_without_trailing_slash(self):
+ def test_url_without_trailing_slash(self):
j = jenkins.Jenkins('http://example.com', 'test', 'test')
self.assertEqual(j.server, 'http://example.com/')
self.assertEqual(j.auth, b'Basic dGVzdDp0ZXN0')
self.assertEqual(j.crumb, None)
- def test_constructor_without_user_or_password(self):
+ def test_without_user_or_password(self):
j = jenkins.Jenkins('http://example.com')
self.assertEqual(j.server, 'http://example.com/')
self.assertEqual(j.auth, None)
self.assertEqual(j.crumb, None)
- def test_constructor_unicode_password(self):
+ def test_unicode_password(self):
j = jenkins.Jenkins('http://example.com',
six.u('nonascii'),
six.u('\xe9\u20ac'))
@@ -92,7 +42,7 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(j.auth, b'Basic bm9uYXNjaWk6w6nigqw=')
self.assertEqual(j.crumb, None)
- def test_constructor_long_user_or_password(self):
+ def test_long_user_or_password(self):
long_str = 'a' * 60
long_str_b64 = 'YWFh' * 20
@@ -102,111 +52,94 @@ class JenkinsTest(unittest.TestCase):
self.assertEqual(j.auth.decode('utf-8'), 'Basic %s' % (
long_str_b64 + 'Om' + long_str_b64[2:] + 'YQ=='))
- def test_constructor_default_timeout(self):
+ def test_default_timeout(self):
j = jenkins.Jenkins('http://example.com')
self.assertEqual(j.timeout, socket._GLOBAL_DEFAULT_TIMEOUT)
- def test_constructor_custom_timeout(self):
+ def test_custom_timeout(self):
j = jenkins.Jenkins('http://example.com', timeout=300)
self.assertEqual(j.timeout, 300)
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_job_config_encodes_job_name(self, jenkins_mock):
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
- j.get_job_config(u'Test Job')
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/job/Test%20Job/config.xml')
- self._check_requests(jenkins_mock.call_args_list)
+class JenkinsMaybeAddCrumbTest(JenkinsTestBase):
@patch('jenkins.urlopen')
- def test_maybe_add_crumb(self, jenkins_mock):
+ def test_simple(self, jenkins_mock):
jenkins_mock.side_effect = jenkins.NotFoundException()
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
request = jenkins.Request('http://example.com/job/TestJob')
- j.maybe_add_crumb(request)
+ self.j.maybe_add_crumb(request)
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
'http://example.com/crumbIssuer/api/json')
- self.assertFalse(j.crumb)
+ self.assertFalse(self.j.crumb)
self.assertFalse('.crumb' in request.headers)
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.urlopen')
- def test_maybe_add_crumb__with_data(self, jenkins_mock):
- crumb_data = {
- "crumb": "dab177f483b3dd93483ef6716d8e792d",
- "crumbRequestField": ".crumb",
- }
- jenkins_mock.return_value = get_mock_urlopen_return_value(crumb_data)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
+ def test_with_data(self, jenkins_mock):
+ jenkins_mock.return_value = get_mock_urlopen_return_value(self.crumb_data)
request = jenkins.Request('http://example.com/job/TestJob')
- j.maybe_add_crumb(request)
+ self.j.maybe_add_crumb(request)
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
'http://example.com/crumbIssuer/api/json')
- self.assertEqual(j.crumb, crumb_data)
- self.assertEqual(request.headers['.crumb'], crumb_data['crumb'])
+ 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_maybe_add_crumb__empty_response(self, jenkins_mock):
+ 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")
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
request = jenkins.Request('http://example.com/job/TestJob')
- j.maybe_add_crumb(request)
+ self.j.maybe_add_crumb(request)
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
'http://example.com/crumbIssuer/api/json')
- self.assertFalse(j.crumb)
+ 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_jenkins_open(self, jenkins_mock):
- crumb_data = {
- "crumb": "dab177f483b3dd93483ef6716d8e792d",
- "crumbRequestField": ".crumb",
- }
+ def test_simple(self, jenkins_mock):
data = {'foo': 'bar'}
jenkins_mock.side_effect = [
- get_mock_urlopen_return_value(crumb_data),
+ get_mock_urlopen_return_value(self.crumb_data),
get_mock_urlopen_return_value(data),
]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
request = jenkins.Request('http://example.com/job/TestJob')
- response = j.jenkins_open(request)
+ response = self.j.jenkins_open(request)
self.assertEqual(
jenkins_mock.call_args[0][0].get_full_url(),
'http://example.com/job/TestJob')
self.assertEqual(response, json.dumps(data))
- self.assertEqual(j.crumb, crumb_data)
- self.assertEqual(request.headers['.crumb'], crumb_data['crumb'])
+ 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_jenkins_open__403(self, jenkins_mock):
+ def test_response_403(self, jenkins_mock):
jenkins_mock.side_effect = jenkins.HTTPError(
'http://example.com/job/TestJob',
code=401,
msg="basic auth failed",
hdrs=[],
fp=None)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
request = jenkins.Request('http://example.com/job/TestJob')
with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.jenkins_open(request, add_crumb=False)
+ self.j.jenkins_open(request, add_crumb=False)
self.assertEqual(
str(context_manager.exception),
'Error in request. Possibly authentication failed [401]: '
@@ -217,18 +150,17 @@ class JenkinsTest(unittest.TestCase):
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.urlopen')
- def test_jenkins_open__404(self, jenkins_mock):
+ def test_response_404(self, jenkins_mock):
jenkins_mock.side_effect = jenkins.HTTPError(
'http://example.com/job/TestJob',
code=404,
msg="basic auth failed",
hdrs=[],
fp=None)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
request = jenkins.Request('http://example.com/job/TestJob')
with self.assertRaises(jenkins.NotFoundException) as context_manager:
- j.jenkins_open(request, add_crumb=False)
+ self.j.jenkins_open(request, add_crumb=False)
self.assertEqual(
str(context_manager.exception),
'Requested item could not be found')
@@ -238,14 +170,13 @@ class JenkinsTest(unittest.TestCase):
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.urlopen')
- def test_jenkins_open__empty_response(self, jenkins_mock):
+ def test_empty_response(self, jenkins_mock):
jenkins_mock.return_value = Mock(**{'read.return_value': None})
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
request = jenkins.Request('http://example.com/job/TestJob')
with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.jenkins_open(request, False)
+ self.j.jenkins_open(request, False)
self.assertEqual(
str(context_manager.exception),
'Error communicating with server[http://example.com/]: '
@@ -256,18 +187,17 @@ class JenkinsTest(unittest.TestCase):
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.urlopen')
- def test_jenkins_open__501(self, jenkins_mock):
+ def test_response_501(self, jenkins_mock):
jenkins_mock.side_effect = jenkins.HTTPError(
'http://example.com/job/TestJob',
code=501,
msg="Not implemented",
hdrs=[],
fp=None)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
request = jenkins.Request('http://example.com/job/TestJob')
with self.assertRaises(HTTPError) as context_manager:
- j.jenkins_open(request, add_crumb=False)
+ self.j.jenkins_open(request, add_crumb=False)
self.assertEqual(
str(context_manager.exception),
'HTTP Error 501: Not implemented')
@@ -277,7 +207,7 @@ class JenkinsTest(unittest.TestCase):
self._check_requests(jenkins_mock.call_args_list)
@patch('jenkins.urlopen')
- def test_jenkins_open__timeout(self, jenkins_mock):
+ def test_timeout(self, jenkins_mock):
jenkins_mock.side_effect = jenkins.URLError(
reason="timed out")
j = jenkins.Jenkins('http://example.com/', 'test', 'test', timeout=1)
@@ -292,1581 +222,3 @@ class JenkinsTest(unittest.TestCase):
jenkins_mock.call_args[0][0].get_full_url(),
'http://example.com/job/TestJob')
self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_assert_job_exists__job_missing(self, jenkins_mock):
- jenkins_mock.side_effect = jenkins.NotFoundException()
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.assert_job_exists('NonExistent')
- self.assertEqual(
- str(context_manager.exception),
- 'job[NonExistent] does not exist')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_assert_job_exists__job_exists(self, jenkins_mock):
- jenkins_mock.side_effect = [
- json.dumps({'name': 'ExistingJob'}),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
- j.assert_job_exists('ExistingJob')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_create_job(self, jenkins_mock):
- config_xml = """
- <matrix-project>
- <actions/>
- <description>Foo</description>
- </matrix-project>"""
- jenkins_mock.side_effect = [
- jenkins.NotFoundException(),
- None,
- json.dumps({'name': 'Test Job'}),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.create_job(u'Test Job', config_xml)
-
- self.assertEqual(
- jenkins_mock.call_args_list[1][0][0].get_full_url(),
- 'http://example.com/createItem?name=Test%20Job')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_create_job__already_exists(self, jenkins_mock):
- config_xml = """
- <matrix-project>
- <actions/>
- <description>Foo</description>
- </matrix-project>"""
- jenkins_mock.side_effect = [
- json.dumps({'name': 'TestJob'}),
- None,
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.create_job(u'TestJob', config_xml)
- self.assertEqual(
- jenkins_mock.call_args_list[0][0][0].get_full_url(),
- 'http://example.com/job/TestJob/api/json?tree=name')
- self.assertEqual(
- str(context_manager.exception),
- 'job[TestJob] already exists')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_create_job__create_failed(self, jenkins_mock):
- config_xml = """
- <matrix-project>
- <actions/>
- <description>Foo</description>
- </matrix-project>"""
- jenkins_mock.side_effect = [
- jenkins.NotFoundException(),
- None,
- jenkins.NotFoundException(),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.create_job(u'TestJob', config_xml)
- self.assertEqual(
- jenkins_mock.call_args_list[0][0][0].get_full_url(),
- 'http://example.com/job/TestJob/api/json?tree=name')
- self.assertEqual(
- jenkins_mock.call_args_list[1][0][0].get_full_url(),
- 'http://example.com/createItem?name=TestJob')
- self.assertEqual(
- str(context_manager.exception),
- 'create[TestJob] failed')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_reconfig_job(self, jenkins_mock):
- config_xml = """
- <matrix-project>
- <actions/>
- <description>Foo</description>
- </matrix-project>"""
- jenkins_mock.side_effect = [
- json.dumps({'name': 'Test Job'}),
- None,
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.reconfig_job(u'Test Job', config_xml)
-
- self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/job/Test%20Job/config.xml')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_build_job(self, jenkins_mock):
- jenkins_mock.side_effect = [
- {'foo': 'bar'},
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- build_info = j.build_job(u'Test Job')
-
- self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/job/Test%20Job/build')
- self.assertEqual(build_info, {'foo': 'bar'})
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_build_job__with_token(self, jenkins_mock):
- jenkins_mock.side_effect = [
- {'foo': 'bar'},
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- build_info = j.build_job(u'TestJob', token='some_token')
-
- self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/job/TestJob/build?token=some_token')
- self.assertEqual(build_info, {'foo': 'bar'})
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_build_job__with_parameters_and_token(self, jenkins_mock):
- jenkins_mock.side_effect = [
- {'foo': 'bar'},
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- build_info = j.build_job(
- u'TestJob',
- 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.assertEqual(build_info, {'foo': 'bar'})
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_run_script(self, jenkins_mock):
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.run_script(u'println(\"Hello World!\")')
-
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/scriptText')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_stop_build(self, jenkins_mock):
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.stop_build(u'Test Job', number=52)
-
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/job/Test%20Job/52/stop')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_build_console_output(self, jenkins_mock):
- jenkins_mock.return_value = "build console output..."
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- build_info = 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].get_full_url(),
- u'http://example.com/job/Test%20Job/52/consoleText')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_build_console_output__None(self, jenkins_mock):
- jenkins_mock.return_value = None
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- 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_get_build_console_output__invalid_json(self, jenkins_mock):
- jenkins_mock.return_value = 'Invalid JSON'
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- console_output = j.get_build_console_output(u'TestJob', number=52)
- self.assertEqual(console_output, jenkins_mock.return_value)
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_build_console_output__HTTPError(self, jenkins_mock):
- jenkins_mock.side_effect = jenkins.HTTPError(
- 'http://example.com/job/TestJob/52/consoleText',
- code=401,
- msg="basic auth failed",
- hdrs=[],
- fp=None)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.get_build_console_output(u'TestJob', number=52)
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/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_get_build_info(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)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- build_info = 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].get_full_url(),
- u'http://example.com/job/Test%20Job/52/api/json?depth=0')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_build_info__None(self, jenkins_mock):
- jenkins_mock.return_value = None
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- 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_get_build_info__invalid_json(self, jenkins_mock):
- jenkins_mock.return_value = 'Invalid JSON'
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- 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.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_build_info__HTTPError(self, jenkins_mock):
- jenkins_mock.side_effect = jenkins.HTTPError(
- 'http://example.com/job/TestJob/api/json?depth=0',
- code=401,
- msg="basic auth failed",
- hdrs=[],
- fp=None)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- 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_get_job_info(self, jenkins_mock):
- job_info_to_return = {
- u'building': False,
- u'msg': u'test',
- u'revision': 66,
- u'user': u'unknown'
- }
- jenkins_mock.return_value = json.dumps(job_info_to_return)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- job_info = j.get_job_info(u'Test Job')
-
- self.assertEqual(job_info, job_info_to_return)
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/job/Test%20Job/api/json?depth=0')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_job_info_regex(self, jenkins_mock):
- jobs = [
- {u'name': u'my-job-1'},
- {u'name': u'my-job-2'},
- {u'name': u'your-job-1'},
- {u'name': u'Your-Job-1'},
- {u'name': u'my-project-1'},
- ]
- job_info_to_return = {u'jobs': jobs}
- jenkins_mock.return_value = json.dumps(job_info_to_return)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- self.assertEqual(len(j.get_job_info_regex('her-job')), 0)
- self.assertEqual(len(j.get_job_info_regex('my-job-1')), 1)
- self.assertEqual(len(j.get_job_info_regex('my-job')), 2)
- self.assertEqual(len(j.get_job_info_regex('job')), 3)
- self.assertEqual(len(j.get_job_info_regex('project')), 1)
- self.assertEqual(len(j.get_job_info_regex('[Yy]our-[Jj]ob-1')), 2)
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_job_info__None(self, jenkins_mock):
- jenkins_mock.return_value = None
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.get_job_info(u'TestJob')
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/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_get_job_info__invalid_json(self, jenkins_mock):
- jenkins_mock.return_value = 'Invalid JSON'
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.get_job_info(u'TestJob')
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/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_get_job_info__HTTPError(self, jenkins_mock):
- jenkins_mock.side_effect = jenkins.HTTPError(
- 'http://example.com/job/TestJob/api/json?depth=0',
- code=401,
- msg="basic auth failed",
- hdrs=[],
- fp=None)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.get_job_info(u'TestJob')
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/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_debug_job_info(self, jenkins_mock):
- job_info_to_return = {
- u'building': False,
- u'msg': u'test',
- u'revision': 66,
- u'user': u'unknown'
- }
- jenkins_mock.return_value = json.dumps(job_info_to_return)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.debug_job_info(u'Test Job')
-
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/job/Test%20Job/api/json?depth=0')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch('jenkins.urlopen')
- def test_get_version__some_version(self, urlopen_mock):
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- 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]
- self.assertEqual(j.get_version(), 'Version42')
- self._check_requests(urlopen_mock.call_args_list)
-
- @patch('jenkins.urlopen')
- def test_get_version__HTTPError(self, urlopen_mock):
- urlopen_mock.side_effect = jenkins.HTTPError(
- 'http://example.com/',
- code=503,
- msg="internal server error",
- hdrs=[],
- fp=None)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
- with self.assertRaises(jenkins.BadHTTPException) as context_manager:
- j.get_version()
- self.assertEqual(
- str(context_manager.exception),
- 'Error communicating with server[http://example.com/]')
- self._check_requests(urlopen_mock.call_args_list)
-
- @patch('jenkins.urlopen')
- def test_get_version__BadStatusLine(self, urlopen_mock):
- urlopen_mock.side_effect = jenkins.BadStatusLine('not a valid status line')
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
- with self.assertRaises(jenkins.BadHTTPException) as context_manager:
- j.get_version()
- self.assertEqual(
- str(context_manager.exception),
- 'Error communicating with server[http://example.com/]')
- self._check_requests(urlopen_mock.call_args_list)
-
- @patch('jenkins.urlopen', return_value=None)
- def test_get_version__empty_response(self, urlopen_mock):
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
- with self.assertRaises(jenkins.EmptyResponseException) as context_manager:
- j.get_version()
- self.assertEqual(
- str(context_manager.exception),
- 'Error communicating with server[http://example.com/]:'
- ' empty response')
- self._check_requests(urlopen_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_jobs_count(self, jenkins_mock):
- jobs = [
- {u'url': u'http://localhost:8080/job/guava/', u'color': u'notbuilt', u'name': u'guava'},
- {u'url': u'http://localhost:8080/job/kiwi/', u'color': u'blue', u'name': u'kiwi'},
- {u'url': u'http://localhost:8080/job/lemon/', u'color': u'red', u'name': u'lemon'}
- ]
- job_info_to_return = {u'jobs': jobs}
- jenkins_mock.return_value = json.dumps(job_info_to_return)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
- self.assertEqual(j.jobs_count(), 3)
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_jobs(self, jenkins_mock):
- jobs = {
- u'url': u'http://your_url_here/job/my_job/',
- u'color': u'blue',
- u'name': u'my_job',
- }
- job_info_to_return = {u'jobs': jobs}
- jenkins_mock.return_value = json.dumps(job_info_to_return)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- job_info = j.get_jobs()
-
- self.assertEqual(job_info, jobs)
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/api/json')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_info(self, jenkins_mock):
- job_info_to_return = {
- u'jobs': {
- u'url': u'http://your_url_here/job/my_job/',
- u'color': u'blue',
- u'name': u'my_job',
- }
- }
- jenkins_mock.return_value = json.dumps(job_info_to_return)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- job_info = j.get_info()
-
- self.assertEqual(job_info, job_info_to_return)
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/api/json')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_plugins_info(self, jenkins_mock):
-
- jenkins_mock.return_value = json.dumps(self.plugin_info_json)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- # expected to return a list of plugins
- plugins_info = j.get_plugins_info()
- self.assertEqual(plugins_info, self.plugin_info_json['plugins'])
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/pluginManager/api/json?depth=2')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_plugins_info_none(self, jenkins_mock):
-
- empty_plugin_info_json = {u"plugins": []}
-
- jenkins_mock.return_value = json.dumps(empty_plugin_info_json)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- plugins_info = j.get_plugins_info()
- self.assertEqual(plugins_info, [])
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_plugins_info_depth(self, jenkins_mock):
-
- jenkins_mock.return_value = json.dumps(self.plugin_info_json)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.get_plugins_info(depth=1)
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/pluginManager/api/json?depth=1')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_plugins_info__BadStatusLine(self, jenkins_mock):
- jenkins_mock.side_effect = jenkins.BadStatusLine('not a valid status line')
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.BadHTTPException) as context_manager:
- j.get_plugins_info()
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/pluginManager/api/json?depth=2')
- self.assertEqual(
- str(context_manager.exception),
- 'Error communicating with server[http://example.com/]')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_plugins_info__ValueError(self, jenkins_mock):
- jenkins_mock.return_value = 'not valid JSON'
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.get_plugins_info()
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/pluginManager/api/json?depth=2')
- self.assertEqual(
- str(context_manager.exception),
- 'Could not parse JSON info for server[http://example.com/]')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_plugins_info__HTTPError(self, jenkins_mock):
- jenkins_mock.side_effect = jenkins.HTTPError(
- 'http://example.com/job/pluginManager/api/json?depth=2',
- code=401,
- msg="basic auth failed",
- hdrs=[],
- fp=None)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.BadHTTPException) as context_manager:
- j.get_plugins_info(depth=52)
- self.assertEqual(
- str(context_manager.exception),
- 'Error communicating with server[http://example.com/]')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_plugin_info_shortname(self, jenkins_mock):
-
- jenkins_mock.return_value = json.dumps(self.plugin_info_json)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- # expected to return info on a single plugin
- plugin_info = j.get_plugin_info("mailer")
- self.assertEqual(plugin_info, self.plugin_info_json['plugins'][0])
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_plugin_info_longname(self, jenkins_mock):
-
- jenkins_mock.return_value = json.dumps(self.plugin_info_json)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- # expected to return info on a single plugin
- plugin_info = j.get_plugin_info("Jenkins Mailer Plugin")
- self.assertEqual(plugin_info, self.plugin_info_json['plugins'][0])
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_plugin_info_none(self, jenkins_mock):
-
- jenkins_mock.return_value = json.dumps(self.plugin_info_json)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- # expected not to find bogus so should return None
- plugin_info = j.get_plugin_info("bogus")
- self.assertEqual(plugin_info, None)
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_plugin_info_depth(self, jenkins_mock):
-
- jenkins_mock.return_value = json.dumps(self.plugin_info_json)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.get_plugin_info('test', depth=1)
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/pluginManager/api/json?depth=1')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_plugin_info__BadStatusLine(self, jenkins_mock):
- jenkins_mock.side_effect = jenkins.BadStatusLine('not a valid status line')
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.get_plugin_info('test')
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/pluginManager/api/json?depth=2')
- self.assertEqual(
- str(context_manager.exception),
- 'Error communicating with server[http://example.com/]')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_plugin_info__ValueError(self, jenkins_mock):
- jenkins_mock.return_value = 'not valid JSON'
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.get_plugin_info('test')
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/pluginManager/api/json?depth=2')
- self.assertEqual(
- str(context_manager.exception),
- 'Could not parse JSON info for server[http://example.com/]')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_plugin_info__HTTPError(self, jenkins_mock):
- jenkins_mock.side_effect = jenkins.HTTPError(
- 'http://example.com/job/pluginManager/api/json?depth=2',
- code=401,
- msg="basic auth failed",
- hdrs=[],
- fp=None)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.get_plugin_info(u'TestPlugin', depth=52)
- self.assertEqual(
- str(context_manager.exception),
- 'Error communicating with server[http://example.com/]')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_info__HTTPError(self, jenkins_mock):
- jenkins_mock.side_effect = jenkins.HTTPError(
- 'http://example.com/job/TestJob/api/json?depth=0',
- code=401,
- msg="basic auth failed",
- hdrs=[],
- fp=None)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.BadHTTPException) as context_manager:
- j.get_info()
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/api/json')
- self.assertEqual(
- str(context_manager.exception),
- 'Error communicating with server[http://example.com/]')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_info__BadStatusLine(self, jenkins_mock):
- jenkins_mock.side_effect = jenkins.BadStatusLine('not a valid status line')
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.BadHTTPException) as context_manager:
- j.get_info()
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/api/json')
- self.assertEqual(
- str(context_manager.exception),
- 'Error communicating with server[http://example.com/]')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_info__ValueError(self, jenkins_mock):
- jenkins_mock.return_value = 'not valid JSON'
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.get_info()
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/api/json')
- self.assertEqual(
- str(context_manager.exception),
- 'Could not parse JSON info for server[http://example.com/]')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_info__empty_response(self, jenkins_mock):
- jenkins_mock.side_effect = jenkins.JenkinsException(
- "Error communicating with server[http://example.com/]: "
- "empty response")
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.get_info()
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/api/json')
- self.assertEqual(
- str(context_manager.exception),
- 'Error communicating with server[http://example.com/]: '
- 'empty response')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_copy_job(self, jenkins_mock):
- jenkins_mock.side_effect = [
- json.dumps({'name': 'Test Job_2'}),
- json.dumps({'name': 'Test Job_2'}),
- json.dumps({'name': 'Test Job_2'}),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.copy_job(u'Test Job', u'Test Job_2')
-
- self.assertEqual(
- jenkins_mock.call_args_list[0][0][0].get_full_url(),
- 'http://example.com/createItem'
- '?name=Test%20Job_2&mode=copy&from=Test%20Job')
- self.assertTrue(j.job_exists('Test Job_2'))
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_copy_job__create_failed(self, jenkins_mock):
- jenkins_mock.side_effect = [
- None,
- jenkins.NotFoundException(),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.copy_job(u'TestJob', u'TestJob_2')
- self.assertEqual(
- jenkins_mock.call_args_list[0][0][0].get_full_url(),
- 'http://example.com/createItem'
- '?name=TestJob_2&mode=copy&from=TestJob')
- self.assertEqual(
- str(context_manager.exception),
- 'create[TestJob_2] failed')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_rename_job(self, jenkins_mock):
- jenkins_mock.side_effect = [
- json.dumps({'name': 'Test Job_2'}),
- json.dumps({'name': 'Test Job_2'}),
- json.dumps({'name': 'Test Job_2'}),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.rename_job(u'Test Job', u'Test Job_2')
-
- self.assertEqual(
- jenkins_mock.call_args_list[0][0][0].get_full_url(),
- 'http://example.com/job/Test%20Job/doRename?newName=Test%20Job_2')
- self.assertTrue(j.job_exists('Test Job_2'))
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_rename_job__rename_failed(self, jenkins_mock):
- jenkins_mock.side_effect = [
- None,
- jenkins.NotFoundException(),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.rename_job(u'TestJob', u'TestJob_2')
- self.assertEqual(
- jenkins_mock.call_args_list[0][0][0].get_full_url(),
- 'http://example.com/job/TestJob/doRename?newName=TestJob_2')
- self.assertEqual(
- str(context_manager.exception),
- 'rename[TestJob_2] failed')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_delete_job(self, jenkins_mock):
- jenkins_mock.side_effect = [
- None,
- jenkins.NotFoundException(),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.delete_job(u'Test Job')
-
- self.assertEqual(
- jenkins_mock.call_args_list[0][0][0].get_full_url(),
- 'http://example.com/job/Test%20Job/doDelete')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_delete_job__delete_failed(self, jenkins_mock):
- jenkins_mock.side_effect = [
- json.dumps({'name': 'TestJob'}),
- json.dumps({'name': 'TestJob'}),
- json.dumps({'name': 'TestJob'}),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.delete_job(u'TestJob')
- self.assertEqual(
- jenkins_mock.call_args_list[0][0][0].get_full_url(),
- 'http://example.com/job/TestJob/doDelete')
- self.assertEqual(
- str(context_manager.exception),
- 'delete[TestJob] failed')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_enable_job(self, jenkins_mock):
- jenkins_mock.side_effect = [
- json.dumps({'name': 'TestJob'}),
- json.dumps({'name': 'TestJob'}),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.enable_job(u'TestJob')
-
- self.assertEqual(
- jenkins_mock.call_args_list[0][0][0].get_full_url(),
- 'http://example.com/job/TestJob/enable')
- self.assertTrue(j.job_exists('TestJob'))
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_disable_job(self, jenkins_mock):
- jenkins_mock.side_effect = [
- json.dumps({'name': 'Test Job'}),
- json.dumps({'name': 'Test Job'}),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.disable_job(u'Test Job')
-
- self.assertEqual(
- jenkins_mock.call_args_list[0][0][0].get_full_url(),
- 'http://example.com/job/Test%20Job/disable')
- self.assertTrue(j.job_exists('Test Job'))
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_job_name(self, jenkins_mock):
- job_name_to_return = {u'name': 'Test Job'}
- jenkins_mock.return_value = json.dumps(job_name_to_return)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- job_name = j.get_job_name(u'Test Job')
-
- self.assertEqual(job_name, 'Test Job')
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/job/Test%20Job/api/json?tree=name')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_job_name__None(self, jenkins_mock):
- jenkins_mock.side_effect = jenkins.NotFoundException()
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- job_name = j.get_job_name(u'TestJob')
-
- self.assertEqual(job_name, None)
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/job/TestJob/api/json?tree=name')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_job_name__unexpected_job_name(self, jenkins_mock):
- job_name_to_return = {u'name': 'not the right name'}
- jenkins_mock.return_value = json.dumps(job_name_to_return)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.get_job_name(u'TestJob')
- self.assertEqual(
- jenkins_mock.call_args_list[0][0][0].get_full_url(),
- 'http://example.com/job/TestJob/api/json?tree=name')
- self.assertEqual(
- str(context_manager.exception),
- 'Jenkins returned an unexpected job name {0} '
- '(expected: {1})'.format(job_name_to_return['name'], 'TestJob'))
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_cancel_queue(self, jenkins_mock):
- job_name_to_return = {u'name': 'TestJob'}
- jenkins_mock.return_value = json.dumps(job_name_to_return)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.cancel_queue(52)
-
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/queue/cancelItem?id=52')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open',
- side_effect=jenkins.NotFoundException('not found'))
- def test_cancel_queue__notfound(self, jenkins_mock):
- job_name_to_return = {u'name': 'TestJob'}
- jenkins_mock.return_value = json.dumps(job_name_to_return)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.cancel_queue(52)
-
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/queue/cancelItem?id=52')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_nodes(self, jenkins_mock):
- jenkins_mock.return_value = json.dumps({
- "computer": [{
- "displayName": "master",
- "offline": False
- }],
- "busyExecutors": 2})
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
- self.assertEqual(j.get_nodes(),
- [{'name': 'master', 'offline': False}])
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_nodes__invalid_json(self, jenkins_mock):
- jenkins_mock.side_effect = [
- 'Invalid JSON',
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.get_nodes()
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- 'http://example.com/computer/api/json')
- self.assertEqual(
- str(context_manager.exception),
- 'Could not parse JSON info for server[http://example.com/]')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch('jenkins.urlopen')
- def test_get_nodes__BadStatusLine(self, urlopen_mock):
- urlopen_mock.side_effect = jenkins.BadStatusLine('not a valid status line')
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
- with self.assertRaises(jenkins.BadHTTPException) as context_manager:
- j.get_nodes()
- self.assertEqual(
- str(context_manager.exception),
- 'Error communicating with server[http://example.com/]')
- self._check_requests(urlopen_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_nodes__HTTPError(self, jenkins_mock):
- jenkins_mock.side_effect = jenkins.HTTPError(
- 'http://example.com/job/TestJob',
- code=401,
- msg="basic auth failed",
- hdrs=[],
- fp=None)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.get_nodes()
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- 'http://example.com/computer/api/json')
- self.assertEqual(
- str(context_manager.exception),
- 'Error communicating with server[http://example.com/]')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_node_info(self, jenkins_mock):
- node_info = {
- 'displayName': 'nodes',
- 'totalExecutors': 5,
- }
- jenkins_mock.side_effect = [
- json.dumps(node_info),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- self.assertEqual(j.get_node_info('test node'), node_info)
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- 'http://example.com/computer/test%20node/api/json?depth=0')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_node_info__invalid_json(self, jenkins_mock):
- jenkins_mock.side_effect = [
- 'Invalid JSON',
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.get_node_info('test_node')
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- 'http://example.com/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_get_node_info__HTTPError(self, jenkins_mock):
- jenkins_mock.side_effect = jenkins.HTTPError(
- 'http://example.com/job/TestJob',
- code=401,
- msg="basic auth failed",
- hdrs=[],
- fp=None)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.get_node_info('test_node')
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- 'http://example.com/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)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_assert_node_exists__node_missing(self, jenkins_mock):
- jenkins_mock.side_effect = [None]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.assert_node_exists('NonExistentNode')
- self.assertEqual(
- str(context_manager.exception),
- 'node[NonExistentNode] does not exist')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_assert_node_exists__node_exists(self, jenkins_mock):
- jenkins_mock.side_effect = [
- json.dumps({'name': 'ExistingNode'})
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
- j.assert_node_exists('ExistingNode')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_delete_node(self, jenkins_mock):
- node_info = {
- 'displayName': 'nodes',
- 'totalExecutors': 5,
- }
- jenkins_mock.side_effect = [
- json.dumps(node_info),
- None,
- None,
- None,
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.delete_node('test node')
-
- self.assertEqual(
- jenkins_mock.call_args_list[1][0][0].get_full_url(),
- 'http://example.com/computer/test%20node/doDelete')
- self.assertFalse(j.node_exists('test node'))
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_delete_node__delete_failed(self, jenkins_mock):
- node_info = {
- 'displayName': 'nodes',
- 'totalExecutors': 5,
- }
- jenkins_mock.side_effect = [
- json.dumps(node_info),
- None,
- json.dumps(node_info),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.delete_node('test_node')
- self.assertEqual(
- jenkins_mock.call_args_list[1][0][0].get_full_url(),
- 'http://example.com/computer/test_node/doDelete')
- self.assertEqual(
- str(context_manager.exception),
- 'delete[test_node] failed')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_create_node(self, jenkins_mock):
- node_info = {
- 'displayName': 'test node',
- 'totalExecutors': 5,
- }
- jenkins_mock.side_effect = [
- None,
- None,
- json.dumps(node_info),
- json.dumps(node_info),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.create_node('test node', exclusive=True)
-
- self.assertEqual(
- jenkins_mock.call_args_list[1][0][0].get_full_url().split('?')[0],
- 'http://example.com/computer/doCreateItem')
- self.assertTrue(j.node_exists('test node'))
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_create_node__node_already_exists(self, jenkins_mock):
- node_info = {
- 'displayName': 'nodes',
- 'totalExecutors': 5,
- }
- jenkins_mock.side_effect = [
- json.dumps(node_info),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- 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_create_node__create_failed(self, jenkins_mock):
- jenkins_mock.side_effect = [
- None,
- None,
- None,
- None,
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.create_node('test_node')
- self.assertEqual(
- jenkins_mock.call_args_list[1][0][0].get_full_url().split('?')[0],
- 'http://example.com/computer/doCreateItem')
- self.assertEqual(
- str(context_manager.exception),
- 'create[test_node] failed')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_enable_node(self, jenkins_mock):
- expected_node_info = {
- 'displayName': 'test node',
- 'totalExecutors': 5,
- 'offline': True,
- }
- jenkins_mock.side_effect = [
- json.dumps(expected_node_info),
- None,
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.enable_node('test node')
-
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- 'http://example.com/computer/test%20node/' +
- 'toggleOffline?offlineMessage=')
-
- expected_node_info = {
- 'displayName': 'test node',
- 'totalExecutors': 5,
- 'offline': False,
- }
- jenkins_mock.side_effect = [json.dumps(expected_node_info)]
- node_info = j.get_node_info('test node')
- self.assertEqual(node_info, expected_node_info)
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_enable_node__offline_False(self, jenkins_mock):
- node_info = {
- 'displayName': 'nodes',
- 'totalExecutors': 5,
- 'offline': False,
- }
- jenkins_mock.side_effect = [
- json.dumps(node_info),
- None,
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.enable_node('test_node')
-
- # 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(),
- 'http://example.com/computer/test_node/api/json?depth=0')
-
- expected_node_info = {
- 'displayName': 'nodes',
- 'totalExecutors': 5,
- 'offline': False,
- }
- jenkins_mock.side_effect = [json.dumps(expected_node_info)]
- node_info = j.get_node_info('test_node')
- self.assertEqual(node_info, expected_node_info)
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_disable_node(self, jenkins_mock):
- node_info = {
- 'displayName': 'test node',
- 'totalExecutors': 5,
- 'offline': False,
- }
- jenkins_mock.side_effect = [
- json.dumps(node_info),
- None,
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.disable_node('test node')
-
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- 'http://example.com/computer/test%20node/' +
- 'toggleOffline?offlineMessage=')
-
- expected_node_info = {
- 'displayName': 'test node',
- 'totalExecutors': 5,
- 'offline': True,
- }
- jenkins_mock.side_effect = [json.dumps(expected_node_info)]
- node_info = j.get_node_info('test node')
- self.assertEqual(node_info, expected_node_info)
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_disable_node__offline_True(self, jenkins_mock):
- node_info = {
- 'displayName': 'nodes',
- 'totalExecutors': 5,
- 'offline': True,
- }
- jenkins_mock.side_effect = [
- json.dumps(node_info),
- None,
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.disable_node('test_node')
-
- # 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(),
- 'http://example.com/computer/test_node/api/json?depth=0')
-
- expected_node_info = {
- 'displayName': 'nodes',
- 'totalExecutors': 5,
- 'offline': True,
- }
- jenkins_mock.side_effect = [json.dumps(expected_node_info)]
- node_info = j.get_node_info('test_node')
- self.assertEqual(node_info, expected_node_info)
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_queue_info(self, jenkins_mock):
- queue_info_to_return = {
- 'items': {
- 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,
- }
- }
- jenkins_mock.return_value = json.dumps(queue_info_to_return)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- queue_info = j.get_queue_info()
-
- self.assertEqual(queue_info, queue_info_to_return['items'])
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/queue/api/json?depth=0')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_view_name(self, jenkins_mock):
- view_name_to_return = {u'name': 'Test View'}
- jenkins_mock.return_value = json.dumps(view_name_to_return)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- view_name = j.get_view_name(u'Test View')
-
- self.assertEqual(view_name, 'Test View')
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/view/Test%20View/api/json?tree=name')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_view_name__None(self, jenkins_mock):
- jenkins_mock.side_effect = jenkins.NotFoundException()
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- view_name = j.get_view_name(u'TestView')
-
- self.assertEqual(view_name, None)
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/view/TestView/api/json?tree=name')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_view_name__unexpected_view_name(self, jenkins_mock):
- view_name_to_return = {u'name': 'not the right name'}
- jenkins_mock.return_value = json.dumps(view_name_to_return)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.get_view_name(u'TestView')
- self.assertEqual(
- jenkins_mock.call_args_list[0][0][0].get_full_url(),
- 'http://example.com/view/TestView/api/json?tree=name')
- self.assertEqual(
- str(context_manager.exception),
- 'Jenkins returned an unexpected view name {0} '
- '(expected: {1})'.format(view_name_to_return['name'], 'TestView'))
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_assert_view_exists__view_missing(self, jenkins_mock):
- jenkins_mock.side_effect = jenkins.NotFoundException()
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.assert_view_exists('NonExistent')
- self.assertEqual(
- str(context_manager.exception),
- 'view[NonExistent] does not exist')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_assert_view_exists__view_exists(self, jenkins_mock):
- jenkins_mock.side_effect = [
- json.dumps({'name': 'ExistingView'}),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
- j.assert_view_exists('ExistingView')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_views(self, jenkins_mock):
- views = {
- u'url': u'http://your_url_here/view/my_view/',
- u'name': u'my_view',
- }
- view_info_to_return = {u'views': views}
- jenkins_mock.return_value = json.dumps(view_info_to_return)
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- view_info = j.get_views()
-
- self.assertEqual(view_info, views)
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/api/json')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_delete_view(self, jenkins_mock):
- jenkins_mock.side_effect = [
- None,
- jenkins.NotFoundException(),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.delete_view(u'Test View')
-
- self.assertEqual(
- jenkins_mock.call_args_list[0][0][0].get_full_url(),
- 'http://example.com/view/Test%20View/doDelete')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_delete_view__delete_failed(self, jenkins_mock):
- jenkins_mock.side_effect = [
- json.dumps({'name': 'TestView'}),
- json.dumps({'name': 'TestView'}),
- json.dumps({'name': 'TestView'}),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.delete_view(u'TestView')
- self.assertEqual(
- jenkins_mock.call_args_list[0][0][0].get_full_url(),
- 'http://example.com/view/TestView/doDelete')
- self.assertEqual(
- str(context_manager.exception),
- 'delete[TestView] failed')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_create_view(self, jenkins_mock):
- config_xml = """
- <listView>
- <description>Foo</description>
- <jobNames />
- </listView>"""
- jenkins_mock.side_effect = [
- jenkins.NotFoundException(),
- None,
- json.dumps({'name': 'Test View'}),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.create_view(u'Test View', config_xml)
-
- self.assertEqual(
- jenkins_mock.call_args_list[1][0][0].get_full_url(),
- 'http://example.com/createView?name=Test%20View')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_create_view__already_exists(self, jenkins_mock):
- config_xml = """
- <listView>
- <description>Foo</description>
- <jobNames />
- </listView>"""
- jenkins_mock.side_effect = [
- json.dumps({'name': 'TestView'}),
- None,
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.create_view(u'TestView', config_xml)
- self.assertEqual(
- jenkins_mock.call_args_list[0][0][0].get_full_url(),
- 'http://example.com/view/TestView/api/json?tree=name')
- self.assertEqual(
- str(context_manager.exception),
- 'view[TestView] already exists')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_create_view__create_failed(self, jenkins_mock):
- config_xml = """
- <listView>
- <description>Foo</description>
- <jobNames />
- </listView>"""
- jenkins_mock.side_effect = [
- jenkins.NotFoundException(),
- None,
- jenkins.NotFoundException(),
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- with self.assertRaises(jenkins.JenkinsException) as context_manager:
- j.create_view(u'TestView', config_xml)
- self.assertEqual(
- jenkins_mock.call_args_list[0][0][0].get_full_url(),
- 'http://example.com/view/TestView/api/json?tree=name')
- self.assertEqual(
- jenkins_mock.call_args_list[1][0][0].get_full_url(),
- 'http://example.com/createView?name=TestView')
- self.assertEqual(
- str(context_manager.exception),
- 'create[TestView] failed')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_reconfig_view(self, jenkins_mock):
- config_xml = """
- <listView>
- <description>Foo</description>
- <jobNames />
- </listView>"""
- jenkins_mock.side_effect = [
- json.dumps({'name': 'Test View'}),
- None,
- ]
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
-
- j.reconfig_view(u'Test View', config_xml)
-
- self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/view/Test%20View/config.xml')
- self._check_requests(jenkins_mock.call_args_list)
-
- @patch.object(jenkins.Jenkins, 'jenkins_open')
- def test_get_view_config_encodes_view_name(self, jenkins_mock):
- j = jenkins.Jenkins('http://example.com/', 'test', 'test')
- j.get_view_config(u'Test View')
-
- self.assertEqual(
- jenkins_mock.call_args[0][0].get_full_url(),
- u'http://example.com/view/Test%20View/config.xml')
- self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/test_node.py b/tests/test_node.py
new file mode 100644
index 0000000..854349e
--- /dev/null
+++ b/tests/test_node.py
@@ -0,0 +1,325 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.base import JenkinsTestBase
+
+
+class JenkinsNodesTestBase(JenkinsTestBase):
+
+ def setUp(self):
+ super(JenkinsNodesTestBase, self).setUp()
+ self.node_info = {
+ 'displayName': 'test node',
+ 'totalExecutors': 5,
+ }
+ self.online_node_info = dict(self.node_info)
+ self.online_node_info['offline'] = False
+ self.offline_node_info = dict(self.node_info)
+ self.offline_node_info['offline'] = True
+
+
+class JenkinsGetNodesTest(JenkinsNodesTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jenkins_mock.return_value = json.dumps({
+ "computer": [{
+ "displayName": "master",
+ "offline": False
+ }],
+ "busyExecutors": 2})
+ self.assertEqual(self.j.get_nodes(),
+ [{'name': 'master', 'offline': False}])
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_return_invalid_json(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ 'Invalid JSON',
+ ]
+
+ with self.assertRaises(jenkins.JenkinsException) as context_manager:
+ self.j.get_nodes()
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ 'http://example.com/computer/api/json')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'Could not parse JSON info for server[http://example.com/]')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch('jenkins.urlopen')
+ def test_raise_BadStatusLine(self, urlopen_mock):
+ urlopen_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[http://example.com/]')
+ self._check_requests(urlopen_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_raise_HTTPError(self, jenkins_mock):
+ jenkins_mock.side_effect = jenkins.HTTPError(
+ 'http://example.com/job/TestJob',
+ code=401,
+ msg="basic auth failed",
+ hdrs=[],
+ fp=None)
+
+ with self.assertRaises(jenkins.JenkinsException) as context_manager:
+ self.j.get_nodes()
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ 'http://example.com/computer/api/json')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'Error communicating with server[http://example.com/]')
+ self._check_requests(jenkins_mock.call_args_list)
+
+
+class JenkinsGetNodeInfoTest(JenkinsNodesTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ json.dumps(self.node_info),
+ ]
+
+ self.assertEqual(self.j.get_node_info('test node'), self.node_info)
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ 'http://example.com/computer/test%20node/api/json?depth=0')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_return_invalid_json(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ 'Invalid JSON',
+ ]
+
+ 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(),
+ 'http://example.com/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(
+ 'http://example.com/job/TestJob',
+ code=401,
+ msg="basic auth failed",
+ hdrs=[],
+ fp=None)
+
+ 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(),
+ 'http://example.com/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):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_node_missing(self, jenkins_mock):
+ jenkins_mock.side_effect = [None]
+
+ with self.assertRaises(jenkins.JenkinsException) as context_manager:
+ self.j.assert_node_exists('NonExistentNode')
+ 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):
+ jenkins_mock.side_effect = [
+ json.dumps({'name': 'ExistingNode'})
+ ]
+ self.j.assert_node_exists('ExistingNode')
+ self._check_requests(jenkins_mock.call_args_list)
+
+
+class JenkinsDeleteNodeTest(JenkinsNodesTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ json.dumps(self.node_info),
+ None,
+ None,
+ None,
+ ]
+
+ self.j.delete_node('test node')
+
+ self.assertEqual(
+ jenkins_mock.call_args_list[1][0][0].get_full_url(),
+ 'http://example.com/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):
+ jenkins_mock.side_effect = [
+ json.dumps(self.node_info),
+ None,
+ json.dumps(self.node_info),
+ ]
+
+ with self.assertRaises(jenkins.JenkinsException) as context_manager:
+ self.j.delete_node('test_node')
+ self.assertEqual(
+ jenkins_mock.call_args_list[1][0][0].get_full_url(),
+ 'http://example.com/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),
+ ]
+
+ self.j.create_node('test node', exclusive=True)
+
+ self.assertEqual(
+ jenkins_mock.call_args_list[1][0][0].get_full_url().split('?')[0],
+ 'http://example.com/computer/doCreateItem')
+ self.assertTrue(self.j.node_exists('test node'))
+ 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),
+ ]
+
+ 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):
+ jenkins_mock.side_effect = [
+ None,
+ None,
+ None,
+ None,
+ ]
+
+ 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],
+ 'http://example.com/computer/doCreateItem')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'create[test_node] failed')
+ self._check_requests(jenkins_mock.call_args_list)
+
+
+class JenkinsEnableNodeTest(JenkinsNodesTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ json.dumps(self.offline_node_info),
+ None,
+ ]
+
+ self.j.enable_node('test node')
+
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ 'http://example.com/computer/test%20node/' +
+ 'toggleOffline?offlineMessage=')
+
+ jenkins_mock.side_effect = [json.dumps(self.online_node_info)]
+ node_info = self.j.get_node_info('test node')
+ self.assertEqual(node_info, self.online_node_info)
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_offline_false(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ json.dumps(self.online_node_info),
+ None,
+ ]
+
+ self.j.enable_node('test_node')
+
+ # 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(),
+ 'http://example.com/computer/test_node/api/json?depth=0')
+
+ jenkins_mock.side_effect = [json.dumps(self.online_node_info)]
+ node_info = self.j.get_node_info('test_node')
+ self.assertEqual(node_info, self.online_node_info)
+ self._check_requests(jenkins_mock.call_args_list)
+
+
+class JenkinsDisableNodeTest(JenkinsNodesTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ json.dumps(self.online_node_info),
+ None,
+ ]
+
+ self.j.disable_node('test node')
+
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ 'http://example.com/computer/test%20node/' +
+ 'toggleOffline?offlineMessage=')
+
+ jenkins_mock.side_effect = [json.dumps(self.offline_node_info)]
+ node_info = self.j.get_node_info('test node')
+ self.assertEqual(node_info, self.offline_node_info)
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_offline_true(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ json.dumps(self.offline_node_info),
+ None,
+ ]
+
+ self.j.disable_node('test_node')
+
+ # 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(),
+ 'http://example.com/computer/test_node/api/json?depth=0')
+
+ jenkins_mock.side_effect = [json.dumps(self.offline_node_info)]
+ node_info = self.j.get_node_info('test_node')
+ self.assertEqual(node_info, self.offline_node_info)
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/test_plugins.py b/tests/test_plugins.py
new file mode 100644
index 0000000..04eae34
--- /dev/null
+++ b/tests/test_plugins.py
@@ -0,0 +1,193 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.base import JenkinsTestBase
+
+
+class JenkinsPluginsBase(JenkinsTestBase):
+
+ plugin_info_json = {
+ u"plugins":
+ [
+ {
+ u"active": u'true',
+ u"backupVersion": u'null',
+ u"bundled": u'true',
+ u"deleted": u'false',
+ u"dependencies": [],
+ u"downgradable": u'false',
+ u"enabled": u'true',
+ u"hasUpdate": u'true',
+ u"longName": u"Jenkins Mailer Plugin",
+ u"pinned": u'false',
+ u"shortName": u"mailer",
+ u"supportsDynamicLoad": u"MAYBE",
+ u"url": u"http://wiki.jenkins-ci.org/display/JENKINS/Mailer",
+ u"version": u"1.5"
+ }
+ ]
+ }
+
+
+class JenkinsPluginsInfoTest(JenkinsPluginsBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jenkins_mock.return_value = json.dumps(self.plugin_info_json)
+
+ # expected to return a list of plugins
+ 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(),
+ u'http://example.com/pluginManager/api/json?depth=2')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_return_none(self, jenkins_mock):
+ empty_plugin_info_json = {u"plugins": []}
+
+ jenkins_mock.return_value = json.dumps(empty_plugin_info_json)
+
+ plugins_info = self.j.get_plugins_info()
+ self.assertEqual(plugins_info, [])
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_depth(self, jenkins_mock):
+ jenkins_mock.return_value = json.dumps(self.plugin_info_json)
+
+ self.j.get_plugins_info(depth=1)
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/pluginManager/api/json?depth=1')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_raise_BadStatusLine(self, jenkins_mock):
+ jenkins_mock.side_effect = jenkins.BadStatusLine('not a valid status line')
+
+ with self.assertRaises(jenkins.BadHTTPException) as context_manager:
+ self.j.get_plugins_info()
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/pluginManager/api/json?depth=2')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'Error communicating with server[http://example.com/]')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_return_invalid_json(self, jenkins_mock):
+ jenkins_mock.return_value = 'not valid JSON'
+
+ with self.assertRaises(jenkins.JenkinsException) as context_manager:
+ self.j.get_plugins_info()
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/pluginManager/api/json?depth=2')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'Could not parse JSON info for server[http://example.com/]')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_raise_HTTPError(self, jenkins_mock):
+ jenkins_mock.side_effect = jenkins.HTTPError(
+ 'http://example.com/job/pluginManager/api/json?depth=2',
+ code=401,
+ msg="basic auth failed",
+ hdrs=[],
+ fp=None)
+
+ 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[http://example.com/]')
+ self._check_requests(jenkins_mock.call_args_list)
+
+
+class JenkinsPluginInfoTest(JenkinsPluginsBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_shortname(self, jenkins_mock):
+ jenkins_mock.return_value = json.dumps(self.plugin_info_json)
+
+ # expected to return info on a single plugin
+ plugin_info = self.j.get_plugin_info("mailer")
+ self.assertEqual(plugin_info, self.plugin_info_json['plugins'][0])
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_longname(self, jenkins_mock):
+ jenkins_mock.return_value = json.dumps(self.plugin_info_json)
+
+ # expected to return info on a single plugin
+ plugin_info = self.j.get_plugin_info("Jenkins Mailer Plugin")
+ self.assertEqual(plugin_info, self.plugin_info_json['plugins'][0])
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_return_none(self, jenkins_mock):
+ jenkins_mock.return_value = json.dumps(self.plugin_info_json)
+
+ # expected not to find bogus so should return None
+ plugin_info = self.j.get_plugin_info("bogus")
+ self.assertEqual(plugin_info, None)
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_depth(self, jenkins_mock):
+ jenkins_mock.return_value = json.dumps(self.plugin_info_json)
+
+ self.j.get_plugin_info('test', depth=1)
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/pluginManager/api/json?depth=1')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_raise_BadStatusLine(self, jenkins_mock):
+ jenkins_mock.side_effect = jenkins.BadStatusLine('not a valid status line')
+
+ 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(),
+ u'http://example.com/pluginManager/api/json?depth=2')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'Error communicating with server[http://example.com/]')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_return_invalid_json(self, jenkins_mock):
+ jenkins_mock.return_value = 'not valid JSON'
+
+ 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(),
+ u'http://example.com/pluginManager/api/json?depth=2')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'Could not parse JSON info for server[http://example.com/]')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_raise_HTTPError(self, jenkins_mock):
+ jenkins_mock.side_effect = jenkins.HTTPError(
+ 'http://example.com/job/pluginManager/api/json?depth=2',
+ code=401,
+ msg="basic auth failed",
+ hdrs=[],
+ fp=None)
+
+ 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[http://example.com/]')
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/test_queue.py b/tests/test_queue.py
new file mode 100644
index 0000000..adda62f
--- /dev/null
+++ b/tests/test_queue.py
@@ -0,0 +1,72 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.base import JenkinsTestBase
+
+
+class JenkinsCancelQueueTest(JenkinsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ job_name_to_return = {u'name': 'TestJob'}
+ jenkins_mock.return_value = json.dumps(job_name_to_return)
+
+ self.j.cancel_queue(52)
+
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/queue/cancelItem?id=52')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open',
+ side_effect=jenkins.NotFoundException('not found'))
+ def test_notfound(self, jenkins_mock):
+ job_name_to_return = {u'name': 'TestJob'}
+ jenkins_mock.return_value = json.dumps(job_name_to_return)
+
+ self.j.cancel_queue(52)
+
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/queue/cancelItem?id=52')
+ self._check_requests(jenkins_mock.call_args_list)
+
+
+class JenkinsQueueInfoTest(JenkinsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ queue_info_to_return = {
+ 'items': {
+ 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,
+ }
+ }
+ jenkins_mock.return_value = json.dumps(queue_info_to_return)
+
+ queue_info = self.j.get_queue_info()
+
+ self.assertEqual(queue_info, queue_info_to_return['items'])
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/queue/api/json?depth=0')
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/test_script.py b/tests/test_script.py
new file mode 100644
index 0000000..e58d8db
--- /dev/null
+++ b/tests/test_script.py
@@ -0,0 +1,16 @@
+from mock import patch
+
+import jenkins
+from tests.base import JenkinsTestBase
+
+
+class JenkinsScriptTest(JenkinsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_run_script(self, jenkins_mock):
+ self.j.run_script(u'println(\"Hello World!\")')
+
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/scriptText')
+ self._check_requests(jenkins_mock.call_args_list)
diff --git a/tests/test_version.py b/tests/test_version.py
new file mode 100644
index 0000000..fa152cb
--- /dev/null
+++ b/tests/test_version.py
@@ -0,0 +1,57 @@
+from mock import patch, Mock
+import six
+
+import jenkins
+from tests.base import JenkinsTestBase
+
+
+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]
+ 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(
+ 'http://example.com/',
+ code=503,
+ msg="internal server error",
+ hdrs=[],
+ fp=None)
+ with self.assertRaises(jenkins.BadHTTPException) as context_manager:
+ self.j.get_version()
+ self.assertEqual(
+ str(context_manager.exception),
+ 'Error communicating with server[http://example.com/]')
+ self._check_requests(urlopen_mock.call_args_list)
+
+ @patch('jenkins.urlopen')
+ def test_raise_BadStatusLine(self, urlopen_mock):
+ urlopen_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[http://example.com/]')
+ self._check_requests(urlopen_mock.call_args_list)
+
+ @patch('jenkins.urlopen', return_value=None)
+ def test_return_empty_response(self, urlopen_mock):
+ with self.assertRaises(jenkins.EmptyResponseException) as context_manager:
+ self.j.get_version()
+ self.assertEqual(
+ str(context_manager.exception),
+ 'Error communicating with server[http://example.com/]:'
+ ' empty response')
+ self._check_requests(urlopen_mock.call_args_list)
diff --git a/tests/test_view.py b/tests/test_view.py
new file mode 100644
index 0000000..e7117bf
--- /dev/null
+++ b/tests/test_view.py
@@ -0,0 +1,218 @@
+import json
+from mock import patch
+
+import jenkins
+from tests.base import JenkinsTestBase
+
+
+class JenkinsViewsTestBase(JenkinsTestBase):
+ config_xml = """
+ <listView>
+ <description>Foo</description>
+ <jobNames />
+ </listView>"""
+
+
+class JenkinsGetViewNameTest(JenkinsViewsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ view_name_to_return = {u'name': 'Test View'}
+ jenkins_mock.return_value = json.dumps(view_name_to_return)
+
+ view_name = self.j.get_view_name(u'Test View')
+
+ self.assertEqual(view_name, 'Test View')
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/view/Test%20View/api/json?tree=name')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_return_none(self, jenkins_mock):
+ jenkins_mock.side_effect = jenkins.NotFoundException()
+
+ view_name = self.j.get_view_name(u'TestView')
+
+ self.assertEqual(view_name, None)
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/view/TestView/api/json?tree=name')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_unexpected_view_name(self, jenkins_mock):
+ view_name_to_return = {u'name': 'not the right name'}
+ jenkins_mock.return_value = json.dumps(view_name_to_return)
+
+ 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(),
+ 'http://example.com/view/TestView/api/json?tree=name')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'Jenkins returned an unexpected view name {0} '
+ '(expected: {1})'.format(view_name_to_return['name'], 'TestView'))
+ self._check_requests(jenkins_mock.call_args_list)
+
+
+class JenkinsAssertViewTest(JenkinsViewsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_view_missing(self, jenkins_mock):
+ jenkins_mock.side_effect = jenkins.NotFoundException()
+
+ with self.assertRaises(jenkins.JenkinsException) as context_manager:
+ self.j.assert_view_exists('NonExistent')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'view[NonExistent] does not exist')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_view_exists(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ json.dumps({'name': 'ExistingView'}),
+ ]
+ self.j.assert_view_exists('ExistingView')
+ self._check_requests(jenkins_mock.call_args_list)
+
+
+class JenkinsGetViewsTest(JenkinsViewsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ views = {
+ u'url': u'http://your_url_here/view/my_view/',
+ u'name': u'my_view',
+ }
+ view_info_to_return = {u'views': views}
+ jenkins_mock.return_value = json.dumps(view_info_to_return)
+
+ view_info = self.j.get_views()
+
+ self.assertEqual(view_info, views)
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/api/json')
+ self._check_requests(jenkins_mock.call_args_list)
+
+
+class JenkinsDeleteViewTest(JenkinsViewsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ None,
+ jenkins.NotFoundException(),
+ ]
+
+ self.j.delete_view(u'Test View')
+
+ self.assertEqual(
+ jenkins_mock.call_args_list[0][0][0].get_full_url(),
+ 'http://example.com/view/Test%20View/doDelete')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_failed(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ json.dumps({'name': 'TestView'}),
+ json.dumps({'name': 'TestView'}),
+ json.dumps({'name': 'TestView'}),
+ ]
+
+ 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(),
+ 'http://example.com/view/TestView/doDelete')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'delete[TestView] failed')
+ self._check_requests(jenkins_mock.call_args_list)
+
+
+class JenkinsCreateViewTest(JenkinsViewsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ jenkins.NotFoundException(),
+ None,
+ json.dumps({'name': 'Test View'}),
+ ]
+
+ self.j.create_view(u'Test View', self.config_xml)
+
+ self.assertEqual(
+ jenkins_mock.call_args_list[1][0][0].get_full_url(),
+ 'http://example.com/createView?name=Test%20View')
+ 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({'name': 'TestView'}),
+ None,
+ ]
+
+ 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(),
+ 'http://example.com/view/TestView/api/json?tree=name')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'view[TestView] already exists')
+ self._check_requests(jenkins_mock.call_args_list)
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_failed(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ jenkins.NotFoundException(),
+ None,
+ jenkins.NotFoundException(),
+ ]
+
+ 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(),
+ 'http://example.com/view/TestView/api/json?tree=name')
+ self.assertEqual(
+ jenkins_mock.call_args_list[1][0][0].get_full_url(),
+ 'http://example.com/createView?name=TestView')
+ self.assertEqual(
+ str(context_manager.exception),
+ 'create[TestView] failed')
+ self._check_requests(jenkins_mock.call_args_list)
+
+
+class JenkinsReconfigViewTest(JenkinsViewsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_simple(self, jenkins_mock):
+ jenkins_mock.side_effect = [
+ json.dumps({'name': 'Test View'}),
+ None,
+ ]
+
+ self.j.reconfig_view(u'Test View', self.config_xml)
+
+ self.assertEqual(jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/view/Test%20View/config.xml')
+ self._check_requests(jenkins_mock.call_args_list)
+
+
+class JenkinsGetViewConfigTest(JenkinsViewsTestBase):
+
+ @patch.object(jenkins.Jenkins, 'jenkins_open')
+ def test_encodes_view_name(self, jenkins_mock):
+ self.j.get_view_config(u'Test View')
+
+ self.assertEqual(
+ jenkins_mock.call_args[0][0].get_full_url(),
+ u'http://example.com/view/Test%20View/config.xml')
+ self._check_requests(jenkins_mock.call_args_list)