-from urlparse import urlparse
-from jenkinsapi.jenkinsbase import JenkinsBase
-from jenkinsapi.job import Job
import urllib
import logging
+from jenkinsapi.jenkinsbase import JenkinsBase
+from jenkinsapi.job import Job
+
+
log = logging.getLogger(__name__)
-class View(JenkinsBase):
+class View(JenkinsBase):
def __init__(self, url, name, jenkins_obj):
self.name = name
self.jenkins_obj = jenkins_obj
def iteritems(self):
for name, url in self.get_job_dict().iteritems():
- api_url = self.python_api_url( url )
- yield name, Job( api_url, name, self.jenkins_obj )
+ api_url = self.python_api_url(url)
+ yield name, Job(api_url, name, self.jenkins_obj)
def values(self):
- return [ a[1] for a in self.iteritems() ]
+ return [a[1] for a in self.iteritems()]
def items(self):
- return [ a for a in self.iteritems() ]
+ return [a for a in self.iteritems()]
- def _get_jobs( self ):
- if not self._data.has_key( "jobs" ):
+ def _get_jobs(self):
+ if not 'jobs' in self._data:
pass
else:
for viewdict in self._data["jobs"]:
yield viewdict["name"], viewdict["url"]
def get_job_dict(self):
- return dict( self._get_jobs() )
+ return dict(self._get_jobs())
def __len__(self):
- return len( self.get_job_dict().keys() )
+ return len(self.get_job_dict().keys())
def get_job_url(self, str_job_name):
try:
"""
Add job to a view
- :param job_name: name of the job to be added
+ :param str_job_name: name of the job to be added
:param job: Job object to be added
:return: True if job has been added, False if job already exists or
job not known to Jenkins
if not job:
if str_job_name in self.get_job_dict():
log.warn('Job %s is already in the view %s' %
- (str_job_name, self.name))
+ (str_job_name, self.name))
return False
else:
# Since this call can be made from nested view,
# Thus let's create top level Jenkins and ask him
# http://jenkins:8080/view/CRT/view/CRT-FB/view/CRT-SCRT-1301/
top_jenkins = self.get_jenkins_obj().get_jenkins_obj_from_url(
- self.baseurl.split('view/')[0])
+ self.baseurl.split('view/')[0])
if not top_jenkins.has_job(str_job_name):
log.error('Job "%s" is not known to Jenkins' % str_job_name)
return False
job = top_jenkins.get_job(str_job_name)
log.info('Creating job %s in view %s' % (str_job_name, self.name))
- jobs = self._data.setdefault('jobs', [])
data = {
- "description":"",
- "statusFilter":"",
- "useincluderegex":"on",
- "includeRegex":"",
+ "description": "",
+ "statusFilter": "",
+ "useincluderegex": "on",
+ "includeRegex": "",
"columns": [{"stapler-class": "hudson.views.StatusColumn",
- "kind": "hudson.views.StatusColumn"},
+ "kind": "hudson.views.StatusColumn"},
{"stapler-class": "hudson.views.WeatherColumn",
- "kind": "hudson.views.WeatherColumn"},
+ "kind": "hudson.views.WeatherColumn"},
{"stapler-class": "hudson.views.JobColumn",
- "kind": "hudson.views.JobColumn"},
+ "kind": "hudson.views.JobColumn"},
{"stapler-class": "hudson.views.LastSuccessColumn",
- "kind": "hudson.views.LastSuccessColumn"},
+ "kind": "hudson.views.LastSuccessColumn"},
{"stapler-class": "hudson.views.LastFailureColumn",
- "kind": "hudson.views.LastFailureColumn"},
+ "kind": "hudson.views.LastFailureColumn"},
{"stapler-class": "hudson.views.LastDurationColumn",
- "kind": "hudson.views.LastDurationColumn"},
+ "kind": "hudson.views.LastDurationColumn"},
{"stapler-class": "hudson.views.BuildButtonColumn",
- "kind": "hudson.views.BuildButtonColumn"}],
- "Submit":"OK",
- }
+ "kind": "hudson.views.BuildButtonColumn"}],
+ "Submit": "OK",
+ }
data["name"] = self.name
# Add existing jobs (if any)
for job_name in self.get_job_dict().keys():
data['json'] = data.copy()
data = urllib.urlencode(data)
self.get_jenkins_obj().requester.post_and_confirm_status(
- '%s/configSubmit' % self.baseurl, data=data)
+ '%s/configSubmit' % self.baseurl, data=data)
self.poll()
log.debug('Job "%s" has been added to a view "%s"' %
- (job.name, self.name))
+ (job.name, self.name))
return True
def _get_nested_views(self):
- if not self._data.has_key( "views" ):
+ if not 'views' in self._data:
pass
else:
for viewdict in self._data["views"]:
yield viewdict["name"], viewdict["url"]
def get_nested_view_dict(self):
- return dict( self._get_nested_views() )
+ return dict(self._get_nested_views())
def views(self):
return self.jenkins_obj.views()
self.assertEquals(ar.exception.message, 'Cannot create job job_new')
+ @mock.patch.object(JenkinsBase, '_poll')
+ @mock.patch.object(Jenkins, '_poll')
+ @mock.patch.object(Job, '_poll')
+ def test_get_jenkins_obj_from_url(self, _base_poll, _poll, _job_poll):
+ _job_poll.return_value = {}
+ _poll.return_value = {'jobs': [
+ {'name': 'job_one',
+ 'url': 'http://localhost:8080/job_one'},
+ {'name': 'job_one',
+ 'url': 'http://localhost:8080/job_one'},
+ ]}
+ _base_poll.return_value = _poll.return_value
+
+ mock_requester = Requester(username='foouser', password='foopassword')
+ mock_requester.post_xml_and_confirm_status = mock.MagicMock(
+ return_value='')
+
+ J = Jenkins('http://localhost:8080/',
+ username='foouser', password='foopassword',
+ requester=mock_requester)
+
+ new_jenkins = J.get_jenkins_obj_from_url('http://localhost:8080/')
+ self.assertEquals(new_jenkins, J)
+
+ new_jenkins = J.get_jenkins_obj_from_url('http://localhost:8080/foo')
+ self.assertNotEquals(new_jenkins, J)
+
+ @mock.patch.object(JenkinsBase, '_poll')
+ @mock.patch.object(Jenkins, '_poll')
+ @mock.patch.object(Job, '_poll')
+ def test_get_jenkins_obj(self, _base_poll, _poll, _job_poll):
+ _job_poll.return_value = {}
+ _poll.return_value = {'jobs': [
+ {'name': 'job_one',
+ 'url': 'http://localhost:8080/job_one'},
+ {'name': 'job_one',
+ 'url': 'http://localhost:8080/job_one'},
+ ]}
+ _base_poll.return_value = _poll.return_value
+
+ mock_requester = Requester(username='foouser', password='foopassword')
+ mock_requester.post_xml_and_confirm_status = mock.MagicMock(
+ return_value='')
+
+ J = Jenkins('http://localhost:8080/',
+ username='foouser', password='foopassword',
+ requester=mock_requester)
+
+ new_jenkins = J.get_jenkins_obj()
+ self.assertEquals(new_jenkins, J)
+
+
class TestJenkinsURLs(unittest.TestCase):
@mock.patch.object(Jenkins, '_poll')
self.assertEquals(
J.get_create_url(), 'http://localhost:8080/createItem')
+
if __name__ == '__main__':
unittest.main()