pass
+class NotBuiltYet(NotFound):
+ """
+ A job has no build data.
+ """
+ pass
+
+
class ArtifactBroken(JenkinsAPIException):
"""
An artifact is broken, wrong
"""
import ast
+import pprint
import logging
from jenkinsapi import config
from jenkinsapi.custom_exceptions import JenkinsAPIException
self._data = self._poll()
if 'jobs' in self._data:
self._data['jobs'] = self.resolve_job_folders(self._data['jobs'])
+ return self
def _poll(self):
url = self.python_api_url(self.baseurl)
logging.exception('Inappropriate content found at %s', url)
raise JenkinsAPIException('Cannot parse %s' % response.content)
+ def pprint(self):
+ """Print out all the data in this object for debugging.
+ """
+ pprint.pprint(self._data)
+
def resolve_job_folders(self, jobs):
for job in list(jobs):
if 'color' not in job.keys():
from collections import defaultdict
from time import sleep
from jenkinsapi.build import Build
-from jenkinsapi.invocation import Invocation
+from jenkinsapi.queue import QueueItem
from jenkinsapi.jenkinsbase import JenkinsBase
from jenkinsapi.queue import QueueItem
from jenkinsapi.mutable_jenkins_thing import MutableJenkinsThing
Represents a jenkins job
A job can hold N builds which are the actual execution environments
"""
+
def __init__(self, url, name, jenkins_obj):
self.name = name
self.jenkins = jenkins_obj
and updates it with the missing builds if needed.'''
if not data.get("builds"):
return data
- # do not call _buildid_for_type here: it would poll and do an infinite loop
+ # do not call _buildid_for_type here: it would poll and do an infinite
+ # loop
oldest_loaded_build_number = data["builds"][-1]["number"]
if not data['firstBuild']:
first_build_number = oldest_loaded_build_number
if all_builds_loaded:
return data
api_url = self.python_api_url(self.baseurl)
- response = self.get_data(api_url, params={'tree': 'allBuilds[number,url]'})
+ response = self.get_data(
+ api_url, params={'tree': 'allBuilds[number,url]'})
data['builds'] = response['allBuilds']
return data
assert isinstance(block, bool)
assert isinstance(skip_if_running, bool)
- # Create a new invocation instance
- invocation = Invocation(self)
# Either copy the params dict or make a new one.
build_params = build_params and dict(
build_params.items()) or {} # Via POSTed JSON
params = {} # Via Get string
- with invocation:
- if len(self.get_params_list()) == 0:
- if self.is_queued():
- raise WillNotBuild('%s is already queued' % repr(self))
-
- elif self.is_running():
- if skip_if_running:
- log.warn(
- "Will not request new build because %s is already running", self.name)
- else:
- log.warn(
- "Will re-schedule %s even though it is already running", self.name)
- elif self.has_queued_build(build_params):
- msg = 'A build with these parameters is already queued.'
- raise WillNotBuild(msg)
-
- log.info("Attempting to start %s on %s", self.name, repr(
- self.get_jenkins_obj()))
-
- url = self.get_build_triggerurl()
- # If job has file parameters - it must be triggered
- # using "/build", not by "/buildWithParameters"
- # "/buildWithParameters" will ignore non-file parameters
- if files:
- url = "%s/build" % self.baseurl
-
- if cause:
- build_params['cause'] = cause
-
- if securitytoken:
- params['token'] = securitytoken
-
- build_params['json'] = self.mk_json_from_build_parameters(build_params, files)
- data = build_params
-
- response = self.jenkins.requester.post_and_confirm_status(
- url,
- data=data,
- params=params,
- files=files,
- valid=[200, 201]
- )
- response = response
- if invoke_pre_check_delay > 0:
- log.info(
- "Waiting for %is to allow Jenkins to catch up", invoke_pre_check_delay)
- sleep(invoke_pre_check_delay)
- if block:
- total_wait = 0
-
- while self.is_queued():
- log.info(
- "Waited %is for %s to begin...", total_wait, self.name)
- sleep(invoke_block_delay)
- total_wait += invoke_block_delay
- if self.is_running():
- running_build = self.get_last_build()
- running_build.block_until_complete(
- delay=invoke_pre_check_delay)
- return invocation
+ url = self.get_build_triggerurl()
+ # If job has file parameters - it must be triggered
+ # using "/build", not by "/buildWithParameters"
+ # "/buildWithParameters" will ignore non-file parameters
+ if files:
+ url = "%s/build" % self.baseurl
+
+ if cause:
+ build_params['cause'] = cause
+
+ if securitytoken:
+ params['token'] = securitytoken
+
+ build_params['json'] = self.mk_json_from_build_parameters(
+ build_params, files)
+ data = build_params
+
+ response = self.jenkins.requester.post_url(
+ url,
+ data=data,
+ params=params,
+ files=files,
+ )
+
+ queue_url = response.headers['location']
+ qi = QueueItem(queue_url, self.jenkins)
+ return qi
def _buildid_for_type(self, buildtype):
"""Gets a buildid for a given type of build"""
"""
from jenkinsapi.jenkinsbase import JenkinsBase
-from jenkinsapi.custom_exceptions import UnknownQueueItem
+from jenkinsapi.custom_exceptions import UnknownQueueItem, NotBuiltYet
import logging
+import time
log = logging.getLogger(__name__)
def iteritems(self):
for item in self._data['items']:
- yield item['id'], QueueItem(self.jenkins, **item)
+ id = item['id']
+ item_baseurl = "%s/item/%i" % (self.baseurl, id)
+ yield item['id'], QueueItem(baseurl=item_baseurl, jenkins_obj=self.jenkins)
def iterkeys(self):
for item in self._data['items']:
self.get_jenkins_obj().requester.post_url(deleteurl)
-class QueueItem(object):
- """
- Flexible class to handle queue items.
- If the Jenkins API changes this support those changes
+class QueueItem(JenkinsBase):
+ """An individual item in the queue
"""
- def __init__(self, jenkins, **kwargs):
- self.jenkins = jenkins
- self.__dict__.update(kwargs)
+ def __init__(self, baseurl, jenkins_obj):
+ self.jenkins = jenkins_obj
+ JenkinsBase.__init__(self, baseurl)
+
+ @property
+ def id(self):
+ return self._data['id']
+
+
+ def get_jenkins_obj(self):
+ return self.jenkins
def get_job(self):
"""
self.__class__.__name__, str(self))
def __str__(self):
- return "%s #%i" % (self.task['name'], self.id)
+ return "%s Queue #%i" % (self._data['task']['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):
+ build = self.block_until_building(delay)
+ return build.block_until_complete(delay=delay)
+
+
+ def block_until_building(self, delay=5):
+ while True:
+ try:
+ return self.poll().get_build()
+ except NotBuiltYet:
+ time.sleep(delay)
+ continue
+
+
+ def is_running(self):
+ """Return True if this queued item is running.
+ """
+ try:
+ return self.get_build().is_running()
+ except NotBuiltYet:
+ return False
+
+ def get_build_number(self):
+ try:
+ return self._data['executable']['number']
+ except KeyError:
+ raise NotBuiltYet()
+
+ def get_job_name(self):
+ try:
+ return self._data['task']['name']
+ except KeyError:
+ raise NotBuiltYet()
+
\ No newline at end of file
self.password = password
self.ssl_verify = ssl_verify
- def get_request_dict(self, params=None, data=None, files=None, headers=None):
- requestKwargs = {}
+ def get_request_dict(self, params=None, data=None, files=None, headers=None, **kwargs):
+ requestKwargs = kwargs
if self.username:
requestKwargs['auth'] = (self.username, self.password)
)
return url
- def get_url(self, url, params=None, headers=None):
- requestKwargs = self.get_request_dict(params=params, headers=headers)
+ def get_url(self, url, params=None, headers=None, allow_redirects=True):
+ requestKwargs = self.get_request_dict(params=params, headers=headers, allow_redirects=allow_redirects)
return requests.get(self._update_url_scheme(url), **requestKwargs)
- def post_url(self, url, params=None, data=None, files=None, headers=None):
- requestKwargs = self.get_request_dict(params=params, data=data, files=files, headers=headers)
+ def post_url(self, url, params=None, data=None, files=None, headers=None, allow_redirects=True):
+ requestKwargs = self.get_request_dict(params=params, data=data, files=files, headers=headers, allow_redirects=allow_redirects)
return requests.post(self._update_url_scheme(url), **requestKwargs)
def post_xml_and_confirm_status(self, url, params=None, data=None, valid=None):
<concurrentBuild>false</concurrentBuild>
<builders>
<hudson.tasks.Shell>
- <command>ping -c 10 localhost</command>
+ <command>ping -c 5 localhost</command>
</hudson.tasks.Shell>
</builders>
<publishers/>
except ImportError:
import unittest
import time
+import logging
from jenkinsapi.build import Build
-from jenkinsapi.invocation import Invocation
+from jenkinsapi.queue import QueueItem
from jenkinsapi_tests.systests.base import BaseSystemTest
from jenkinsapi_tests.test_utils.random_strings import random_string
from jenkinsapi_tests.systests.job_configs import LONG_RUNNING_JOB
from jenkinsapi_tests.systests.job_configs import SHORTISH_JOB, EMPTY_JOB
+log = logging.getLogger(__name__)
+
+
class TestInvocation(BaseSystemTest):
def test_invocation_object(self):
job_name = 'create_%s' % random_string()
- job = self.jenkins.create_job(job_name, LONG_RUNNING_JOB)
- ii = job.invoke(invoke_pre_check_delay=7)
- self.assertIsInstance(ii, Invocation)
+ job = self.jenkins.create_job(job_name, SHORTISH_JOB)
+ qq = job.invoke(invoke_pre_check_delay=7)
+ self.assertIsInstance(qq, QueueItem)
# Let Jenkins catchup
- time.sleep(3)
- self.assertTrue(ii.is_queued_or_running())
- self.assertEquals(ii.get_build_number(), 1)
+ qq.block_until_building()
+ self.assertEquals(qq.get_build_number(), 1)
def test_get_block_until_build_running(self):
job_name = 'create_%s' % random_string()
job = self.jenkins.create_job(job_name, LONG_RUNNING_JOB)
- ii = job.invoke(invoke_pre_check_delay=7)
+ qq = job.invoke(invoke_pre_check_delay=7)
time.sleep(3)
- bn = ii.get_build_number()
+ bn = qq.block_until_building(delay=3).get_number()
self.assertIsInstance(bn, int)
- ii.block(until='not_queued')
- self.assertTrue(ii.is_running())
- b = ii.get_build()
+
+ b = qq.get_build()
self.assertIsInstance(b, Build)
- ii.stop()
- self.assertFalse(ii.is_running())
- self.assertIsInstance(ii.get_build().get_console(), str)
- self.assertIn('Started by user', ii.get_build().get_console())
-
+ self.assertTrue(b.is_running())
+
+ b.stop()
+ time.sleep(1)
+ self.assertFalse(b.poll().is_running())
+ console = b.get_console()
+ self.assertIsInstance(console, str)
+ self.assertIn('Started by user', console)
+
def test_get_block_until_build_complete(self):
job_name = 'create_%s' % random_string()
job = self.jenkins.create_job(job_name, SHORTISH_JOB)
- ii = job.invoke()
- ii.block(until='completed')
- self.assertFalse(ii.is_running())
-
+ qq = job.invoke()
+ qq.block_until_complete()
+ self.assertFalse(qq.get_build().is_running())
+
def test_multiple_invocations_and_get_last_build(self):
job_name = 'create_%s' % random_string()
-
+
job = self.jenkins.create_job(job_name, SHORTISH_JOB)
-
+
for _ in range(3):
ii = job.invoke()
- ii.block(until='completed')
-
+ ii.block_until_complete(delay=2)
+
build_number = job.get_last_good_buildnumber()
self.assertEquals(build_number, 3)
-
+
build = job.get_build(build_number)
self.assertIsInstance(build, Build)
-
+
def test_multiple_invocations_and_get_build_number(self):
job_name = 'create_%s' % random_string()
-
+
job = self.jenkins.create_job(job_name, EMPTY_JOB)
-
+
for invocation in range(3):
- ii = job.invoke()
- ii.block(until='completed')
- build_number = ii.get_build_number()
+ qq = job.invoke()
+ qq.block_until_complete(delay=1)
+ build_number = qq.get_build_number()
self.assertEquals(build_number, invocation + 1)
if __name__ == '__main__':
+# logging.basicConfig()
+# logging.getLogger("").setLevel(logging.INFO)
unittest.main()
self.assertTrue(j.is_queued_or_running())
queue = self.jenkins.get_queue()
+
reprString = repr(queue)
self.assertIn(queue.baseurl, reprString)
projects / tools / python-jenkinsapi.git / commitdiff