Repository: lepture/authlib

Issue:
HTTPX OAuth1 implementation does not set Content-Length header
**Describe the bug**
The OAuth1 implementation for HTTPX does not set the content-length header, which can cause problems when using body signatures and talking to servers who block body POSTs where the length header is not specified.

**Expected behavior**

Content-Length header is set.

**Environment:**

 - OS: linux
 - Python Version: 3.8
 - Authlib Version: 0.14.3

Incoming PR fixes this.


Discussion:

Patch:
diff --git a/.github/workflows/python.yml b/.github/workflows/python.yml
index 42de65d..4db46eb 100644
--- a/.github/workflows/python.yml
+++ b/.github/workflows/python.yml
@@ -24,11 +24,11 @@ jobs:
         - version: 2.7
           toxenv: py27,py27-flask
         - version: 3.6
-          toxenv: py36,flask,django,async
+          toxenv: py36,flask,django,py3
         - version: 3.7
-          toxenv: py37,flask,django,async
+          toxenv: py37,flask,django,py3
         - version: 3.8
-          toxenv: py38,flask,django,async
+          toxenv: py38,flask,django,py3
 
     steps:
     - uses: actions/checkout@v2
diff --git a/authlib/integrations/httpx_client/__init__.py b/authlib/integrations/httpx_client/__init__.py
index 21d641e..6b4b9d6 100644
--- a/authlib/integrations/httpx_client/__init__.py
+++ b/authlib/integrations/httpx_client/__init__.py
@@ -6,12 +6,12 @@ from authlib.oauth1 import (
     SIGNATURE_TYPE_QUERY,
     SIGNATURE_TYPE_BODY,
 )
-from .oauth1_client import OAuth1Auth, AsyncOAuth1Client
+from .oauth1_client import OAuth1Auth, AsyncOAuth1Client, OAuth1Client
 from .oauth2_client import (
-    OAuth2Auth, OAuth2ClientAuth,
+    OAuth2Auth, OAuth2Client, OAuth2ClientAuth,
     AsyncOAuth2Client,
 )
-from .assertion_client import AsyncAssertionClient
+from .assertion_client import AssertionClient, AsyncAssertionClient
 from ..base_client import OAuthError
 
 
diff --git a/authlib/integrations/httpx_client/assertion_client.py b/authlib/integrations/httpx_client/assertion_client.py
index 17228d3..62f81b7 100644
--- a/authlib/integrations/httpx_client/assertion_client.py
+++ b/authlib/integrations/httpx_client/assertion_client.py
@@ -1,4 +1,5 @@
-from httpx import AsyncClient
+from httpx import AsyncClient, Client
+from httpx._config import UNSET
 from authlib.oauth2.rfc7521 import AssertionClient as _AssertionClient
 from authlib.oauth2.rfc7523 import JWTBearerGrant
 from authlib.oauth2 import OAuth2Error
@@ -31,7 +32,7 @@ class AsyncAssertionClient(_AssertionClient, AsyncClient):
 
     async def request(self, method, url, withhold_token=False, auth=None, **kwargs):
         """Send request with auto refresh token feature."""
-        if not withhold_token and auth is None:
+        if not withhold_token and auth is UNSET:
             if not self.token or self.token.is_expired():
                 await self.refresh_token()
 
@@ -51,3 +52,35 @@ class AsyncAssertionClient(_AssertionClient, AsyncClient):
             )
         self.token = token
         return self.token
+
+
+class AssertionClient(_AssertionClient, Client):
+    token_auth_class = OAuth2Auth
+    JWT_BEARER_GRANT_TYPE = JWTBearerGrant.GRANT_TYPE
+    ASSERTION_METHODS = {
+        JWT_BEARER_GRANT_TYPE: JWTBearerGrant.sign,
+    }
+    DEFAULT_GRANT_TYPE = JWT_BEARER_GRANT_TYPE
+
+    def __init__(self, token_endpoint, issuer, subject, audience=None, grant_type=None,
+                 claims=None, token_placement='header', scope=None, **kwargs):
+
+        client_kwargs = extract_client_kwargs(kwargs)
+        Client.__init__(self, **client_kwargs)
+
+        _AssertionClient.__init__(
+            self, session=self,
+            token_endpoint=token_endpoint, issuer=issuer, subject=subject,
+            audience=audience, grant_type=grant_type, claims=claims,
+            token_placement=token_placement, scope=scope, **kwargs
+        )
+
+    def request(self, method, url, withhold_token=False, auth=None, **kwargs):
+        """Send request with auto refresh token feature."""
+        if not withhold_token and auth is UNSET:
+            if not self.token or self.token.is_expired():
+                self.refresh_token()
+
+            auth = self.token_auth
+        return super(AssertionClient, self).request(
+            method, url, auth=auth, **kwargs)
diff --git a/authlib/integrations/httpx_client/oauth1_client.py b/authlib/integrations/httpx_client/oauth1_client.py
index 7bb1ccb..6d755bb 100644
--- a/authlib/integrations/httpx_client/oauth1_client.py
+++ b/authlib/integrations/httpx_client/oauth1_client.py
@@ -1,5 +1,5 @@
 import typing
-from httpx import AsyncClient, Auth, Request, Response
+from httpx import AsyncClient, Auth, Client, Request, Response
 from authlib.oauth1 import (
     SIGNATURE_HMAC_SHA1,
     SIGNATURE_TYPE_HEADER,
@@ -18,6 +18,7 @@ class OAuth1Auth(Auth, ClientAuth):
     def auth_flow(self, request: Request) -> typing.Generator[Request, Response, None]:
         url, headers, body = self.prepare(
             request.method, str(request.url), request.headers, request.content)
+        headers['Content-Length'] = str(len(body))
         yield Request(method=request.method, url=url, headers=headers, data=body)
 
 
@@ -72,3 +73,28 @@ class AsyncOAuth1Client(_OAuth1Client, AsyncClient):
     @staticmethod
     def handle_error(error_type, error_description):
         raise OAuthError(error_type, error_description)
+
+class OAuth1Client(_OAuth1Client, Client):
+    auth_class = OAuth1Auth
+
+    def __init__(self, client_id, client_secret=None,
+                 token=None, token_secret=None,
+                 redirect_uri=None, rsa_key=None, verifier=None,
+                 signature_method=SIGNATURE_HMAC_SHA1,
+                 signature_type=SIGNATURE_TYPE_HEADER,
+                 force_include_body=False, **kwargs):
+
+        _client_kwargs = extract_client_kwargs(kwargs)
+        Client.__init__(self, **_client_kwargs)
+
+        _OAuth1Client.__init__(
+            self, self,
+            client_id=client_id, client_secret=client_secret,
+            token=token, token_secret=token_secret,
+            redirect_uri=redirect_uri, rsa_key=rsa_key, verifier=verifier,
+            signature_method=signature_method, signature_type=signature_type,
+            force_include_body=force_include_body, **kwargs)
+
+    @staticmethod
+    def handle_error(error_type, error_description):
+        raise OAuthError(error_type, error_description)
diff --git a/authlib/integrations/httpx_client/oauth2_client.py b/authlib/integrations/httpx_client/oauth2_client.py
index 0e2f81f..387560b 100644
--- a/authlib/integrations/httpx_client/oauth2_client.py
+++ b/authlib/integrations/httpx_client/oauth2_client.py
@@ -1,6 +1,7 @@
 import asyncio
 import typing
-from httpx import AsyncClient, Auth, Request, Response
+from httpx import AsyncClient, Auth, Client, Request, Response
+from httpx._config import UNSET
 from authlib.common.urls import url_decode
 from authlib.oauth2.client import OAuth2Client as _OAuth2Client
 from authlib.oauth2.auth import ClientAuth, TokenAuth
@@ -78,7 +79,7 @@ class AsyncOAuth2Client(_OAuth2Client, AsyncClient):
         raise OAuthError(error_type, error_description)
 
     async def request(self, method, url, withhold_token=False, auth=None, **kwargs):
-        if not withhold_token and auth is None:
+        if not withhold_token and auth is UNSET:
             if not self.token:
                 raise MissingTokenError()
 
@@ -150,3 +151,60 @@ class AsyncOAuth2Client(_OAuth2Client, AsyncClient):
         return self.post(
             url, data=dict(url_decode(body)),
             headers=headers, auth=auth, **kwargs)
+
+class OAuth2Client(_OAuth2Client, Client):
+    SESSION_REQUEST_PARAMS = HTTPX_CLIENT_KWARGS
+
+    client_auth_class = OAuth2ClientAuth
+    token_auth_class = OAuth2Auth
+
+    def __init__(self, client_id=None, client_secret=None,
+                 token_endpoint_auth_method=None,
+                 revocation_endpoint_auth_method=None,
+                 scope=None, redirect_uri=None,
+                 token=None, token_placement='header',
+                 update_token=None, **kwargs):
+
+        # extract httpx.Client kwargs
+        client_kwargs = self._extract_session_request_params(kwargs)
+        Client.__init__(self, **client_kwargs)
+
+        _OAuth2Client.__init__(
+            self, session=self,
+            client_id=client_id, client_secret=client_secret,
+            token_endpoint_auth_method=token_endpoint_auth_method,
+            revocation_endpoint_auth_method=revocation_endpoint_auth_method,
+            scope=scope, redirect_uri=redirect_uri,
+            token=token, token_placement=token_placement,
+            update_token=update_token, **kwargs
+        )
+
+    @staticmethod
+    def handle_error(error_type, error_description):
+        raise OAuthError(error_type, error_description)
+
+    def request(self, method, url, withhold_token=False, auth=None, **kwargs):
+        if not withhold_token and auth is UNSET:
+            if not self.token:
+                raise MissingTokenError()
+
+            if self.token.is_expired():
+                self.ensure_active_token()
+
+            auth = self.token_auth
+
+        return super(OAuth2Client, self).request(
+            method, url, auth=auth, **kwargs)
+
+    def ensure_active_token(self):
+        refresh_token = self.token.get('refresh_token')
+        url = self.metadata.get('token_endpoint')
+        if refresh_token and url:
+            self.refresh_token(url, refresh_token=refresh_token)
+        elif self.metadata.get('grant_type') == 'client_credentials':
+            access_token = self.token['access_token']
+            token = self.fetch_token(url, grant_type='client_credentials')
+            if self.update_token:
+                self.update_token(token, access_token=access_token)
+        else:
+            raise InvalidTokenError()
diff --git a/docs/client/httpx.rst b/docs/client/httpx.rst
index d97f6bd..48412f1 100644
--- a/docs/client/httpx.rst
+++ b/docs/client/httpx.rst
@@ -15,6 +15,9 @@ OAuth for HTTPX
 HTTPX is a next-generation HTTP client for Python. Authlib enables OAuth 1.0
 and OAuth 2.0 for HTTPX with its async versions:
 
+* :class:`OAuth1Client`
+* :class:`OAuth2Client`
+* :class:`AssertionClient`
 * :class:`AsyncOAuth1Client`
 * :class:`AsyncOAuth2Client`
 * :class:`AsyncAssertionClient`
diff --git a/tox.ini b/tox.ini
index 587d38b..a8c5a35 100644
--- a/tox.ini
+++ b/tox.ini
@@ -1,7 +1,7 @@
 [tox]
 envlist =
     py{27,36,37,38}
-    {py36,py37,py38}-async
+    {py36,py37,py38}
     {py27,py36,py37,py38}-flask
     {py36,py37,py38}-django
     coverage
@@ -12,10 +12,11 @@ deps =
     py27: unittest2
     flask: Flask
     flask: Flask-SQLAlchemy
-    async: httpx==0.14.1
-    async: pytest-asyncio
-    async: starlette
-    async: itsdangerous
+    py3: httpx==0.14.3
+    py3: pytest-asyncio
+    py3: starlette
+    py3: itsdangerous
+    py3: werkzeug
     django: Django
     django: pytest-django
 
@@ -23,7 +24,7 @@ setenv =
     TESTPATH=tests/core
     RCFILE=setup.cfg
     py27: RCFILE=.py27conf
-    async: TESTPATH=tests/py3
+    py3: TESTPATH=tests/py3
     flask: TESTPATH=tests/flask
     django: TESTPATH=tests/django
 commands =


Test patch:
diff --git a/tests/py3/test_httpx_client/test_assertion_client.py b/tests/py3/test_httpx_client/test_assertion_client.py
new file mode 100644
index 0000000..91b0529
--- /dev/null
+++ b/tests/py3/test_httpx_client/test_assertion_client.py
@@ -0,0 +1,65 @@
+import time
+import pytest
+from authlib.integrations.httpx_client import AssertionClient
+from tests.py3.utils import MockDispatch
+
+
+default_token = {
+    'token_type': 'Bearer',
+    'access_token': 'a',
+    'refresh_token': 'b',
+    'expires_in': '3600',
+    'expires_at': int(time.time()) + 3600,
+}
+
+
+@pytest.mark.asyncio
+def test_refresh_token():
+    def verifier(request):
+        content = request.form
+        if str(request.url) == 'https://i.b/token':
+            assert 'assertion' in content
+
+    with AssertionClient(
+        'https://i.b/token',
+        grant_type=AssertionClient.JWT_BEARER_GRANT_TYPE,
+        issuer='foo',
+        subject='foo',
+        audience='foo',
+        alg='HS256',
+        key='secret',
+        app=MockDispatch(default_token, assert_func=verifier)
+    ) as client:
+        client.get('https://i.b')
+
+    # trigger more case
+    now = int(time.time())
+    with AssertionClient(
+        'https://i.b/token',
+        issuer='foo',
+        subject=None,
+        audience='foo',
+        issued_at=now,
+        expires_at=now + 3600,
+        header={'alg': 'HS256'},
+        key='secret',
+        scope='email',
+        claims={'test_mode': 'true'},
+        app=MockDispatch(default_token, assert_func=verifier)
+    ) as client:
+        client.get('https://i.b')
+        client.get('https://i.b')
+
+
+@pytest.mark.asyncio
+def test_without_alg():
+    with AssertionClient(
+        'https://i.b/token',
+        issuer='foo',
+        subject='foo',
+        audience='foo',
+        key='secret',
+        app=MockDispatch(default_token)
+    ) as client:
+        with pytest.raises(ValueError):
+            client.get('https://i.b')
diff --git a/tests/py3/test_httpx_client/test_async_assertion_client.py b/tests/py3/test_httpx_client/test_async_assertion_client.py
index fa51e14..46286bf 100644
--- a/tests/py3/test_httpx_client/test_async_assertion_client.py
+++ b/tests/py3/test_httpx_client/test_async_assertion_client.py
@@ -1,7 +1,7 @@
 import time
 import pytest
 from authlib.integrations.httpx_client import AsyncAssertionClient
-from tests.py3.utils import MockDispatch
+from tests.py3.utils import AsyncMockDispatch
 
 
 default_token = {
@@ -28,7 +28,7 @@ async def test_refresh_token():
         audience='foo',
         alg='HS256',
         key='secret',
-        app=MockDispatch(default_token, assert_func=verifier)
+        app=AsyncMockDispatch(default_token, assert_func=verifier)
     ) as client:
         await client.get('https://i.b')
 
@@ -45,7 +45,7 @@ async def test_refresh_token():
         key='secret',
         scope='email',
         claims={'test_mode': 'true'},
-        app=MockDispatch(default_token, assert_func=verifier)
+        app=AsyncMockDispatch(default_token, assert_func=verifier)
     ) as client:
         await client.get('https://i.b')
         await client.get('https://i.b')
@@ -59,6 +59,7 @@ async def test_without_alg():
         subject='foo',
         audience='foo',
         key='secret',
+        app=AsyncMockDispatch()
     ) as client:
         with pytest.raises(ValueError):
             await client.get('https://i.b')
diff --git a/tests/py3/test_httpx_client/test_async_oauth1_client.py b/tests/py3/test_httpx_client/test_async_oauth1_client.py
index bf8542d..7570356 100644
--- a/tests/py3/test_httpx_client/test_async_oauth1_client.py
+++ b/tests/py3/test_httpx_client/test_async_oauth1_client.py
@@ -5,7 +5,7 @@ from authlib.integrations.httpx_client import (
     SIGNATURE_TYPE_BODY,
     SIGNATURE_TYPE_QUERY,
 )
-from tests.py3.utils import MockDispatch
+from tests.py3.utils import AsyncMockDispatch
 
 oauth_url = 'https://example.com/oauth'
 
@@ -19,7 +19,7 @@ async def test_fetch_request_token_via_header():
         assert 'oauth_consumer_key="id"' in auth_header
         assert 'oauth_signature=' in auth_header
 
-    app = MockDispatch(request_token, assert_func=assert_func)
+    app = AsyncMockDispatch(request_token, assert_func=assert_func)
     async with AsyncOAuth1Client('id', 'secret', app=app) as client:
         response = await client.fetch_request_token(oauth_url)
 
@@ -38,7 +38,7 @@ async def test_fetch_request_token_via_body():
         assert b'oauth_consumer_key=id' in content
         assert b'&oauth_signature=' in content
 
-    mock_response = MockDispatch(request_token, assert_func=assert_func)
+    mock_response = AsyncMockDispatch(request_token, assert_func=assert_func)
 
     async with AsyncOAuth1Client(
         'id', 'secret', signature_type=SIGNATURE_TYPE_BODY,
@@ -61,7 +61,7 @@ async def test_fetch_request_token_via_query():
         assert 'oauth_consumer_key=id' in url
         assert '&oauth_signature=' in url
 
-    mock_response = MockDispatch(request_token, assert_func=assert_func)
+    mock_response = AsyncMockDispatch(request_token, assert_func=assert_func)
 
     async with AsyncOAuth1Client(
         'id', 'secret', signature_type=SIGNATURE_TYPE_QUERY,
@@ -83,7 +83,7 @@ async def test_fetch_access_token():
         assert 'oauth_consumer_key="id"' in auth_header
         assert 'oauth_signature=' in auth_header
 
-    mock_response = MockDispatch(request_token, assert_func=assert_func)
+    mock_response = AsyncMockDispatch(request_token, assert_func=assert_func)
     async with AsyncOAuth1Client(
         'id', 'secret', token='foo', token_secret='bar',
         app=mock_response,
@@ -98,7 +98,7 @@ async def test_fetch_access_token():
 
 @pytest.mark.asyncio
 async def test_get_via_header():
-    mock_response = MockDispatch(b'hello')
+    mock_response = AsyncMockDispatch(b'hello')
     async with AsyncOAuth1Client(
         'id', 'secret', token='foo', token_secret='bar',
         app=mock_response,
@@ -121,7 +121,7 @@ async def test_get_via_body():
         assert b'oauth_consumer_key=id' in content
         assert b'oauth_signature=' in content
 
-    mock_response = MockDispatch(b'hello', assert_func=assert_func)
+    mock_response = AsyncMockDispatch(b'hello', assert_func=assert_func)
     async with AsyncOAuth1Client(
         'id', 'secret', token='foo', token_secret='bar',
         signature_type=SIGNATURE_TYPE_BODY,
@@ -138,7 +138,7 @@ async def test_get_via_body():
 
 @pytest.mark.asyncio
 async def test_get_via_query():
-    mock_response = MockDispatch(b'hello')
+    mock_response = AsyncMockDispatch(b'hello')
     async with AsyncOAuth1Client(
         'id', 'secret', token='foo', token_secret='bar',
         signature_type=SIGNATURE_TYPE_QUERY,
diff --git a/tests/py3/test_httpx_client/test_async_oauth2_client.py b/tests/py3/test_httpx_client/test_async_oauth2_client.py
index 9de1a51..2333d2e 100644
--- a/tests/py3/test_httpx_client/test_async_oauth2_client.py
+++ b/tests/py3/test_httpx_client/test_async_oauth2_client.py
@@ -9,7 +9,7 @@ from authlib.integrations.httpx_client import (
     OAuthError,
     AsyncOAuth2Client,
 )
-from tests.py3.utils import MockDispatch
+from tests.py3.utils import AsyncMockDispatch
 
 
 default_token = {
@@ -28,7 +28,7 @@ async def test_add_token_to_header():
         auth_header = request.headers.get('authorization')
         assert auth_header == token
 
-    mock_response = MockDispatch({'a': 'a'}, assert_func=assert_func)
+    mock_response = AsyncMockDispatch({'a': 'a'}, assert_func=assert_func)
     async with AsyncOAuth2Client(
             'foo',
             token=default_token,
@@ -46,7 +46,7 @@ async def test_add_token_to_body():
         content = await request.body()
         assert default_token['access_token'] in content.decode()
 
-    mock_response = MockDispatch({'a': 'a'}, assert_func=assert_func)
+    mock_response = AsyncMockDispatch({'a': 'a'}, assert_func=assert_func)
     async with AsyncOAuth2Client(
             'foo',
             token=default_token,
@@ -64,7 +64,7 @@ async def test_add_token_to_uri():
     async def assert_func(request):
         assert default_token['access_token'] in str(request.url)
 
-    mock_response = MockDispatch({'a': 'a'}, assert_func=assert_func)
+    mock_response = AsyncMockDispatch({'a': 'a'}, assert_func=assert_func)
     async with AsyncOAuth2Client(
             'foo',
             token=default_token,
@@ -124,7 +124,7 @@ async def test_fetch_token_post():
         assert 'client_id=' in content
         assert 'grant_type=authorization_code' in content
 
-    mock_response = MockDispatch(default_token, assert_func=assert_func)
+    mock_response = AsyncMockDispatch(default_token, assert_func=assert_func)
     async with AsyncOAuth2Client('foo', app=mock_response) as client:
         token = await client.fetch_token(url, authorization_response='https://i.b/?code=v')
         assert token == default_token
@@ -137,7 +137,7 @@ async def test_fetch_token_post():
         token = await client.fetch_token(url, code='v')
         assert token == default_token
 
-    mock_response = MockDispatch({'error': 'invalid_request'})
+    mock_response = AsyncMockDispatch({'error': 'invalid_request'})
     async with AsyncOAuth2Client('foo', app=mock_response) as client:
         with pytest.raises(OAuthError):
             await client.fetch_token(url)
@@ -153,7 +153,7 @@ async def test_fetch_token_get():
         assert 'client_id=' in url
         assert 'grant_type=authorization_code' in url
 
-    mock_response = MockDispatch(default_token, assert_func=assert_func)
+    mock_response = AsyncMockDispatch(default_token, assert_func=assert_func)
     async with AsyncOAuth2Client('foo', app=mock_response) as client:
         authorization_response = 'https://i.b/?code=v'
         token = await client.fetch_token(
@@ -184,7 +184,7 @@ async def test_token_auth_method_client_secret_post():
         assert 'client_secret=bar' in content
         assert 'grant_type=authorization_code' in content
 
-    mock_response = MockDispatch(default_token, assert_func=assert_func)
+    mock_response = AsyncMockDispatch(default_token, assert_func=assert_func)
     async with AsyncOAuth2Client(
             'foo', 'bar',
             token_endpoint_auth_method='client_secret_post',
@@ -204,7 +204,7 @@ async def test_access_token_response_hook():
         return resp
 
     access_token_response_hook = mock.Mock(side_effect=_access_token_response_hook)
-    app = MockDispatch(default_token)
+    app = AsyncMockDispatch(default_token)
     async with AsyncOAuth2Client('foo', token=default_token, app=app) as sess:
         sess.register_compliance_hook(
             'access_token_response',
@@ -225,7 +225,7 @@ async def test_password_grant_type():
         assert 'scope=profile' in content
         assert 'grant_type=password' in content
 
-    app = MockDispatch(default_token, assert_func=assert_func)
+    app = AsyncMockDispatch(default_token, assert_func=assert_func)
     async with AsyncOAuth2Client('foo', scope='profile', app=app) as sess:
         token = await sess.fetch_token(url, username='v', password='v')
         assert token == default_token
@@ -245,7 +245,7 @@ async def test_client_credentials_type():
         assert 'scope=profile' in content
         assert 'grant_type=client_credentials' in content
 
-    app = MockDispatch(default_token, assert_func=assert_func)
+    app = AsyncMockDispatch(default_token, assert_func=assert_func)
     async with AsyncOAuth2Client('foo', scope='profile', app=app) as sess:
         token = await sess.fetch_token(url)
         assert token == default_token
@@ -263,7 +263,7 @@ async def test_cleans_previous_token_before_fetching_new_one():
     new_token['expires_at'] = now + 3600
     url = 'https://example.com/token'
 
-    app = MockDispatch(new_token)
+    app = AsyncMockDispatch(new_token)
     with mock.patch('time.time', lambda: now):
         async with AsyncOAuth2Client('foo', token=default_token, app=app) as sess:
             assert await sess.fetch_token(url) == new_token
@@ -289,7 +289,7 @@ async def test_auto_refresh_token():
         token_type='bearer', expires_at=100
     )
 
-    app = MockDispatch(default_token)
+    app = AsyncMockDispatch(default_token)
     async with AsyncOAuth2Client(
             'foo', token=old_token, token_endpoint='https://i.b/token',
             update_token=update_token, app=app
@@ -325,7 +325,7 @@ async def test_auto_refresh_token2():
         expires_at=100
     )
 
-    app = MockDispatch(default_token)
+    app = AsyncMockDispatch(default_token)
 
     async with AsyncOAuth2Client(
             'foo', token=old_token,
@@ -359,7 +359,7 @@ async def test_auto_refresh_token3():
         expires_at=100
     )
 
-    app = MockDispatch(default_token)
+    app = AsyncMockDispatch(default_token)
 
     async with AsyncOAuth2Client(
             'foo', token=old_token, token_endpoint='https://i.b/token',
@@ -382,7 +382,7 @@ async def test_auto_refresh_token4():
         expires_at=100
     )
 
-    app = MockDispatch(default_token)
+    app = AsyncMockDispatch(default_token)
 
     async with AsyncOAuth2Client(
             'foo', token=old_token, token_endpoint='https://i.b/token',
@@ -396,7 +396,7 @@ async def test_auto_refresh_token4():
 @pytest.mark.asyncio
 async def test_revoke_token():
     answer = {'status': 'ok'}
-    app = MockDispatch(answer)
+    app = AsyncMockDispatch(answer)
 
     async with AsyncOAuth2Client('a', app=app) as sess:
         resp = await sess.revoke_token('https://i.b/token', 'hi')
@@ -411,6 +411,6 @@ async def test_revoke_token():
 
 @pytest.mark.asyncio
 async def test_request_without_token():
-    async with AsyncOAuth2Client('a') as client:
+    async with AsyncOAuth2Client('a', app=AsyncMockDispatch()) as client:
         with pytest.raises(OAuthError):
             await client.get('https://i.b/token')
diff --git a/tests/py3/test_httpx_client/test_oauth1_client.py b/tests/py3/test_httpx_client/test_oauth1_client.py
new file mode 100644
index 0000000..a5f34df
--- /dev/null
+++ b/tests/py3/test_httpx_client/test_oauth1_client.py
@@ -0,0 +1,157 @@
+import pytest
+from authlib.integrations.httpx_client import (
+    OAuthError,
+    OAuth1Client,
+    SIGNATURE_TYPE_BODY,
+    SIGNATURE_TYPE_QUERY,
+)
+from tests.py3.utils import MockDispatch
+
+oauth_url = 'https://example.com/oauth'
+
+
+@pytest.mark.asyncio
+def test_fetch_request_token_via_header():
+    request_token = {'oauth_token': '1', 'oauth_token_secret': '2'}
+
+    def assert_func(request):
+        auth_header = request.headers.get('authorization')
+        assert 'oauth_consumer_key="id"' in auth_header
+        assert 'oauth_signature=' in auth_header
+
+    app = MockDispatch(request_token, assert_func=assert_func)
+    with OAuth1Client('id', 'secret', app=app) as client:
+        response = client.fetch_request_token(oauth_url)
+
+    assert response == request_token
+
+
+@pytest.mark.asyncio
+def test_fetch_request_token_via_body():
+    request_token = {'oauth_token': '1', 'oauth_token_secret': '2'}
+
+    def assert_func(request):
+        auth_header = request.headers.get('authorization')
+        assert auth_header is None
+
+        content = request.form
+        assert content.get('oauth_consumer_key') == 'id'
+        assert 'oauth_signature' in content
+
+    mock_response = MockDispatch(request_token, assert_func=assert_func)
+
+    with OAuth1Client(
+        'id', 'secret', signature_type=SIGNATURE_TYPE_BODY,
+        app=mock_response,
+    ) as client:
+        response = client.fetch_request_token(oauth_url)
+
+    assert response == request_token
+
+
+@pytest.mark.asyncio
+def test_fetch_request_token_via_query():
+    request_token = {'oauth_token': '1', 'oauth_token_secret': '2'}
+
+    def assert_func(request):
+        auth_header = request.headers.get('authorization')
+        assert auth_header is None
+
+        url = str(request.url)
+        assert 'oauth_consumer_key=id' in url
+        assert '&oauth_signature=' in url
+
+    mock_response = MockDispatch(request_token, assert_func=assert_func)
+
+    with OAuth1Client(
+        'id', 'secret', signature_type=SIGNATURE_TYPE_QUERY,
+        app=mock_response,
+    ) as client:
+        response = client.fetch_request_token(oauth_url)
+
+    assert response == request_token
+
+
+@pytest.mark.asyncio
+def test_fetch_access_token():
+    request_token = {'oauth_token': '1', 'oauth_token_secret': '2'}
+
+    def assert_func(request):
+        auth_header = request.headers.get('authorization')
+        assert 'oauth_verifier="d"' in auth_header
+        assert 'oauth_token="foo"' in auth_header
+        assert 'oauth_consumer_key="id"' in auth_header
+        assert 'oauth_signature=' in auth_header
+
+    mock_response = MockDispatch(request_token, assert_func=assert_func)
+    with OAuth1Client(
+        'id', 'secret', token='foo', token_secret='bar',
+        app=mock_response,
+    ) as client:
+        with pytest.raises(OAuthError):
+            client.fetch_access_token(oauth_url)
+
+        response = client.fetch_access_token(oauth_url, verifier='d')
+
+    assert response == request_token
+
+
+@pytest.mark.asyncio
+def test_get_via_header():
+    mock_response = MockDispatch(b'hello')
+    with OAuth1Client(
+        'id', 'secret', token='foo', token_secret='bar',
+        app=mock_response,
+    ) as client:
+        response = client.get('https://example.com/')
+
+    assert response.content == b'hello'
+    request = response.request
+    auth_header = request.headers.get('authorization')
+    assert 'oauth_token="foo"' in auth_header
+    assert 'oauth_consumer_key="id"' in auth_header
+    assert 'oauth_signature=' in auth_header
+
+
+@pytest.mark.asyncio
+def test_get_via_body():
+    def assert_func(request):
+        content = request.form
+        assert content.get('oauth_token') == 'foo'
+        assert content.get('oauth_consumer_key') == 'id'
+        assert 'oauth_signature' in content
+
+    mock_response = MockDispatch(b'hello', assert_func=assert_func)
+    with OAuth1Client(
+        'id', 'secret', token='foo', token_secret='bar',
+        signature_type=SIGNATURE_TYPE_BODY,
+        app=mock_response,
+    ) as client:
+        response = client.post('https://example.com/')
+
+    assert response.content == b'hello'
+
+    request = response.request
+    auth_header = request.headers.get('authorization')
+    assert auth_header is None
+
+
+@pytest.mark.asyncio
+def test_get_via_query():
+    mock_response = MockDispatch(b'hello')
+    with OAuth1Client(
+        'id', 'secret', token='foo', token_secret='bar',
+        signature_type=SIGNATURE_TYPE_QUERY,
+        app=mock_response,
+    ) as client:
+        response = client.get('https://example.com/')
+
+    assert response.content == b'hello'
+    request = response.request
+    auth_header = request.headers.get('authorization')
+    assert auth_header is None
+
+    url = str(request.url)
+    assert 'oauth_token=foo' in url
+    assert 'oauth_consumer_key=id' in url
+    assert 'oauth_signature=' in url
diff --git a/tests/py3/test_httpx_client/test_oauth2_client.py b/tests/py3/test_httpx_client/test_oauth2_client.py
new file mode 100644
index 0000000..7bd3938
--- /dev/null
+++ b/tests/py3/test_httpx_client/test_oauth2_client.py
@@ -0,0 +1,374 @@
+import mock
+import time
+import pytest
+from copy import deepcopy
+from authlib.common.security import generate_token
+from authlib.common.urls import url_encode
+from authlib.integrations.httpx_client import (
+    OAuthError,
+    OAuth2Client,
+)
+from tests.py3.utils import MockDispatch
+
+
+default_token = {
+    'token_type': 'Bearer',
+    'access_token': 'a',
+    'refresh_token': 'b',
+    'expires_in': '3600',
+    'expires_at': int(time.time()) + 3600,
+}
+
+
+def test_add_token_to_header():
+    def assert_func(request):
+        token = 'Bearer ' + default_token['access_token']
+        auth_header = request.headers.get('authorization')
+        assert auth_header == token
+
+    mock_response = MockDispatch({'a': 'a'}, assert_func=assert_func)
+    with OAuth2Client(
+            'foo',
+            token=default_token,
+            app=mock_response
+    ) as client:
+        resp = client.get('https://i.b')
+
+    data = resp.json()
+    assert data['a'] == 'a'
+
+
+def test_add_token_to_body():
+    def assert_func(request):
+        content = request.data
+        content = content.decode()
+        assert content == 'access_token=%s' % default_token['access_token']
+
+    mock_response = MockDispatch({'a': 'a'}, assert_func=assert_func)
+    with OAuth2Client(
+            'foo',
+            token=default_token,
+            token_placement='body',
+            app=mock_response
+    ) as client:
+        resp = client.get('https://i.b')
+
+    data = resp.json()
+    assert data['a'] == 'a'
+
+
+def test_add_token_to_uri():
+    def assert_func(request):
+        assert default_token['access_token'] in str(request.url)
+
+    mock_response = MockDispatch({'a': 'a'}, assert_func=assert_func)
+    with OAuth2Client(
+            'foo',
+            token=default_token,
+            token_placement='uri',
+            app=mock_response
+    ) as client:
+        resp = client.get('https://i.b')
+
+    data = resp.json()
+    assert data['a'] == 'a'
+
+
+def test_create_authorization_url():
+    url = 'https://example.com/authorize?foo=bar'
+
+    sess = OAuth2Client(client_id='foo')
+    auth_url, state = sess.create_authorization_url(url)
+    assert state in auth_url
+    assert 'client_id=foo' in auth_url
+    assert 'response_type=code' in auth_url
+
+    sess = OAuth2Client(client_id='foo', prompt='none')
+    auth_url, state = sess.create_authorization_url(
+        url, state='foo', redirect_uri='https://i.b', scope='profile')
+    assert state == 'foo'
+    assert 'i.b' in auth_url
+    assert 'profile' in auth_url
+    assert 'prompt=none' in auth_url
+
+
+def test_code_challenge():
+    sess = OAuth2Client('foo', code_challenge_method='S256')
+
+    url = 'https://example.com/authorize'
+    auth_url, _ = sess.create_authorization_url(
+        url, code_verifier=generate_token(48))
+    assert 'code_challenge=' in auth_url
+    assert 'code_challenge_method=S256' in auth_url
+
+
+def test_token_from_fragment():
+    sess = OAuth2Client('foo')
+    response_url = 'https://i.b/callback#' + url_encode(default_token.items())
+    assert sess.token_from_fragment(response_url) == default_token
+    token = sess.fetch_token(authorization_response=response_url)
+    assert token == default_token
+
+
+def test_fetch_token_post():
+    url = 'https://example.com/token'
+
+    def assert_func(request):
+        content = request.form
+        assert content.get('code') == 'v'
+        assert content.get('client_id') == 'foo'
+        assert content.get('grant_type') == 'authorization_code'
+
+    mock_response = MockDispatch(default_token, assert_func=assert_func)
+    with OAuth2Client('foo', app=mock_response) as client:
+        token = client.fetch_token(url, authorization_response='https://i.b/?code=v')
+        assert token == default_token
+
+    with OAuth2Client(
+            'foo',
+            token_endpoint_auth_method='none',
+            app=mock_response
+    ) as client:
+        token = client.fetch_token(url, code='v')
+        assert token == default_token
+
+    mock_response = MockDispatch({'error': 'invalid_request'})
+    with OAuth2Client('foo', app=mock_response) as client:
+        with pytest.raises(OAuthError):
+            client.fetch_token(url)
+
+
+def test_fetch_token_get():
+    url = 'https://example.com/token'
+
+    def assert_func(request):
+        url = str(request.url)
+        assert 'code=v' in url
+        assert 'client_id=' in url
+        assert 'grant_type=authorization_code' in url
+
+    mock_response = MockDispatch(default_token, assert_func=assert_func)
+    with OAuth2Client('foo', app=mock_response) as client:
+        authorization_response = 'https://i.b/?code=v'
+        token = client.fetch_token(
+            url, authorization_response=authorization_response, method='GET')
+        assert token == default_token
+
+    with OAuth2Client(
+            'foo',
+            token_endpoint_auth_method='none',
+            app=mock_response
+    ) as client:
+        token = client.fetch_token(url, code='v', method='GET')
+        assert token == default_token
+
+        token = client.fetch_token(url + '?q=a', code='v', method='GET')
+        assert token == default_token
+
+
+def test_token_auth_method_client_secret_post():
+    url = 'https://example.com/token'
+
+    def assert_func(request):
+        content = request.form
+        assert content.get('code') == 'v'
+        assert content.get('client_id') == 'foo'
+        assert content.get('client_secret') == 'bar'
+        assert content.get('grant_type') == 'authorization_code'
+
+    mock_response = MockDispatch(default_token, assert_func=assert_func)
+    with OAuth2Client(
+            'foo', 'bar',
+            token_endpoint_auth_method='client_secret_post',
+            app=mock_response
+    ) as client:
+        token = client.fetch_token(url, code='v')
+
+    assert token == default_token
+
+
+def test_access_token_response_hook():
+    url = 'https://example.com/token'
+
+    def _access_token_response_hook(resp):
+        assert resp.json() == default_token
+        return resp
+
+    access_token_response_hook = mock.Mock(side_effect=_access_token_response_hook)
+    app = MockDispatch(default_token)
+    with OAuth2Client('foo', token=default_token, app=app) as sess:
+        sess.register_compliance_hook(
+            'access_token_response',
+            access_token_response_hook
+        )
+        assert sess.fetch_token(url) == default_token
+        assert access_token_response_hook.called is True
+
+
+def test_password_grant_type():
+    url = 'https://example.com/token'
+
+    def assert_func(request):
+        content = request.form
+        assert content.get('username') == 'v'
+        assert content.get('scope') == 'profile'
+        assert content.get('grant_type') == 'password'
+
+    app = MockDispatch(default_token, assert_func=assert_func)
+    with OAuth2Client('foo', scope='profile', app=app) as sess:
+        token = sess.fetch_token(url, username='v', password='v')
+        assert token == default_token
+
+        token = sess.fetch_token(
+            url, username='v', password='v', grant_type='password')
+        assert token == default_token
+
+
+def test_client_credentials_type():
+    url = 'https://example.com/token'
+
+    def assert_func(request):
+        content = request.form
+        assert content.get('scope') == 'profile'
+        assert content.get('grant_type') == 'client_credentials'
+
+    app = MockDispatch(default_token, assert_func=assert_func)
+    with OAuth2Client('foo', scope='profile', app=app) as sess:
+        token = sess.fetch_token(url)
+        assert token == default_token
+
+        token = sess.fetch_token(url, grant_type='client_credentials')
+        assert token == default_token
+
+
+def test_cleans_previous_token_before_fetching_new_one():
+    now = int(time.time())
+    new_token = deepcopy(default_token)
+    past = now - 7200
+    default_token['expires_at'] = past
+    new_token['expires_at'] = now + 3600
+    url = 'https://example.com/token'
+
+    app = MockDispatch(new_token)
+    with mock.patch('time.time', lambda: now):
+        with OAuth2Client('foo', token=default_token, app=app) as sess:
+            assert sess.fetch_token(url) == new_token
+
+
+def test_token_status():
+    token = dict(access_token='a', token_type='bearer', expires_at=100)
+    sess = OAuth2Client('foo', token=token)
+    assert sess.token.is_expired() is True
+
+
+def test_auto_refresh_token():
+
+    def _update_token(token, refresh_token=None, access_token=None):
+        assert refresh_token == 'b'
+        assert token == default_token
+
+    update_token = mock.Mock(side_effect=_update_token)
+
+    old_token = dict(
+        access_token='a', refresh_token='b',
+        token_type='bearer', expires_at=100
+    )
+
+    app = MockDispatch(default_token)
+    with OAuth2Client(
+            'foo', token=old_token, token_endpoint='https://i.b/token',
+            update_token=update_token, app=app
+    ) as sess:
+        sess.get('https://i.b/user')
+        assert update_token.called is True
+
+    old_token = dict(
+        access_token='a',
+        token_type='bearer',
+        expires_at=100
+    )
+    with OAuth2Client(
+            'foo', token=old_token, token_endpoint='https://i.b/token',
+            update_token=update_token, app=app
+    ) as sess:
+        with pytest.raises(OAuthError):
+            sess.get('https://i.b/user')
+
+
+def test_auto_refresh_token2():
+
+    def _update_token(token, refresh_token=None, access_token=None):
+        assert access_token == 'a'
+        assert token == default_token
+
+    update_token = mock.Mock(side_effect=_update_token)
+
+    old_token = dict(
+        access_token='a',
+        token_type='bearer',
+        expires_at=100
+    )
+
+    app = MockDispatch(default_token)
+
+    with OAuth2Client(
+            'foo', token=old_token,
+            token_endpoint='https://i.b/token',
+            grant_type='client_credentials',
+            app=app,
+    ) as client:
+        client.get('https://i.b/user')
+        assert update_token.called is False
+
+    with OAuth2Client(
+            'foo', token=old_token, token_endpoint='https://i.b/token',
+            update_token=update_token, grant_type='client_credentials',
+            app=app,
+    ) as client:
+        client.get('https://i.b/user')
+        assert update_token.called is True
+
+
+def test_auto_refresh_token3():
+    def _update_token(token, refresh_token=None, access_token=None):
+        assert access_token == 'a'
+        assert token == default_token
+
+    update_token = mock.Mock(side_effect=_update_token)
+
+    old_token = dict(
+        access_token='a',
+        token_type='bearer',
+        expires_at=100
+    )
+
+    app = MockDispatch(default_token)
+
+    with OAuth2Client(
+            'foo', token=old_token, token_endpoint='https://i.b/token',
+            update_token=update_token, grant_type='client_credentials',
+            app=app,
+    ) as client:
+        client.post('https://i.b/user', json={'foo': 'bar'})
+        assert update_token.called is True
+
+
+def test_revoke_token():
+    answer = {'status': 'ok'}
+    app = MockDispatch(answer)
+
+    with OAuth2Client('a', app=app) as sess:
+        resp = sess.revoke_token('https://i.b/token', 'hi')
+        assert resp.json() == answer
+
+        resp = sess.revoke_token(
+            'https://i.b/token', 'hi',
+            token_type_hint='access_token'
+        )
+        assert resp.json() == answer
+
+
+def test_request_without_token():
+    with OAuth2Client('a', app=MockDispatch()) as client:
+        with pytest.raises(OAuthError):
+            client.get('https://i.b/token')
diff --git a/tests/py3/test_starlette_client/test_oauth_client.py b/tests/py3/test_starlette_client/test_oauth_client.py
index 4722e4e..68654fc 100644
--- a/tests/py3/test_starlette_client/test_oauth_client.py
+++ b/tests/py3/test_starlette_client/test_oauth_client.py
@@ -2,7 +2,7 @@ import pytest
 from starlette.config import Config
 from starlette.requests import Request
 from authlib.integrations.starlette_client import OAuth
-from tests.py3.utils import PathMapDispatch
+from tests.py3.utils import AsyncPathMapDispatch
 from tests.client_base import get_bearer_token
 
 
@@ -39,7 +39,7 @@ def test_register_with_overwrite():
 @pytest.mark.asyncio
 async def test_oauth1_authorize():
     oauth = OAuth()
-    app = PathMapDispatch({
+    app = AsyncPathMapDispatch({
         '/request-token': {'body': 'oauth_token=foo&oauth_verifier=baz'},
         '/token': {'body': 'oauth_token=a&oauth_token_secret=b'},
     })
@@ -74,7 +74,7 @@ async def test_oauth1_authorize():
 @pytest.mark.asyncio
 async def test_oauth2_authorize():
     oauth = OAuth()
-    app = PathMapDispatch({
+    app = AsyncPathMapDispatch({
         '/token': {'body': get_bearer_token()}
     })
     client = oauth.register(
@@ -113,7 +113,7 @@ async def test_oauth2_authorize():
 
 @pytest.mark.asyncio
 async def test_oauth2_authorize_code_challenge():
-    app = PathMapDispatch({
+    app = AsyncPathMapDispatch({
         '/token': {'body': get_bearer_token()}
     })
     oauth = OAuth()
@@ -163,7 +163,7 @@ async def test_with_fetch_token_in_register():
     async def fetch_token(request):
         return {'access_token': 'dev', 'token_type': 'bearer'}
 
-    app = PathMapDispatch({
+    app = AsyncPathMapDispatch({
         '/user': {'body': {'sub': '123'}}
     })
     oauth = OAuth()
@@ -191,7 +191,7 @@ async def test_with_fetch_token_in_oauth():
     async def fetch_token(name, request):
         return {'access_token': 'dev', 'token_type': 'bearer'}
 
-    app = PathMapDispatch({
+    app = AsyncPathMapDispatch({
         '/user': {'body': {'sub': '123'}}
     })
     oauth = OAuth(fetch_token=fetch_token)
@@ -216,7 +216,7 @@ async def test_with_fetch_token_in_oauth():
 @pytest.mark.asyncio
 async def test_request_withhold_token():
     oauth = OAuth()
-    app = PathMapDispatch({
+    app = AsyncPathMapDispatch({
         '/user': {'body': {'sub': '123'}}
     })
     client = oauth.register(
@@ -252,7 +252,7 @@ async def test_oauth2_authorize_with_metadata():
         await client.create_authorization_url(req)
 
 
-    app = PathMapDispatch({
+    app = AsyncPathMapDispatch({
         '/.well-known/openid-configuration': {'body': {
             'authorization_endpoint': 'https://i.b/authorize'
         }}
diff --git a/tests/py3/test_starlette_client/test_user_mixin.py b/tests/py3/test_starlette_client/test_user_mixin.py
index 2e34801..f9e32b5 100644
--- a/tests/py3/test_starlette_client/test_user_mixin.py
+++ b/tests/py3/test_starlette_client/test_user_mixin.py
@@ -5,7 +5,7 @@ from authlib.jose import jwk
 from authlib.jose.errors import InvalidClaimError
 from authlib.oidc.core.grants.util import generate_id_token
 from tests.util import read_file_path
-from tests.py3.utils import PathMapDispatch
+from tests.py3.utils import AsyncPathMapDispatch
 from tests.client_base import get_bearer_token
 
 
@@ -15,7 +15,7 @@ async def run_fetch_userinfo(payload, compliance_fix=None):
     async def fetch_token(request):
         return get_bearer_token()
 
-    app = PathMapDispatch({
+    app = AsyncPathMapDispatch({
         '/userinfo': {'body': payload}
     })
 
@@ -125,7 +125,7 @@ async def test_force_fetch_jwks_uri():
         aud='dev', exp=3600, nonce='n',
     )
 
-    app = PathMapDispatch({
+    app = AsyncPathMapDispatch({
         '/jwks': {'body': read_file_path('jwks_public.json')}
     })
 
diff --git a/tests/py3/utils.py b/tests/py3/utils.py
index 807cea0..9416e10 100644
--- a/tests/py3/utils.py
+++ b/tests/py3/utils.py
@@ -1,9 +1,11 @@
 import json
-from starlette.requests import Request
-from starlette.responses import Response
+from starlette.requests import Request as ASGIRequest
+from starlette.responses import Response as ASGIResponse
+from werkzeug.wrappers import Request as WSGIRequest
+from werkzeug.wrappers import Response as WSGIResponse
 
 
-class MockDispatch:
+class AsyncMockDispatch:
     def __init__(self, body=b'', status_code=200, headers=None,
                  assert_func=None):
         if headers is None:
@@ -22,12 +24,12 @@ class MockDispatch:
         self.assert_func = assert_func
 
     async def __call__(self, scope, receive, send):
-        request = Request(scope, receive=receive)
+        request = ASGIRequest(scope, receive=receive)
 
         if self.assert_func:
             await self.assert_func(request)
 
-        response = Response(
+        response = ASGIResponse(
             status_code=self.status_code,
             content=self.body,
             headers=self.headers,
@@ -35,12 +37,12 @@ class MockDispatch:
         await response(scope, receive, send)
 
 
-class PathMapDispatch:
+class AsyncPathMapDispatch:
     def __init__(self, path_maps):
         self.path_maps = path_maps
 
     async def __call__(self, scope, receive, send):
-        request = Request(scope, receive=receive)
+        request = ASGIRequest(scope, receive=receive)
 
         rv = self.path_maps[request.url.path]
         status_code = rv.get('status_code', 200)
@@ -54,9 +56,66 @@ class PathMapDispatch:
                 body = body.encode()
             headers['Content-Type'] = 'application/x-www-form-urlencoded'
 
-        response = Response(
+        response = ASGIResponse(
             status_code=status_code,
             content=body,
             headers=headers,
         )
         await response(scope, receive, send)
+
+class MockDispatch:
+    def __init__(self, body=b'', status_code=200, headers=None,
+                 assert_func=None):
+        if headers is None:
+            headers = {}
+        if isinstance(body, dict):
+            body = json.dumps(body).encode()
+            headers['Content-Type'] = 'application/json'
+        else:
+            if isinstance(body, str):
+                body = body.encode()
+            headers['Content-Type'] = 'application/x-www-form-urlencoded'
+
+        self.body = body
+        self.status_code = status_code
+        self.headers = headers
+        self.assert_func = assert_func
+
+    def __call__(self, environ, start_response):
+        request = WSGIRequest(environ)
+
+        if self.assert_func:
+            self.assert_func(request)
+
+        response = WSGIResponse(
+            status=self.status_code,
+            response=self.body,
+            headers=self.headers,
+        )
+        return response(environ, start_response)
+
+
+class PathMapDispatch:
+    def __init__(self, path_maps):
+        self.path_maps = path_maps
+
+    def __call__(self, environ, start_response):
+        request = WSGIRequest(environ)
+
+        rv = self.path_maps[request.url.path]
+        status_code = rv.get('status_code', 200)
+        body = rv.get('body', b'')
+        headers = rv.get('headers', {})
+        if isinstance(body, dict):
+            body = json.dumps(body).encode()
+            headers['Content-Type'] = 'application/json'
+        else:
+            if isinstance(body, str):
+                body = body.encode()
+            headers['Content-Type'] = 'application/x-www-form-urlencoded'
+        response = WSGIResponse(
+            status=status_code,
+            response=body,
+            headers=headers,
+        )
+        return response(environ, start_response)