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_request_json_error(self, mock_post, mock_request_headers, mock_cert, mock_create_req):
mock_response = mock.Mock()
mock_response.status_code = 200
mock_response.json.side_effect = ValueError()
mock_post.return_value = mock_response
with self.assertRaises(InvalidResponse):
self.base_endpoint.request(None, None, None)
def test_request_json_error(self, mock_get, mock_request_headers, mock_cert):
mock_response = mock.Mock()
mock_response.status_code = 200
mock_response.json.side_effect = ValueError()
mock_get.return_value = mock_response
with self.assertRaises(InvalidResponse):
self.navigation.request()
def test_request_json_error(self, mock_post, mock_login_headers, mock_cert):
mock_response = mock.Mock()
mock_response.status_code = 200
mock_response.json.side_effect = ValueError()
mock_post.return_value = mock_response
with self.assertRaises(InvalidResponse):
self.login.request()
def test_request_error(
self,
mock_get,
mock_login_headers,
mock_create_req,
mock_check_status_code,
mock_json_loads,
):
response = mock.Mock()
mock_get.return_value = response
with self.assertRaises(InvalidResponse):
self.race_card.request()
def test_request_json_error(self, mock_post, mock_keep_alive_headers):
mock_response = mock.Mock()
mock_response.status_code = 200
mock_response.json.side_effect = ValueError()
mock_post.return_value = mock_response
with self.assertRaises(InvalidResponse):
self.keep_alive.request()
session = session or self.client.session
date_time_sent = datetime.datetime.utcnow()
url = "%s%s" % (self.url, method)
try:
response = session.get(url, params=params, headers=self.headers)
except requests.ConnectionError as e:
raise APIError(None, method, params, e)
except Exception as e:
raise APIError(None, method, params, e)
elapsed_time = (datetime.datetime.utcnow() - date_time_sent).total_seconds()
check_status_code(response)
try:
response_data = json_loads(response.text)
except ValueError:
raise InvalidResponse(response.text)
return response_data, elapsed_time
def request(self, method=None, params=None, session=None):
session = session or self.client.session
date_time_sent = datetime.datetime.utcnow()
try:
response = session.post(self.url, data=self.data, headers=self.client.login_headers, cert=self.client.cert)
except ConnectionError:
raise APIError(None, exception='ConnectionError')
except Exception as e:
raise APIError(None, exception=e)
elapsed_time = (datetime.datetime.utcnow() - date_time_sent).total_seconds()
check_status_code(response)
try:
response_data = response.json()
except ValueError:
raise InvalidResponse(response.text)
if self._error_handler:
self._error_handler(response_data)
return response_data, elapsed_time
def request(self, payload=None, params=None, session=None):
session = session or self.client.session
date_time_sent = datetime.datetime.utcnow()
try:
response = session.post(self.url, headers=self.client.keep_alive_headers)
except ConnectionError:
raise APIError(None, exception='ConnectionError')
except Exception as e:
raise APIError(None, exception=e)
elapsed_time = (datetime.datetime.utcnow() - date_time_sent).total_seconds()
check_status_code(response)
try:
response_data = response.json()
except ValueError:
raise InvalidResponse(response.text)
if self._error_handler:
self._error_handler(response_data)
return response_data, elapsed_time
'%s%s' % (self.url, method),
data=json.dumps(params),
headers=self.headers,
timeout=(self.connect_timeout, self.read_timeout)
)
except ConnectionError:
raise APIError(None, method, params, 'ConnectionError')
except Exception as e:
raise APIError(None, method, params, e)
elapsed_time = (datetime.datetime.utcnow()-date_time_sent).total_seconds()
check_status_code(response)
try:
response_data = response.json()
except ValueError:
raise InvalidResponse(response.text)
return response_data, elapsed_time