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_modifying_retry_status_codes(self):
ApiConfig.retry_status_codes = [1, 2, 3]
retries = Connection.get_session().get_adapter(ApiConfig.api_protocol).max_retries
self.assertEqual(retries.status_forcelist, ApiConfig.retry_status_codes)
def test_modifying_number_of_retries(self):
ApiConfig.number_of_retries = 3000
retries = Connection.get_session().get_adapter(ApiConfig.api_protocol).max_retries
self.assertEqual(retries.total, ApiConfig.number_of_retries)
self.assertEqual(retries.connect, ApiConfig.number_of_retries)
self.assertEqual(retries.read, ApiConfig.number_of_retries)
def setUp(self):
self.default_use_retries = ApiConfig.use_retries
self.default_number_of_retries = ApiConfig.number_of_retries
self.default_retry_backoff_factor = ApiConfig.retry_backoff_factor
self.default_max_wait_between_retries = ApiConfig.max_wait_between_retries
self.default_retry_status_codes = ApiConfig.retry_status_codes
def test_bulk_download_table_raises_exception_when_error_response(self, request_method):
if request_method == 'POST':
RequestType.USE_GET_REQUEST = False
httpretty.reset()
ApiConfig.number_of_retries = 2
error_responses = [httpretty.Response(
body=json.dumps({'quandl_error': {'code': 'QEMx01',
'message': 'something went wrong'}}),
status=500)]
httpretty.register_uri(getattr(httpretty, request_method),
re.compile(
'https://www.quandl.com/api/v3/datatables/*'),
responses=error_responses)
self.assertRaises(
InternalServerError, lambda: self.datatable.download_file('.', params={}))
def test_modifying_use_retries(self):
ApiConfig.use_retries = False
retries = Connection.get_session().get_adapter(ApiConfig.api_protocol).max_retries
self.assertEqual(retries.total, 0)
def test_quandl_exceptions_no_retries(self, request_method):
ApiConfig.use_retries = False
quandl_errors = [('QELx04', 429, LimitExceededError),
('QEMx01', 500, InternalServerError),
('QEAx01', 400, AuthenticationError),
('QEPx02', 403, ForbiddenError),
('QESx03', 422, InvalidRequestError),
('QECx05', 404, NotFoundError),
('QEXx01', 503, ServiceUnavailableError),
('QEZx02', 400, QuandlError)]
httpretty.register_uri(getattr(httpretty, request_method),
"https://www.quandl.com/api/v3/databases",
responses=[httpretty.Response(body=json.dumps(
{'quandl_error':
{'code': x[0], 'message': 'something went wrong'}}),
status=x[1]) for x in quandl_errors]
)
def test_modifying_retry_backoff_factor(self):
ApiConfig.retry_backoff_factor = 3000
retries = Connection.get_session().get_adapter(ApiConfig.api_protocol).max_retries
self.assertEqual(retries.backoff_factor, ApiConfig.retry_backoff_factor)
def bulk_download_url(self, **options):
url = self._bulk_download_path()
url = ApiConfig.api_base + '/' + url
if 'params' not in options:
options['params'] = {}
if ApiConfig.api_key:
options['params']['api_key'] = ApiConfig.api_key
if ApiConfig.api_version:
options['params']['api_version'] = ApiConfig.api_version
if list(options.keys()):
url += '?' + urlencode(options['params'])
return url