From 924726103a0befe98448e5a27453b36562c59649 Mon Sep 17 00:00:00 2001 From: Aleksey Maksimov Date: Sat, 23 Nov 2013 18:06:34 +0800 Subject: [PATCH] Updated api module Added optional username and password to all methods Fixed #203 Fixed #201 --- jenkinsapi/api.py | 108 +++++++++++++++++++++++++++++++++--------------------- 1 file changed, 66 insertions(+), 42 deletions(-) diff --git a/jenkinsapi/api.py b/jenkinsapi/api.py index 108b567..948f4de 100644 --- a/jenkinsapi/api.py +++ b/jenkinsapi/api.py @@ -1,6 +1,8 @@ """ -This module is a collection of helpful, high-level functions for automating common tasks. -Many of these functions were designed to be exposed to the command-line, hence the have simple string arguments. +This module is a collection of helpful, high-level functions +for automating common tasks. +Many of these functions were designed to be exposed to the command-line, +hence they have simple string arguments. """ import os import time @@ -15,67 +17,76 @@ from jenkinsapi.custom_exceptions import ArtifactsMissing, TimeOut, BadURL log = logging.getLogger(__name__) -def get_latest_test_results(jenkinsurl, jobname): +def get_latest_test_results(jenkinsurl, jobname, username=None, password=None): """ - A convenience function to fetch down the very latest test results from a jenkins job. + 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, username=username, + password=password) res = latestbuild.get_resultset() return res -def get_latest_build(jenkinsurl, jobname): +def get_latest_build(jenkinsurl, jobname, username=None, password=None): """ - A convenience function to fetch down the very latest test results from a jenkins job. + A convenience function to fetch down the very latest test results + from a jenkins job. """ - jenkinsci = Jenkins(jenkinsurl) + jenkinsci = Jenkins(jenkinsurl, username=username, password=password) job = jenkinsci[jobname] return job.get_last_build() -def get_latest_complete_build(jenkinsurl, jobname): +def get_latest_complete_build(jenkinsurl, jobname, + username=None, password=None): """ - A convenience function to fetch down the very latest test results from a jenkins job. + A convenience function to fetch down the very latest test results + from a jenkins job. """ - jenkinsci = Jenkins(jenkinsurl) + jenkinsci = Jenkins(jenkinsurl, username=username, password=password) job = jenkinsci[jobname] return job.get_last_completed_build() -def get_build(jenkinsurl, jobname, build_no): +def get_build(jenkinsurl, jobname, build_no, username=None, password=None): """ - A convenience function to fetch down the test results from a jenkins job by build number. + A convenience function to fetch down the test results + from a jenkins job by build number. """ - jenkinsci = Jenkins(jenkinsurl) + jenkinsci = Jenkins(jenkinsurl, username=username, password=password) job = jenkinsci[jobname] 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, + username=None, password=None): """ Find all the artifacts for the latest build of a job. """ - jenkinsci = Jenkins(jenkinsurl, proxyhost, proxyport, proxyuser, proxypass) + jenkinsci = Jenkins(jenkinsurl, username=username, password=password) job = jenkinsci[jobid] if build_no: build = job.get_build(build_no) else: build = job.get_last_good_build() artifacts = build.get_artifact_dict() - log.info(msg="Found %i artifacts in '%s'" % (len(artifacts.keys()), build_no)) + log.info(msg="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, + username=None, password=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 + 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) + jenkinsci = Jenkins(jenkinsurl, username=username, password=password) job = jenkinsci[jobid] build_ids = job.get_build_ids() for build_id in build_ids: @@ -84,24 +95,29 @@ def search_artifacts(jenkinsurl, jobid, artifact_ids=None): 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(msg="Artifacts %s missing from %s #%i" % (", ".join(missing_artifacts), jobid, build_id)) + log.debug(msg="Artifacts %s missing from %s #%i" + % (", ".join(missing_artifacts), jobid, build_id)) #noinspection PyUnboundLocalVariable raise ArtifactsMissing(missing_artifacts) -def grab_artifact(jenkinsurl, jobid, artifactid, targetdir): +def grab_artifact(jenkinsurl, jobid, artifactid, targetdir, + username=None, password=None): """ - 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. + 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) + artifacts = get_artifacts(jenkinsurl, jobid, + username=username, password=password) artifact = artifacts[artifactid] if not os.path.exists(targetdir): os.makedirs(targetdir) - artifact.savetodir(targetdir) + artifact.save_to_dir(targetdir) -def block_until_complete(jenkinsurl, jobs, maxwait=12000, interval=30, raise_on_timeout=True): +def block_until_complete(jenkinsurl, jobs, maxwait=12000, interval=30, + raise_on_timeout=True, username=None, password=None): """ Wait until all of the jobs in the list are complete. """ @@ -109,21 +125,23 @@ def block_until_complete(jenkinsurl, jobs, maxwait=12000, interval=30, raise_on_ assert maxwait > interval assert interval > 0 - obj_jenkins = Jenkins(jenkinsurl) + obj_jenkins = Jenkins(jenkinsurl, username=username, password=password) obj_jobs = [obj_jenkins[jid] for jid in jobs] for time_left in xrange(maxwait, 0, -interval): still_running = [j for j in obj_jobs if j.is_queued_or_running()] if not still_running: return str_still_running = ", ".join('"%s"' % str(a) for a in still_running) - log.warn(msg="Waiting for jobs %s to complete. Will wait another %is" % (str_still_running, time_left)) + log.warn(msg="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) + raise TimeOut("Waited too long for these jobs to complete: %s" + % str_still_running) -def get_view_from_url(url): +def get_view_from_url(url, username=None, password=None): """ Factory method """ @@ -131,18 +149,18 @@ def get_view_from_url(url): if not matched: raise BadURL("Cannot parse URL %s" % url) jenkinsurl, view_name = matched.groups() - jenkinsci = Jenkins(jenkinsurl) + jenkinsci = Jenkins(jenkinsurl, username=username, password=password) return jenkinsci.views[view_name] -def get_nested_view_from_url(url): +def get_nested_view_from_url(url, username=None, password=None): """ Returns View based on provided URL. Convenient for nested views. """ matched = constants.RE_SPLIT_VIEW_URL.search(url) if not matched: raise BadURL("Cannot parse URL %s" % url) - jenkinsci = Jenkins(matched.group(0)) + jenkinsci = Jenkins(matched.group(0), username=username, password=password) return jenkinsci.get_view_by_url(url) @@ -165,7 +183,8 @@ def install_artifacts(artifacts, dirstruct, installdir, basestaticurl): # The artifact must be loaded from jenkins theartifact = artifacts[artifactname] else: - # It's probably a static file, we can get it from the static collection + # It's probably a static file, + # we can get it from the static collection staticurl = urlparse.urljoin(basestaticurl, artifactname) theartifact = Artifact(artifactname, staticurl) theartifact.save(destpath) @@ -173,16 +192,21 @@ def install_artifacts(artifacts, dirstruct, installdir, basestaticurl): return installed -def search_artifact_by_regexp(jenkinsurl, jobid, artifactRegExp): +def search_artifact_by_regexp(jenkinsurl, jobid, artifactRegExp, + username=None, password=None): ''' - 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. + 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. @param jenkinsurl: The base URL of the jenkins server @param jobid: The name of the job we are to search through - @param artifactRegExp: A compiled regular expression object (not a re-string) + @param artifactRegExp: A compiled regular expression object + (not a re-string) + @param username: Jenkins login user name, optional + @param password: Jenkins login password, optional ''' - job = Jenkins(jenkinsurl) + job = Jenkins(jenkinsurl, username=username, password=password) j = job[jobid] build_ids = j.get_build_ids() -- 2.7.4