def _poll(self):
url = self.python_api_url(self.baseurl)
- try:
- return self.get_data(url)
- except Exception as e:
- import pdb
- pdb.set_trace()
+ return self.get_data(url)
+
def get_data(self, url, params=None):
requester = self.get_jenkins_obj().requester
"""
Queue module for jenkinsapi
"""
-
+from requests import HTTPError
from jenkinsapi.jenkinsbase import JenkinsBase
from jenkinsapi.custom_exceptions import UnknownQueueItem, NotBuiltYet
import logging
self.__class__.__name__, str(self))
def __str__(self):
- return "%s Queue #%i" % (self._data['task']['name'], self._data['id'])
+ return "%s Queue #%i" % (self._data['name'], self._data['id'])
def get_build(self):
build_number = self.get_build_number()
job_name = self.get_job_name()
return self.jenkins[job_name][build_number]
- def block_until_complete(self, delay=15):
+ def block_until_complete(self, delay=5):
build = self.block_until_building(delay)
return build.block_until_complete(delay=delay)
while True:
try:
return self.poll().get_build()
- except NotBuiltYet:
+ except (NotBuiltYet, HTTPError):
time.sleep(delay)
continue
class TestNodes(BaseSystemTest):
def test_invoke_job_parameterized(self):
- node_name = random_string()
- self.jenkins.create_node(node_name)
- self.assertTrue(self.jenkins.has_node(node_name))
-
- N = self.jenkins.get_node(node_name)
- self.assertEquals(N.baseurl, self.jenkins.get_node_url(node_name))
-
- self.jenkins.delete_node(node_name)
- self.assertFalse(self.jenkins.has_node(node_name))
+ node_name = "param_" + random_string()
+ try:
+ self.jenkins.create_node(node_name)
+ self.assertTrue(self.jenkins.has_node(node_name))
+
+ N = self.jenkins.get_node(node_name)
+ self.assertEquals(N.baseurl, self.jenkins.get_node_url(node_name))
+
+ finally:
+ self.jenkins.delete_node(node_name)
+ self.assertFalse(self.jenkins.has_node(node_name))
def test_online_offline(self):
"""