Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
requester = self.get_jenkins_obj().requester
if tree:
if not params:
params = {'tree': tree}
else:
params.update({'tree': tree})
response = requester.get(url, params)
if response.status_code != 200:
if response.status_code == 404 and url.endswith(self.JENKINS_API):
# if url is python api, and response status code is 404,
# it means current jenkins server does not support this api.
raise exception.UnsupportedAPI(url)
elif response.status_code == 401:
# unauthorized error, username or password is invalid
raise exception.UnauthorizedError(
response.url, "GET", 401,
msg="username or password is invalid")
logging.error("fail to request at %s with params: %s %s",
url, params, tree if tree else '')
raise exception.InvalidRequestStatus(response.url, "GET", response.status_code)
# try to parse original api info
try:
return ast.literal_eval(response.text)
except Exception:
pass
# replace invalid chars
api_info = response.text
for char in INVALID_CHARS:
if char in api_info:
def get(self, url, params=None, headers=None, allow_redirects=True):
"""request url in GET method.
Returns:
Response object.
"""
requests_kwargs = self.__build_request(params=params, headers=headers,
allow_redirects=allow_redirects)
try:
return requests.get(url, **requests_kwargs)
except requests.RequestException as err:
raise exception.RequestError(url, "GET", err=err)
def _poll(self, tree=None):
"""poll out api info.
If there is no promotion, api will return 404, catch it and build empty data.
"""
url = self.python_api_url(self.url)
params = {} if self.depth is None else {'depth': self.depth}
try:
return self.get_data(url, params=params, tree=tree)
except exception.UnsupportedAPI:
return {}
UnknownPromotion = _exception.UnknownPromotion
UnknownQueueItem = _exception.UnknownQueueItem
NotBuiltYet = _exception.NotBuiltYet
NoBuildData = _exception.NoBuildData
DeletedBuild = _exception.DeletedBuild
NoArtifacts = _exception.NoArtifacts
JenkinsAPIError = _exception.JenkinsAPIError
UnsupportedAPI = _exception.UnsupportedAPI
NotStopYet = _exception.NotStopYet
ImappropriateMethod = _exception.ImappropriateMethod
ImappropriateMethodInStaticMode = _exception.ImappropriateMethodInStaticMode
NotImplementedMethod = _exception.NotImplementedMethod
OSIOError = _exception.OSIOError
RequestError = _exception.RequestError
PostRequired = _exception.PostRequired
InvalidRequestStatus = _exception.InvalidRequestStatus
UnauthorizedError = _exception.UnauthorizedError
NetworkError = _exception.NetworkError
FtpError = _exception.FtpError
def get_jenkins_obj_by_url(url, username=None, password=None):
"""get Jenkins object by url.
Args:
url: jenkins url.
username: username to login jenkins.
password: password or API token of username.
Returns:
Jenkins object.
"""
Args:
delay: check status every `delay` seconds, default is 15s.
timeout: wait `timeout` seconds at most, default is forever.
Returns:
True if stopped, False if still running.
"""
if self.is_static:
raise exception.ImappropriateMethodInStaticMode("block_until_complete")
stime = time.time()
while self.is_running:
waited = time.time() - stime
logging.info("waited %is for %s to complete" % (waited, str(self)))
if timeout is not None and timeout < waited:
raise exception.RunTimeout("wait %is > %is, timeout!" % (waited, timeout))
time.sleep(delay)
# update info
self.poll()
return True
def __getitem__(self, job_name):
"""get job by name."""
for row in self.jenkins._data.get('jobs', []):
if row['name'] == job_name:
return self.jenkins.Job(row['url'], row['name'], self.jenkins)
raise exception.UnknownJob(job_name)
Returns:
new job object.
"""
if job_name in self:
return self[job_name]
params = {'name': job_name}
self.jenkins.requester.post_xml_and_confirm_status(
self.jenkins.get_create_url(),
data=config,
params=params)
self.jenkins.poll()
if job_name not in self:
raise exception.JenkinsAPIError('Cannot create job %s' % job_name)
return self[job_name]