<s> from jenkinsapi . jenkinsbase import JenkinsBase <EOL> from jenkinsapi . fingerprint import Fingerprint <EOL> from jenkinsapi . job import Job <EOL> from jenkinsapi . view import View <EOL> from jenkinsapi . node import Node <EOL> from jenkinsapi . exceptions import UnknownJob , NotAuthorized <EOL> from utils . urlopener import mkurlopener , mkopener , NoAuto302Handler <EOL> import logging <EOL> import time <EOL> import urllib2 <EOL> import urllib <EOL> import urlparse <EOL> import cookielib <EOL> try : <EOL> import json <EOL> except ImportError : <EOL> import simplejson as json <EOL> log = logging . getLogger ( __name__ ) <EOL> class Jenkins ( JenkinsBase ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , baseurl , username = None , password = None , proxyhost = None , proxyport = None , proxyuser = None , proxypass = None , formauth = False ) : <EOL> """<STR_LIT>""" <EOL> self . username = username <EOL> self . password = password <EOL> self . proxyhost = proxyhost <EOL> self . proxyport = proxyport <EOL> self . proxyuser = proxyuser <EOL> self . proxypass = proxypass <EOL> JenkinsBase . __init__ ( self , baseurl , formauth = formauth ) <EOL> def get_proxy_auth ( self ) : <EOL> return self . proxyhost , self . proxyport , self . proxyuser , self . proxypass <EOL> def get_jenkins_auth ( self ) : <EOL> return self . username , self . password , self . baseurl <EOL> def get_auth ( self ) : <EOL> auth_args = [ ] <EOL> auth_args . extend ( self . get_jenkins_auth ( ) ) <EOL> auth_args . extend ( self . get_proxy_auth ( ) ) <EOL> log . debug ( "<STR_LIT>" % auth_args ) <EOL> return auth_args <EOL> def get_opener ( self ) : <EOL> if self . formauth : <EOL> return self . get_login_opener ( ) <EOL> return mkurlopener ( * self . get_auth ( ) ) <EOL> def get_login_opener ( self ) : <EOL> hdrs = [ ] <EOL> if getattr ( self , '<STR_LIT>' , False ) : <EOL> mcj = cookielib . MozillaCookieJar ( ) <EOL> for c in self . _cookies : <EOL> mcj . set_cookie ( c ) <EOL> hdrs . append ( urllib2 . HTTPCookieProcessor ( mcj ) ) <EOL> return mkopener ( * hdrs ) <EOL> def login ( self ) : <EOL> formdata = dict ( j_username = self . username , j_password = self . password , <EOL> remember_me = True , form = '<STR_LIT:/>' ) <EOL> formdata . update ( dict ( json = json . dumps ( formdata ) , Submit = '<STR_LIT>' ) ) <EOL> formdata = urllib . urlencode ( formdata ) <EOL> loginurl = urlparse . urljoin ( self . baseurl , '<STR_LIT>' ) <EOL> mcj = cookielib . MozillaCookieJar ( ) <EOL> cookiehandler = urllib2 . HTTPCookieProcessor ( mcj ) <EOL> urlopen = mkopener ( NoAuto302Handler , cookiehandler ) <EOL> res = urlopen ( loginurl , data = formdata ) <EOL> self . _cookies = [ c for c in mcj ] <EOL> return res . getcode ( ) == <NUM_LIT> <EOL> def validate_fingerprint ( self , id ) : <EOL> obj_fingerprint = Fingerprint ( self . baseurl , id , jenkins_obj = self ) <EOL> obj_fingerprint . validate ( ) <EOL> log . info ( "<STR_LIT>" % id ) <EOL> def reload ( self ) : <EOL> '''<STR_LIT>''' <EOL> try : <EOL> self . hit_url ( "<STR_LIT>" % self . __dict__ ) <EOL> except urllib2 . HTTPError , e : <EOL> if e . code == <NUM_LIT> : <EOL> raise NotAuthorized ( "<STR_LIT>" ) <EOL> raise <EOL> def get_artifact_data ( self , id ) : <EOL> obj_fingerprint = Fingerprint ( self . baseurl , id , jenkins_obj = self ) <EOL> obj_fingerprint . validate ( ) <EOL> return obj_fingerprint . get_info ( ) <EOL> def validate_fingerprint_for_build ( self , digest , filename , job , build ) : <EOL> obj_fingerprint = Fingerprint ( self . baseurl , digest , jenkins_obj = self ) <EOL> return obj_fingerprint . validate_for_build ( filename , job , build ) <EOL> def get_jenkins_obj ( self ) : <EOL> return self <EOL> def get_jobs ( self ) : <EOL> """<STR_LIT>""" <EOL> for info in self . _data [ "<STR_LIT>" ] : <EOL> yield info [ "<STR_LIT:name>" ] , Job ( info [ "<STR_LIT:url>" ] , info [ "<STR_LIT:name>" ] , jenkins_obj = self ) <EOL> def get_jobs_info ( self ) : <EOL> """<STR_LIT>""" <EOL> for info in self . _data [ "<STR_LIT>" ] : <EOL> yield info [ "<STR_LIT:url>" ] , info [ "<STR_LIT:name>" ] <EOL> def get_jobs_list ( self ) : <EOL> """<STR_LIT>""" <EOL> jobs = [ ] <EOL> for info in self . _data [ "<STR_LIT>" ] : <EOL> jobs . append ( info [ "<STR_LIT:name>" ] ) <EOL> return jobs <EOL> def get_job ( self , jobname ) : <EOL> """<STR_LIT>""" <EOL> return self [ jobname ] <EOL> def has_job ( self , jobname ) : <EOL> """<STR_LIT>""" <EOL> return jobname in self . get_jobs_list ( ) <EOL> def copy_job ( self , jobname , newjobname ) : <EOL> """<STR_LIT>""" <EOL> qs = urllib . urlencode ( { '<STR_LIT:name>' : newjobname , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : jobname } ) <EOL> copy_job_url = urlparse . urljoin ( self . baseurl , "<STR_LIT>" % qs ) <EOL> self . post_data ( copy_job_url , '<STR_LIT>' ) <EOL> newjk = Jenkins ( self . baseurl , username = self . username , <EOL> password = self . password , proxyhost = self . proxyhost , <EOL> proxyport = self . proxyport , proxyuser = self . proxyuser , <EOL> proxypass = self . proxypass , formauth = self . formauth ) <EOL> return newjk . get_job ( newjobname ) <EOL> def delete_job ( self , jobname ) : <EOL> """<STR_LIT>""" <EOL> delete_job_url = urlparse . urljoin ( Jenkins ( self . baseurl ) . get_job ( jobname ) . baseurl , "<STR_LIT>" ) <EOL> self . post_data ( delete_job_url , '<STR_LIT>' ) <EOL> newjk = Jenkins ( self . baseurl , username = self . username , <EOL> password = self . password , proxyhost = self . proxyhost , <EOL> proxyport = self . proxyport , proxyuser = self . proxyuser , <EOL> proxypass = self . proxypass , formauth = self . formauth ) <EOL> return newjk <EOL> def iteritems ( self ) : <EOL> return self . get_jobs ( ) <EOL> def iterkeys ( self ) : <EOL> for info in self . _data [ "<STR_LIT>" ] : <EOL> yield info [ "<STR_LIT:name>" ] <EOL> def keys ( self ) : <EOL> return [ a for a in self . iterkeys ( ) ] <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" % self . baseurl <EOL> def _get_views ( self ) : <EOL> if not self . _data . has_key ( "<STR_LIT>" ) : <EOL> pass <EOL> else : <EOL> for viewdict in self . _data [ "<STR_LIT>" ] : <EOL> yield viewdict [ "<STR_LIT:name>" ] , viewdict [ "<STR_LIT:url>" ] <EOL> def get_view_dict ( self ) : <EOL> return dict ( self . _get_views ( ) ) <EOL> def get_view_url ( self , str_view_name ) : <EOL> try : <EOL> view_dict = self . get_view_dict ( ) <EOL> return view_dict [ str_view_name ] <EOL> except KeyError : <EOL> all_views = "<STR_LIT:U+002CU+0020>" . join ( view_dict . keys ( ) ) <EOL> raise KeyError ( "<STR_LIT>" % ( str_view_name , all_views ) ) <EOL> def get_view ( self , str_view_name ) : <EOL> view_url = self . get_view_url ( str_view_name ) <EOL> view_api_url = self . python_api_url ( view_url ) <EOL> return View ( view_url , str_view_name , jenkins_obj = self ) <EOL> def get_view_by_url ( self , str_view_url ) : <EOL> str_view_name = str_view_url . split ( '<STR_LIT>' ) [ - <NUM_LIT:1> ] . replace ( '<STR_LIT:/>' , '<STR_LIT>' ) <EOL> return View ( str_view_url , str_view_name , jenkins_obj = self ) <EOL> def delete_view_by_url ( self , str_url ) : <EOL> url = "<STR_LIT>" % str_url <EOL> self . post_data ( url , '<STR_LIT>' ) <EOL> newjk = Jenkins ( self . baseurl , username = self . username , <EOL> password = self . password , proxyhost = self . proxyhost , <EOL> proxyport = self . proxyport , proxyuser = self . proxyuser , <EOL> proxypass = self . proxypass , formauth = self . formauth ) <EOL> return newjk <EOL> def create_view ( self , str_view_name , people = None ) : <EOL> """<STR_LIT>""" <EOL> url = urlparse . urljoin ( self . baseurl , "<STR_LIT>" % people ) if people else self . baseurl <EOL> qs = urllib . urlencode ( { '<STR_LIT:value>' : str_view_name } ) <EOL> viewExistsCheck_url = urlparse . urljoin ( url , "<STR_LIT>" % qs ) <EOL> fn_urlopen = self . get_jenkins_obj ( ) . get_opener ( ) <EOL> try : <EOL> r = fn_urlopen ( viewExistsCheck_url ) . read ( ) <EOL> except urllib2 . HTTPError , e : <EOL> log . debug ( "<STR_LIT>" % viewExistsCheck_url ) <EOL> log . exception ( e ) <EOL> raise <EOL> """<STR_LIT>""" <EOL> if len ( r ) > <NUM_LIT:7> : <EOL> return '<STR_LIT>' % ( str_view_name ) <EOL> else : <EOL> data = { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT:OK>" } <EOL> data [ '<STR_LIT:name>' ] = str_view_name <EOL> data [ '<STR_LIT>' ] = data . copy ( ) <EOL> params = urllib . urlencode ( data ) <EOL> try : <EOL> createView_url = urlparse . urljoin ( url , "<STR_LIT>" ) <EOL> result = self . post_data ( createView_url , params ) <EOL> except urllib2 . HTTPError , e : <EOL> log . debug ( "<STR_LIT>" % createView_url ) <EOL> log . exception ( e ) <EOL> return urlparse . urljoin ( url , "<STR_LIT>" % str_view_name ) <EOL> def __getitem__ ( self , jobname ) : <EOL> """<STR_LIT>""" <EOL> for url , name in self . get_jobs_info ( ) : <EOL> if name == jobname : <EOL> return Job ( url , name , jenkins_obj = self ) <EOL> raise UnknownJob ( jobname ) <EOL> def get_node_dict ( self ) : <EOL> """<STR_LIT>""" <EOL> url = self . python_api_url ( self . get_node_url ( ) ) <EOL> node_dict = dict ( self . get_data ( url ) ) <EOL> return dict ( <EOL> ( node [ '<STR_LIT>' ] , self . python_api_url ( self . get_node_url ( node [ '<STR_LIT>' ] ) ) ) <EOL> for node in node_dict [ '<STR_LIT>' ] ) <EOL> def get_node ( self , nodename ) : <EOL> """<STR_LIT>""" <EOL> node_url = self . python_api_url ( self . get_node_url ( nodename ) ) <EOL> return Node ( node_url , nodename , jenkins_obj = self ) <EOL> def get_node_url ( self , nodename = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> url = "<STR_LIT>" % { '<STR_LIT>' : self . baseurl , '<STR_LIT>' : urllib . quote ( nodename ) } <EOL> return url <EOL> def has_node ( self , nodename ) : <EOL> """<STR_LIT>""" <EOL> return nodename in self . get_node_dict ( ) <EOL> def delete_node ( self , nodename ) : <EOL> """<STR_LIT>""" <EOL> assert self . has_node ( nodename ) , "<STR_LIT>" % nodename <EOL> assert nodename != "<STR_LIT>" , "<STR_LIT>" <EOL> url = "<STR_LIT>" % self . get_node_url ( nodename ) <EOL> fn_urlopen = self . get_jenkins_obj ( ) . get_opener ( ) <EOL> try : <EOL> fn_urlopen ( url ) . read ( ) <EOL> except urllib2 . HTTPError , e : <EOL> log . debug ( "<STR_LIT>" % url ) <EOL> log . exception ( e ) <EOL> raise <EOL> return not self . has_node ( nodename ) <EOL> def create_node ( self , name , num_executors = <NUM_LIT:2> , node_description = None , <EOL> remote_fs = '<STR_LIT>' , labels = None , exclusive = False ) : <EOL> """<STR_LIT>""" <EOL> NODE_TYPE = '<STR_LIT>' <EOL> MODE = '<STR_LIT>' <EOL> if self . has_node ( name ) : <EOL> return Node ( nodename = name , baseurl = self . get_node_url ( nodename = name ) , jenkins_obj = self ) <EOL> if exclusive : <EOL> MODE = '<STR_LIT>' <EOL> params = { <EOL> '<STR_LIT:name>' : name , <EOL> '<STR_LIT:type>' : NODE_TYPE , <EOL> '<STR_LIT>' : json . dumps ( { <EOL> '<STR_LIT:name>' : name , <EOL> '<STR_LIT>' : node_description , <EOL> '<STR_LIT>' : num_executors , <EOL> '<STR_LIT>' : remote_fs , <EOL> '<STR_LIT>' : labels , <EOL> '<STR_LIT>' : MODE , <EOL> '<STR_LIT:type>' : NODE_TYPE , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT:true>' } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } <EOL> } ) <EOL> } <EOL> url = "<STR_LIT>" % { <EOL> '<STR_LIT>' : self . get_node_url ( ) , <EOL> '<STR_LIT>' : urllib . urlencode ( params ) <EOL> } <EOL> print url <EOL> fn_urlopen = self . get_jenkins_obj ( ) . get_opener ( ) <EOL> try : <EOL> fn_urlopen ( url ) . read ( ) <EOL> except urllib2 . HTTPError , e : <EOL> log . debug ( "<STR_LIT>" % url ) <EOL> log . exception ( e ) <EOL> raise <EOL> return Node ( nodename = name , baseurl = self . get_node_url ( nodename = name ) , jenkins_obj = self ) </s>
