from jenkinsapi import exceptions
from jenkinsapi.mutable_jenkins_thing import MutableJenkinsThing
-from jenkinsapi.exceptions import NoBuildData, NotFound, NotInQueue
+from jenkinsapi.exceptions import NoBuildData, NotFound, NotInQueue, WillNotBuild
log = logging.getLogger(__name__)
self._element_tree = ET.fromstring(self._config)
return self._element_tree
- def get_build_triggerurl(self, token=None, params=None):
+ def get_build_triggerurl(self):
return "%s/build" % self.baseurl
def invoke(self, securitytoken=None, block=False, skip_if_running=False, invoke_pre_check_delay=3, invoke_block_delay=15, params=None, cause=None):
params = params and dict(params.items()) or {}
if self.is_queued():
- log.warn( "Will not request new build because %s is already queued", self.name )
+ raise WillNotBuild('%s is already queued' % repr(self))
+
elif self.is_running():
if skip_if_running:
log.warn( "Will not request new build because %s is already running", self.name )
log.warn("Will re-schedule %s even though it is already running", self.name )
original_build_no = self.get_last_buildnumber()
log.info( "Attempting to start %s on %s", self.name, repr(self.get_jenkins_obj()) )
+
url = self.get_build_triggerurl()
if cause:
assert len( response.text ) > 0
if invoke_pre_check_delay > 0:
- log.info("Waiting for %is to allow Jenkins to catch up" % invoke_pre_check_delay )
+ log.info("Waiting for %is to allow Jenkins to catch up" , invoke_pre_check_delay )
sleep( invoke_pre_check_delay )
if block:
total_wait = 0
while self.is_queued():
- log.info( "Waited %is for %s to begin..." % ( total_wait, self.name ))
+ log.info( "Waited %is for %s to begin...", total_wait, self.name )
sleep( invoke_block_delay )
total_wait += invoke_block_delay
if self.is_running():
assert self.get_last_buildnumber() > original_build_no, "Job does not appear to have run."
else:
if self.is_queued():
- log.info( "%s has been queued." % self.name )
+ log.info( "%s has been queued.", self.name )
elif self.is_running():
- log.info( "%s is running." % self.name )
+ log.info( "%s is running.", self.name )
elif original_build_no < self.get_last_buildnumber():
- log.info( "%s has completed." % self.name )
+ log.info( "%s has completed.", self.name )
else:
raise AssertionError("The job did not schedule.")
"""
return reversed( sorted( self.get_build_dict().keys() ) )
+ def get_next_build_number(self):
+ """
+ Return the next build number that Jenkins will assign.
+ """
+ return self._data.get('nextBuildNumber', 0)
+
def get_last_good_build( self ):
"""
Get the last good build
if build is not None:
return build.is_running()
except NoBuildData:
- log.info("No build info available for %s, assuming not running." % str(self) )
+ log.info("No build info available for %s, assuming not running.", str(self) )
return False
def get_config(self):
'''Returns the config.xml from the job'''
- return self.jenkins.requester.get_and_confirm_status("%(baseurl)s/config.xml" % self.__dict__)
+ response = self.jenkins.requester.get_and_confirm_status("%(baseurl)s/config.xml" % self.__dict__)
+ return response.text
def load_config(self):
self._config = self.get_config()
System tests for `jenkinsapi.jenkins` module.
'''
import unittest
+from jenkinsapi.build import Build
from jenkinsapi_tests.test_utils.random_strings import random_string
from jenkinsapi_tests.systests.base import BaseSystemTest, EMPTY_JOB_CONFIG
self.jenkins.create_job(job_name, EMPTY_JOB_CONFIG)
self.assertJobIsPresent(job_name)
+ def test_get_job_config(self):
+ job_name = 'config_%s' % random_string()
+ self.jenkins.create_job(job_name, EMPTY_JOB_CONFIG)
+ self.assertJobIsPresent(job_name)
+ config = self.jenkins[job_name].get_config()
+ self.assertEquals(config.strip(), EMPTY_JOB_CONFIG.strip())
+
def test_invoke_job(self):
job_name = 'create_%s' % random_string()
job = self.jenkins.create_job(job_name, EMPTY_JOB_CONFIG)
-
- build = job.invoke()
+ job.invoke()
def test_get_jobs_list(self):
job1_name = 'first_%s' % random_string()