Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_stop_job_raise_ApiClientException_when_api_return_error_code(self, mock_requests):
job_id = 'jobdfsfnqwerty1234'
self.mock_response.status_code = 500
self.mock_response.json.return_value = {'result': 'success'}
mock_requests.post.return_value = self.mock_response
self.assertRaises(exceptions.ApiClientException, self.job_manager.start_job, job_id)
def create(self, doc, action_id=''):
action_id = action_id or doc.get('action_id', '')
endpoint = self.endpoint + action_id
r = requests.post(endpoint,
data=json.dumps(doc),
headers=self.headers,
verify=self.verify)
if r.status_code != 201:
raise exceptions.ApiClientException(r)
action_id = r.json()['action_id']
return action_id
:param job_id: the id of the job to start
:return: the response obj:
{
result: string 'success' or 'already stopped'
}
"""
# endpoint /v1/jobs/{job_id}/event
endpoint = '{0}{1}/event'.format(self.endpoint, job_id)
doc = {"stop": None}
r = requests.post(endpoint,
headers=self.headers,
data=json.dumps(doc),
verify=self.verify)
if r.status_code != 202:
raise exceptions.ApiClientException(r)
return r.json()
def create(self, doc, session_id=''):
session_id = session_id or doc.get('session_id', '')
endpoint = self.endpoint + session_id
r = requests.post(endpoint,
data=json.dumps(doc),
headers=self.headers,
verify=self.verify)
if r.status_code != 201:
raise exceptions.ApiClientException(r)
session_id = r.json()['session_id']
return session_id
def get(self, session_id):
endpoint = self.endpoint + session_id
r = requests.get(endpoint, headers=self.headers, verify=self.verify)
if r.status_code == 200:
return r.json()
if r.status_code == 404:
return None
raise exceptions.ApiClientException(r)
def delete(self, action_id):
endpoint = self.endpoint + action_id
r = requests.delete(endpoint,
headers=self.headers,
verify=self.verify)
if r.status_code != 204:
raise exceptions.ApiClientException(r)
def delete(self, client_id):
endpoint = self.endpoint + client_id
r = requests.delete(endpoint, headers=self.headers,
verify=self.verify)
if r.status_code != 204:
raise exceptions.ApiClientException(r)
def get(self, backup_id):
endpoint = self.endpoint + backup_id
r = requests.get(endpoint, headers=self.headers, verify=self.verify)
if r.status_code == 200:
return r.json()
if r.status_code == 404:
return None
raise exceptions.ApiClientException(r)