self.depth = depth
JenkinsBase.__init__(self, url)
- def _poll(self):
+ 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)
+ return self.get_data(url, tree=tree)
def __str__(self):
return self._data['fullDisplayName']
url = url[:-1]
return url
- def poll(self):
- self._data = self._poll()
+ 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
- def _poll(self):
+ def _poll(self, tree=None):
url = self.python_api_url(self.baseurl)
- return self.get_data(url)
+ return self.get_data(url, tree=tree)
- def get_data(self, url, params=None):
+ def get_data(self, url, params=None, tree=None):
requester = self.get_jenkins_obj().requester
+ if tree:
+ if not params:
+ params = {'tree': tree}
+ else:
+ params.update({'tree', tree})
+
response = requester.get_url(url, params)
if response.status_code != 200:
- logging.error('Failed request at %s with params: %s', url, params)
+ logging.error('Failed request at %s with params: %s %s',
+ url, params, tree if tree else '')
response.raise_for_status()
try:
return ast.literal_eval(response.text)
branches.append(hg_default_branch)
return branches
- def _poll(self):
- data = JenkinsBase._poll(self)
+ 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 get_jenkins_obj(self):
return self.jenkins_obj
- def _poll(self):
- return self.get_data(self.baseurl)
+ def _poll(self, tree=None):
+ return self.get_data(self.baseurl, tree=tree)
def keys(self):
return self.get_plugins_dict().keys()
@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')
+ get_data_mock.assert_called_with('http://halob:8080/job/foo/98/api/python?depth=0', tree=None)
def test_get_revision_no_scm(self):
""" with no scm, get_revision should return None """
]
# Mock function
- def second_call_poll():
+ def second_call_poll(tree=None):
return TestJenkins.create_job_returns.pop(0)
# Patch Jenkins with mock function
(JOB3_API_URL, str({'tree': 'allBuilds[number,url]'})): JOB3_ALL_BUILDS_DATA,
}
- def fakeGetData(self, url, params=None):
+ def fakeGetData(self, url, params=None, tree=None):
TestJobGetAllBuilds.__get_data_call_count += 1
if params is None:
try:
URL_DATA = {'http://halob:8080/job/foo/%s' % config.JENKINS_API: JOB_DATA}
- def fakeGetData(self, url, *args):
+ def fakeGetData(self, url, *args, **kwargs):
try:
return TestHgJob.URL_DATA[url]
except KeyError:
Assert that attempting to create a view which
already exists simply returns the same view.
"""
- def mockGetData(JJ, url):
+ def mockGetData(JJ, url, tree=None):
DATA = {}
DATA['http://localhost:8080/%s' % config.JENKINS_API] = \
{'views': [dict(name='NewView', url='http://xxxxx/yyyy')]}