Repository: prkumar/uplink
Pull request #250 (2022-01-24 19:32:59)

Release v0.9.6
Added
-------
- Add a new base class, ``uplink.retry.RetryBackoff``, which can be extended to
  implement custom backoff strategies. An instance of a ``RetryBackoff`` subclass
  can be provided through the ``backoff`` argument of the ``@retry`` decorator.
  (#238)

Changed
---------
- Bump minimum version of ``six`` to ``1.13.0``. (#246)

Fixed
------
- Fix ``@returns.json`` to cast JSON response (or field referenced by the ``key``
  argument) using the ``type`` argument when the given type is callable. This
  restores behavior that was inadvertently changed in v0.9.3. (#215)
- Remove all usages of ``asyncio.coroutine`` in the library code to fix warnings
  related to the function's deprecation in Python 3.8+. (#203)



Patch:
diff --git a/.gitignore b/.gitignore
index 0f971d7..28bb755 100644
--- a/.gitignore
+++ b/.gitignore
@@ -123,3 +123,5 @@ Pipfile.lock
 # macOS
 .DS_Store
 
+# VS Code
+.vscode/
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 28d9862..a65940d 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -6,6 +6,27 @@ All notable changes to this project will be documented in this file.
 The format is based on `Keep a Changelog`_, and this project adheres to the
 `Semantic Versioning`_ scheme.
 
+Unreleased
+==========
+Added
+-----
+- Add a new base class, ``uplink.retry.RetryBackoff``, which can be extended to
+  implement custom backoff strategies. An instance of a ``RetryBackoff`` subclass
+  can be provided through the ``backoff`` argument of the ``@retry`` decorator.
+  (`#238`_)
+
+Changed
+-------
+- Bump minimum version of ``six`` to ``1.13.0``. (`#246`_)
+
+Fixed
+-----
+- Fix ``@returns.json`` to cast the JSON response (or field referenced by the
+  ``key`` argument) using the ``type`` argument when it is a callable type.
+  This effectively reverts a change in the decorator's behavior that was
+  introduced in v0.9.3. (`#215`_)
+
+
 0.9.5_ - 2022-01-04
 ====================
 Added
@@ -405,9 +426,12 @@ Added
 .. _#204: https://github.com/prkumar/uplink/pull/204
 .. _#207: https://github.com/prkumar/uplink/pull/207
 .. _#209: https://github.com/prkumar/uplink/pull/209
+.. _#215: https://github.com/prkumar/uplink/issues/215
 .. _#217: https://github.com/prkumar/uplink/issues/217
 .. _#221: https://github.com/prkumar/uplink/issues/221
 .. _#237: https://github.com/prkumar/uplink/discussions/237
+.. _#238: https://github.com/prkumar/uplink/issues/238
+.. _#246: https://github.com/prkumar/uplink/issues/246
 
 .. Commits
 .. _3653a672ee: https://github.com/prkumar/uplink/commit/3653a672ee0703119720d0077bb450649af5459c
@@ -421,4 +445,4 @@ Added
 .. _@cognifloyd: https://github.com/cognifloyd
 .. _@gmcrocetti: https://github.com/gmcrocetti
 .. _@leiserfg: https://github.com/leiserfg
-.. _@lust4life: https://github.com/lust4life
\ No newline at end of file
+.. _@lust4life: https://github.com/lust4life
diff --git a/README.rst b/README.rst
index 442711d..ac35cb6 100644
--- a/README.rst
+++ b/README.rst
@@ -119,7 +119,7 @@ Extra! Extra!
 -------------
 
 Further, uplink has optional integrations and features. You can view a full list
-of available extras `here <http://uplink.readthedocs.io/en/latest/install.html#extras>`_.
+of available extras `here <https://uplink.readthedocs.io/en/latest/user/install.html#extras>`_.
 
 When installing Uplink with ``pip``, you can select extras using the format:
 
diff --git a/docs/source/dev/decorators.rst b/docs/source/dev/decorators.rst
index 3539017..739ff63 100644
--- a/docs/source/dev/decorators.rst
+++ b/docs/source/dev/decorators.rst
@@ -150,6 +150,23 @@ to specify one of the alternative approaches exposed through the
         def get_user(self, user):
             """Get user by username."""
 
+You can implement a custom backoff strategy by extending the class
+:class:`uplink.retry.RetryBackoff`:
+
+.. code-block:: python
+   :emphasize-lines: 3,7
+
+    from uplink.retry import RetryBackoff
+
+    class MyCustomBackoff(RetryBackoff):
+        ...
+
+    class GitHub(uplink.Consumer):
+        @uplink.retry(backoff=MyCustomBackoff())
+        @uplink.get("/users/{user}")
+        def get_user(self, user):
+            pass
+
 .. automodule:: uplink.retry.backoff
     :members:
 
diff --git a/setup.py b/setup.py
index 3121928..1b1f701 100644
--- a/setup.py
+++ b/setup.py
@@ -15,7 +15,7 @@ with open(os.path.join("uplink", "__about__.py")) as fp:
     exec(fp.read(), about)
     about = dict((k.strip("_"), about[k]) for k in about)
 
-install_requires = ["requests>=2.18.0", "six>=1.12.0", "uritemplate>=3.0.0"]
+install_requires = ["requests>=2.18.0", "six>=1.13.0", "uritemplate>=3.0.0"]
 
 extras_require = {
     "marshmallow": ["marshmallow>=2.15.0"],
@@ -29,6 +29,7 @@ extras_require = {
     "twisted:python_version == '3.4'": "twisted<=19.2.1",
     "typing": ["typing>=3.6.4"],
     "tests": ["pytest", "pytest-mock", "pytest-cov", "pytest-twisted"],
+    "tests:python_version >= '3.5'": ["pytest-asyncio"],
 }
 
 metadata = {
diff --git a/uplink/clients/aiohttp_.py b/uplink/clients/aiohttp_.py
index f260b37..cc64976 100644
--- a/uplink/clients/aiohttp_.py
+++ b/uplink/clients/aiohttp_.py
@@ -19,14 +19,11 @@ from uplink.clients import exceptions, io, interfaces, register
 
 
 def threaded_callback(callback):
-    coroutine_callback = asyncio.coroutine(callback)
-
-    @asyncio.coroutine
-    def new_callback(response):
+    async def new_callback(response):
         if isinstance(response, aiohttp.ClientResponse):
-            yield from response.text()
+            await response.text()
             response = ThreadedResponse(response)
-        response = yield from coroutine_callback(response)
+        response = callback(response)
         if isinstance(response, ThreadedResponse):
             return response.unwrap()
         else:
@@ -68,6 +65,7 @@ class AiohttpClient(interfaces.HttpClientAdapter):
         self._sync_callback_adapter = threaded_callback
 
     def __del__(self):
+        # TODO: Consider replacing this with a close method
         if self._auto_created_session:
             # aiohttp v3.0 has made ClientSession.close a coroutine,
             # so we check whether it is one here and register it
@@ -79,8 +77,7 @@ class AiohttpClient(interfaces.HttpClientAdapter):
             else:
                 self._session.close()
 
-    @asyncio.coroutine
-    def session(self):
+    async def session(self):
         """Returns the underlying `aiohttp.ClientSession`."""
         if isinstance(self._session, self.__ARG_SPEC):
             args, kwargs = self._session
@@ -128,11 +125,10 @@ class AiohttpClient(interfaces.HttpClientAdapter):
         session_build_args = cls._create_session(*args, **kwargs)
         return AiohttpClient(session=session_build_args)
 
-    @asyncio.coroutine
-    def send(self, request):
+    async def send(self, request):
         method, url, extras = request
-        session = yield from self.session()
-        response = yield from session.request(method, url, **extras)
+        session = await self.session()
+        response = await session.request(method, url, **extras)
 
         # Make `aiohttp` response "quack" like a `requests` response
         response.status_code = response.status
diff --git a/uplink/clients/io/asyncio_strategy.py b/uplink/clients/io/asyncio_strategy.py
index 5c19386..103e9e3 100644
--- a/uplink/clients/io/asyncio_strategy.py
+++ b/uplink/clients/io/asyncio_strategy.py
@@ -11,29 +11,22 @@ __all__ = ["AsyncioStrategy"]
 class AsyncioStrategy(interfaces.IOStrategy):
     """A non-blocking execution strategy using asyncio."""
 
-    @asyncio.coroutine
-    def invoke(self, func, args, kwargs, callback):
+    async def invoke(self, func, args, kwargs, callback):
         try:
-            response = yield from func(*args, **kwargs)
+            response = await func(*args, **kwargs)
         except Exception as error:
             tb = sys.exc_info()[2]
-            response = yield from callback.on_failure(type(error), error, tb)
+            response = await callback.on_failure(type(error), error, tb)
         else:
-            response = yield from callback.on_success(response)
+            response = await callback.on_success(response)
         return response
 
-    @asyncio.coroutine
-    def sleep(self, duration, callback):
-        yield from asyncio.sleep(duration)
-        response = yield from callback.on_success()
-        return response
+    async def sleep(self, duration, callback):
+        await asyncio.sleep(duration)
+        return await callback.on_success()
 
-    @asyncio.coroutine
-    def finish(self, response):
-        yield
+    async def finish(self, response):
         return response
 
-    @asyncio.coroutine
-    def execute(self, executable):
-        response = yield from executable.execute()
-        return response
+    async def execute(self, executable):
+        return await executable.execute()
diff --git a/uplink/retry/__init__.py b/uplink/retry/__init__.py
index 36057f1..f3d4c4c 100644
--- a/uplink/retry/__init__.py
+++ b/uplink/retry/__init__.py
@@ -1,4 +1,5 @@
 from uplink.retry.retry import retry
 from uplink.retry.when import RetryPredicate
+from uplink.retry.backoff import RetryBackoff
 
-__all__ = ["retry", "RetryPredicate"]
+__all__ = ["retry", "RetryPredicate", "RetryBackoff"]
diff --git a/uplink/retry/backoff.py b/uplink/retry/backoff.py
index b5eb93b..07f9648 100644
--- a/uplink/retry/backoff.py
+++ b/uplink/retry/backoff.py
@@ -8,7 +8,132 @@ MAX_VALUE = sys.maxsize / 2
 __all__ = ["jittered", "exponential", "fixed"]
 
 
-def jittered(base=2, multiplier=1, minimum=0, maximum=MAX_VALUE):
+def from_iterable(iterable):
+    """Creates a retry strategy from an iterable of timeouts"""
+
+    class IterableRetryBackoff(_IterableBackoff):
+        def __iter__(self):
+            return iter(iterable)
+
+    return IterableRetryBackoff()
+
+
+def from_iterable_factory(iterable_factory):
+    """
+    Creates a retry strategy from a function that returns an iterable
+    of timeouts.
+    """
+
+    class IterableRetryBackoff(_IterableBackoff):
+        def __iter__(self):
+            return iter(iterable_factory())
+
+    return IterableRetryBackoff()
+
+
+class RetryBackoff(object):
+    """
+    Base class for a strategy that calculates the timeout between
+    retry attempts.
+
+    You can compose two ``RetryBackoff`` instances by using the ``|``
+    operator::
+
+        CustomBackoffA() | CustomBackoffB()
+
+    The resulting backoff strategy will first compute the timeout using
+    the left-hand instance. If that timeout is ``None``, the strategy
+    will try to compute a fallback using the right-hand instance. If
+    both instances return ``None``, the resulting strategy will also
+    return ``None``.
+    """
+
+    def get_timeout_after_response(self, request, response):
+        """
+        Returns the number of seconds to wait before retrying the
+        request, or ``None`` to indicate that the given response should
+        be returned.
+        """
+        raise NotImplementedError  # pragma: no cover
+
+    def get_timeout_after_exception(self, request, exc_type, exc_val, exc_tb):
+        """
+        Returns the number of seconds to wait before retrying the
+        request, or ``None`` to indicate that the given exception
+        should be raised.
+        """
+        raise NotImplementedError  # pragma: no cover
+
+    def handle_after_final_retry(self):
+        """
+        Handles any clean-up necessary following the final retry
+        attempt.
+        """
+        pass  # pragma: no cover
+
+    def __or__(self, other):
+        """Composes the current strategy with another."""
+        assert isinstance(
+            other, RetryBackoff
+        ), "Both objects should be backoffs."
+        return _Or(self, other)
+
+
+class _Or(RetryBackoff):
+    def __init__(self, left, right):
+        self._left = left
+        self._right = right
+
+    def get_timeout_after_response(self, request, response):
+        timeout = self._left.get_timeout_after_response(request, response)
+        if timeout is None:
+            return self._right.get_timeout_after_response(request, response)
+        return timeout
+
+    def get_timeout_after_exception(self, request, exc_type, exc_val, exc_tb):
+        timeout = self._left.get_timeout_after_exception(
+            request, exc_type, exc_val, exc_tb
+        )
+        if timeout is None:
+            return self._right.get_timeout_after_exception(
+                request, exc_type, exc_val, exc_tb
+            )
+        return timeout
+
+    def handle_after_final_retry(self):
+        self._left.handle_after_final_retry()
+        self._right.handle_after_final_retry()
+
+
+class _IterableBackoff(RetryBackoff):
+    __iterator = None
+
+    def __iter__(self):
+        raise NotImplementedError  # pragma: no cover
+
+    def __call__(self):
+        return iter(self)
+
+    def __next(self):
+        if self.__iterator is None:
+            self.__iterator = iter(self)
+
+        try:
+            return next(self.__iterator)
+        except StopIteration:
+            return None
+
+    def get_timeout_after_response(self, request, response):
+        return self.__next()
+
+    def get_timeout_after_exception(self, request, exc_type, exc_val, exc_tb):
+        return self.__next()
+
+    def handle_after_final_retry(self):
+        self.__iterator = None
+
+
+class jittered(_IterableBackoff):
     """
     Waits using capped exponential backoff and full jitter.
 
@@ -18,35 +143,43 @@ def jittered(base=2, multiplier=1, minimum=0, maximum=MAX_VALUE):
     time of competing clients in a distributed system experiencing
     high contention.
     """
-    exp_backoff = exponential(base, multiplier, minimum, maximum)
-    return lambda *_: iter(
-        random.uniform(0, 1) * delay for delay in exp_backoff()
-    )  # pragma: no cover
 
+    def __init__(self, base=2, multiplier=1, minimum=0, maximum=MAX_VALUE):
+        self._exp_backoff = exponential(base, multiplier, minimum, maximum)
+
+    def __iter__(self):
+        for delay in self._exp_backoff():  # pragma: no branch
+            yield random.uniform(0, 1) * delay
 
-def exponential(base=2, multiplier=1, minimum=0, maximum=MAX_VALUE):
+
+class exponential(_IterableBackoff):
     """
     Waits using capped exponential backoff, meaning that the delay
     is multiplied by a constant ``base`` after each attempt, up to
     an optional ``maximum`` value.
     """
 
-    def wait_iterator():
-        delay = multiplier
-        while minimum > delay:
-            delay *= base
-        while True:
-            yield min(delay, maximum)
-            delay *= base
+    def __init__(self, base=2, multiplier=1, minimum=0, maximum=MAX_VALUE):
+        self._base = base
+        self._multiplier = multiplier
+        self._minimum = minimum
+        self._maximum = maximum
 
-    return wait_iterator
+    def __iter__(self):
+        delay = self._multiplier
+        while self._minimum > delay:
+            delay *= self._base
+        while True:
+            yield min(delay, self._maximum)
+            delay *= self._base
 
 
-def fixed(seconds):
+class fixed(_IterableBackoff):
     """Waits for a fixed number of ``seconds`` before each retry."""
 
-    def wait_iterator():
-        while True:
-            yield seconds
+    def __init__(self, seconds):
+        self._seconds = seconds
 
-    return wait_iterator
+    def __iter__(self):
+        while True:
+            yield self._seconds
diff --git a/uplink/retry/retry.py b/uplink/retry/retry.py
index 1f7bee6..9f06a14 100644
--- a/uplink/retry/retry.py
+++ b/uplink/retry/retry.py
@@ -11,40 +11,6 @@ from uplink.retry._helpers import ClientExceptionProxy
 __all__ = ["retry"]
 
 
-class RetryTemplate(RequestTemplate):
-    def __init__(self, backoff, retry_condition):
-        self._backoff = backoff
-        self._backoff_iterator = None
-        self._condition = retry_condition
-        self._reset()
-
-    def _next_delay(self):
-        try:
-            delay = next(self._backoff_iterator)
-        except StopIteration:
-            # Fallback to the default behavior
-            pass
-        else:
-            return transitions.sleep(delay)
-
-    def _reset(self):
-        self._backoff_iterator = self._backoff()
-
-    def after_response(self, request, response):
-        if self._condition.should_retry_after_response(response):
-            return self._next_delay()
-        else:
-            self._reset()
-
-    def after_exception(self, request, exc_type, exc_val, exc_tb):
-        if self._condition.should_retry_after_exception(
-            exc_type, exc_val, exc_tb
-        ):
-            return self._next_delay()
-        else:
-            self._reset()
-
-
 # noinspection PyPep8Naming
 class retry(decorators.MethodAnnotation):
     """
@@ -87,13 +53,18 @@ class retry(decorators.MethodAnnotation):
             that should prompt a retry attempt.
         stop (:obj:`callable`, optional): A function that creates
             predicates that decide when to stop retrying a request.
-        backoff (:obj:`callable`, optional): A function that creates
-            an iterator over the ordered sequence of timeouts between
-            retries. If not specified, exponential backoff is used.
+        backoff (:class:`RetryBackoff`, :obj:`callable`, optional): A
+            backoff strategy or a function that creates an iterator
+            over the ordered sequence of timeouts between retries. If
+            not specified, exponential backoff is used.
     """
 
     _DEFAULT_PREDICATE = when_mod.raises(Exception)
 
+    stop = stop_mod
+    backoff = backoff_mod
+    when = when_mod
+
     def __init__(
         self,
         when=None,
@@ -102,22 +73,27 @@ class retry(decorators.MethodAnnotation):
         stop=None,
         backoff=None,
     ):
-        if stop is not None:
-            self._stop = stop
-        elif max_attempts is not None:
-            self._stop = stop_mod.after_attempt(max_attempts)
-        else:
-            self._stop = stop_mod.NEVER
-
-        self._predicate = when
+        if stop is None:
+            if max_attempts is not None:
+                stop = stop_mod.after_attempt(max_attempts)
+            else:
+                stop = stop_mod.NEVER
 
         if on_exception is not None:
-            self._predicate = when_mod.raises(on_exception) | self._predicate
+            when = when_mod.raises(on_exception) | when
+
+        if when is None:
+            when = self._DEFAULT_PREDICATE
 
-        if self._predicate is None:
-            self._predicate = self._DEFAULT_PREDICATE
+        if backoff is None:
+            backoff = backoff_mod.jittered()
 
-        self._backoff = backoff_mod.jittered() if backoff is None else backoff
+        if not isinstance(backoff, backoff_mod.RetryBackoff):
+            backoff = backoff_mod.from_iterable_factory(backoff)
+
+        self._when = when
+        self._backoff = backoff
+        self._stop = stop
 
     BASE_CLIENT_EXCEPTION = ClientExceptionProxy(
         lambda ex: ex.BaseClientException
@@ -129,22 +105,43 @@ class retry(decorators.MethodAnnotation):
 
     def modify_request(self, request_builder):
         request_builder.add_request_template(
-            self._create_template(request_builder)
+            _RetryTemplate(
+                self._when(request_builder),
+                self._backoff,
+                self._stop,
+            )
         )
 
-    def _create_template(self, request_builder):
-        return RetryTemplate(
-            self._backoff_iterator, self._predicate(request_builder)
-        )
 
-    def _backoff_iterator(self):
-        stop_gen = self._stop()
-        for delay in self._backoff():
-            next(stop_gen)
-            if stop_gen.send(delay):
-                break
-            yield delay
+class _RetryTemplate(RequestTemplate):
+    def __init__(self, condition, backoff, stop):
+        self._condition = condition
+        self._backoff = backoff
+        self._stop = stop
+        self._stop_iter = self._stop()
+
+    def _process_timeout(self, timeout):
+        next(self._stop_iter)
+        if timeout is None or self._stop_iter.send(timeout):
+            self._backoff.handle_after_final_retry()
+            self._stop_iter = self._stop()
+            return None
+        return transitions.sleep(timeout)
 
-    stop = stop_mod
-    backoff = backoff_mod
-    when = when_mod
+    def after_response(self, request, response):
+        if not self._condition.should_retry_after_response(response):
+            return self._process_timeout(None)
+        return self._process_timeout(
+            self._backoff.get_timeout_after_response(request, response)
+        )
+
+    def after_exception(self, request, exc_type, exc_val, exc_tb):
+        if not self._condition.should_retry_after_exception(
+            exc_type, exc_val, exc_tb
+        ):
+            return self._process_timeout(None)
+        return self._process_timeout(
+            self._backoff.get_timeout_after_exception(
+                request, exc_type, exc_val, exc_tb
+            )
+        )
diff --git a/uplink/returns.py b/uplink/returns.py
index 7851fde..904525e 100644
--- a/uplink/returns.py
+++ b/uplink/returns.py
@@ -44,9 +44,6 @@ class _ReturnsBase(decorators.MethodAnnotation):
     def return_type(self):  # pragma: no cover
         raise NotImplementedError
 
-    def _get_return_type(self, return_type):  # pragma: no cover
-        return return_type
-
     def _make_strategy(self, converter):  # pragma: no cover
         pass
 
@@ -56,20 +53,24 @@ class _ReturnsBase(decorators.MethodAnnotation):
             definition.return_type, self
         )
 
+    def _get_converter(self, request_builder, return_type):  # pragma: no cover
+        return request_builder.get_converter(
+            keys.CONVERT_FROM_RESPONSE_BODY, return_type.type
+        )
+
     def modify_request(self, request_builder):
         return_type = request_builder.return_type
         if not return_type.is_applicable(self):
             return
 
-        converter = request_builder.get_converter(
-            keys.CONVERT_FROM_RESPONSE_BODY,
-            self._get_return_type(return_type.type),
+        converter = self._get_converter(request_builder, return_type)
+        if converter is None:
+            return
+
+        # Found a converter that can handle the return type.
+        request_builder.return_type = return_type.with_strategy(
+            self._make_strategy(converter)
         )
-        if converter is not None:
-            # Found a converter that can handle the return type.
-            request_builder.return_type = return_type.with_strategy(
-                self._make_strategy(converter)
-            )
 
 
 class JsonStrategy(object):
@@ -90,9 +91,6 @@ class JsonStrategy(object):
         content = self._converter(content)
         return content
 
-    def unwrap(self):
-        return self._converter
-
 
 # noinspection PyPep8Naming
 class json(_ReturnsBase):
@@ -121,20 +119,30 @@ class json(_ReturnsBase):
             :emphasize-lines: 2
 
             {
-                "data": { "user": "prkumar", "id": 140232 },
+                "data": { "user": "prkumar", "id": "140232" },
                 "errors": []
             }
 
         If returning the list of errors is unnecessary, we can use the
-        :py:attr:`key` argument to strictly return the inner field
-        :py:attr:`data`:
+        :py:attr:`key` argument to strictly return the nested field
+        :py:attr:`data.id`:
 
         .. code-block:: python
 
-            @returns.json(key="data")
+            @returns.json(key=("data", "id"))
+            @get("/users/{username}")
+            def get_user_id(self, username):
+                \"""Get a specific user's ID.\"""
+
+        We can also configure Uplink to convert the field before it's
+        returned by also specifying the``type`` argument:
+
+       .. code-block:: python
+
+            @returns.json(key=("data", "id"), type=int)
             @get("/users/{username}")
-            def get_user(self, username):
-                \"""Get a specific user.\"""
+            def get_user_id(self, username):
+                \"""Get a specific user's ID.\"""
 
     .. versionadded:: v0.5.0
     """
@@ -145,6 +153,13 @@ class json(_ReturnsBase):
         def convert(self, response):
             return response
 
+    class _CastConverter(interfaces.Converter):
+        def __init__(self, cast):
+            self._cast = cast
+
+        def convert(self, response):
+            return self._cast(response)
+
     __dummy_converter = _DummyConverter()
 
     def __init__(self, type=None, key=(), model=None, member=()):
@@ -167,14 +182,24 @@ class json(_ReturnsBase):
     def return_type(self):
         return self._type
 
-    def _get_return_type(self, return_type):
-        # If return_type is None, the strategy should directly return
-        # the JSON body of the HTTP response, instead of trying to
+    def _get_converter(self, request_builder, return_type):
+        converter = super(json, self)._get_converter(
+            request_builder, return_type
+        )
+
+        if converter:
+            return converter
+
+        if callable(return_type.type):
+            return self._CastConverter(return_type.type)
+
+        # If the return_type cannot be converted, the strategy should directly
+        # return the JSON body of the HTTP response, instead of trying to
         # deserialize it into a certain type. In this case, by
         # defaulting the return type to the dummy converter, which
         # implements this pass-through behavior, we ensure that
         # _make_strategy is called.
-        return self.__dummy_converter if return_type is None else return_type
+        return self.__dummy_converter
 
     def _make_strategy(self, converter):
         return JsonStrategy(converter, self._key)


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..a809aa9
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,10 @@
+import sys
+
+collect_ignore = []
+if sys.version_info.major < 3:
+    collect_ignore.extend(
+        [
+            "unit/test_aiohttp_client.py",
+            "integration/test_retry_aiohttp.py",
+        ]
+    )
diff --git a/tests/integration/test_retry.py b/tests/integration/test_retry.py
index 851f4c7..13ba45c 100644
--- a/tests/integration/test_retry.py
+++ b/tests/integration/test_retry.py
@@ -5,7 +5,6 @@ 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/"
@@ -29,7 +28,9 @@ class GitHub(Consumer):
     def get_issue(self, user, repo, issue):
         pass
 
-    @retry(max_attempts=3, on_exception=retry.CONNECTION_TIMEOUT)
+    @retry(
+        stop=retry.stop.after_attempt(3), on_exception=retry.CONNECTION_TIMEOUT
+    )
     @get("repos/{user}/{repo}/project/{project}")
     def get_project(self, user, repo, project):
         pass
@@ -123,30 +124,6 @@ def test_retry_with_status_501(mock_client, mock_response):
     assert len(mock_client.history) == 2
 
 
-@requires_python34
-def test_retry_with_asyncio(mock_client, mock_response):
-    import asyncio
-
-    @asyncio.coroutine
-    def coroutine():
-        return mock_response
-
-    # Setup
-    mock_response.with_json({"id": 123, "name": "prkumar"})
-    mock_client.with_side_effect([Exception, coroutine()])
-    mock_client.with_io(io.AsyncioStrategy())
-    github = GitHub(base_url=BASE_URL, client=mock_client)
-
-    # Run
-    awaitable = github.get_user("prkumar")
-    loop = asyncio.get_event_loop()
-    response = loop.run_until_complete(asyncio.ensure_future(awaitable))
-
-    # Verify
-    assert len(mock_client.history) == 2
-    assert response.json() == {"id": 123, "name": "prkumar"}
-
-
 @pytest_twisted.inlineCallbacks
 def test_retry_with_twisted(mock_client, mock_response):
     from twisted.internet import defer
diff --git a/tests/integration/test_retry_aiohttp.py b/tests/integration/test_retry_aiohttp.py
new file mode 100644
index 0000000..05895cd
--- /dev/null
+++ b/tests/integration/test_retry_aiohttp.py
@@ -0,0 +1,29 @@
+# Third-party imports
+import pytest
+
+# Local imports.
+from uplink.clients import io
+
+from . import test_retry
+
+
+@pytest.mark.asyncio
+async def test_retry_with_asyncio(mock_client, mock_response):
+    import asyncio
+
+    @asyncio.coroutine
+    def coroutine():
+        return mock_response
+
+    # Setup
+    mock_response.with_json({"id": 123, "name": "prkumar"})
+    mock_client.with_side_effect([Exception, coroutine()])
+    mock_client.with_io(io.AsyncioStrategy())
+    github = test_retry.GitHub(base_url=test_retry.BASE_URL, client=mock_client)
+
+    # Run
+    response = await github.get_user("prkumar")
+
+    # Verify
+    assert len(mock_client.history) == 2
+    assert response.json() == {"id": 123, "name": "prkumar"}
diff --git a/tests/integration/test_returns.py b/tests/integration/test_returns.py
index b6e4c7e..d640b45 100644
--- a/tests/integration/test_returns.py
+++ b/tests/integration/test_returns.py
@@ -49,15 +49,47 @@ class GitHub(uplink.Consumer):
     def get_repos(self, user):
         pass
 
+    @uplink.returns.from_json(key=("data", 0, "size"))
+    @uplink.get("/users/{user}/repos")
+    def get_first_repo_size(self, user):
+        pass
+
+    @uplink.returns.from_json(key=("data", 0, "stars"), type=int)
+    @uplink.get("/users/{user}/repos")
+    def get_first_repo_stars(self, user):
+        pass
+
     @uplink.json
     @uplink.post("/users/{user}/repos", args={"repo": uplink.Body(Repo)})
     def create_repo(self, user, repo):
         pass
 
+    @uplink.returns(object)
+    @uplink.get("/users")
+    def list_users(self):
+        pass
+
 
 # Tests
 
 
+def test_returns_response_when_type_has_no_converter(
+    mock_client, mock_response
+):
+    # Setup
+    mock_response.with_json({"id": 123, "name": "prkumar"})
+    mock_client.with_response(mock_response)
+    github = GitHub(
+        base_url=BASE_URL, client=mock_client, converters=user_reader
+    )
+
+    # Run
+    response = github.list_users()
+
+    # Verify
+    assert response == mock_response
+
+
 def test_returns_with_type(mock_client, mock_response):
     # Setup
     mock_response.with_json({"id": 123, "name": "prkumar"})
@@ -114,6 +146,48 @@ def test_returns_json_with_list(mock_client, mock_response):
     ] == repo
 
 
+def test_returns_json_by_key(mock_client, mock_response):
+    # Setup
+    mock_response.with_json(
+        {
+            "data": [
+                {"owner": "prkumar", "name": "uplink", "size": 300},
+                {"owner": "prkumar", "name": "uplink-protobuf", "size": 400},
+            ],
+            "errors": [],
+        }
+    )
+    mock_client.with_response(mock_response)
+    github = GitHub(base_url=BASE_URL, client=mock_client)
+
+    # Run
+    size = github.get_first_repo_size("prkumar")
+
+    # Verify
+    assert size == 300
+
+
+def test_returns_json_with_key_and_type(mock_client, mock_response):
+    # Setup
+    mock_response.with_json(
+        {
+            "data": [
+                {"owner": "prkumar", "name": "uplink", "stars": "300"},
+                {"owner": "prkumar", "name": "uplink-protobuf", "stars": "400"},
+            ],
+            "errors": [],
+        }
+    )
+    mock_client.with_response(mock_response)
+    github = GitHub(base_url=BASE_URL, client=mock_client)
+
+    # Run
+    stars = github.get_first_repo_stars("prkumar")
+
+    # Verify
+    assert stars == 300
+
+
 def test_post_json(mock_client):
     # Setup
     github = GitHub(
diff --git a/tests/unit/test_aiohttp_client.py b/tests/unit/test_aiohttp_client.py
new file mode 100644
index 0000000..c27a64b
--- /dev/null
+++ b/tests/unit/test_aiohttp_client.py
@@ -0,0 +1,263 @@
+# Standard library imports
+import contextlib
+
+# Third-party imports
+import aiohttp
+import asyncio
+import pytest
+
+# Local imports
+from uplink.clients import (
+    AiohttpClient,
+    aiohttp_,
+    register,
+    io,
+)
+
+
+@contextlib.contextmanager
+def _patch(obj, attr, value):
+    if obj is not None:
+        old_value = getattr(obj, attr)
+        setattr(obj, attr, value)
+    yield
+    if obj is not None:
+        setattr(obj, attr, old_value)
+
+
+@pytest.fixture
+def aiohttp_session_mock(mocker):
+    return mocker.Mock(spec=aiohttp.ClientSession)
+
+
+class AsyncMock(object):
+    def __init__(self, result=None):
+        self._result = result
+        self._calls = 0
+
+    async def __call__(self, *args, **kwargs):
+        self._calls += 1
+        f = asyncio.Future()
+        f.set_result(self._result)
+        return f
+
+    @property
+    def called(self):
+        return self._calls > 0
+
+
+class TestAiohttp(object):
+    def test_init_when_aiohttp_is_not_installed(self):
+        with _patch(aiohttp_, "aiohttp", None):
+            with pytest.raises(NotImplementedError):
+                AiohttpClient()
+
+    def test_init_with_session_None(self, mocker):
+        mocker.spy(AiohttpClient, "_create_session")
+        AiohttpClient(kwarg1="value")
+        AiohttpClient._create_session.assert_called_with(kwarg1="value")
+
+    def test_get_client(self, aiohttp_session_mock):
+        client = register.get_client(aiohttp_session_mock)
+        assert isinstance(client, aiohttp_.AiohttpClient)
+
+    @pytest.mark.asyncio
+    async def test_request_send(self, mocker, aiohttp_session_mock):
+        # Setup
+        expected_response = mocker.Mock()
+
+        async def request(*args, **kwargs):
+            return expected_response
+
+        aiohttp_session_mock.request = request
+        client = aiohttp_.AiohttpClient(aiohttp_session_mock)
+
+        # Run
+        response = await client.send((1, 2, {}))
+
+        # Verify
+        assert response == expected_response
+
+    @pytest.mark.asyncio
+    async def test_callback(self, mocker, aiohttp_session_mock):
+        # Setup
+        expected_response = mocker.Mock(spec=aiohttp_.aiohttp.ClientResponse)
+        expected_response.text = AsyncMock()
+
+        async def request(*args, **kwargs):
+            return expected_response
+
+        aiohttp_session_mock.request = request
+        client = aiohttp_.AiohttpClient(aiohttp_session_mock)
+
+        # Run
+        async def call():
+            response = await client.send((1, 2, {}))
+            response = await client.apply_callback(lambda x: 2, response)
+            return response
+
+        value = await call()
+
+        # Verify
+        assert value == 2
+        assert expected_response.text.called
+
+    def test_wrap_callback(self, mocker):
+        # Setup
+        c = AiohttpClient()
+        mocker.spy(c, "_sync_callback_adapter")
+
+        # Run: with callback that is not a coroutine
+        def callback(*_):
+            pass
+
+        c.wrap_callback(callback)
+
+        # Verify: Should wrap it
+        c._sync_callback_adapter.assert_called_with(callback)
+
+        # Run: with coroutine callback
+        async def awaitable_callback():
+            pass
+
+        assert c.wrap_callback(awaitable_callback) is awaitable_callback
+
+    @pytest.mark.asyncio
+    async def test_threaded_callback(self, mocker):
+        def callback(response):
+            return response
+
+        # Mock response.
+        response = mocker.Mock(spec=aiohttp_.aiohttp.ClientResponse)
+        response.text = AsyncMock()
+
+        # Run
+        new_callback = aiohttp_.threaded_callback(callback)
+        return_value = await new_callback(response)
+
+        # Verify
+        assert response.text.called
+        assert return_value == response
+
+        # Run: Verify with callback that returns new value
+        def callback(*_):
+            return 1
+
+        new_callback = aiohttp_.threaded_callback(callback)
+        value = await new_callback(response)
+        assert value == 1
+        assert response.text.called
+
+        # Run: Verify with response that is not ClientResponse (should not be wrapped)
+        response = mocker.Mock()
+        await new_callback(response)
+        assert not response.text.called
+
+    def test_threaded_coroutine(self):
+        async def coroutine():
+            return 1
+
+        threaded_coroutine = aiohttp_.ThreadedCoroutine(coroutine)
+
+        # Run -- should block
+        response = threaded_coroutine()
+
+        # Verify
+        assert response == 1
+
+    def test_threaded_response(self, mocker):
+        async def coroutine():
+            return 1
+
+        def not_a_coroutine():
+            return 2
+
+        response = mocker.Mock()
+        response.coroutine = coroutine
+        response.not_coroutine = not_a_coroutine
+        threaded_response = aiohttp_.ThreadedResponse(response)
+
+        # Run
+        threaded_coroutine = threaded_response.coroutine
+        return_value = threaded_coroutine()
+
+        # Verify
+        assert isinstance(threaded_coroutine, aiohttp_.ThreadedCoroutine)
+        assert return_value == 1
+        assert threaded_response.not_coroutine is not_a_coroutine
+
+    @pytest.mark.asyncio
+    async def test_create(self, mocker):
+        session_cls_mock = mocker.patch("aiohttp.ClientSession")
+        positionals = [1]
+        keywords = {"keyword": 2}
+
+        # Run: Create client
+        client = aiohttp_.AiohttpClient.create(*positionals, **keywords)
+
+        # Verify: session hasn't been created yet.
+        assert not session_cls_mock.called
+
+        # Run: Get session
+        await client.session()
+
+        # Verify: session created with args
+        session_cls_mock.assert_called_with(*positionals, **keywords)
+
+    @pytest.mark.asyncio
+    async def test_close_auto_created_session(self, mocker):
+        # Setup
+        import gc
+
+        mock_session = mocker.Mock(spec=aiohttp.ClientSession)
+        session_cls_mock = mocker.patch("aiohttp.ClientSession")
+        session_cls_mock.return_value = mock_session
+
+        positionals = [1]
+        keywords = {"keyword": 2}
+
+        # Run: Create client
+        client = aiohttp_.AiohttpClient.create(*positionals, **keywords)
+
+        # Run: Get session
+        await client.session()
+
+        # Verify: session created with args
+        session_cls_mock.assert_called_with(*positionals, **keywords)
+
+        # Verify: session closed on garbage collection
+        del client
+        gc.collect()
+        session_cls_mock.return_value.close.assert_called_with()
+
+    def test_exceptions(self):
+        import aiohttp
+
+        exceptions = aiohttp_.AiohttpClient.exceptions
+
+        with pytest.raises(exceptions.BaseClientException):
+            raise aiohttp.ClientError()
+
+        with pytest.raises(exceptions.BaseClientException):
+            # Test polymorphism
+            raise aiohttp.InvalidURL("invalid")
+
+        with pytest.raises(exceptions.ConnectionError):
+            raise aiohttp.ClientConnectionError()
+
+        with pytest.raises(exceptions.ConnectionTimeout):
+            raise aiohttp.ClientConnectorError.__new__(
+                aiohttp.ClientConnectorError
+            )
+
+        with pytest.raises(exceptions.ServerTimeout):
+            raise aiohttp.ServerTimeoutError()
+
+        with pytest.raises(exceptions.SSLError):
+            raise aiohttp.ClientSSLError.__new__(aiohttp.ClientSSLError)
+
+        with pytest.raises(exceptions.InvalidURL):
+            raise aiohttp.InvalidURL("invalid")
+
+    def test_io(self):
+        assert isinstance(aiohttp_.AiohttpClient.io(), io.AsyncioStrategy)
diff --git a/tests/unit/test_clients.py b/tests/unit/test_clients.py
index 4fe9985..9cdd467 100644
--- a/tests/unit/test_clients.py
+++ b/tests/unit/test_clients.py
@@ -6,7 +6,6 @@ import pytest
 
 # Local imports
 from uplink.clients import (
-    AiohttpClient,
     interfaces,
     requests_,
     twisted_,
@@ -14,16 +13,6 @@ from uplink.clients import (
     io,
 )
 
-try:
-    from uplink.clients import aiohttp_
-except (ImportError, SyntaxError):
-    aiohttp_ = None
-
-
-requires_python34 = pytest.mark.skipif(
-    not aiohttp_, reason="Requires Python 3.4 or above"
-)
-
 
 @contextlib.contextmanager
 def _patch(obj, attr, value):
@@ -195,274 +184,3 @@ class TestTwisted(object):
 
     def test_io(self):
         assert isinstance(twisted_.TwistedClient.io(), io.TwistedStrategy)
-
-
-@pytest.fixture
-def aiohttp_session_mock(mocker):
-    import aiohttp
-
-    return mocker.Mock(spec=aiohttp.ClientSession)
-
-
-class TestAiohttp(object):
-    @requires_python34
-    def test_init_when_aiohttp_is_not_installed(self):
-        with _patch(aiohttp_, "aiohttp", None):
-            with pytest.raises(NotImplementedError):
-                AiohttpClient()
-
-    @requires_python34
-    def test_init_with_session_None(self, mocker):
-        mocker.spy(AiohttpClient, "_create_session")
-        AiohttpClient(kwarg1="value")
-        AiohttpClient._create_session.assert_called_with(kwarg1="value")
-
-    @requires_python34
-    def test_get_client(self, aiohttp_session_mock):
-        client = register.get_client(aiohttp_session_mock)
-        assert isinstance(client, aiohttp_.AiohttpClient)
-
-    @requires_python34
-    def test_request_send(self, mocker, aiohttp_session_mock):
-        # Setup
-        import asyncio
-
-        expected_response = mocker.Mock()
-
-        @asyncio.coroutine
-        def request(*args, **kwargs):
-            return expected_response
-
-        aiohttp_session_mock.request = request
-        client = aiohttp_.AiohttpClient(aiohttp_session_mock)
-
-        # Run
-        response = client.send((1, 2, {}))
-        loop = asyncio.get_event_loop()
-        value = loop.run_until_complete(asyncio.ensure_future(response))
-
-        # Verify
-        assert value == expected_response
-
-    @requires_python34
-    def test_callback(self, mocker, aiohttp_session_mock):
-        # Setup
-        import asyncio
-
-        expected_response = mocker.Mock(spec=aiohttp_.aiohttp.ClientResponse)
-
-        @asyncio.coroutine
-        def request(*args, **kwargs):
-            return expected_response
-
-        aiohttp_session_mock.request = request
-        client = aiohttp_.AiohttpClient(aiohttp_session_mock)
-        client._sync_callback_adapter = asyncio.coroutine
-
-        # Run
-        globals_ = dict(globals(), client=client)
-        locals_ = {"asyncio": asyncio}
-        exec(
-            """@asyncio.coroutine
-def call():
-    response = yield from client.send((1, 2, {}))
-    response = yield from client.apply_callback(lambda x: 2, response)
-    return response
-""",
-            globals_,
-            locals_,
-        )
-
-        loop = asyncio.get_event_loop()
-        value = loop.run_until_complete(
-            asyncio.ensure_future(locals_["call"]())
-        )
-
-        # Verify
-        assert value == 2
-
-    @requires_python34
-    def test_wrap_callback(self, mocker):
-        import asyncio
-
-        # Setup
-        c = AiohttpClient()
-        mocker.spy(c, "_sync_callback_adapter")
-
-        # Run: with callback that is not a coroutine
-        def callback(*_):
-            pass
-
-        c.wrap_callback(callback)
-
-        # Verify: Should wrap it
-        c._sync_callback_adapter.assert_called_with(callback)
-
-        # Run: with coroutine callback
-        coroutine_callback = asyncio.coroutine(callback)
-        assert c.wrap_callback(coroutine_callback) is coroutine_callback
-
-    @requires_python34
-    def test_threaded_callback(self, mocker):
-        import asyncio
-
-        def callback(response):
-            return response
-
-        # Mock response.
-        response = mocker.Mock(spec=aiohttp_.aiohttp.ClientResponse)
-        response.text = asyncio.coroutine(mocker.stub())
-
-        # Run
-        new_callback = aiohttp_.threaded_callback(callback)
-        return_value = new_callback(response)
-        loop = asyncio.get_event_loop()
-        value = loop.run_until_complete(asyncio.ensure_future(return_value))
-
-        # Verify
-        response.text.assert_called_with()
-        assert value == response
-
-        # Run: Verify with callback that returns new value
-        def callback(*_):
-            return 1
-
-        new_callback = aiohttp_.threaded_callback(callback)
-        awaitable = new_callback(response)
-        loop = asyncio.get_event_loop()
-        value = loop.run_until_complete(asyncio.ensure_future(awaitable))
-        assert value == 1
-        assert response.text.called
-
-        # Run: Verify with response that is not ClientResponse (should not be wrapped)
-        response = mocker.Mock()
-        awaitable = new_callback(response)
-        loop = asyncio.get_event_loop()
-        loop.run_until_complete(asyncio.ensure_future(awaitable))
-        assert not response.text.called
-
-    @requires_python34
-    def test_threaded_coroutine(self):
-        # Setup
-        import asyncio
-
-        @asyncio.coroutine
-        def coroutine():
-            return 1
-
-        threaded_coroutine = aiohttp_.ThreadedCoroutine(coroutine)
-
-        # Run -- should block
-        response = threaded_coroutine()
-
-        # Verify
-        assert response == 1
-
-    @requires_python34
-    def test_threaded_response(self, mocker):
-        # Setup
-        import asyncio
-
-        @asyncio.coroutine
-        def coroutine():
-            return 1
-
-        def not_a_coroutine():
-            return 2
-
-        response = mocker.Mock()
-        response.coroutine = coroutine
-        response.not_coroutine = not_a_coroutine
-        threaded_response = aiohttp_.ThreadedResponse(response)
-
-        # Run
-        threaded_coroutine = threaded_response.coroutine
-        return_value = threaded_coroutine()
-
-        # Verify
-        assert isinstance(threaded_coroutine, aiohttp_.ThreadedCoroutine)
-        assert return_value == 1
-        assert threaded_response.not_coroutine is not_a_coroutine
-
-    @requires_python34
-    def test_create(self, mocker):
-        # Setup
-        import asyncio
-
-        session_cls_mock = mocker.patch("aiohttp.ClientSession")
-        positionals = [1]
-        keywords = {"keyword": 2}
-
-        # Run: Create client
-        client = aiohttp_.AiohttpClient.create(*positionals, **keywords)
-
-        # Verify: session hasn't been created yet.
-        assert not session_cls_mock.called
-
-        # Run: Get session
-        loop = asyncio.get_event_loop()
-        loop.run_until_complete(asyncio.ensure_future(client.session()))
-
-        # Verify: session created with args
-        session_cls_mock.assert_called_with(*positionals, **keywords)
-
-    @requires_python34
-    def test_close_auto_created_session(self, mocker):
-        # Setup
-        import asyncio
-        import gc
-        import aiohttp
-
-        mock_session = mocker.Mock(spec=aiohttp.ClientSession)
-        session_cls_mock = mocker.patch("aiohttp.ClientSession")
-        session_cls_mock.return_value = mock_session
-
-        positionals = [1]
-        keywords = {"keyword": 2}
-
-        # Run: Create client
-        client = aiohttp_.AiohttpClient.create(*positionals, **keywords)
-
-        # Run: Get session
-        loop = asyncio.get_event_loop()
-        loop.run_until_complete(asyncio.ensure_future(client.session()))
-
-        # Verify: session created with args
-        session_cls_mock.assert_called_with(*positionals, **keywords)
-        del client
-        gc.collect()
-        session_cls_mock.return_value.close.assert_called_with()
-
-    @requires_python34
-    def test_exceptions(self):
-        import aiohttp
-
-        exceptions = aiohttp_.AiohttpClient.exceptions
-
-        with pytest.raises(exceptions.BaseClientException):
-            raise aiohttp.ClientError()
-
-        with pytest.raises(exceptions.BaseClientException):
-            # Test polymorphism
-            raise aiohttp.InvalidURL("invalid")
-
-        with pytest.raises(exceptions.ConnectionError):
-            raise aiohttp.ClientConnectionError()
-
-        with pytest.raises(exceptions.ConnectionTimeout):
-            raise aiohttp.ClientConnectorError.__new__(
-                aiohttp.ClientConnectorError
-            )
-
-        with pytest.raises(exceptions.ServerTimeout):
-            raise aiohttp.ServerTimeoutError()
-
-        with pytest.raises(exceptions.SSLError):
-            raise aiohttp.ClientSSLError.__new__(aiohttp.ClientSSLError)
-
-        with pytest.raises(exceptions.InvalidURL):
-            raise aiohttp.InvalidURL("invalid")
-
-    @requires_python34
-    def test_io(self):
-        assert isinstance(aiohttp_.AiohttpClient.io(), io.AsyncioStrategy)
diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py
index 02ecc0a..0fd7b9e 100644
--- a/tests/unit/test_retry.py
+++ b/tests/unit/test_retry.py
@@ -35,6 +35,26 @@ def test_fixed_backoff():
     assert next(iterator) == 10
 
 
+def test_compose_backoff(mocker):
+    left = backoff.from_iterable([0, 1])
+    right = backoff.from_iterable([2])
+    mocker.spy(left, "handle_after_final_retry")
+    mocker.spy(right, "handle_after_final_retry")
+    strategy = left | right
+
+    # Should return None after both strategies are exhausted
+    assert strategy.get_timeout_after_response(None, None) == 0
+    assert strategy.get_timeout_after_exception(None, None, None, None) == 1
+    assert strategy.get_timeout_after_response(None, None) == 2
+    assert strategy.get_timeout_after_exception(None, None, None, None) is None
+
+    # Should invoke both strategies after_stop method
+    strategy.handle_after_final_retry()
+
+    left.handle_after_final_retry.assert_called_once_with()
+    right.handle_after_final_retry.assert_called_once_with()
+
+
 def test_retry_stop_default():
     decorator = retry()
     assert stop.NEVER == decorator._stop
@@ -75,22 +95,6 @@ def test_stop_or_with_none():
     assert stop1 is (stop1 | None)
 
 
-def test_retry_custom_stop():
-    def custom_stop(*_):
-        return True
-
-    decorator = retry(stop=custom_stop)
-    assert decorator._stop == custom_stop
-
-
-def test_retry_backoff():
-    def custom_backoff(*_):
-        return True
-
-    decorator = retry(backoff=custom_backoff)
-    assert decorator._backoff == custom_backoff
-
-
 def test_retry_decorator_exposes_submodules_as_properties():
     assert retry.backoff is backoff
     assert retry.stop is stop
diff --git a/tests/unit/test_returns.py b/tests/unit/test_returns.py
index 72a4a1c..f5a2e91 100644
--- a/tests/unit/test_returns.py
+++ b/tests/unit/test_returns.py
@@ -19,9 +19,7 @@ def test_returns_with_multiple_decorators(request_builder, mocker):
     first_type = returns.ReturnType.with_decorator(None, decorator1)
     second_type = (
         request_builder.return_type
-    ) = returns.ReturnType.with_decorator(
-        first_type, decorator2
-    )
+    ) = returns.ReturnType.with_decorator(first_type, decorator2)
 
     # Verify that the return type doesn't change after being handled by first decorator
     decorator1.modify_request(request_builder)
@@ -59,14 +57,73 @@ def test_returns_json(request_builder, mocker):
         mock_response.json()
     )
 
-    # Verify: Doesn't apply to unsupported types
+    # Verify: Returns JSON when type cannot be converted
     request_builder.get_converter.return_value = None
-    returns_json = returns.json(str, ())
+    returns_json = returns.json(None, ())
     request_builder.return_type = returns.ReturnType.with_decorator(
         None, returns_json
     )
     returns_json.modify_request(request_builder)
-    assert not callable(request_builder.return_type)
+    assert callable(request_builder.return_type)
+    assert request_builder.return_type(mock_response) == mock_response.json()
+
+
+def test_returns_json_builtin_type(request_builder, mocker):
+    mock_response = mocker.Mock()
+    mock_response.json.return_value = {"key": "1"}
+    request_builder.get_converter.return_value = None
+    returns_json = returns.json(type=int, key="key")
+    request_builder.return_type = returns.ReturnType.with_decorator(
+        None, returns_json
+    )
+    returns_json.modify_request(request_builder)
+    print(request_builder.return_type)
+    assert callable(request_builder.return_type)
+    assert request_builder.return_type(mock_response) == 1
+
+
+class TestReturnsJsonCast(object):
+    default_value = {"key": "1"}
+
+    @staticmethod
+    def prepare_test(request_builder, mocker, value=default_value, **kwargs):
+        mock_response = mocker.Mock()
+        mock_response.json.return_value = value
+        request_builder.get_converter.return_value = None
+        returns_json = returns.json(**kwargs)
+        request_builder.return_type = returns.ReturnType.with_decorator(
+            None, returns_json
+        )
+        returns_json.modify_request(request_builder)
+        return mock_response
+
+    def test_without_type(self, request_builder, mocker):
+        mock_response = self.prepare_test(request_builder, mocker, key="key")
+        assert request_builder.return_type(mock_response) == "1"
+
+    def test_with_callable_type(self, request_builder, mocker):
+        mock_response = self.prepare_test(
+            request_builder,
+            mocker,
+            type=lambda _: "test",
+        )
+        assert request_builder.return_type(mock_response) == "test"
+
+    def test_with_builtin_type(self, request_builder, mocker):
+        mock_response = self.prepare_test(request_builder, mocker, type=str)
+        assert request_builder.return_type(mock_response) == str(
+            self.default_value
+        )
+
+    def test_with_builtin_type_and_key(self, request_builder, mocker):
+        mock_response = self.prepare_test(
+            request_builder, mocker, key="key", type=int
+        )
+        assert request_builder.return_type(mock_response) == 1
+
+    def test_with_not_callable_cast(self, request_builder, mocker):
+        mock_response = self.prepare_test(request_builder, mocker, type=1)
+        assert request_builder.return_type(mock_response) == self.default_value
 
 
 def test_returns_JsonStrategy(mocker):
@@ -77,5 +134,3 @@ def test_returns_JsonStrategy(mocker):
 
     converter = returns.JsonStrategy(lambda y: y + "!", "hello")
     assert converter(response) == "world!"
-
-    assert returns.JsonStrategy(1).unwrap() == 1