else:
raise UnknownQueueItem(item_id)
- def get_queue_items_for_job(self, job_name=''):
- if not job_name:
- return [QueueItem(self.jenkins, **item)
- for item in self._data['items']]
- else:
- return [QueueItem(self.jenkins, **item)
- for item in self._data['items']
- if item['task']['name'] == job_name]
-
+ def _get_queue_items_for_job(self, job_name):
+ for item in self._data["items"]:
+ if item['task']['name'] == job_name:
+ yield QueueItem(self.get_queue_item_url(item), jenkins_obj=self.jenkins)
+
+ def get_queue_items_for_job(self, job_name):
+ return list(self._get_queue_items_for_job(job_name))
+
+ def get_queue_item_url(self, item):
+ return "%s/item/%i" % (self.baseurl, item["id"])
+
def delete_item(self, queue_item):
self.delete_item_by_id(queue_item.id)
"""
Return the job associated with this queue item
"""
- return self.jenkins[self.task['name']]
+ return self.jenkins[self._data['task']['name']]
def get_parameters(self):
"""returns parameters of queue item"""
- actions = getattr(self, 'actions', [])
+ actions = self._data.get('actions', [])
for action in actions:
if type(action) is dict and 'parameters' in action:
parameters = action['parameters']
class FourOhFourError(Exception):
+
"""
Missing fake data
"""
class TestQueue(unittest.TestCase):
-
- @classmethod
- def mockGetData(self, url):
- try:
- return TestQueue.URL_DATA[url]
- except KeyError:
- raise FourOhFourError(url)
-
- URL_DATA = {}
-
- URL_DATA['http://localhost:8080/%s' % config.JENKINS_API] = {
- 'jobs': [
- {
- 'name': 'utmebvpxrw',
- 'color': 'blue',
- 'url': 'http://localhost/job/utmebvpxrw'
- }
- ]
- }
-
- URL_DATA['http://localhost/job/utmebvpxrw/%s' % config.JENKINS_API] = {}
-
- URL_DATA['http://localhost:8080/queue/%s' % config.JENKINS_API] = {
+ QUEUE_DATA = {
'items': [
{
'actions': [
'userId': None,
'userName': 'anonymous'
}
- ]
+ ],
+ 'parameters':
+ [{'name': 'name1', 'value': 'value1'},
+ {'name': 'node'}]
}
],
'blocked': False,
]
}
+ @classmethod
+ def mockGetData(self, url):
+ try:
+ return TestQueue.URL_DATA[url]
+ except KeyError:
+ raise FourOhFourError(url)
+
+ URL_DATA = {}
+
+ URL_DATA['http://localhost:8080/%s' % config.JENKINS_API] = {
+ 'jobs': [
+ {
+ 'name': 'utmebvpxrw',
+ 'color': 'blue',
+ 'url': 'http://localhost/job/utmebvpxrw'
+ }
+ ]
+ }
+
+ URL_DATA['http://localhost/job/utmebvpxrw/%s' % config.JENKINS_API] = {}
+
+ URL_DATA['http://localhost:8080/queue/%s' %
+ config.JENKINS_API] = QUEUE_DATA
+
+ for item in QUEUE_DATA['items']:
+ url = 'http://localhost:8080/queue/item/%i/%s' % (
+ item['id'], config.JENKINS_API)
+ print(url)
+ URL_DATA[url] = item
+
@mock.patch.object(JenkinsBase, 'get_data', mockGetData)
def setUp(self):
self.J = Jenkins('http://localhost:8080') # Jenkins
def test_list_items(self):
self.assertEquals(set(self.q.keys()), set([40, 41, 42]))
+ @mock.patch.object(JenkinsBase, 'get_data', mockGetData)
def test_getitem(self):
item40 = self.q[40]
self.assertIsInstance(item40, QueueItem)
self.assertEquals(len(item40), 1)
self.assertIsInstance(item40[0], QueueItem)
- item40 = self.q.get_queue_items_for_job()
- self.assertIsInstance(item40, list)
- self.assertEquals(len(item40), 3)
+ @mock.patch.object(JenkinsBase, 'get_data', mockGetData)
def test_qi_get_parameters(self):
- act = [{'parameters':
- [{'name': 'name1', 'value': 'value1'},
- {'name': 'node'}]}]
- qi = QueueItem(jenkins=None, actions=act)
-
+ qi = self.q[42]
self.assertEquals(qi.get_parameters(), {'name1': 'value1',
'node': None})