"""
Represents a jenkins environment.
"""
- def __init__(self, baseurl, username=None, password=None, requester=None):
+ def __init__(self, baseurl, username=None, password=None, requester=None, lazy=False):
"""
:param baseurl: baseurl for jenkins instance including port, str
:param username: username for jenkins auth, str
self.username = username
self.password = password
self.requester = requester or Requester(username, password, baseurl=baseurl)
- JenkinsBase.__init__(self, baseurl)
+ self.lazy = lazy
+ JenkinsBase.__init__(self, baseurl, poll=not lazy)
+
+ def _poll_if_needed(self):
+ if self.lazy and self._data is None:
+ self.poll()
def _clone(self):
return Jenkins(self.baseurl, username=self.username,
"""
Fetch all the build-names on this Jenkins server.
"""
+ self._poll_if_needed()
for info in self._data["jobs"]:
yield info["name"], \
Job(info["url"], info["name"], jenkins_obj=self)
Get the jobs information
:return url, name
"""
+ self._poll_if_needed()
for info in self._data["jobs"]:
yield info["url"], info["name"]
return self.jobs.rename(jobname, newjobname)
def iterkeys(self):
+ self._poll_if_needed()
for info in self._data["jobs"]:
yield info["name"]
:param jobname: name of job, str
:return: Job obj
"""
+ self._poll_if_needed()
+
for info in self._data["jobs"]:
if info["name"] == jobname:
return Job(info["url"], info["name"], jenkins_obj=self)
raise UnknownJob(jobname)
def __len__(self):
+ self._poll_if_needed()
return len(self._data["jobs"])
def __contains__(self, jobname):
self.assertEquals(job.name, _poll.return_value['jobs'][idx]['name'])
self.assertEquals(job.baseurl, _poll.return_value['jobs'][idx]['url'])
+ @mock.patch.object(JenkinsBase, '_poll')
+ @mock.patch.object(Jenkins, '_poll')
+ @mock.patch.object(Job, '_poll')
+ def test_lazy_loading(self, _base_poll, _poll, _job_poll):
+ _poll.return_value = {
+ 'jobs': [
+ {'name': 'job_one', 'url': 'http://localhost:8080/job_one', 'color': 'blue'},
+ {'name': 'job_two', 'url': 'http://localhost:8080/job_two', 'color': 'blue'},
+ ]
+ }
+ _base_poll.return_value = _poll.return_value
+ _job_poll.return_value = {}
+ J = Jenkins('http://localhost:8080/',
+ username='foouser', password='foopassword', lazy=True)
+
+ self.assertEquals(J._data, None)
+
+ for idx, (job_name, job) in enumerate(J.get_jobs()):
+ self.assertEquals(job_name, _poll.return_value['jobs'][idx]['name'])
+ self.assertTrue(isinstance(job, Job))
+ self.assertEquals(job.name, _poll.return_value['jobs'][idx]['name'])
+ self.assertEquals(job.baseurl, _poll.return_value['jobs'][idx]['url'])
+
@mock.patch.object(JenkinsBase, '_poll')
@mock.patch.object(Jenkins, '_poll')
@mock.patch.object(Job, '_poll')