"fingerprint", "jenkins", "jenkinsbase", "job", "node", "result_set", "result", "view"
]
__docformat__ = "epytext"
-__version__ = pkg_resources.working_set.by_key['jenkinsapi'].version
+
+import sys
+# In case of jenkinsapi is not installed in 'develop' mode
+if not sys.argv[0].endswith('nosetests'):
+ __version__ = pkg_resources.working_set.by_key['jenkinsapi'].version
+else:
+ # Return bogus version
+ __version__ = '99.99.99'
def get_progress(self):
"""Returns percentage"""
- self.poll()
- return self._data['progress']
+ return self.poll(tree='progress')['progress']
def get_number(self):
"""
Get Executor number.
"""
- self.poll()
- return self._data['number']
+ return self.poll(tree='number')['number']
def is_idle(self):
"""
Returns Boolean: whether Executor is idle or not.
"""
- self.poll()
- return self._data['idle']
+ return self.poll(tree='idle')['idle']
def likely_stuck(self):
"""
Returns Boolean: whether Executor is likely stuck or not.
"""
- self.poll()
- return self._data['likelyStuck']
+ return self.poll(tree='likelyStuck')['likelyStuck']
def get_current_executable(self):
"""
Returns the current Queue.Task this executor is running.
"""
- self.poll()
- return self._data['currentExecutable']
+ return self.poll(tree='currentExecutable')['currentExecutable']
url = self.get_build_triggerurl(files)
if cause:
build_params['cause'] = cause
-
+
# Build require params as form fields
# and as Json.
data = {'json': self.mk_json_from_build_parameters(
- build_params, files)}
+ build_params, files)}
data.update(build_params)
- response = self.jenkins.requester.post_url(
+ response = self.jenkins.requester.post_and_confirm_status(
url,
data=data,
params=params,
files=files,
+ valid=[200, 201, 303],
+ allow_redirects=False
)
redirect_url = response.headers['location']
if not redirect_url.startswith("%s/queue/item" % self.jenkins.baseurl):
- raise ValueError("Not a Queue URL: %s" % redirect_url)
+ if files:
+ raise ValueError('Builds with file parameters are not '
+ 'supported by this jenkinsapi version. '
+ 'Please use previous version.')
+ else:
+ raise ValueError("Not a Queue URL: %s" % redirect_url)
qi = QueueItem(redirect_url, self.jenkins)
if block:
return self.name
def is_online(self):
- self.poll()
- return not self._data['offline']
+ return not self.poll(tree='offline')['offline']
def is_temporarily_offline(self):
- self.poll()
- return self._data['temporarilyOffline']
+ return self.poll(tree='temporarilyOffline')['temporarilyOffline']
def is_jnlpagent(self):
return self._data['jnlpAgent']
"""
if not self._data['offline']:
self.toggle_temporarily_offline(message)
- self.poll()
- if not self._data['offline']:
+ data = self.poll(tree='offline,temporarilyOffline')
+ if not data['offline']:
raise AssertionError("The node state is still online:" +
"offline = %s , temporarilyOffline = %s" %
- (self._data['offline'], self._data['temporarilyOffline']))
+ (data['offline'], data['temporarilyOffline']))
def toggle_temporarily_offline(self, message="requested from jenkinsapi"):
"""
def block_until_building(self, delay=5):
while True:
try:
- return self.poll().get_build()
+ self.poll()
+ return self.get_build()
except (NotBuiltYet, HTTPError):
time.sleep(delay)
continue
headers = {'Content-Type': 'text/xml'}
return self.post_and_confirm_status(url, params=params, data=data, headers=headers, valid=valid)
- def post_and_confirm_status(self, url, params=None, data=None, files=None, headers=None, valid=None):
+ def post_and_confirm_status(self, url, params=None, data=None, files=None, headers=None, valid=None, allow_redirects=True):
valid = valid or self.VALID_STATUS_CODES
assert isinstance(data, (
str, dict)), \
if not headers and not files:
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
- response = self.post_url(url, params, data, files, headers)
+ response = self.post_url(url, params, data, files, headers, allow_redirects)
if response.status_code not in valid:
raise JenkinsAPIException('Operation failed. url={0}, data={1}, headers={2}, status={3}, text={4}'.format(
response.url, data, headers, response.status_code, response.text.encode('UTF-8')))
time.sleep(3)
bn = qq.block_until_building(delay=3).get_number()
self.assertIsInstance(bn, int)
-
+
b = qq.get_build()
self.assertIsInstance(b, Build)
self.assertTrue(b.is_running())
+ b.stop()
# if we call next line right away - Jenkins have no time to stop job
# so we wait a bit
time.sleep(1)
- self.assertFalse(b.poll().is_running())
+ self.assertFalse(b.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 = 'Ccreate_%s' % random_string()
job = self.jenkins.create_job(job_name, SHORTISH_JOB)
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 = 'Dcreate_%s' % random_string()
-
+
job = self.jenkins.create_job(job_name, SHORTISH_JOB)
-
+
for _ in range(3):
ii = job.invoke()
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 = 'Ecreate_%s' % random_string()
-
+
job = self.jenkins.create_job(job_name, EMPTY_JOB)
-
+
for invocation in range(3):
qq = job.invoke()
qq.block_until_complete(delay=1)
job_name = 'Ecreate_%s' % random_string()
job = self.jenkins.create_job(job_name, EMPTY_JOB)
with self.assertRaises(BadParams):
- job.invoke(build_params={'foo':'bar', 'baz':99})
+ job.invoke(build_params={'foo': 'bar', 'baz': 99})
if __name__ == '__main__':
job_name = 'create1_%s' % random_string()
job = self.jenkins.create_job(job_name, JOB_WITH_FILE)
- item = job.invoke(block=True, files={'file.txt': param_file})
- build = job.poll().get_last_build()
- while build.is_running():
- time.sleep(0.25)
+ with self.assertRaises(ValueError) as ve:
+ item = job.invoke(block=True, files={'file.txt': param_file})
- artifacts = build.get_artifact_dict()
- self.assertIsInstance(artifacts, dict)
- art_file = artifacts['file.txt']
- self.assertTrue(art_file.get_data().strip(), file_data)
+ # Following test is disabled because file parameters do not work
+ #
+ # build = job.get_last_build()
+ # while build.is_running():
+ # time.sleep(0.25)
+
+ # artifacts = build.get_artifact_dict()
+ # self.assertIsInstance(artifacts, dict)
+ # art_file = artifacts['file.txt']
+ # self.assertTrue(art_file.get_data().strip(), file_data)
def test_invoke_job_parameterized(self):
param_B = random_string()
-
+
job_name = 'create2_%s' % random_string()
job = self.jenkins.create_job(job_name, JOB_WITH_PARAMETERS)
job.invoke(block=True, build_params={'B': param_B})
artB.get_data().strip(),
param_B,
)
-
+
self.assertIn(param_B, build.get_console())
-
+
def test_parameterized_job_build_queuing(self):
"""Accept multiple builds of parameterized jobs with unique
parameters."""
job_name = 'create_%s' % random_string()
job = self.jenkins.create_job(job_name, JOB_WITH_PARAMETERS)
-
+
for i in range(3):
param_B = random_string()
params = {'B': param_B}
job.invoke(build_params=params)
time.sleep(0.25)
-
+
self.assertTrue(job.has_queued_build(params))
-
+
while job.has_queued_build(params):
time.sleep(0.25)
-
+
build = job.get_last_build()
while build.is_running():
time.sleep(0.25)
-
+
artifacts = build.get_artifact_dict()
self.assertIsInstance(artifacts, dict)
artB = artifacts['b.txt']
self.assertTrue(artB.get_data().strip(), param_B)
-
+
self.assertIn(param_B, build.get_console())
-
+
def test_parameterized_multiple_builds_get_the_same_queue_item(self):
"""Multiple attempts to run the same parameteized
build will get the same queue item."""
job_name = 'create_%s' % random_string()
job = self.jenkins.create_job(job_name, JOB_WITH_PARAMETERS)
-
+
for i in range(3):
params = {'B': random_string()}
qq0 = job.invoke(build_params=params)
-
-
+
+
qq1 = job.invoke(build_params=params)
self.assertEqual(qq0, qq1)
-
+
def test_invoke_job_with_file_and_params(self):
file_data = random_string()
param_data = random_string()
param_file = StringIO(file_data)
-
+
job_name = 'create_%s' % random_string()
job = self.jenkins.create_job(job_name, JOB_WITH_FILE_AND_PARAMS)
- job.invoke(
- block=True,
- files={'file.txt': param_file},
- build_params={'B': param_data}
- )
- build = job.get_last_build()
- artifacts = build.get_artifact_dict()
- self.assertIsInstance(artifacts, dict)
- art_file = artifacts['file.txt']
- self.assertTrue(art_file.get_data().strip(), file_data)
- art_param = artifacts['file1.txt']
- self.assertTrue(art_param.get_data().strip(), param_data)
+ with self.assertRaises(ValueError) as ve:
+ job.invoke(
+ block=True,
+ files={'file.txt': param_file},
+ build_params={'B': param_data}
+ )
+
+ # Following test is disabled because file parameters do not work
+ #
+ # build = job.get_last_build()
+ # artifacts = build.get_artifact_dict()
+ # self.assertIsInstance(artifacts, dict)
+ # art_file = artifacts['file.txt']
+ # self.assertTrue(art_file.get_data().strip(), file_data)
+ # art_param = artifacts['file1.txt']
+ # self.assertTrue(art_param.get_data().strip(), param_data)
if __name__ == '__main__':
}
@classmethod
- def mockGetData(self, url):
+ def mockGetData(self, url, tree):
try:
return TestQueue.URL_DATA[url]
except KeyError:
projects / tools / python-jenkinsapi.git / commitdiff