Project Authors
===============
- * Salim Fadhley (sal@stodge.org)
- * Ramon van Alteren (ramon@vanalteren.nl)
- * Ruslan Lutsenko (ruslan.lutcenko@gmail.com)
+ * Salim Fadhley (sal@stodge.org)
+ * Ramon van Alteren (ramon@vanalteren.nl)
+ * Ruslan Lutsenko (ruslan.lutcenko@gmail.com)
Current code lives on github: https://github.com/salimfadhley/jenkinsapi
"""
-__all__= [ "command_line", "utils",
+__all__= ["command_line", "utils",
"api", "artifact", "build", "config", "constants", "exceptions", "fingerprint",
"jenkins", "jenkinsbase", "job", "node", "result_set", "result", "view"]
-__docformat__ = "epytext"
\ No newline at end of file
+__docformat__ = "epytext"
log = logging.getLogger(__name__)
-def get_latest_test_results( jenkinsurl, jobname ):
+
+def get_latest_test_results(jenkinsurl, jobname):
"""
A convenience function to fetch down the very latest test results from a jenkins job.
"""
- latestbuild = get_latest_build( jenkinsurl, jobname )
+ latestbuild = get_latest_build(jenkinsurl, jobname)
res = latestbuild.get_resultset()
return res
+
def get_latest_build(jenkinsurl, jobname):
"""
A convenience function to fetch down the very latest test results from a jenkins job.
job = jenkinsci[jobname]
return job.get_last_build()
+
def get_latest_complete_build(jenkinsurl, jobname):
"""
A convenience function to fetch down the very latest test results from a jenkins job.
job = jenkinsci[jobname]
return job.get_last_completed_build()
+
def get_build(jenkinsurl, jobname, build_no):
"""
A convenience function to fetch down the test results from a jenkins job by build number.
"""
jenkinsci = Jenkins(jenkinsurl)
job = jenkinsci[jobname]
- return job.get_build( build_no )
+ return job.get_build(build_no)
+
-def get_artifacts( jenkinsurl, jobid=None, build_no=None, proxyhost=None, proxyport=None, proxyuser=None, proxypass=None ):
+def get_artifacts(jenkinsurl, jobid=None, build_no=None, proxyhost=None, proxyport=None, proxyuser=None, proxypass=None):
"""
Find all the artifacts for the latest build of a job.
"""
jenkinsci = Jenkins(jenkinsurl, proxyhost, proxyport, proxyuser, proxypass)
job = jenkinsci[jobid]
if build_no:
- build = job.get_build( build_no )
+ build = job.get_build(build_no)
else:
build = job.get_last_good_build()
artifacts = build.get_artifact_dict()
- log.info("Found %i artifacts in '%s'" % ( len(artifacts.keys() ), build_no ))
+ log.info("Found %i artifacts in '%s'" % (len(artifacts.keys()), build_no))
return artifacts
-def search_artifacts(jenkinsurl, jobid, artifact_ids=None ):
+
+def search_artifacts(jenkinsurl, jobid, artifact_ids=None):
"""
Search the entire history of a jenkins job for a list of artifact names. If same_build
is true then ensure that all artifacts come from the same build of the job
"""
if len(artifact_ids) == 0 or artifact_ids is None:
return []
-
- jenkinsci = Jenkins( jenkinsurl )
- job = jenkinsci[ jobid ]
+
+ jenkinsci = Jenkins(jenkinsurl)
+ job = jenkinsci[jobid]
build_ids = job.get_build_ids()
for build_id in build_ids:
- build = job.get_build( build_id )
+ build = job.get_build(build_id)
artifacts = build.get_artifact_dict()
- if set( artifact_ids ).issubset( set( artifacts.keys() ) ):
- return dict( ( a,artifacts[a] ) for a in artifact_ids )
- missing_artifacts = set( artifact_ids ) - set( artifacts.keys() )
- log.debug("Artifacts %s missing from %s #%i" % ( ", ".join( missing_artifacts ), jobid, build_id ))
+ if set(artifact_ids).issubset(set(artifacts.keys())):
+ return dict((a,artifacts[a]) for a in artifact_ids)
+ missing_artifacts = set(artifact_ids) - set(artifacts.keys())
+ log.debug("Artifacts %s missing from %s #%i" % (", ".join(missing_artifacts), jobid, build_id))
#noinspection PyUnboundLocalVariable
- raise ArtifactsMissing( missing_artifacts )
+ raise ArtifactsMissing(missing_artifacts)
+
def grab_artifact(jenkinsurl, jobid, artifactid, targetdir):
"""
Convenience method to find the latest good version of an artifact and save it
to a target directory. Directory is made automatically if not exists.
"""
- artifacts = get_artifacts( jenkinsurl, jobid )
- artifact = artifacts[ artifactid ]
- if not os.path.exists( targetdir ):
- os.makedirs( targetdir )
- artifact.savetodir( targetdir)
+ artifacts = get_artifacts(jenkinsurl, jobid)
+ artifact = artifacts[artifactid]
+ if not os.path.exists(targetdir):
+ os.makedirs(targetdir)
+ artifact.savetodir(targetdir)
+
def block_until_complete(jenkinsurl, jobs, maxwait=12000, interval=30, raise_on_timeout=True):
"""
if not still_running:
return
str_still_running = ", ".join('"%s"' % str(a) for a in still_running)
- log.warn( "Waiting for jobs %s to complete. Will wait another %is" % (str_still_running, time_left ))
+ log.warn("Waiting for jobs %s to complete. Will wait another %is" % (str_still_running, time_left))
time.sleep(interval)
if raise_on_timeout:
#noinspection PyUnboundLocalVariable
raise TimeOut("Waited too long for these jobs to complete: %s" % str_still_running)
+
def get_view_from_url(url):
"""
Factory method
jenkinsci = Jenkins(jenkinsurl)
return jenkinsci.views[view_name]
+
def get_nested_view_from_url(url):
"""
Returns View based on provided URL. Convenient for nested views.
jenkinsci = Jenkins(matched.group(0))
return jenkinsci.get_view_by_url(url)
+
def install_artifacts(artifacts, dirstruct, installdir, basestaticurl):
"""
Install the artifacts.
else:
assert os.path.isdir(destdir)
for artifactname in artifactnames:
- destpath = os.path.abspath(os.path.join( destdir, artifactname))
+ destpath = os.path.abspath(os.path.join(destdir, artifactname))
if artifactname in artifacts.keys():
# The artifact must be loaded from jenkins
theartifact = artifacts[artifactname]
theartifact.save(destpath)
installed.append(destpath)
return installed
-
-def search_artifact_by_regexp( jenkinsurl, jobid, artifactRegExp ):
+
+
+def search_artifact_by_regexp(jenkinsurl, jobid, artifactRegExp):
'''
@param jenkinsurl: The base URL of the jenkins server
@param jobid: The name of the job we are to search through
Search the entire history of a hudson job for a build which has an artifact whose
name matches a supplied regular expression. Return only that artifact.
"""
- J = Jenkins( jenkinsurl )
- j = J[ jobid ]
-
+ J = Jenkins(jenkinsurl)
+ j = J[jobid]
+
build_ids = j.get_build_ids()
-
+
for build_id in build_ids:
- build = j.get_build( build_id )
-
+ build = j.get_build(build_id)
+
artifacts = build.get_artifact_dict()
-
+
for name, art in artifacts.iteritems():
- md_match = artifactRegExp.search( name )
-
+ md_match = artifactRegExp.search(name)
+
if md_match:
return art
-
- raise ArtifactsMissing( )
+
+ raise ArtifactsMissing()
log = logging.getLogger(__name__)
+
class Artifact(object):
"""
Represents a single Jenkins artifact, usually some kind of file
outputfilepath = os.path.join(dirpath, self.filename)
return self.save(outputfilepath)
-
def __repr__(self):
"""
Produce a handy repr-string.
log = logging.getLogger(__name__)
+
class Build(JenkinsBase):
"""
Represents a jenkins build, executed in context of a job.
STR_TOTALCOUNT = "totalCount"
STR_TPL_NOTESTS_ERR = "%s has status %s, and does not have any test results"
- def __init__( self, url, buildno, job ):
+ def __init__(self, url, buildno, job):
assert type(buildno) == int
self.buildno = buildno
self.job = job
- JenkinsBase.__init__( self, url )
+ JenkinsBase.__init__(self, url)
def __str__(self):
return self._data['fullDisplayName']
def get_duration(self):
return self._data["duration"]
- def get_artifacts( self ):
+ def get_artifacts(self):
for afinfo in self._data["artifacts"]:
- url = "%s/artifact/%s" % ( self.baseurl, afinfo["relativePath"] )
- af = Artifact( afinfo["fileName"], url, self )
+ url = "%s/artifact/%s" % (self.baseurl, afinfo["relativePath"])
+ af = Artifact(afinfo["fileName"], url, self)
yield af
def get_artifact_dict(self):
for rinfo in self._data["runs"]:
yield Build(rinfo["url"], rinfo["number"], self.job)
- def is_running( self ):
+ def is_running(self):
"""
Return a bool if running.
"""
while self.is_running():
time.sleep(1)
- def is_good( self ):
+ def is_good(self):
"""
Return a bool, true if the build was good.
If the build is still running, return False.
"""
- return ( not self.is_running() ) and self._data["result"] == STATUS_SUCCESS
+ return (not self.is_running()) and self._data["result"] == STATUS_SUCCESS
def block_until_complete(self, delay=15):
- assert isinstance( delay, int )
+ assert isinstance(delay, int)
count = 0
while self.is_running():
total_wait = delay * count
- log.info("Waited %is for %s #%s to complete" % ( total_wait, self.job.name, self.name ) )
- sleep( delay )
+ log.info("Waited %is for %s #%s to complete" % (total_wait, self.job.name, self.name))
+ sleep(delay)
count += 1
def get_jenkins_obj(self):
Return the URL for the object which provides the job's result summary.
"""
url_tpl = r"%stestReport/%s"
- return url_tpl % ( self._data["url"] , config.JENKINS_API )
+ return url_tpl % (self._data["url"], config.JENKINS_API)
def get_resultset(self):
"""
"""
result_url = self.get_result_url()
if self.STR_TOTALCOUNT not in self.get_actions():
- raise NoResults( "%s does not have any published results" % str(self) )
+ raise NoResults("%s does not have any published results" % str(self))
buildstatus = self.get_status()
if not self.get_actions()[self.STR_TOTALCOUNT]:
- raise NoResults( self.STR_TPL_NOTESTS_ERR % ( str(self), buildstatus ) )
- obj_results = ResultSet( result_url, build=self )
+ raise NoResults(self.STR_TPL_NOTESTS_ERR % (str(self), buildstatus))
+ obj_results = ResultSet(result_url, build=self)
return obj_results
def has_resultset(self):
all_actions = {}
for dct_action in self._data["actions"]:
if dct_action is None: continue
- all_actions.update( dct_action )
+ all_actions.update(dct_action)
return all_actions
def get_timestamp(self):
log = logging.getLogger(__name__)
+
class jenkins_invoke(object):
@classmethod
def mkparser(cls):
parser = optparse.OptionParser()
- DEFAULT_BASEURL=os.environ.get( "JENKINS_URL", "http://localhost/jenkins" )
+ DEFAULT_BASEURL=os.environ.get("JENKINS_URL", "http://localhost/jenkins")
parser.help_text = "Execute a number of jenkins jobs on the server of your choice. Optionally block until the jobs are complete."
parser.add_option("-J", "--jenkinsbase", dest="baseurl",
help="Base URL for the Jenkins server, default is %s" % DEFAULT_BASEURL,
job.invoke(securitytoken=token, block=block)
-def main( ):
+def main():
logging.basicConfig()
logging.getLogger("").setLevel(logging.INFO)
- jenkins_invoke.main()
\ No newline at end of file
+ jenkins_invoke.main()
RESULTSTATUS_SKIPPED = "SKIPPED"
STR_RE_SPLIT_VIEW = "(.*)/view/([^/]*)/?"
-RE_SPLIT_VIEW_URL = re.compile( STR_RE_SPLIT_VIEW )
+RE_SPLIT_VIEW_URL = re.compile(STR_RE_SPLIT_VIEW)
Base class for all errors
"""
+
class NotFound(JenkinsAPIException):
"""
Resource cannot be found
"""
+
class ArtifactsMissing(NotFound):
"""
Cannot find a build with all of the required artifacts.
"""
-class UnknownJob( KeyError, NotFound):
+
+class UnknownJob(KeyError, NotFound):
"""
Jenkins does not recognize the job requested.
"""
-class UnknownView( KeyError, NotFound):
+
+class UnknownView(KeyError, NotFound):
"""
Jenkins does not recognize the view requested.
"""
-class UnknownNode( KeyError, NotFound):
+
+class UnknownNode(KeyError, NotFound):
"""
Jenkins does not recognize the node requested.
"""
-class UnknownQueueItem( KeyError, NotFound):
+
+class UnknownQueueItem(KeyError, NotFound):
"""
Jenkins does not recognize the requested queue item
"""
+
class NoBuildData(NotFound):
"""
A job has no build data.
"""
+
class ArtifactBroken(JenkinsAPIException):
"""
An artifact is broken, wrong
"""
-class TimeOut( JenkinsAPIException ):
+
+class TimeOut(JenkinsAPIException):
"""
Some jobs have taken too long to complete.
"""
+
class WillNotBuild(JenkinsAPIException):
"""
Cannot trigger a new build.
"""
-
class NoResults(JenkinsAPIException):
"""
A build did not publish any results.
"""
+
class FailedNoResults(NoResults):
"""
A build did not publish any results because it failed
"""
+
class BadURL(ValueError,JenkinsAPIException):
"""
A URL appears to be broken
"""
-
class NotAuthorized(JenkinsAPIException):
"""Not Authorized to access resource"""
# Usually thrown when we get a 403 returned
+
class NotSupportSCM(JenkinsAPIException):
"""
It's a SCM that does not supported by current version of jenkinsapi
"""
+
class NotConfiguredSCM(JenkinsAPIException):
"""
It's a job that doesn't have configured SCM
"""
+
+
class NotInQueue(JenkinsAPIException):
"""
It's a job that is not in the queue
import logging
-log = logging.getLogger( __name__ )
+log = logging.getLogger(__name__)
+
class Fingerprint(JenkinsBase):
"""
def __init__(self, baseurl, id, jenkins_obj):
logging.basicConfig()
self.jenkins_obj = jenkins_obj
- assert self.RE_MD5.search( id ), "%s does not look like a valid id" % id
- url = "%s/fingerprint/%s/" % ( baseurl, id )
- JenkinsBase.__init__( self, url, poll=False )
+ assert self.RE_MD5.search(id), "%s does not look like a valid id" % id
+ url = "%s/fingerprint/%s/" % (baseurl, id)
+ JenkinsBase.__init__(self, url, poll=False)
self.id = id
def get_jenkins_obj(self):
if self._data["original"]["number"] == build:
return True
if self._data["fileName"] != filename:
- log.info("Filename from jenkins (%s) did not match provided (%s)" % ( self._data["fileName"], filename ) )
+ log.info("Filename from jenkins (%s) did not match provided (%s)" % (self._data["fileName"], filename))
return False
for usage_item in self._data["usage"]:
if usage_item["name"] == job:
for range in usage_item["ranges"]["ranges"]:
if range["start"] <= build <= range["end"]:
- log.info("This artifact was generated by %s between build %i and %i" % ( job, range["start"], range["end"] ) )
+ log.info("This artifact was generated by %s between build %i and %i" % (job, range["start"], range["end"]))
return True
return False
try:
assert self.valid()
except AssertionError:
- raise ArtifactBroken( "Artifact %s seems to be broken, check %s" % ( self.id, self.baseurl ) )
+ raise ArtifactBroken("Artifact %s seems to be broken, check %s" % (self.id, self.baseurl))
except urllib2.HTTPError:
- raise ArtifactBroken( "Unable to validate artifact id %s using %s" % ( self.id, self.baseurl ) )
+ raise ArtifactBroken("Unable to validate artifact id %s using %s" % (self.id, self.baseurl))
return True
- def get_info( self ):
+ def get_info(self):
"""
Returns a tuple of build-name, build# and artifiact filename for a good build.
"""
import datetime
from jenkinsapi.exceptions import UnknownQueueItem, TimeOut
+
class Invocation(object):
"""
Represents the state and consequences of a single attempt to start a job.
self.queue_item = None
self.build_number = None
-
def __enter__(self):
"""
Start watching the job
self.job.poll()
self.initial_builds = set(self.job.get_build_dict().keys())
-
def __exit__(self, type, value, traceback):
"""
Finish watching the job - it will track which new queue items or builds have
return self.job[self.get_build_number()]
def block_until_not_queued(self, timeout, delay):
- self.__block(lambda : self.is_queued(), False, timeout, delay )
+ self.__block(lambda: self.is_queued(), False, timeout, delay)
def block_until_completed(self, timeout, delay):
- self.__block(lambda : self.is_running(), False, timeout, delay )
+ self.__block(lambda: self.is_running(), False, timeout, delay)
@staticmethod
def __block(fn, expectation, timeout, delay=2):
if datetime.datetime.now() > endTime:
raise TimeOut()
-
def block(self, until='completed', timeout=200, delay=2):
"""
Block this item until a condition is met.
:param newjobname: name of new job, str
:return: new Job obj
"""
- params = { 'name': newjobname,
+ params = {'name': newjobname,
'mode': 'copy',
'from': jobname}
return list(self.get_jobs())
def keys(self):
- return [ a for a in self.iterkeys() ]
+ return [a for a in self.iterkeys()]
# This is a function alias we retain for historical compatibility
get_jobs_list = keys
def get_view_by_url(self, str_view_url):
#for nested view
str_view_name = str_view_url.split('/view/')[-1].replace('/', '')
- return View(str_view_url , str_view_name, jenkins_obj=self)
+ return View(str_view_url, str_view_name, jenkins_obj=self)
def __getitem__(self, jobname):
"""
if exclusive:
MODE = 'EXCLUSIVE'
params = {
- 'name' : name,
- 'type' : NODE_TYPE,
- 'json' : json.dumps ({
- 'name' : name,
- 'nodeDescription' : node_description,
- 'numExecutors' : num_executors,
- 'remoteFS' : remote_fs,
- 'labelString' : labels,
- 'mode' : MODE,
- 'type' : NODE_TYPE,
- 'retentionStrategy' : { 'stapler-class' : 'hudson.slaves.RetentionStrategy$Always' },
- 'nodeProperties' : { 'stapler-class-bag' : 'true' },
- 'launcher' : { 'stapler-class' : 'hudson.slaves.JNLPLauncher' }
+ 'name': name,
+ 'type': NODE_TYPE,
+ 'json': json.dumps ({
+ 'name': name,
+ 'nodeDescription': node_description,
+ 'numExecutors': num_executors,
+ 'remoteFS': remote_fs,
+ 'labelString': labels,
+ 'mode': MODE,
+ 'type': NODE_TYPE,
+ 'retentionStrategy': {'stapler-class': 'hudson.slaves.RetentionStrategy$Always'},
+ 'nodeProperties': {'stapler-class-bag': 'true'},
+ 'launcher': {'stapler-class': 'hudson.slaves.JNLPLauncher'}
})
}
url = self.get_node_url() + "doCreateItem?%s" % urllib.urlencode(params)
from jenkinsapi.exceptions import JenkinsAPIException
log = logging.getLogger(__name__)
+
class JenkinsBase(object):
"""
This appears to be the base object that all other jenkins objects are inherited from
def __repr__(self):
return """<%s.%s %s>""" % (self.__class__.__module__,
self.__class__.__name__,
- str( self ))
+ str(self))
def __str__(self):
raise NotImplementedError
"""
return self._buildid_for_type("lastSuccessfulBuild")
- def get_last_failed_buildnumber( self ):
+ def get_last_failed_buildnumber(self):
"""
Get the numerical ID of the last good build.
"""
return self._buildid_for_type(buildtype="lastFailedBuild")
- def get_last_buildnumber( self ):
+ def get_last_buildnumber(self):
"""
Get the numerical ID of the last build.
"""
log = logging.getLogger(__name__)
+
class Node(JenkinsBase):
"""
Class to hold information on nodes that are attached as slaves to the master jenkins instance
def is_idle(self):
return self._data['idle']
-
def set_online(self):
"""
Set node online.
log = logging.getLogger(__name__)
+
class Nodes(JenkinsBase):
"""
Class to hold information on a collection of nodes
assert(isinstance(plugin_dict, dict))
self.__dict__ = plugin_dict
- def __eq__(self, other):
+ def __eq__(self, other):
return self.__dict__ == other.__dict__
log = logging.getLogger(__name__)
+
class Queue(JenkinsBase):
"""
Class that represents the Jenkins queue
class Result(object):
- def __init__(self, **kwargs ):
+ def __init__(self, **kwargs):
"""
"""
- self.__dict__.update( kwargs )
+ self.__dict__.update(kwargs)
def __str__(self):
- return "%s %s %s" % ( self.className, self.name, self.status )
+ return "%s %s %s" % (self.className, self.name, self.status)
def __repr__(self):
module_name = self.__class__.__module__
class_name = self.__class__.__name__
- self_str = str( self )
- return "<%s.%s %s>" % ( module_name , class_name , self_str )
+ self_str = str(self)
+ return "<%s.%s %s>" % (module_name, class_name, self_str)
def id(self):
"""
Calculate an ID for this object.
"""
- return "%s.%s" % ( self.className, self.name )
+ return "%s.%s" % (self.className, self.name)
from jenkinsapi.jenkinsbase import JenkinsBase
from jenkinsapi.result import Result
+
class ResultSet(JenkinsBase):
"""
Represents a result from a completed Jenkins run.
"""
- def __init__(self, url, build ):
+ def __init__(self, url, build):
"""
Init a resultset
:param url: url for a build, str
return self.build.job.get_jenkins_obj()
def __str__(self):
- return "Test Result for %s" % str( self.build )
+ return "Test Result for %s" % str(self.build)
@property
def name(self):
return str(self)
def keys(self):
- return [ a[0] for a in self.iteritems() ]
+ return [a[0] for a in self.iteritems()]
def items(self):
return [a for a in self.iteritems()]
def iteritems(self):
- for suite in self._data.get("suites", [] ):
+ for suite in self._data.get("suites", []):
for case in suite["cases"]:
- R = Result( **case )
+ R = Result(**case)
yield R.id(), R
- for report_set in self._data.get( "childReports", [] ):
+ for report_set in self._data.get("childReports", []):
for suite in report_set["result"]["suites"]:
for case in suite["cases"]:
- R = Result( **case )
+ R = Result(**case)
yield R.id(), R
def __len__(self):
requestKwargs = self.get_request_dict(url, params, None, headers)
return requests.get(url, **requestKwargs)
-
def post_url(self, url, params=None, data=None, headers=None):
requestKwargs = self.get_request_dict(url, params, data, headers)
return requests.post(url, **requestKwargs)
log = logging.getLogger(__name__)
+
class Views(object):
# TODO @lechat 20130702: Add check that plugin for view actually exists in Jenkins
projects / tools / python-jenkinsapi.git / commitdiff