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_too_many_requests_retry_header_combination_one(self):
# If the first response didn't have a header, second one has, and third one doesn't have, what should happen?
with patch('time.sleep') as mock_time:
fcn = MagicMock()
fcn.side_effect = [
TooManyRequests(retry_after_seconds=2),
TooManyRequests(),
TooManyRequests(retry_after_seconds=2),
self.response,
]
self.assertIs(self.response, _translate_and_retry(fcn, 4))
self.assertEqual([call(2), call(1.5), call(2)], mock_time.mock_calls)
def test_too_many_requests_failed_after_sleep(self):
with patch('time.sleep') as mock_time:
fcn = MagicMock()
fcn.side_effect = [
TooManyRequests(retry_after_seconds=2),
TooManyRequests(retry_after_seconds=5),
]
with self.assertRaises(TooManyRequests):
_translate_and_retry(fcn, 2)
self.assertEqual([call(2)], mock_time.mock_calls)
def test_too_many_requests_retry_header_combination_two(self):
# If the first response had header, and the second did not, but the third has header again, what should happen?
with patch('time.sleep') as mock_time:
fcn = MagicMock()
fcn.side_effect = [
TooManyRequests(),
TooManyRequests(retry_after_seconds=5),
TooManyRequests(),
self.response,
]
self.assertIs(self.response, _translate_and_retry(fcn, 4))
self.assertEqual([call(1.0), call(5), call(2.25)], mock_time.mock_calls)
def test_too_many_requests(self):
response = MagicMock()
response.status_code = 429
response.headers = {'retry-after': 1}
response.content = six.b(
'{"status": 429, "code": "Too Many requests", "message": "retry after some time"}'
)
with self.assertRaises(TooManyRequests):
_translate_errors(lambda: response)
def test_too_many_requests_retry_header_combination_two(self):
# If the first response had header, and the second did not, but the third has header again, what should happen?
with patch('time.sleep') as mock_time:
fcn = MagicMock()
fcn.side_effect = [
TooManyRequests(),
TooManyRequests(retry_after_seconds=5),
TooManyRequests(),
self.response,
]
self.assertIs(self.response, _translate_and_retry(fcn, 4))
self.assertEqual([call(1.0), call(5), call(2.25)], mock_time.mock_calls)
def test_too_many_requests_works_after_sleep(self):
with patch('time.sleep') as mock_time:
fcn = MagicMock()
fcn.side_effect = [TooManyRequests(retry_after_seconds=2), self.response]
self.assertIs(self.response, _translate_and_retry(fcn, 3))
self.assertEqual([call(2)], mock_time.mock_calls)
def test_too_many_requests_retry_header_combination_one(self):
# If the first response didn't have a header, second one has, and third one doesn't have, what should happen?
with patch('time.sleep') as mock_time:
fcn = MagicMock()
fcn.side_effect = [
TooManyRequests(retry_after_seconds=2),
TooManyRequests(),
TooManyRequests(retry_after_seconds=2),
self.response,
]
self.assertIs(self.response, _translate_and_retry(fcn, 4))
self.assertEqual([call(2), call(1.5), call(2)], mock_time.mock_calls)
def test_too_many_requests_failed_after_sleep(self):
with patch('time.sleep') as mock_time:
fcn = MagicMock()
fcn.side_effect = [
TooManyRequests(retry_after_seconds=2),
TooManyRequests(retry_after_seconds=5),
]
with self.assertRaises(TooManyRequests):
_translate_and_retry(fcn, 2)
self.assertEqual([call(2)], mock_time.mock_calls)
return UploadTokenUsedConcurrently(token)
return BadRequest(message, code)
elif status == 400:
return BadRequest(message, code)
elif status == 401 and code in ("bad_auth_token", "expired_auth_token"):
return InvalidAuthToken(message, code)
elif status == 401:
return Unauthorized(message, code)
elif status == 403 and code == "storage_cap_exceeded":
return StorageCapExceeded()
elif status == 409:
return Conflict()
elif status == 416 and code == "range_not_satisfiable":
return UnsatisfiableRange()
elif status == 429:
return TooManyRequests(retry_after_seconds=response_headers.get('retry-after'))
elif 500 <= status < 600:
return ServiceError('%d %s %s' % (status, code, message))
return UnknownError('%d %s %s' % (status, code, message))
def __init__(self, retry_after_seconds=None):
super(TooManyRequests, self).__init__()
self.retry_after_seconds = retry_after_seconds