diff --git a/jenkins/__init__.py b/jenkins/__init__.py index 8674222..620ce14 100755 --- a/jenkins/__init__.py +++ b/jenkins/__init__.py @@ -123,6 +123,8 @@ BUILD_ENV_VARS = '%(folder_url)sjob/%(short_name)s/%(number)d/injectedEnvVars/ap '?depth=%(depth)s' BUILD_TEST_REPORT = '%(folder_url)sjob/%(short_name)s/%(number)d/testReport/api/json' + \ '?depth=%(depth)s' +BUILD_ARTIFACT = '%(folder_url)sjob/%(short_name)s/%(number)d/artifact/%(artifact)s' +BUILD_STAGES = '%(folder_url)sjob/%(short_name)s/%(number)d/wfapi/describe/' DELETE_BUILD = '%(folder_url)sjob/%(short_name)s/%(number)s/doDelete' WIPEOUT_JOB_WORKSPACE = '%(folder_url)sjob/%(short_name)s/doWipeOutWorkspace' NODE_LIST = 'computer/api/json?depth=%(depth)s' @@ -350,7 +352,8 @@ class Jenkins(object): def _get_encoded_params(self, params): for k, v in params.items(): if k in ["name", "msg", "short_name", "from_short_name", - "to_short_name", "folder_url", "from_folder_url", "to_folder_url"]: + "to_short_name", "folder_url", "from_folder_url", "to_folder_url", + "artifact"]: params[k] = quote(v.encode('utf8')) return params @@ -712,6 +715,59 @@ class Jenkins(object): # This can happen if the test report wasn't generated for any reason return None + def get_build_artifact(self, name, number, artifact): + """Get artifacts from job + + :param name: Job name, ``str`` + :param number: Build number, ``int`` + :param artifact: Artifact relative path, ``str`` + :returns: artifact to download, ``dict`` + """ + folder_url, short_name = self._get_job_folder(name) + + try: + response = self.jenkins_open(requests.Request( + 'GET', self._build_url(BUILD_ARTIFACT, locals()))) + + if response: + return json.loads(response) + else: + raise JenkinsException('job[%s] number[%d] does not exist' % (name, number)) + except requests.exceptions.HTTPError: + raise JenkinsException('job[%s] number[%d] does not exist' % (name, number)) + except ValueError: + raise JenkinsException( + 'Could not parse JSON info for job[%s] number[%d]' % (name, number)) + except NotFoundException: + # This can happen if the artifact is not found + return None + + def get_build_stages(self, name, number): + """Get stages info from job + + :param name: Job name, ``str`` + :param number: Build number, ``int`` + :returns: dictionary of stages in the job, ``dict`` + """ + folder_url, short_name = self._get_job_folder(name) + + try: + response = self.jenkins_open(requests.Request( + 'GET', self._build_url(BUILD_STAGES, locals()))) + + if response: + return json.loads(response) + else: + raise JenkinsException('job[%s] number[%d] does not exist' % (name, number)) + except requests.exceptions.HTTPError: + raise JenkinsException('job[%s] number[%d] does not exist' % (name, number)) + except ValueError: + raise JenkinsException( + 'Could not parse JSON info for job[%s] number[%d]' % (name, number)) + except NotFoundException: + # This can happen if this isn't a stages/pipeline job + return None + def get_queue_info(self): ''':returns: list of job dictionaries, ``[dict]`` diff --git a/tests/test_build.py b/tests/test_build.py index afb36bb..ea3e5b4 100644 --- a/tests/test_build.py +++ b/tests/test_build.py @@ -649,3 +649,163 @@ class JenkinsBuildTestReportUrlTest(JenkinsTestBase): self.assertEqual( str(context_manager.exception), 'Error in request. Possibly authentication failed [401]: Not Authorised') + + +class JenkinsBuildArtifactUrlTest(JenkinsTestBase): + + @patch.object(jenkins.Jenkins, 'jenkins_open') + def test_simple(self, jenkins_mock): + jenkins_mock.return_value = '{}' + ret = self.j.get_build_artifact(u'Test Job', number=52, artifact="filename") + self.assertEqual(ret, json.loads(jenkins_mock.return_value)) + self.assertEqual( + jenkins_mock.call_args[0][0].url, + self.make_url('job/Test%20Job/52/artifact/filename')) + self._check_requests(jenkins_mock.call_args_list) + + @patch.object(jenkins.Jenkins, 'jenkins_open') + def test_in_folder(self, jenkins_mock): + jenkins_mock.return_value = '{}' + ret = self.j.get_build_artifact(u'a Folder/Test Job', number=52, artifact="file name") + self.assertEqual(ret, json.loads(jenkins_mock.return_value)) + self.assertEqual( + jenkins_mock.call_args[0][0].url, + self.make_url('job/a%20Folder/job/Test%20Job/52/artifact/file%20name')) + self._check_requests(jenkins_mock.call_args_list) + + @patch('jenkins.requests.Session.send', autospec=True) + def test_return_none(self, session_send_mock): + session_send_mock.side_effect = iter([ + build_response_mock(404, reason="Not Found"), # crumb + build_response_mock(404, reason="Not Found"), # request + ]) + ret = self.j.get_build_artifact(u'TestJob', number=52, artifact="filename") + self.assertIsNone(ret) + + @patch.object(jenkins.Jenkins, 'jenkins_open') + def test_open_return_none(self, jenkins_mock): + jenkins_mock.return_value = None + + with self.assertRaises(jenkins.JenkinsException) as context_manager: + self.j.get_build_artifact(u'TestJob', number=52, artifact="filename") + self.assertEqual( + str(context_manager.exception), + 'job[TestJob] number[52] does not exist') + self._check_requests(jenkins_mock.call_args_list) + + @patch.object(jenkins.Jenkins, 'jenkins_open') + def test_return_invalid_json(self, jenkins_mock): + jenkins_mock.return_value = 'Invalid JSON' + + with self.assertRaises(jenkins.JenkinsException) as context_manager: + self.j.get_build_artifact(u'TestJob', number=52, artifact="filename") + self.assertEqual( + str(context_manager.exception), + 'Could not parse JSON info for job[TestJob] number[52]') + self._check_requests(jenkins_mock.call_args_list) + + @patch('jenkins.requests.Session.send', autospec=True) + def test_raise_HTTPError(self, session_send_mock): + session_send_mock.side_effect = iter([ + build_response_mock(401, reason="Not Authorised"), # crumb + build_response_mock(401, reason="Not Authorised"), # request + ]) + + with self.assertRaises(jenkins.JenkinsException) as context_manager: + self.j.get_build_artifact(u'TestJob', number=52, artifact="filename") + self.assertEqual( + str(context_manager.exception), + 'Error in request. Possibly authentication failed [401]: Not Authorised') + + @patch('jenkins.requests.Session.send', autospec=True) + def test_in_folder_raise_HTTPError(self, session_send_mock): + session_send_mock.side_effect = iter([ + build_response_mock(401, reason="Not Authorised"), # crumb + build_response_mock(401, reason="Not Authorised"), # request + ]) + + with self.assertRaises(jenkins.JenkinsException) as context_manager: + self.j.get_build_artifact(u'a Folder/TestJob', number=52, artifact="filename") + self.assertEqual( + str(context_manager.exception), + 'Error in request. Possibly authentication failed [401]: Not Authorised') + + +class JenkinsBuildStagesUrlTest(JenkinsTestBase): + + @patch.object(jenkins.Jenkins, 'jenkins_open') + def test_simple(self, jenkins_mock): + jenkins_mock.return_value = '{}' + ret = self.j.get_build_stages(u'Test Job', number=52) + self.assertEqual(ret, json.loads(jenkins_mock.return_value)) + self.assertEqual( + jenkins_mock.call_args[0][0].url, + self.make_url('job/Test%20Job/52/wfapi/describe/')) + self._check_requests(jenkins_mock.call_args_list) + + @patch.object(jenkins.Jenkins, 'jenkins_open') + def test_in_folder(self, jenkins_mock): + jenkins_mock.return_value = '{}' + ret = self.j.get_build_stages(u'a Folder/Test Job', number=52) + self.assertEqual(ret, json.loads(jenkins_mock.return_value)) + self.assertEqual( + jenkins_mock.call_args[0][0].url, + self.make_url('job/a%20Folder/job/Test%20Job/52/wfapi/describe/')) + self._check_requests(jenkins_mock.call_args_list) + + @patch('jenkins.requests.Session.send', autospec=True) + def test_return_none(self, session_send_mock): + session_send_mock.side_effect = iter([ + build_response_mock(404, reason="Not Found"), # crumb + build_response_mock(404, reason="Not Found"), # request + ]) + ret = self.j.get_build_stages(u'TestJob', number=52) + self.assertIsNone(ret) + + @patch.object(jenkins.Jenkins, 'jenkins_open') + def test_open_return_none(self, jenkins_mock): + jenkins_mock.return_value = None + + with self.assertRaises(jenkins.JenkinsException) as context_manager: + self.j.get_build_stages(u'TestJob', number=52) + self.assertEqual( + str(context_manager.exception), + 'job[TestJob] number[52] does not exist') + self._check_requests(jenkins_mock.call_args_list) + + @patch.object(jenkins.Jenkins, 'jenkins_open') + def test_return_invalid_json(self, jenkins_mock): + jenkins_mock.return_value = 'Invalid JSON' + + with self.assertRaises(jenkins.JenkinsException) as context_manager: + self.j.get_build_stages(u'TestJob', number=52) + self.assertEqual( + str(context_manager.exception), + 'Could not parse JSON info for job[TestJob] number[52]') + self._check_requests(jenkins_mock.call_args_list) + + @patch('jenkins.requests.Session.send', autospec=True) + def test_raise_HTTPError(self, session_send_mock): + session_send_mock.side_effect = iter([ + build_response_mock(401, reason="Not Authorised"), # crumb + build_response_mock(401, reason="Not Authorised"), # request + ]) + + with self.assertRaises(jenkins.JenkinsException) as context_manager: + self.j.get_build_stages(u'TestJob', number=52) + self.assertEqual( + str(context_manager.exception), + 'Error in request. Possibly authentication failed [401]: Not Authorised') + + @patch('jenkins.requests.Session.send', autospec=True) + def test_in_folder_raise_HTTPError(self, session_send_mock): + session_send_mock.side_effect = iter([ + build_response_mock(401, reason="Not Authorised"), # crumb + build_response_mock(401, reason="Not Authorised"), # request + ]) + + with self.assertRaises(jenkins.JenkinsException) as context_manager: + self.j.get_build_stages(u'a Folder/TestJob', number=52) + self.assertEqual( + str(context_manager.exception), + 'Error in request. Possibly authentication failed [401]: Not Authorised')