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"
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())
+ 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)
fp = Fingerprint(self.build.job.jenkins.baseurl, local_md5, self.build.job.jenkins)
return fp.validate_for_build(os.path.basename(fspath), self.build.job.name, self.build.buildno)
- def _md5sum(self, fspath, chunksize=2**20):
+ def _md5sum(self, fspath, chunksize=2 ** 20):
"""
A MD5 hashing function intended to produce the same results as that used by
Jenkins.
"""
md5 = hashlib.md5()
try:
- with open(fspath,'rb') as f:
+ with open(fspath, 'rb') as f:
for chunk in iter(lambda: f.read(chunksize), ''):
md5.update(chunk)
except:
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):
"""
Returns build timestamp in UTC
'''
# Java timestamps are given in miliseconds since the epoch start!
- naive_timestamp = datetime.datetime(*time.gmtime(self._data['timestamp']/1000.0)[:6])
+ naive_timestamp = datetime.datetime(*time.gmtime(self._data['timestamp'] / 1000.0)[:6])
return pytz.utc.localize(naive_timestamp)
def get_console(self):
@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,
help="password for jenkins user auth", type='str', default=None)
parser.add_option("-b", "--block", dest="block", action="store_true", default=False,
help="Block until each of the jobs is complete.")
- parser.add_option("-t", "--token", dest="token",help="Optional security token.",
+ parser.add_option("-t", "--token", dest="token", help="Optional security token.",
default=None)
return parser
"""
-class BadURL(ValueError,JenkinsAPIException):
+class BadURL(ValueError, JenkinsAPIException):
"""
A URL appears to be broken
"""
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)
+ url = "%s/fingerprint/%s/" % (baseurl, id)
JenkinsBase.__init__(self, url, poll=False)
self.id = id
"""
assert until in ['completed', 'not_queued'], 'Unknown block condition: %s' % until
self.block_until_not_queued(timeout, delay)
- if until=='completed':
+ if until == 'completed':
self.block_until_completed(timeout, delay)
def stop(self):
:param exclusive: tied to specific job, boolean
:return: node obj
"""
- NODE_TYPE = 'hudson.slaves.DumbSlave$DescriptorImpl'
+ NODE_TYPE = 'hudson.slaves.DumbSlave$DescriptorImpl'
MODE = 'NORMAL'
if self.has_node(name):
return Node(nodename=name, baseurl=self.get_node_url(nodename=name), jenkins_obj=self)
params = {
'name': name,
'type': NODE_TYPE,
- 'json': json.dumps ({
+ 'json': json.dumps({
'name': name,
'nodeDescription': node_description,
'numExecutors': num_executors,
return url
else:
if url.endswith(r"/"):
- fmt="%s%s"
+ fmt = "%s%s"
else:
fmt = "%s/%s"
return fmt % (url, config.JENKINS_API)
This default class can handle simple authentication only.
"""
- VALID_STATUS_CODES = [200,]
+ VALID_STATUS_CODES = [200, ]
def __init__(self, username=None, password=None, ssl_verify=True):
if username: