-import artifact\r
-import constants\r
-import jenkins\r
-from exceptions import ArtifactsMissing, TimeOut, BadURL\r
+from pyjenkinsci.artifact import Artifact\r
+from pyjenkinsci import constants\r
+from pyjenkinsci.jenkins import Jenkins\r
+from pyjenkinsci.exceptions import ArtifactsMissing, TimeOut, BadURL\r
from urllib2 import urlparse\r
\r
import os\r
res = latestbuild.get_resultset()\r
return res\r
\r
-def get_latest_build( jenkinsurl, jobname ):\r
+def get_latest_build(jenkinsurl, jobname):\r
"""\r
A convenience function to fetch down the very latest test results from a jenkins job.\r
"""\r
- jenkinsci = jenkins( jenkinsurl )\r
- job = jenkinsci[ jobname ]\r
+ jenkinsci = Jenkins(jenkinsurl)\r
+ job = jenkinsci[jobname]\r
return job.get_last_build()\r
\r
-def get_latest_complete_build( jenkinsurl, jobname ):\r
+def get_latest_complete_build(jenkinsurl, jobname):\r
"""\r
A convenience function to fetch down the very latest test results from a jenkins job.\r
"""\r
- jenkinsci = jenkins( jenkinsurl )\r
- job = jenkinsci[ jobname ]\r
+ jenkinsci = Jenkins(jenkinsurl)\r
+ job = jenkinsci[jobname]\r
return job.get_last_completed_build()\r
\r
def get_artifacts( jenkinsurl, jobid=None, build_no=None, proxyhost=None, proxyport=None, proxyuser=None, proxypass=None ):\r
"""\r
Find all the artifacts for the latest build of a job.\r
"""\r
- jenkinsci = jenkins( jenkinsurl, proxyhost, proxyport, proxyuser, proxypass )\r
- job = jenkinsci[ jobid ]\r
+ jenkinsci = Jenkins(jenkinsurl, proxyhost, proxyport, proxyuser, proxypass)\r
+ job = jenkinsci[jobid]\r
if build_no:\r
build = job.get_build( build_no )\r
else:\r
build = job.get_last_good_build()\r
- artifacts = dict( (artifact.filename, artifact) for artifact in build.get_artifacts() )\r
- log.info("Found %i artifacts in '%s'" % ( len(artifacts.keys() ), build_no ) )\r
+ artifacts = dict((artifact.filename, artifact) for artifact in build.get_artifacts())\r
+ log.info("Found %i artifacts in '%s'" % ( len(artifacts.keys() ), build_no ))\r
return artifacts\r
\r
def search_artifacts(jenkinsurl, jobid, artifact_ids=None, same_build=True, build_search_limit=None):\r
Search the entire history of a jenkins job for a list of artifact names. If same_build\r
is true then ensure that all artifacts come from the same build of the job\r
"""\r
- if len( artifact_ids ) == 0 or artifact_ids is None:\r
+ if len(artifact_ids) == 0 or artifact_ids is None:\r
return []\r
assert same_build, "same_build==False not supported yet"\r
- jenkinsci = jenkins( jenkinsurl )\r
+ jenkinsci = Jenkins( jenkinsurl )\r
job = jenkinsci[ jobid ]\r
build_ids = job.get_build_ids()\r
for build_id in build_ids:\r
log.debug("Artifacts %s missing from %s #%i" % ( ", ".join( missing_artifacts ), jobid, build_id ))\r
raise ArtifactsMissing( missing_artifacts )\r
\r
-def grab_artifact( jenkinsurl, jobid, artifactid, targetdir ):\r
+def grab_artifact(jenkinsurl, jobid, artifactid, targetdir):\r
"""\r
Convenience method to find the latest good version of an artifact and save it\r
to a target directory. Directory is made automatically if not exists.\r
os.makedirs( targetdir )\r
artifact.savetodir( targetdir)\r
\r
-def block_until_complete( jenkinsurl, jobs, maxwait=12000, interval=30, raise_on_timeout=True ):\r
+def block_until_complete(jenkinsurl, jobs, maxwait=12000, interval=30, raise_on_timeout=True):\r
"""\r
Wait until all of the jobs in the list are complete.\r
"""\r
assert maxwait > interval\r
assert interval > 0\r
\r
- obj_jenkins = jenkins( jenkinsurl )\r
- obj_jobs = [ obj_jenkins[ jid ] for jid in jobs ]\r
- for time_left in xrange( maxwait, 0, -interval ):\r
- still_running = [ j for j in obj_jobs if j.is_queued_or_running() ]\r
+ obj_jenkins = Jenkins(jenkinsurl)\r
+ obj_jobs = [obj_jenkins[jid] for jid in jobs]\r
+ for time_left in xrange(maxwait, 0, -interval):\r
+ still_running = [j for j in obj_jobs if j.is_queued_or_running()]\r
if not still_running:\r
return\r
- str_still_running = ", ".join( '"%s"' % str(a) for a in still_running )\r
- log.warn( "Waiting for jobs %s to complete. Will wait another %is" % ( str_still_running, time_left ) )\r
- time.sleep( interval )\r
+ str_still_running = ", ".join('"%s"' % str(a) for a in still_running)\r
+ log.warn( "Waiting for jobs %s to complete. Will wait another %is" % (str_still_running, time_left ))\r
+ time.sleep(interval)\r
if raise_on_timeout:\r
- raise TimeOut( "Waited too long for these jobs to complete: %s" % str_still_running )\r
+ raise TimeOut("Waited too long for these jobs to complete: %s" % str_still_running)\r
\r
-def get_view_from_url( url ):\r
+def get_view_from_url(url):\r
"""\r
Factory method\r
"""\r
matched = constants.RE_SPLIT_VIEW_URL.search(url)\r
if not matched:\r
- raise BadURL("Cannot parse URL %s" % url )\r
+ raise BadURL("Cannot parse URL %s" % url)\r
jenkinsurl, view_name = matched.groups()\r
- jenkinsci = jenkins( jenkinsurl )\r
- return jenkinsci.get_view( view_name )\r
+ jenkinsci = Jenkins(jenkinsurl)\r
+ return jenkinsci.get_view(view_name)\r
\r
-def install_artifacts( artifacts, dirstruct, installdir, basestaticurl ):\r
+def install_artifacts(artifacts, dirstruct, installdir, basestaticurl):\r
"""\r
Install the artifacts.\r
"""\r
assert basestaticurl.endswith("/"), "Basestaticurl should end with /"\r
installed = []\r
for reldir, artifactnames in dirstruct.items():\r
- destdir = os.path.join( installdir, reldir )\r
- if not os.path.exists( destdir ):\r
- log.warn( "Making install directory %s" % destdir )\r
- os.makedirs( destdir )\r
+ destdir = os.path.join(installdir, reldir)\r
+ if not os.path.exists(destdir):\r
+ log.warn("Making install directory %s" % destdir)\r
+ os.makedirs(destdir)\r
else:\r
- assert os.path.isdir( destdir )\r
+ assert os.path.isdir(destdir)\r
for artifactname in artifactnames:\r
- destpath = os.path.abspath( os.path.join( destdir, artifactname ) )\r
+ destpath = os.path.abspath(os.path.join( destdir, artifactname))\r
if artifactname in artifacts.keys():\r
# The artifact must be loaded from jenkins\r
- theartifact = artifacts[ artifactname ]\r
+ theartifact = artifacts[artifactname]\r
else:\r
# It's probably a static file, we can get it from the static collection\r
- staticurl = urlparse.urljoin( basestaticurl, artifactname )\r
- theartifact = artifact( artifactname, staticurl )\r
- theartifact.save( destpath )\r
- installed.append( destpath )\r
+ staticurl = urlparse.urljoin(basestaticurl, artifactname)\r
+ theartifact = Artifact(artifactname, staticurl)\r
+ theartifact.save(destpath)\r
+ installed.append(destpath)\r
return installed\r
import zipfile\r
import cPickle\r
import datetime\r
-import config\r
+from pyjenkinsci import config\r
\r
-from utils.retry import retry_function\r
-from exceptions import ArtifactBroken\r
-from utils.md5hash import new_digest\r
+from pyjenkinsci.utils.retry import retry_function\r
+from pyjenkinsci.exceptions import ArtifactBroken\r
+from pyjenkinsci.utils.md5hash import new_digest\r
\r
log = logging.getLogger( __name__ )\r
\r
-class artifact( object ):\r
+class Artifact(object):\r
\r
@staticmethod\r
def timedelta_to_seconds( td ):\r
-import artifact\r
-import config\r
-import jenkinsobject\r
-import time\r
+from pyjenkinsci.artifact import Artifact\r
+from pyjenkinsci import config\r
+from pyjenkinsci.jenkinsbase import JenkinsBase\r
+from pyjenkinsci.exceptions import NoResults, FailedNoResults\r
+from pyjenkinsci.constants import STATUS_FAIL, STATUS_ABORTED, RESULTSTATUS_FAILURE\r
+from pyjenkinsci.result_set import ResultSet\r
+\r
+from datetime import time\r
import logging\r
-from exceptions import NoResults, FailedNoResults\r
-from constants import STATUS_FAIL, STATUS_ABORTED, RESULTSTATUS_FAILURE\r
-import result_set\r
\r
log = logging.getLogger(__name__)\r
\r
-class build(jenkinsobject):\r
+class Build(JenkinsBase):\r
"""\r
Represents a jenkins build, executed in context of a job.\r
"""\r
assert type(buildno) == int\r
self.buildno = buildno\r
self.job = job\r
- jenkinsobject.__init__( self, url )\r
+ JenkinsBase.__init__( self, url )\r
\r
def __str__(self):\r
return self._data['fullDisplayName']\r
def get_artifacts( self ):\r
for afinfo in self._data["artifacts"]:\r
url = "%sartifact/%s" % ( self.baseurl, afinfo["relativePath"] )\r
- af = artifact( afinfo["fileName"], url, self )\r
+ af = Artifact( afinfo["fileName"], url, self )\r
yield af\r
del af, url\r
\r
raise FailedNoResults( self.STR_TPL_NOTESTS_ERR % ( str(self), buildstatus ) )\r
if self.get_actions()[ self.STR_TOTALCOUNT ] == 0:\r
raise NoResults( self.STR_TPL_NOTESTS_ERR % ( str(self), buildstatus ) )\r
- obj_results = result_set( result_url, build=self )\r
+ obj_results = ResultSet( result_url, build=self )\r
return obj_results\r
\r
def has_resultset(self):\r
-import jenkinsobject\r
-from exceptions import ArtifactBroken\r
+from pyjenkinsci.jenkinsbase import JenkinsBase\r
+from pyjenkinsci.exceptions import ArtifactBroken\r
\r
import urllib2\r
import re\r
\r
log = logging.getLogger( __name__ )\r
\r
-class fingerprint(jenkinsobject):\r
+class Fingerprint(JenkinsBase):\r
"""\r
Represents a jenkins fingerprint on a single artifact file ??\r
"""\r
RE_MD5 = re.compile("^([0-9a-z]{32})$")\r
\r
- def __init__( self, baseurl, id, jenkins_obj ):\r
+ def __init__(self, baseurl, id, jenkins_obj):\r
logging.basicConfig()\r
self.jenkins_obj = jenkins_obj\r
assert self.RE_MD5.search( id ), "%s does not look like a valid id" % id\r
url = "%s/fingerprint/%s/" % ( baseurl, id )\r
- jenkinsobject.__init__( self, url, poll=False )\r
+ JenkinsBase.__init__( self, url, poll=False )\r
self.id = id\r
\r
def get_jenkins_obj(self):\r
"""\r
self.poll()\r
return self._data["original"]["name"], self._data["original"]["number"], self._data["fileName"]\r
-\r
-\r
-if __name__ == "__main__":\r
- ff = fingerprint( "http://localhost:8080/hudson/", "0f37cbb6545b8778bc0700d90be66bf3" )\r
- print repr(ff)\r
- print ff.baseurl\r
- print ff.valid()\r
- print ff.get_info( )\r
-import fingerprint\r
-import jenkinsobject\r
-import job\r
+from pyjenkinsci.jenkinsbase import JenkinsBase\r
+from pyjenkinsci.fingerprint import Fingerprint\r
+from pyjenkinsci.job import Job\r
+from pyjenkinsci.view import View\r
from exceptions import UnknownJob\r
from utils.urlopener import mkurlopener\r
import logging\r
import time\r
-import view\r
\r
log = logging.getLogger(__name__)\r
\r
-class jenkins(jenkinsobject):\r
+class Jenkins(JenkinsBase):\r
"""\r
Represents a jenkins environment.\r
"""\r
self.proxyport = proxyport\r
self.proxyuser = proxyuser\r
self.proxypass = proxypass\r
- jenkinsobject.__init__( self, baseurl )\r
+ JenkinsBase.__init__( self, baseurl )\r
\r
def get_proxy_auth(self):\r
return (self.proxyhost, self.proxyport, self.proxyuser, self.proxypass)\r
return mkurlopener(*self.get_proxy_auth())\r
\r
def validate_fingerprint( self, id ):\r
- obj_fingerprint = fingerprint(self.baseurl, id, jenkins_obj=self)\r
+ obj_fingerprint = Fingerprint(self.baseurl, id, jenkins_obj=self)\r
obj_fingerprint.validate()\r
log.info("Jenkins says %s is valid" % id)\r
\r
def get_artifact_data(self, id):\r
- obj_fingerprint = fingerprint(self.baseurl, id, jenkins_obj=self)\r
+ obj_fingerprint = Fingerprint(self.baseurl, id, jenkins_obj=self)\r
obj_fingerprint.validate()\r
return obj_fingerprint.get_info()\r
\r
def validate_fingerprint_for_build(self, digest, filename, job, build ):\r
- obj_fingerprint = fingerprint( self.baseurl, digest, jenkins_obj=self )\r
+ obj_fingerprint = Fingerprint( self.baseurl, digest, jenkins_obj=self )\r
return obj_fingerprint.validate_for_build( filename, job, build )\r
\r
def get_jenkins_obj(self):\r
Fetch all the build-names on this Hudson server.\r
"""\r
for info in self._data["jobs"]:\r
- yield info["name"], job( info["url"], info["name"], jenkins_obj=self)\r
+ yield info["name"], Job( info["url"], info["name"], jenkins_obj=self)\r
\r
def iteritems(self):\r
return self.get_jobs()\r
def get_view(self, str_view_name ):\r
view_url = self.get_view_url( str_view_name )\r
view_api_url = self.python_api_url( view_url )\r
- return view(view_api_url , str_view_name, jenkins_obj=self)\r
+ return View(view_api_url , str_view_name, jenkins_obj=self)\r
\r
def __getitem__( self, buildname ):\r
"""\r
--- /dev/null
+import urllib2\r
+import logging\r
+import pprint\r
+from pyjenkinsci import config\r
+from pyjenkinsci.utils.retry import retry_function\r
+\r
+log = logging.getLogger( __name__ )\r
+\r
+class JenkinsBase(object):\r
+ """\r
+ This appears to be the base object that all other jenkins objects are inherited from\r
+ """\r
+ RETRY_ATTEMPTS = 5\r
+\r
+ def __repr__( self ):\r
+ return """<%s.%s %s>""" % ( self.__class__.__module__,\r
+ self.__class__.__name__,\r
+ str( self ) )\r
+\r
+ def print_data(self):\r
+ pprint.pprint( self._data )\r
+\r
+ def __str__(self):\r
+ raise NotImplemented\r
+\r
+ def __init__( self, baseurl, poll=True ):\r
+ """\r
+ Initialize a jenkins connection\r
+ """\r
+ self.baseurl = baseurl\r
+ if poll:\r
+ try:\r
+ self.poll()\r
+ except urllib2.HTTPError, hte:\r
+ log.exception(hte)\r
+ log.warn( "Failed to conenct to %s" % baseurl )\r
+ raise\r
+\r
+ def poll(self):\r
+ self._data = self._poll()\r
+\r
+ def _poll(self):\r
+ url = self.python_api_url( self.baseurl )\r
+ return retry_function( self.RETRY_ATTEMPTS , self.get_data, url )\r
+\r
+ def get_jenkins_obj(self):\r
+ """Not implemented, abstract method implemented by child classes"""\r
+ raise NotImplemented("Abstract method, implemented by child classes")\r
+\r
+ @classmethod\r
+ def python_api_url( cls, url ):\r
+ if url.endswith( config.JENKINS_API ):\r
+ return url\r
+ else:\r
+ if url.endswith( r"/" ):\r
+ fmt="%s%s"\r
+ else:\r
+ fmt = "%s/%s"\r
+ return fmt % (url, config.JENKINS_API)\r
+\r
+ def get_data( self, url ):\r
+ """\r
+ Find out how to connect, and then grab the data.\r
+ """\r
+ fn_urlopen = self.get_jenkins_obj().get_opener()\r
+ try:\r
+ stream = fn_urlopen( url )\r
+ result = eval( stream.read() )\r
+ except urllib2.HTTPError, e:\r
+ log.warn( "Error reading %s" % url )\r
+ raise\r
+ return result\r
+++ /dev/null
-import urllib2\r
-import logging\r
-import pprint\r
-import config\r
-from utils.retry import retry_function\r
-\r
-log = logging.getLogger( __name__ )\r
-\r
-class jenkinsobject( object ):\r
- """\r
- This appears to be the base object that all other jenkins objects are inherited from\r
- """\r
- RETRY_ATTEMPTS = 5\r
-\r
- def __repr__( self ):\r
- return """<%s.%s %s>""" % ( self.__class__.__module__,\r
- self.__class__.__name__,\r
- str( self ) )\r
-\r
- def print_data(self):\r
- pprint.pprint( self._data )\r
-\r
- def __str__(self):\r
- raise NotImplemented\r
-\r
- def __init__( self, baseurl, poll=True ):\r
- """\r
- Initialize a jenkins connection\r
- """\r
- self.baseurl = baseurl\r
- if poll:\r
- try:\r
- self.poll()\r
- except urllib2.HTTPError, hte:\r
- log.exception(hte)\r
- log.warn( "Failed to conenct to %s" % baseurl )\r
- raise\r
-\r
- def poll(self):\r
- self._data = self._poll()\r
-\r
- def _poll(self):\r
- url = self.python_api_url( self.baseurl )\r
- return retry_function( self.RETRY_ATTEMPTS , self.get_data, url )\r
-\r
- @classmethod\r
- def python_api_url( cls, url ):\r
- if url.endswith( config.JENKINS_API ):\r
- return url\r
- else:\r
- if url.endswith( r"/" ):\r
- fmt="%s%s"\r
- else:\r
- fmt = "%s/%s"\r
- return fmt % (url, config.JENKINS_API)\r
-\r
- def get_data( self, url ):\r
- """\r
- Find out how to connect, and then grab the data.\r
- """\r
- fn_urlopen = self.getHudsonObject().get_opener()\r
- try:\r
- stream = fn_urlopen( url )\r
- result = eval( stream.read() )\r
- except urllib2.HTTPError, e:\r
- log.warn( "Error reading %s" % url )\r
- raise\r
- return result\r
import logging\r
import urlparse\r
import urllib2\r
-import time\r
-import build\r
-import jenkinsobject\r
+from datetime import time\r
+from pyjenkinsci.build import Build\r
+from pyjenkinsci.jenkinsbase import JenkinsBase\r
\r
from exceptions import NoBuildData\r
\r
log = logging.getLogger(__name__)\r
\r
-class job(jenkinsobject):\r
+class Job(JenkinsBase):\r
"""\r
Represents a jenkins job\r
A job can hold N builds which are the actual execution environments\r
def __init__( self, url, name, jenkins_obj ):\r
self.name = name\r
self.jenkins = jenkins_obj\r
- jenkinsobject.__init__( self, url )\r
+ JenkinsBase.__init__( self, url )\r
\r
def id( self ):\r
return self._data["name"]\r
def get_build( self, buildnumber ):\r
assert type(buildnumber) == int\r
url = self.get_build_dict()[ buildnumber ]\r
- return build( url, buildnumber, job=self )\r
+ return Build( url, buildnumber, job=self )\r
\r
def __getitem__( self, buildnumber ):\r
return self.get_build(buildnumber)\r
-class result( object ):\r
+class Result(object):\r
def __init__(self, **kwargs ):\r
"""\r
\r
-import jenkinsobject\r
-import result\r
+from pyjenkinsci.jenkinsbase import JenkinsBase\r
+from pyjenkinsci.result import Result\r
\r
-class result_set(jenkinsobject):\r
+class ResultSet(JenkinsBase):\r
"""\r
Represents a result from a completed Hudson run.\r
"""\r
- def getHudsonObject(self):\r
+ def get_jenkins_obj(self):\r
return self.build.job.get_jenkins_obj()\r
\r
def __init__(self, url, build ):\r
"""\r
"""\r
self.build = build\r
- jenkinsobject.__init__( self, url )\r
+ JenkinsBase.__init__( self, url )\r
\r
def __str__(self):\r
return "Test Result for %s" % str( self.build )\r
def iteritems(self):\r
for suite in self._data.get("suites", [] ):\r
for case in suite["cases"]:\r
- R = result( **case )\r
+ R = Result( **case )\r
yield R.id(), R\r
\r
for report_set in self._data.get( "childReports", [] ):\r
for suite in report_set["result"]["suites"]:\r
for case in suite["cases"]:\r
- R = result( **case )\r
+ R = Result( **case )\r
yield R.id(), R\r
\r
def __len__(self):\r
-import jenkinsobject\r
-import job\r
+from pyjenkinsci.jenkinsbase import JenkinsBase\r
+from pyjenkinsci.job import Job\r
\r
-class view(jenkinsobject):\r
+class View(JenkinsBase):\r
\r
def __init__(self, url, name, jenkins_obj):\r
self.name = name\r
self.jenkins_obj = jenkins_obj\r
- jenkinsobject.__init__(self, url)\r
+ JenkinsBase.__init__(self, url)\r
\r
def __str__(self):\r
return self.name\r
def __getitem__(self, str_job_id ):\r
assert isinstance( str_job_id, str )\r
api_url = self.python_api_url( self.get_job_url( str_job_id ) )\r
- return job( api_url, str_job_id, self.jenkins_obj )\r
+ return Job( api_url, str_job_id, self.jenkins_obj )\r
\r
def keys(self):\r
return self.get_job_dict().keys()\r
def iteritems(self):\r
for name, url in self.get_job_dict().iteritems():\r
api_url = self.python_api_url( url )\r
- yield name, job( api_url, name, self.jenkins_obj )\r
+ yield name, Job( api_url, name, self.jenkins_obj )\r
\r
def values(self):\r
return [ a[1] for a in self.iteritems() ]\r
projects / tools / python-jenkinsapi.git / commitdiff