-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 pyjenkinsci.node import Node\r
-from pyjenkinsci.exceptions import UnknownJob\r
-from utils.urlopener import mkurlopener\r
-import logging\r
-import time\r
-import urllib2\r
-import urllib\r
-import json\r
-\r
-log = logging.getLogger(__name__)\r
-\r
-class Jenkins(JenkinsBase):\r
- """\r
- Represents a jenkins environment.\r
- """\r
- def __init__(self, baseurl, username=None, password=None, proxyhost=None, proxyport=None, proxyuser=None, proxypass=None):\r
- """\r
-\r
- :param baseurl: baseurl for jenkins instance including port, str\r
- :param username: username for jenkins auth, str\r
- :param password: password for jenkins auth, str\r
- :param proxyhost: proxyhostname, str\r
- :param proxyport: proxyport, int\r
- :param proxyuser: proxyusername for proxy auth, str\r
- :param proxypass: proxypassword for proxyauth, str\r
- :return: a Jenkins obj\r
- """\r
- self.username = username\r
- self.password = password\r
- self.proxyhost = proxyhost\r
- self.proxyport = proxyport\r
- self.proxyuser = proxyuser\r
- self.proxypass = proxypass\r
- JenkinsBase.__init__( self, baseurl )\r
-\r
- def get_proxy_auth(self):\r
- return self.proxyhost, self.proxyport, self.proxyuser, self.proxypass\r
-\r
- def get_jenkins_auth(self):\r
- return self.username, self.password, self.baseurl\r
-\r
- def get_auth(self):\r
- auth_args = []\r
- auth_args.extend(self.get_jenkins_auth())\r
- auth_args.extend(self.get_proxy_auth())\r
- log.debug("args: %s" % auth_args)\r
- return auth_args\r
-\r
-\r
- def get_opener( self ):\r
- return mkurlopener(*self.get_auth())\r
-\r
- def validate_fingerprint( self, id ):\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.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
- return obj_fingerprint.validate_for_build( filename, job, build )\r
-\r
- def get_jenkins_obj(self):\r
- return self\r
-\r
- def get_jobs(self):\r
- """\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
-\r
- def iteritems(self):\r
- return self.get_jobs()\r
-\r
- def iterkeys(self):\r
- for info in self._data["jobs"]:\r
- yield info["name"]\r
-\r
- def keys(self):\r
- return [ a for a in self.iterkeys() ]\r
-\r
- def __str__(self):\r
- return "Jenkins server at %s" % self.baseurl\r
-\r
- def _get_views( self ):\r
- if not self._data.has_key( "views" ):\r
- pass\r
- else:\r
- for viewdict in self._data["views"]:\r
- yield viewdict["name"], viewdict["url"]\r
-\r
- def get_view_dict(self):\r
- return dict( self._get_views() )\r
-\r
- def get_view_url( self, str_view_name ):\r
- try:\r
- view_dict = self.get_view_dict()\r
- return view_dict[ str_view_name ]\r
- except KeyError, ke:\r
- all_views = ", ".join( view_dict.keys() )\r
- raise KeyError("View %s is not known - available: %s" % ( str_view_name, all_views ) )\r
-\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
-\r
- def __getitem__( self, buildname ):\r
- """\r
- Get a build\r
- """\r
- for name, job in self.get_jobs():\r
- if name == buildname:\r
- return job\r
- raise UnknownJob(buildname)\r
-\r
- def get_node_dict(self):\r
- """Get registered slave nodes on this instance"""\r
- url = self.python_api_url(self.get_node_url())\r
- node_dict = dict(self.get_data(url))\r
- return dict(\r
- (node['displayName'], self.python_api_url(self.get_node_url(node['displayName'])))\r
- for node in node_dict['computer'])\r
-\r
- def get_node(self, nodename):\r
- """Get a node object for a specific node"""\r
- node_url = self.python_api_url(self.get_node_url(nodename))\r
- return Node(node_url, nodename, jenkins_obj=self)\r
-\r
- def get_node_url(self, nodename=""):\r
- """Return the url for nodes"""\r
- url = "%(baseurl)s/computer/%(nodename)s" % {'baseurl': self.baseurl, 'nodename': nodename}\r
- return url\r
-\r
- def has_node(self, nodename):\r
- """\r
- Does a node by the name specified exist\r
- :param nodename: string, hostname\r
- :return: boolean\r
- """\r
- return nodename in self.get_node_dict()\r
-\r
- def delete_node(self, nodename):\r
- """\r
- Remove a node from the managed slave list\r
- Please note that you cannot remove the master node\r
-\r
- :param nodename: string holding a hostname\r
- :return: None\r
- """\r
- assert self.has_node(nodename), "This node: %s is not registered as a slave" % nodename\r
- assert nodename != "master", "you cannot delete the master node"\r
- url = "%s/doDelete" % self.get_node_url(nodename)\r
- fn_urlopen = self.get_jenkins_obj().get_opener()\r
- try:\r
- stream = fn_urlopen(url)\r
- html_result = stream.read()\r
- except urllib2.HTTPError, e:\r
- log.debug("Error reading %s" % url)\r
- raise\r
- return not self.has_node(nodename)\r
-\r
- def create_node(self, name, num_executors=2, node_description=None,\r
- remote_fs='/var/lib/jenkins', labels=None, exclusive=False):\r
- """\r
- Create a new slave node by name.\r
-\r
- :param name: fqdn of slave, str\r
- :param num_executors: number of executors, int\r
- :param node_description: a freetext field describing the node\r
- :param remote_fs: jenkins path, str\r
- :param labels: labels to associate with slave, str\r
- :param exclusive: tied to specific job, boolean\r
- :return: node obj\r
- """\r
- NODE_TYPE = 'hudson.slaves.DumbSlave$DescriptorImpl'\r
- MODE = 'NORMAL'\r
- if self.has_node(name):\r
- return Node(nodename=name, baseurl=self.get_node_url(nodename=name), jenkins_obj=self)\r
- if exclusive:\r
- MODE = 'EXCLUSIVE'\r
- params = {\r
- 'name' : name,\r
- 'type' : NODE_TYPE,\r
- 'json' : json.dumps ({\r
- 'name' : name,\r
- 'nodeDescription' : node_description,\r
- 'numExecutors' : num_executors,\r
- 'remoteFS' : remote_fs,\r
- 'labelString' : labels,\r
- 'mode' : MODE,\r
- 'type' : NODE_TYPE,\r
- 'retentionStrategy' : { 'stapler-class' : 'hudson.slaves.RetentionStrategy$Always' },\r
- 'nodeProperties' : { 'stapler-class-bag' : 'true' },\r
- 'launcher' : { 'stapler-class' : 'hudson.slaves.JNLPLauncher' }\r
- })\r
- }\r
- url = "%(nodeurl)s/doCreateItem?%(params)s" % {\r
- 'nodeurl': self.get_node_url(),\r
- 'params': urllib.urlencode(params)\r
- }\r
- print url\r
- fn_urlopen = self.get_jenkins_obj().get_opener()\r
- try:\r
- stream = fn_urlopen(url)\r
- html_result = stream.read()\r
- except urllib2.HTTPError, e:\r
- log.debug("Error reading %s" % url)\r
- log.exception(e)\r
- raise\r
- return Node(nodename=name, baseurl=self.get_node_url(nodename=name), jenkins_obj=self)\r
+from pyjenkinsci.jenkinsbase import JenkinsBase
+from pyjenkinsci.fingerprint import Fingerprint
+from pyjenkinsci.job import Job
+from pyjenkinsci.view import View
+from pyjenkinsci.node import Node
+from pyjenkinsci.exceptions import UnknownJob
+from utils.urlopener import mkurlopener
+import logging
+import time
+import urllib2
+import urllib
+try:
+ import json
+except ImportError:
+ import simplejson as json
+
+log = logging.getLogger(__name__)
+
+class Jenkins(JenkinsBase):
+ """
+ Represents a jenkins environment.
+ """
+ def __init__(self, baseurl, username=None, password=None, proxyhost=None, proxyport=None, proxyuser=None, proxypass=None):
+ """
+
+ :param baseurl: baseurl for jenkins instance including port, str
+ :param username: username for jenkins auth, str
+ :param password: password for jenkins auth, str
+ :param proxyhost: proxyhostname, str
+ :param proxyport: proxyport, int
+ :param proxyuser: proxyusername for proxy auth, str
+ :param proxypass: proxypassword for proxyauth, str
+ :return: a Jenkins obj
+ """
+ self.username = username
+ self.password = password
+ self.proxyhost = proxyhost
+ self.proxyport = proxyport
+ self.proxyuser = proxyuser
+ self.proxypass = proxypass
+ JenkinsBase.__init__( self, baseurl )
+
+ def get_proxy_auth(self):
+ return self.proxyhost, self.proxyport, self.proxyuser, self.proxypass
+
+ def get_jenkins_auth(self):
+ return self.username, self.password, self.baseurl
+
+ def get_auth(self):
+ auth_args = []
+ auth_args.extend(self.get_jenkins_auth())
+ auth_args.extend(self.get_proxy_auth())
+ log.debug("args: %s" % auth_args)
+ return auth_args
+
+
+ def get_opener( self ):
+ return mkurlopener(*self.get_auth())
+
+ def validate_fingerprint( self, id ):
+ obj_fingerprint = Fingerprint(self.baseurl, id, jenkins_obj=self)
+ obj_fingerprint.validate()
+ log.info("Jenkins says %s is valid" % id)
+
+ def get_artifact_data(self, id):
+ obj_fingerprint = Fingerprint(self.baseurl, id, jenkins_obj=self)
+ obj_fingerprint.validate()
+ return obj_fingerprint.get_info()
+
+ def validate_fingerprint_for_build(self, digest, filename, job, build ):
+ obj_fingerprint = Fingerprint( self.baseurl, digest, jenkins_obj=self )
+ return obj_fingerprint.validate_for_build( filename, job, build )
+
+ def get_jenkins_obj(self):
+ return self
+
+ def get_jobs(self):
+ """
+ Fetch all the build-names on this Hudson server.
+ """
+ for info in self._data["jobs"]:
+ yield info["name"], Job( info["url"], info["name"], jenkins_obj=self)
+
+ def iteritems(self):
+ return self.get_jobs()
+
+ def iterkeys(self):
+ for info in self._data["jobs"]:
+ yield info["name"]
+
+ def keys(self):
+ return [ a for a in self.iterkeys() ]
+
+ def __str__(self):
+ return "Jenkins server at %s" % self.baseurl
+
+ def _get_views( self ):
+ if not self._data.has_key( "views" ):
+ pass
+ else:
+ for viewdict in self._data["views"]:
+ yield viewdict["name"], viewdict["url"]
+
+ def get_view_dict(self):
+ return dict( self._get_views() )
+
+ def get_view_url( self, str_view_name ):
+ try:
+ view_dict = self.get_view_dict()
+ return view_dict[ str_view_name ]
+ except KeyError, ke:
+ all_views = ", ".join( view_dict.keys() )
+ raise KeyError("View %s is not known - available: %s" % ( str_view_name, all_views ) )
+
+ def get_view(self, str_view_name ):
+ view_url = self.get_view_url( str_view_name )
+ view_api_url = self.python_api_url( view_url )
+ return View(view_api_url , str_view_name, jenkins_obj=self)
+
+ def __getitem__( self, buildname ):
+ """
+ Get a build
+ """
+ for name, job in self.get_jobs():
+ if name == buildname:
+ return job
+ raise UnknownJob(buildname)
+
+ def get_node_dict(self):
+ """Get registered slave nodes on this instance"""
+ url = self.python_api_url(self.get_node_url())
+ node_dict = dict(self.get_data(url))
+ return dict(
+ (node['displayName'], self.python_api_url(self.get_node_url(node['displayName'])))
+ for node in node_dict['computer'])
+
+ def get_node(self, nodename):
+ """Get a node object for a specific node"""
+ node_url = self.python_api_url(self.get_node_url(nodename))
+ return Node(node_url, nodename, jenkins_obj=self)
+
+ def get_node_url(self, nodename=""):
+ """Return the url for nodes"""
+ url = "%(baseurl)s/computer/%(nodename)s" % {'baseurl': self.baseurl, 'nodename': nodename}
+ return url
+
+ def has_node(self, nodename):
+ """
+ Does a node by the name specified exist
+ :param nodename: string, hostname
+ :return: boolean
+ """
+ return nodename in self.get_node_dict()
+
+ def delete_node(self, nodename):
+ """
+ Remove a node from the managed slave list
+ Please note that you cannot remove the master node
+
+ :param nodename: string holding a hostname
+ :return: None
+ """
+ assert self.has_node(nodename), "This node: %s is not registered as a slave" % nodename
+ assert nodename != "master", "you cannot delete the master node"
+ url = "%s/doDelete" % self.get_node_url(nodename)
+ fn_urlopen = self.get_jenkins_obj().get_opener()
+ try:
+ stream = fn_urlopen(url)
+ html_result = stream.read()
+ except urllib2.HTTPError, e:
+ log.debug("Error reading %s" % url)
+ raise
+ return not self.has_node(nodename)
+
+ def create_node(self, name, num_executors=2, node_description=None,
+ remote_fs='/var/lib/jenkins', labels=None, exclusive=False):
+ """
+ Create a new slave node by name.
+
+ :param name: fqdn of slave, str
+ :param num_executors: number of executors, int
+ :param node_description: a freetext field describing the node
+ :param remote_fs: jenkins path, str
+ :param labels: labels to associate with slave, str
+ :param exclusive: tied to specific job, boolean
+ :return: node obj
+ """
+ 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)
+ 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' }
+ })
+ }
+ url = "%(nodeurl)s/doCreateItem?%(params)s" % {
+ 'nodeurl': self.get_node_url(),
+ 'params': urllib.urlencode(params)
+ }
+ print url
+ fn_urlopen = self.get_jenkins_obj().get_opener()
+ try:
+ stream = fn_urlopen(url)
+ html_result = stream.read()
+ except urllib2.HTTPError, e:
+ log.debug("Error reading %s" % url)
+ log.exception(e)
+ raise
+ return Node(nodename=name, baseurl=self.get_node_url(nodename=name), jenkins_obj=self)
projects / tools / python-jenkinsapi.git / commitdiff