Jenkins does not recognize the job requested.
"""
+class UnknownView( KeyError, JenkinsAPIException):
+ """
+ Jenkins does not recognize the view requested.
+ """
+
class ArtifactBroken(JenkinsAPIException):
"""
An artifact is broken, wrong
url = self.get_nodes_url()
return Nodes(url, self)
- def delete_view(self, str_view_name, view=None, person=None):
- """
- Delete a view
- :param str_view_name: name of the view, str
- :param view: View object to be deleted, jenkinsapi.View
- :param person: Person name (to create personal view), str
- :return: True if view has been deleted, False if view does not exist
- """
- url = urlparse.urljoin(self.baseurl, "user/%s/my-views/" % person) if person else self.baseurl
- qs = urllib.urlencode({'value': str_view_name})
- viewExistsCheck_url = urlparse.urljoin(url, "viewExistsCheck?%s" % qs)
- log.debug('viewExistsCheck_url=%s' % viewExistsCheck_url)
- result = self.requester.get_url(viewExistsCheck_url)
- log.debug('result=%s' % result)
- # Jenkins returns "<div/>" if view does not exist
- if len(result) == len('<div/>'):
- log.error('A view the name "%s" does not exist' % (str_view_name))
- return False
- else:
- self.delete_view_by_url(urlparse.urljoin(url, 'view/%s' % str_view_name))
- # We changed Jenkins config - need to update ourself
- self.poll()
- # TODO: add check here that the view actually been deleted
- return True
-
def __getitem__(self, jobname):
"""
Get a job by name
:param jobname: name of job, str
:return: Job obj
"""
- for url, name in self.get_jobs_info():
+ for name, job in self.get_jobs():
if name == jobname:
- preferred_scheme = urlparse.urlsplit(self.baseurl).scheme
- url_split = urlparse.urlsplit(url)
- preferred_url = urlparse.urlunsplit([preferred_scheme, url_split.netloc, url_split.path, url_split.query, url_split.fragment])
- return Job(preferred_url, name, jenkins_obj=self)
+ return job
raise UnknownJob(jobname)
def __contains__(self, jobname):
if poll:
self.poll()
+ def get_jenkins_obj(self):
+ raise NotImplementedError('Please implement this method on %s' % self.__class__.__name__)
+
def __eq__(self, other):
"""
Return true if the other object represents a connection to the same server
self.name = name
self.jenkins_obj = jenkins_obj
JenkinsBase.__init__(self, url)
+ self.deleted = False
def __str__(self):
return self.name
api_url = self.python_api_url( self.get_job_url( str_job_id ) )
return Job( api_url, str_job_id, self.jenkins_obj )
+ def delete(self):
+ """
+ Remove this view object
+ """
+ url = "%s/doDelete" % self.baseurl
+ self.jenkins_obj.requester.post_and_confirm_status(url, data='')
+ self.jenkins_obj.poll()
+ self.deleted = True
+
def keys(self):
return self.get_job_dict().keys()
def add_job(self, str_job_name, job=None):
"""
Add job to a view
-
+
: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
return False
job = self.jenkins_obj.get_job(str_job_name)
-
+
jobs = self._data.setdefault('jobs', [])
jobs.append({'name': job.name, 'url': job.baseurl})
data = {
else:
for viewdict in self._data["views"]:
yield viewdict["name"], viewdict["url"]
-
+
def get_nested_view_dict(self):
return dict( self._get_nested_views() )
import json
import urllib
from jenkinsapi.view import View
+from jenkinsapi.exceptions import UnknownView
class Views(object):
"""
def __init__(self, jenkins):
self.jenkins = jenkins
+ def __len__(self):
+ return len(self.keys())
+
+ def __delitem__(self, view_name):
+ if view_name == 'All':
+ raise ValueError('Cannot delete this view: %s' % view_name)
+
+ self[view_name].delete()
+ self.jenkins.poll
+
+ def __setitem__(self, name, value):
+ raise NotImplementedError()
+
def __getitem__(self, view_name):
for row in self.jenkins._data['views']:
if row['name'] == view_name:
row['url'],
row['name'],
self.jenkins)
- raise KeyError(view_name)
+ raise UnknownView(view_name)
def __iteritems__(self):
"""
def delete_view_by_url(self, str_url):
url = "%s/doDelete" % str_url
response = self.requester.get_url(url, data='')
- self.poll()
+ self.jenkins.poll()
return self
def create(self, str_view_name):
def setUp(self):
self.jenkins = Jenkins('http://localhost:8080')
self._delete_all_jobs()
+ self._delete_all_views()
def tearDown(self):
pass
for name in self.jenkins.get_jobs_list():
self.jenkins.delete_job(name)
+ def _delete_all_views(self):
+ all_view_names = self.jenkins.views().keys()[1:]
+ for name in all_view_names:
+ del self.jenkins.views()[name]
+
def _create_job(self, name='whatever', config=EMPTY_JOB_CONFIG):
job = self.jenkins.create_job(name, config)
self.jenkins.poll()
self.assertJobIsPresent(template_job_name)
self.assertJobIsPresent(copied_job_name)
-class NodeTests(BaseSystemTest):
- """
- """
-
- # def test_get_node_dict(self):
- # self.assertEqual(self.jenkins.get_node_dict(), {
- # 'master': 'http://localhost:8080/computer/master/api/python/'})
-
if __name__ == '__main__':
unittest.main()
self.assertIn(view_name, self.jenkins.views())
self.assertIsInstance(v, View)
+ def test_create_and_delete_views(self):
+ self._create_job()
+ view1_name = random_string()
+ new_view = self.jenkins.views().create(view1_name)
+ self.assertIn(view1_name, self.jenkins.views())
+ del self.jenkins.views()[view1_name]
+ self.assertNotIn(view1_name, self.jenkins.views())
+
+ def test_delete_view_which_does_not_exist(self):
+ self._create_job()
+ view1_name = random_string()
+ new_view = self.jenkins.views().create(view1_name)
+ self.assertIn(view1_name, self.jenkins.views())
+ del self.jenkins.views()[view1_name]
+
+ with self.assertRaises(KeyError):
+ del self.jenkins.views()[view1_name]
+
+
+
if __name__ == '__main__':
logging.basicConfig()
unittest.main()