builds is broken.
--- /dev/null
+import logging
+logging.basicConfig()
+
+from jenkinsapi.jenkins import Jenkins
+
+j = Jenkins('http://localhost:8080')
+
+for node_id, _ in j.get_nodes().iteritems():
+ if not node_id == 'master':
+ print(node_id)
+ j.delete_node(node_id)
pass
-class WillNotBuild(JenkinsAPIException):
- """
- Cannot trigger a new build.
- """
- pass
-
-
class NoResults(JenkinsAPIException):
"""
A build did not publish any results.
import xml.etree.ElementTree as ET
from collections import defaultdict
-from time import sleep
from jenkinsapi.build import Build
from jenkinsapi.queue import QueueItem
from jenkinsapi.jenkinsbase import JenkinsBase
-from jenkinsapi.queue import QueueItem
from jenkinsapi.mutable_jenkins_thing import MutableJenkinsThing
from jenkinsapi.custom_exceptions import (
NoBuildData,
NotFound,
NotInQueue,
NotSupportSCM,
- WillNotBuild,
UnknownQueueItem,
)
file_params)
return json.dumps(to_json_structure)
- def invoke(self, securitytoken=None, block=False, build_params=None, cause=None, files=None):
+ def invoke(self, securitytoken=None, block=False, build_params=None, cause=None, files=None, delay=5):
assert isinstance(block, bool)
# Either copy the params dict or make a new one.
files=files,
)
- queue_url = response.headers['location']
- qi = QueueItem(queue_url, self.jenkins)
-
- if block:
- qi.block_until_complete(delay=10)
- return qi
+ redirect_url = response.headers['location']
+ if redirect_url.startswith("%s/queue/item" % self.jenkins.baseurl):
+ qi = QueueItem(redirect_url, self.jenkins)
+ if block:
+ qi.block_until_complete(delay=delay)
+ return qi
+ raise ValueError("Not a Queue URL: %s" % redirect_url)
def _buildid_for_type(self, buildtype):
"""Gets a buildid for a given type of build"""
nodeurl = '%s/(%s)' % (self.baseurl, nodename)
else:
nodeurl = '%s/%s' % (self.baseurl, nodename)
- try:
- yield item['displayName'], Node(nodeurl, nodename, self.jenkins)
- except Exception:
- import ipdb
- ipdb.set_trace()
+ yield item['displayName'], Node(nodeurl, nodename, self.jenkins)
def __getitem__(self, nodename):
self_as_dict = dict(self.iteritems())
JenkinsBase.__init__(self, baseurl)
@property
- def id(self):
+ def queue_id(self):
return self._data['id']
+ @property
+ def name(self):
+ return self._data['task']['name']
+
def get_jenkins_obj(self):
return self.jenkins
self.__class__.__name__, str(self))
def __str__(self):
- return "%s Queue #%i" % (self._data['name'], self._data['id'])
+ return "%s Queue #%i" % (self.name, self.queue_id)
def get_build(self):
build_number = self.get_build_number()
class TestInvocation(BaseSystemTest):
def test_invocation_object(self):
- job_name = 'create_%s' % random_string()
+ job_name = 'Acreate_%s' % random_string()
job = self.jenkins.create_job(job_name, SHORTISH_JOB)
qq = job.invoke()
self.assertIsInstance(qq, QueueItem)
self.assertEquals(qq.get_build_number(), 1)
def test_get_block_until_build_running(self):
- job_name = 'create_%s' % random_string()
+ job_name = 'Bcreate_%s' % random_string()
job = self.jenkins.create_job(job_name, LONG_RUNNING_JOB)
qq = job.invoke()
time.sleep(3)
self.assertIn('Started by user', console)
def test_get_block_until_build_complete(self):
- job_name = 'create_%s' % random_string()
+ 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 = 'create_%s' % random_string()
+ job_name = 'Dcreate_%s' % random_string()
job = self.jenkins.create_job(job_name, SHORTISH_JOB)
self.assertIsInstance(build, Build)
def test_multiple_invocations_and_get_build_number(self):
- job_name = 'create_%s' % random_string()
+ job_name = 'Ecreate_%s' % random_string()
job = self.jenkins.create_job(job_name, EMPTY_JOB)
from jenkinsapi_tests.systests.job_configs import JOB_WITH_FILE
from jenkinsapi_tests.systests.job_configs import JOB_WITH_FILE_AND_PARAMS
from jenkinsapi_tests.systests.job_configs import JOB_WITH_PARAMETERS
-from jenkinsapi.custom_exceptions import WillNotBuild
class TestParameterizedBuilds(BaseSystemTest):
file_data = random_string()
param_file = StringIO(file_data)
- job_name = 'create_%s' % random_string()
+ job_name = 'create1_%s' % random_string()
job = self.jenkins.create_job(job_name, JOB_WITH_FILE)
- job.invoke(block=True, files={'file.txt': param_file})
+ item = job.invoke(block=True, files={'file.txt': param_file})
- build = job.get_last_build()
+ build = job.poll().get_last_build()
while build.is_running():
time.sleep(0.25)
# def test_invoke_job_parameterized(self):
# param_B = random_string()
-#
-# job_name = 'create_%s' % 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})
-#
+#
# 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_job_build_queuing(self):