Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@retry(when=retry.when.status_5xx(), backoff=backoff_default)
@get("repos/{user}/{repo}/issues")
def get_issues(self, user, repo):
pass
def test_basic_client_exception(self, request_builder):
exc = self._get_exception(retry.BASE_CLIENT_EXCEPTION, request_builder)
assert request_builder.client.exceptions.BaseClientException == exc
@retry(max_attempts=3, on_exception=retry.CONNECTION_TIMEOUT)
@get("repos/{user}/{repo}/project/{project}")
def get_project(self, user, repo, project):
pass
import pytest_twisted
# Local imports.
from uplink import get, Consumer, retry
from uplink.clients import io
from tests import requires_python34
# Constants
BASE_URL = "https://api.github.com/"
def backoff_once():
yield 0.1
backoff_default = retry.backoff.exponential(multiplier=0.1, minimum=0.1)
class GitHub(Consumer):
@retry(max_attempts=2, backoff=backoff_default)
@get("/users/{user}")
def get_user(self, user):
pass
@retry(max_attempts=3, backoff=backoff_once)
@get("repos/{user}/{repo}/issues/{issue}")
def get_issue(self, user, repo, issue):
pass
@retry(max_attempts=3, on_exception=retry.CONNECTION_TIMEOUT)
@get("repos/{user}/{repo}/project/{project}")
def get_project(self, user, repo, project):
def test_retry_backoff():
def custom_backoff(*_):
return True
decorator = retry(backoff=custom_backoff)
assert decorator._backoff == custom_backoff
def test_connection_error(self, request_builder):
exc = self._get_exception(retry.CONNECTION_ERROR, request_builder)
assert request_builder.client.exceptions.BaseClientException == exc
@retry(max_attempts=3, backoff=backoff_once)
@get("repos/{user}/{repo}/issues/{issue}")
def get_issue(self, user, repo, issue):
pass
def test_ssl_error(self, request_builder):
exc = self._get_exception(retry.SSL_ERROR, request_builder)
assert request_builder.client.exceptions.BaseClientException == exc
def test_connection_timeout(self, request_builder):
exc = self._get_exception(retry.CONNECTION_TIMEOUT, request_builder)
assert request_builder.client.exceptions.BaseClientException == exc
def test_server_timeout(self, request_builder):
exc = self._get_exception(retry.SERVER_TIMEOUT, request_builder)
assert request_builder.client.exceptions.BaseClientException == exc