def _poll(self, tree=None):
# For build's we need more information for downstream and upstream builds
# so we override the poll to get at the extra data for build objects
- url = '%s?depth=%s' % (self.python_api_url(self.baseurl), self.depth)
- return self.get_data(url, tree=tree)
+ url = self.python_api_url(self.baseurl)
+ return self.get_data(url, params={'depth': self.depth}, tree=tree)
def __str__(self):
return self._data['fullDisplayName']
return datetime.timedelta(milliseconds=self._data["duration"])
def get_artifacts(self):
- for afinfo in self._data["artifacts"]:
+ data = self.poll(tree='artifacts[relativePath,fileName]')
+ for afinfo in data["artifacts"]:
url = "%s/artifact/%s" % (self.baseurl, afinfo["relativePath"])
af = Artifact(afinfo["fileName"], url, self)
yield af
"""
Return a bool if running.
"""
- self.poll()
- return self._data["building"]
+ data = self.poll(tree='building')
+ return data.get('building', False)
def block(self):
while self.is_running():
return url
def poll(self, tree=None):
- self._data = self._poll(tree=tree)
- if 'jobs' in self._data:
- self._data['jobs'] = self.resolve_job_folders(self._data['jobs'])
- return self
+ data = self._poll(tree=tree)
+ if 'jobs' in data:
+ data['jobs'] = self.resolve_job_folders(data['jobs'])
+ if not tree:
+ self._data = data
+ else:
+ return data
def _poll(self, tree=None):
url = self.python_api_url(self.baseurl)
if not params:
params = {'tree': tree}
else:
- params.update({'tree', tree})
+ params.update({'tree': tree})
response = requester.get_url(url, params)
if response.status_code != 200:
JenkinsBase.__init__(self, url)
def __str__(self):
- return self._data["name"]
+ return self.name
def get_description(self):
return self._data["description"]
branches.append(hg_default_branch)
return branches
- def _poll(self, tree=None):
- data = JenkinsBase._poll(self, tree=tree)
- # jenkins loads only the first 100 builds, load more if needed
- data = self._add_missing_builds(data)
- return data
+ def poll(self, tree=None):
+ data = super(Job, self).poll(tree=tree)
+ if not tree:
+ self._data = self._add_missing_builds(self._data)
+ else:
+ return data
# pylint: disable=E1123
# Unexpected keyword arg 'params'
# 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']:
+ if not self._data['firstBuild']:
first_build_number = oldest_loaded_build_number
else:
- first_build_number = data["firstBuild"]["number"]
+ first_build_number = self._data["firstBuild"]["number"]
all_builds_loaded = (oldest_loaded_build_number == first_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.poll(tree='allBuilds[number,url]')
data['builds'] = response['allBuilds']
return data
def _buildid_for_type(self, buildtype):
"""Gets a buildid for a given type of build"""
- self.poll()
KNOWNBUILDTYPES = [
"lastStableBuild",
"lastSuccessfulBuild",
"lastFailedBuild"]
assert buildtype in KNOWNBUILDTYPES, 'Unknown build info type: %s' % buildtype
- if not self._data.get(buildtype):
+ data = self.poll(tree='%s[number]' % buildtype)
+
+ if not data.get(buildtype):
raise NoBuildData(buildtype)
- return self._data[buildtype]["number"]
+ return data[buildtype]["number"]
def get_first_buildnumber(self):
"""
return self._buildid_for_type("lastCompletedBuild")
def get_build_dict(self):
- if "builds" not in self._data:
+ builds = self.poll(tree='builds[number,url]')
+ if not builds:
raise NoBuildData(repr(self))
- builds = self._data["builds"]
- last_build = self._data['lastBuild']
+ builds = self._add_missing_builds(builds)
+ builds = builds['builds']
+ last_build = self.poll(tree='lastBuild[number,url]')['lastBuild']
if builds and last_build and builds[0]['number'] != last_build['number']:
builds = [last_build] + builds
# FIXME SO how is this supposed to work if build is false-y?
return self.is_queued() or self.is_running()
def is_queued(self):
- self.poll()
- return self._data["inQueue"]
+ data = self.poll(tree='inQueue')
+ return data.get('inQueue', False)
def get_queue_item(self):
"""
return QueueItem(self.jenkins, **self._data['queueItem'])
def is_running(self):
- self.poll()
+ # self.poll()
try:
build = self.get_last_build_or_none()
if build is not None:
return upstream_jobs
def is_enabled(self):
- self.poll()
- return self._data["color"] != 'disabled'
+ data = self.poll(tree='color')
+ return data.get('color', None) != 'disabled'
def disable(self):
'''Disable job'''
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())
console = b.get_console()
@mock.patch.object(Build, 'get_data')
def test_build_depth(self, get_data_mock):
- build = Build('http://halob:8080/job/foo/98', 98, self.j, depth=0)
- get_data_mock.assert_called_with('http://halob:8080/job/foo/98/api/python?depth=0', tree=None)
+ Build('http://halob:8080/job/foo/98', 98, self.j, depth=0)
+ get_data_mock.assert_called_with('http://halob:8080/job/foo/98/api/'
+ 'python',
+ tree=None, params={'depth': 0})
def test_get_revision_no_scm(self):
""" with no scm, get_revision should return None """
# expected = ['SingleJob','MultipleJobs']
# self.assertEquals(self.b.get_downstream_job_names(), expected)
+
def main():
unittest.main(verbosity=2)
def second_call_poll(tree=None):
return TestJenkins.create_job_returns.pop(0)
+ def job_second_call_poll(tree=None):
+ return {}
+
# Patch Jenkins with mock function
@mock.patch.object(Jenkins, '_poll', side_effect=second_call_poll)
- @mock.patch.object(Job, '_poll')
+ @mock.patch.object(Job, '_poll', side_effect=job_second_call_poll)
def test_create_new_job(self, _poll, _job_poll):
_job_poll.return_value = {}
except KeyError:
raise Exception("Missing data for %s" % url)
+ def fakeGetDataTree(self, url, **args):
+ try:
+ if 'builds' in args['tree']:
+ return {'builds': TestJob.URL_DATA[url]['builds']}
+ else:
+ return {'lastBuild': TestJob.URL_DATA[url]['lastBuild']}
+ except KeyError:
+ raise Exception("Missing data for %s" % url)
+
+ def fake_get_data_tree_empty(self, url, **args):
+ return {}
+
@mock.patch.object(JenkinsBase, 'get_data', fakeGetData)
def setUp(self):
ret = self.j.get_last_completed_buildnumber()
self.assertEquals(ret, 3)
+ @mock.patch.object(JenkinsBase, 'get_data', fakeGetDataTree)
def test_get_build_dict(self):
ret = self.j.get_build_dict()
self.assertTrue(isinstance(ret, dict))
self.assertEquals(len(ret), 4)
- @mock.patch.object(Job, '_poll')
- def test_nobuilds_get_build_dict(self, _poll):
- # Bare minimum build dict, we only testing dissapearance of 'builds'
- _poll.return_value = {"name": "foo"}
-
+ @mock.patch.object(JenkinsBase, 'get_data', fake_get_data_tree_empty)
+ def test_nobuilds_get_build_dict(self):
j = Job('http://halob:8080/job/foo/', 'foo', self.J)
with self.assertRaises(NoBuildData):
j.get_build_dict()
+ @mock.patch.object(JenkinsBase, 'get_data', fakeGetDataTree)
def test_get_build_ids(self):
# We don't want to deal with listreverseiterator here
# So we convert result to a list
self.assertEquals(len(params), 2)
self.assertEquals(params, ['param1', 'param2'])
+ @mock.patch.object(JenkinsBase, 'get_data', fakeGetDataTree)
+ # @mock.patch.object(JenkinsBase, 'get_data', fakeGetLastBuild)
def test_get_build(self):
buildnumber = 1
with mock.patch('jenkinsapi.job.Build') as build_mock:
build_mock.assert_called_with('http://halob:8080/job/foo/1/',
buildnumber, job=self.j)
+ @mock.patch.object(JenkinsBase, 'get_data', fakeGetDataTree)
def test_get_build_metadata(self):
buildnumber = 1
with mock.patch('jenkinsapi.job.Build') as build_mock:
URL_DATA = {
JOB1_API_URL: JOB1_DATA,
- (JOB1_API_URL, str({'tree': 'allBuilds[number,url]'})): JOB1_ALL_BUILDS_DATA,
+ (JOB1_API_URL, 'allBuilds[number,url]'): JOB1_ALL_BUILDS_DATA,
JOB2_API_URL: JOB2_DATA,
JOB3_API_URL: JOB3_DATA,
# this one below should never be used
- (JOB3_API_URL, str({'tree': 'allBuilds[number,url]'})): JOB3_ALL_BUILDS_DATA,
+ (JOB3_API_URL, 'allBuilds[number,url]'): JOB3_ALL_BUILDS_DATA,
}
def fakeGetData(self, url, params=None, tree=None):
except KeyError:
raise Exception("Missing data for url: %s with parameters %s" % (url, repr(params)))
- @mock.patch.object(JenkinsBase, 'get_data', fakeGetData)
+ def fakeGetDataTree(self, url, **args):
+ TestJobGetAllBuilds.__get_data_call_count += 1
+ try:
+ if args['tree']:
+ if 'builds' in args['tree']:
+ return {'builds': TestJobGetAllBuilds.URL_DATA[url]['builds']}
+ elif 'allBuilds' in args['tree']:
+ return TestJobGetAllBuilds.URL_DATA[(url, args['tree'])]
+ elif 'lastBuild' in args['tree']:
+ return {'lastBuild': TestJobGetAllBuilds.URL_DATA[url]['lastBuild']}
+ else:
+ return dict(TestJobGetAllBuilds.URL_DATA[url])
+ except KeyError:
+ raise Exception("Missing data for %s" % url)
+
+ @mock.patch.object(JenkinsBase, 'get_data', fakeGetDataTree)
def setUp(self):
TestJobGetAllBuilds.__get_data_call_count = 0
self.J = mock.MagicMock() # Jenkins object
self.j = Job('http://halob:8080/job/foo/', 'foo', self.J)
+ @mock.patch.object(JenkinsBase, 'get_data', fakeGetDataTree)
def test_get_build_dict(self):
# The job data contains only one build, so we expect that the
# remaining jobs will be fetched automatically
self.assertTrue(isinstance(ret, dict))
self.assertEquals(len(ret), 4)
- @mock.patch.object(JenkinsBase, 'get_data', fakeGetData)
+ @mock.patch.object(JenkinsBase, 'get_data', fakeGetDataTree)
def test_incomplete_builds_list_will_call_jenkins_twice(self):
# The job data contains only one build, so we expect that the
# remaining jobs will be fetched automatically, and to have two calls to
self.j = Job('http://halob:8080/job/foo/', 'foo', self.J)
self.assertEquals(TestJobGetAllBuilds.__get_data_call_count, 2)
- @mock.patch.object(JenkinsBase, 'get_data', fakeGetData)
+ @mock.patch.object(JenkinsBase, 'get_data', fakeGetDataTree)
def test_complete_builds_list_will_call_jenkins_once(self):
# The job data contains all builds, so we will not gather remaining builds
TestJobGetAllBuilds.__get_data_call_count = 0
self.j = Job('http://halob:8080/job/fullfoo/', 'fullfoo', self.J)
self.assertEquals(TestJobGetAllBuilds.__get_data_call_count, 1)
- @mock.patch.object(JenkinsBase, 'get_data', fakeGetData)
+ @mock.patch.object(JenkinsBase, 'get_data', fakeGetDataTree)
def test_nobuilds_get_build_dict(self):
j = Job('http://halob:8080/job/look_ma_no_builds/', 'look_ma_no_builds', self.J)
self.assertTrue(isinstance(ret, dict))
self.assertEquals(len(ret), 0)
+ @mock.patch.object(JenkinsBase, 'get_data', fakeGetDataTree)
def test_get_build_ids(self):
# The job data contains only one build, so we expect that the
# remaining jobs will be fetched automatically
projects / tools / python-jenkinsapi.git / commitdiff