encode/httpx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add httpx.MockTransport() (#1401)

Browse Source 
* Add httpx.MockTransport

* Add docs on MockTransport

* Add pointer to RESPX

* Add note on pytest-httpx

* Tweak existing docs example to use 'httpx.MockTransport'

Co-authored-by: Florimond Manca <florimond.manca@gmail.com>
This commit is contained in:
Tom Christie 2021-01-06 11:04:26 +00:00 committed by GitHub
parent 3bf18637c1
commit 9c7c2ace99
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 198 additions and 158 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
docs/advanced.md
View File

@ -1062,6 +1062,31 @@ Which we can use in the same way:
{"text": "Hello, world!"}
```
### Mock transports
During testing it can often be useful to be able to mock out a transport,
and return pre-determined responses, rather than making actual network requests.
The `httpx.MockTransport` class accepts a handler function, which can be used
to map requests onto pre-determined responses:
```python
def handler(request):
return httpx.Response(200, json={"text": "Hello, world!"})
# Switch to a mock transport, if the TESTING environment variable is set.
if os.environ['TESTING'].upper() == "TRUE":
transport = httpx.MockTransport(handler)
else:
transport = httpx.HTTPTransport()
client = httpx.Client(transport=transport)
```
For more advanced use-cases you might want to take a look at either [the third-party
mocking library, RESPX](https://lundberg.github.io/respx/), or the [pytest-httpx library](https://github.com/Colin-b/pytest_httpx).
### Mounting transports
You can also mount transports against given schemes or domains, to control
@ -1101,7 +1126,10 @@ Mocking requests to a given domain:
```python
# All requests to "example.org" should be mocked out.
# Other requests occur as usual.
mounts = {"all://example.org": MockTransport()}
def handler(request):
return httpx.Response(200, json={"text": "Hello, World!"})
mounts = {"all://example.org": httpx.MockTransport(handler)}
client = httpx.Client(mounts=mounts)
```

2
httpx/__init__.py
View File

@ -36,6 +36,7 @@ from ._exceptions import (
from ._models import URL, Cookies, Headers, QueryParams, Request, Response
from ._status_codes import StatusCode, codes
from ._transports.asgi import ASGITransport
from ._transports.mock import MockTransport
from ._transports.wsgi import WSGITransport
__all__ = [
@ -65,6 +66,7 @@ __all__ = [
"InvalidURL",
"Limits",
"LocalProtocolError",
"MockTransport",
"NetworkError",
"options",
"patch",

56
httpx/_transports/mock.py Normal file
View File

@ -0,0 +1,56 @@
from typing import Callable, List, Optional, Tuple
import httpcore
from .._models import Request
class MockTransport(httpcore.SyncHTTPTransport, httpcore.AsyncHTTPTransport):
def __init__(self, handler: Callable) -> None:
self.handler = handler
def request(
self,
method: bytes,
url: Tuple[bytes, bytes, Optional[int], bytes],
headers: List[Tuple[bytes, bytes]] = None,
stream: httpcore.SyncByteStream = None,
ext: dict = None,
) -> Tuple[int, List[Tuple[bytes, bytes]], httpcore.SyncByteStream, dict]:
request = Request(
method=method,
url=url,
headers=headers,
stream=stream,
)
request.read()
response = self.handler(request)
return (
response.status_code,
response.headers.raw,
response.stream,
response.ext,
)
async def arequest(
self,
method: bytes,
url: Tuple[bytes, bytes, Optional[int], bytes],
headers: List[Tuple[bytes, bytes]] = None,
stream: httpcore.AsyncByteStream = None,
ext: dict = None,
) -> Tuple[int, List[Tuple[bytes, bytes]], httpcore.AsyncByteStream, dict]:
request = Request(
method=method,
url=url,
headers=headers,
stream=stream,
)
await request.aread()
response = self.handler(request)
return (
response.status_code,
response.headers.raw,
response.stream,
response.ext,
)

11
tests/client/test_async_client.py
View File

@ -5,7 +5,6 @@ import httpcore
import pytest
import httpx
from tests.utils import MockTransport
@pytest.mark.usefixtures("async_environment")
@ -247,7 +246,7 @@ def hello_world(request):
@pytest.mark.usefixtures("async_environment")
async def test_client_closed_state_using_implicit_open():
client = httpx.AsyncClient(transport=MockTransport(hello_world))
client = httpx.AsyncClient(transport=httpx.MockTransport(hello_world))
assert not client.is_closed
await client.get("http://example.com")
@ -262,7 +261,7 @@ async def test_client_closed_state_using_implicit_open():
@pytest.mark.usefixtures("async_environment")
async def test_client_closed_state_using_with_block():
async with httpx.AsyncClient(transport=MockTransport(hello_world)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(hello_world)) as client:
assert not client.is_closed
await client.get("http://example.com")
@ -273,7 +272,7 @@ async def test_client_closed_state_using_with_block():
@pytest.mark.usefixtures("async_environment")
async def test_deleting_unclosed_async_client_causes_warning():
client = httpx.AsyncClient(transport=MockTransport(hello_world))
client = httpx.AsyncClient(transport=httpx.MockTransport(hello_world))
await client.get("http://example.com")
with pytest.warns(UserWarning):
del client
@ -291,8 +290,8 @@ def mounted(request: httpx.Request) -> httpx.Response:
@pytest.mark.usefixtures("async_environment")
async def test_mounted_transport():
transport = MockTransport(unmounted)
mounts = {"custom://": MockTransport(mounted)}
transport = httpx.MockTransport(unmounted)
mounts = {"custom://": httpx.MockTransport(mounted)}
async with httpx.AsyncClient(transport=transport, mounts=mounts) as client:
response = await client.get("https://www.example.com")

73
tests/client/test_auth.py
View File

@ -13,7 +13,6 @@ import pytest
import httpx
from httpx import URL, Auth, BasicAuth, DigestAuth, ProtocolError, Request, Response
from tests.utils import MockTransport
from ..common import FIXTURES_DIR
@ -155,7 +154,7 @@ async def test_basic_auth() -> None:
auth = ("tomchristie", "password123")
app = App()
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
response = await client.get(url, auth=auth)
assert response.status_code == 200
@ -171,7 +170,9 @@ async def test_basic_auth_with_stream() -> None:
auth = ("tomchristie", "password123")
app = App()
async with httpx.AsyncClient(transport=MockTransport(app), auth=auth) as client:
async with httpx.AsyncClient(
transport=httpx.MockTransport(app), auth=auth
) as client:
async with client.stream("GET", url) as response:
await response.aread()
@ -184,7 +185,7 @@ async def test_basic_auth_in_url() -> None:
url = "https://tomchristie:password123@example.org/"
app = App()
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
response = await client.get(url)
assert response.status_code == 200
@ -197,7 +198,9 @@ async def test_basic_auth_on_session() -> None:
auth = ("tomchristie", "password123")
app = App()
async with httpx.AsyncClient(transport=MockTransport(app), auth=auth) as client:
async with httpx.AsyncClient(
transport=httpx.MockTransport(app), auth=auth
) as client:
response = await client.get(url)
assert response.status_code == 200
@ -213,7 +216,7 @@ async def test_custom_auth() -> None:
request.headers["Authorization"] = "Token 123"
return request
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
response = await client.get(url, auth=auth)
assert response.status_code == 200
@ -226,7 +229,7 @@ async def test_netrc_auth() -> None:
url = "http://netrcexample.org"
app = App()
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
response = await client.get(url)
assert response.status_code == 200
@ -241,7 +244,7 @@ async def test_auth_header_has_priority_over_netrc() -> None:
url = "http://netrcexample.org"
app = App()
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
response = await client.get(url, headers={"Authorization": "Override"})
assert response.status_code == 200
@ -255,7 +258,7 @@ async def test_trust_env_auth() -> None:
app = App()
async with httpx.AsyncClient(
transport=MockTransport(app), trust_env=False
transport=httpx.MockTransport(app), trust_env=False
) as client:
response = await client.get(url)
@ -263,7 +266,7 @@ async def test_trust_env_auth() -> None:
assert response.json() == {"auth": None}
async with httpx.AsyncClient(
transport=MockTransport(app), trust_env=True
transport=httpx.MockTransport(app), trust_env=True
) as client:
response = await client.get(url)
@ -279,7 +282,9 @@ async def test_auth_disable_per_request() -> None:
auth = ("tomchristie", "password123")
app = App()
async with httpx.AsyncClient(transport=MockTransport(app), auth=auth) as client:
async with httpx.AsyncClient(
transport=httpx.MockTransport(app), auth=auth
) as client:
response = await client.get(url, auth=None)
assert response.status_code == 200
@ -299,7 +304,7 @@ async def test_auth_hidden_header() -> None:
auth = ("example-username", "example-password")
app = App()
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
response = await client.get(url, auth=auth)
assert "'authorization': '[secure]'" in str(response.request.headers)
@ -309,7 +314,7 @@ async def test_auth_hidden_header() -> None:
async def test_auth_property() -> None:
app = App()
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
assert client.auth is None
client.auth = ("tomchristie", "password123") # type: ignore
@ -327,11 +332,11 @@ async def test_auth_invalid_type() -> None:
with pytest.raises(TypeError):
client = httpx.AsyncClient(
transport=MockTransport(app),
transport=httpx.MockTransport(app),
auth="not a tuple, not a callable", # type: ignore
)
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
with pytest.raises(TypeError):
await client.get(auth="not a tuple, not a callable") # type: ignore
@ -345,7 +350,7 @@ async def test_digest_auth_returns_no_auth_if_no_digest_header_in_response() ->
auth = DigestAuth(username="tomchristie", password="password123")
app = App()
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
response = await client.get(url, auth=auth)
assert response.status_code == 200
@ -359,7 +364,7 @@ def test_digest_auth_returns_no_auth_if_alternate_auth_scheme() -> None:
auth_header = "Token ..."
app = App(auth_header=auth_header, status_code=401)
client = httpx.Client(transport=MockTransport(app))
client = httpx.Client(transport=httpx.MockTransport(app))
response = client.get(url, auth=auth)
assert response.status_code == 401
@ -374,7 +379,7 @@ async def test_digest_auth_200_response_including_digest_auth_header() -> None:
auth_header = 'Digest realm="realm@host.com",qop="auth",nonce="abc",opaque="xyz"'
app = App(auth_header=auth_header, status_code=200)
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
response = await client.get(url, auth=auth)
assert response.status_code == 200
@ -388,7 +393,7 @@ async def test_digest_auth_401_response_without_digest_auth_header() -> None:
auth = DigestAuth(username="tomchristie", password="password123")
app = App(auth_header="", status_code=401)
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
response = await client.get(url, auth=auth)
assert response.status_code == 401
@ -417,7 +422,7 @@ async def test_digest_auth(
auth = DigestAuth(username="tomchristie", password="password123")
app = DigestApp(algorithm=algorithm)
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
response = await client.get(url, auth=auth)
assert response.status_code == 200
@ -448,7 +453,7 @@ async def test_digest_auth_no_specified_qop() -> None:
auth = DigestAuth(username="tomchristie", password="password123")
app = DigestApp(qop="")
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
response = await client.get(url, auth=auth)
assert response.status_code == 200
@ -480,7 +485,7 @@ async def test_digest_auth_qop_including_spaces_and_auth_returns_auth(qop: str)
auth = DigestAuth(username="tomchristie", password="password123")
app = DigestApp(qop=qop)
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
response = await client.get(url, auth=auth)
assert response.status_code == 200
@ -493,7 +498,7 @@ async def test_digest_auth_qop_auth_int_not_implemented() -> None:
auth = DigestAuth(username="tomchristie", password="password123")
app = DigestApp(qop="auth-int")
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
with pytest.raises(NotImplementedError):
await client.get(url, auth=auth)
@ -504,7 +509,7 @@ async def test_digest_auth_qop_must_be_auth_or_auth_int() -> None:
auth = DigestAuth(username="tomchristie", password="password123")
app = DigestApp(qop="not-auth")
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
with pytest.raises(ProtocolError):
await client.get(url, auth=auth)
@ -515,7 +520,7 @@ async def test_digest_auth_incorrect_credentials() -> None:
auth = DigestAuth(username="tomchristie", password="password123")
app = DigestApp(send_response_after_attempt=2)
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
response = await client.get(url, auth=auth)
assert response.status_code == 401
@ -537,7 +542,7 @@ async def test_async_digest_auth_raises_protocol_error_on_malformed_header(
auth = DigestAuth(username="tomchristie", password="password123")
app = App(auth_header=auth_header, status_code=401)
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
with pytest.raises(ProtocolError):
await client.get(url, auth=auth)
@ -556,7 +561,7 @@ def test_sync_digest_auth_raises_protocol_error_on_malformed_header(
auth = DigestAuth(username="tomchristie", password="password123")
app = App(auth_header=auth_header, status_code=401)
with httpx.Client(transport=MockTransport(app)) as client:
with httpx.Client(transport=httpx.MockTransport(app)) as client:
with pytest.raises(ProtocolError):
client.get(url, auth=auth)
@ -571,7 +576,7 @@ async def test_async_auth_history() -> None:
auth = RepeatAuth(repeat=2)
app = App(auth_header="abc")
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
response = await client.get(url, auth=auth)
assert response.status_code == 200
@ -597,7 +602,7 @@ def test_sync_auth_history() -> None:
auth = RepeatAuth(repeat=2)
app = App(auth_header="abc")
with httpx.Client(transport=MockTransport(app)) as client:
with httpx.Client(transport=httpx.MockTransport(app)) as client:
response = client.get(url, auth=auth)
assert response.status_code == 200
@ -623,7 +628,7 @@ async def test_digest_auth_unavailable_streaming_body():
async def streaming_body():
yield b"Example request body" # pragma: nocover
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
with pytest.raises(httpx.StreamConsumed):
await client.post(url, data=streaming_body(), auth=auth)
@ -638,7 +643,7 @@ async def test_async_auth_reads_response_body() -> None:
auth = ResponseBodyAuth("xyz")
app = App()
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
response = await client.get(url, auth=auth)
assert response.status_code == 200
@ -654,7 +659,7 @@ def test_sync_auth_reads_response_body() -> None:
auth = ResponseBodyAuth("xyz")
app = App()
with httpx.Client(transport=MockTransport(app)) as client:
with httpx.Client(transport=httpx.MockTransport(app)) as client:
response = client.get(url, auth=auth)
assert response.status_code == 200
@ -672,7 +677,7 @@ async def test_async_auth() -> None:
auth = SyncOrAsyncAuth()
app = App()
async with httpx.AsyncClient(transport=MockTransport(app)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
response = await client.get(url, auth=auth)
assert response.status_code == 200
@ -687,7 +692,7 @@ def test_sync_auth() -> None:
auth = SyncOrAsyncAuth()
app = App()
with httpx.Client(transport=MockTransport(app)) as client:
with httpx.Client(transport=httpx.MockTransport(app)) as client:
response = client.get(url, auth=auth)
assert response.status_code == 200

15
tests/client/test_client.py
View File

@ -5,7 +5,6 @@ import httpcore
import pytest
import httpx
from tests.utils import MockTransport
def test_get(server):
@ -292,7 +291,7 @@ def hello_world(request):
def test_client_closed_state_using_implicit_open():
client = httpx.Client(transport=MockTransport(hello_world))
client = httpx.Client(transport=httpx.MockTransport(hello_world))
assert not client.is_closed
client.get("http://example.com")
@ -306,7 +305,7 @@ def test_client_closed_state_using_implicit_open():
def test_client_closed_state_using_with_block():
with httpx.Client(transport=MockTransport(hello_world)) as client:
with httpx.Client(transport=httpx.MockTransport(hello_world)) as client:
assert not client.is_closed
client.get("http://example.com")
@ -330,7 +329,9 @@ def test_raw_client_header():
url = "http://example.org/echo_headers"
headers = {"Example-Header": "example-value"}
client = httpx.Client(transport=MockTransport(echo_raw_headers), headers=headers)
client = httpx.Client(
transport=httpx.MockTransport(echo_raw_headers), headers=headers
)
response = client.get(url)
assert response.status_code == 200
@ -355,8 +356,8 @@ def mounted(request: httpx.Request) -> httpx.Response:
def test_mounted_transport():
transport = MockTransport(unmounted)
mounts = {"custom://": MockTransport(mounted)}
transport = httpx.MockTransport(unmounted)
mounts = {"custom://": httpx.MockTransport(mounted)}
client = httpx.Client(transport=transport, mounts=mounts)
@ -370,7 +371,7 @@ def test_mounted_transport():
def test_all_mounted_transport():
mounts = {"all://": MockTransport(mounted)}
mounts = {"all://": httpx.MockTransport(mounted)}
client = httpx.Client(mounts=mounts)

13
tests/client/test_cookies.py
View File

@ -1,7 +1,6 @@
from http.cookiejar import Cookie, CookieJar
import httpx
from tests.utils import MockTransport
def get_and_set_cookies(request: httpx.Request) -> httpx.Response:
@ -21,7 +20,7 @@ def test_set_cookie() -> None:
url = "http://example.org/echo_cookies"
cookies = {"example-name": "example-value"}
client = httpx.Client(transport=MockTransport(get_and_set_cookies))
client = httpx.Client(transport=httpx.MockTransport(get_and_set_cookies))
response = client.get(url, cookies=cookies)
assert response.status_code == 200
@ -56,7 +55,7 @@ def test_set_cookie_with_cookiejar() -> None:
)
cookies.set_cookie(cookie)
client = httpx.Client(transport=MockTransport(get_and_set_cookies))
client = httpx.Client(transport=httpx.MockTransport(get_and_set_cookies))
response = client.get(url, cookies=cookies)
assert response.status_code == 200
@ -91,7 +90,7 @@ def test_setting_client_cookies_to_cookiejar() -> None:
)
cookies.set_cookie(cookie)
client = httpx.Client(transport=MockTransport(get_and_set_cookies))
client = httpx.Client(transport=httpx.MockTransport(get_and_set_cookies))
client.cookies = cookies # type: ignore
response = client.get(url)
@ -108,7 +107,7 @@ def test_set_cookie_with_cookies_model() -> None:
cookies = httpx.Cookies()
cookies["example-name"] = "example-value"
client = httpx.Client(transport=MockTransport(get_and_set_cookies))
client = httpx.Client(transport=httpx.MockTransport(get_and_set_cookies))
response = client.get(url, cookies=cookies)
assert response.status_code == 200
@ -118,7 +117,7 @@ def test_set_cookie_with_cookies_model() -> None:
def test_get_cookie() -> None:
url = "http://example.org/set_cookie"
client = httpx.Client(transport=MockTransport(get_and_set_cookies))
client = httpx.Client(transport=httpx.MockTransport(get_and_set_cookies))
response = client.get(url)
assert response.status_code == 200
@ -130,7 +129,7 @@ def test_cookie_persistence() -> None:
"""
Ensure that Client instances persist cookies between requests.
"""
client = httpx.Client(transport=MockTransport(get_and_set_cookies))
client = httpx.Client(transport=httpx.MockTransport(get_and_set_cookies))
response = client.get("http://example.org/echo_cookies")
assert response.status_code == 200

19
tests/client/test_event_hooks.py
View File

@ -1,7 +1,6 @@
import pytest
import httpx
from tests.utils import MockTransport
def app(request: httpx.Request) -> httpx.Response:
@ -25,7 +24,9 @@ def test_event_hooks():
event_hooks = {"request": [on_request], "response": [on_response]}
with httpx.Client(event_hooks=event_hooks, transport=MockTransport(app)) as http:
with httpx.Client(
event_hooks=event_hooks, transport=httpx.MockTransport(app)
) as http:
http.get("http://127.0.0.1:8000/", auth=("username", "password"))
assert events == [
@ -53,7 +54,9 @@ def test_event_hooks_raising_exception(server):
event_hooks = {"response": [raise_on_4xx_5xx]}
with httpx.Client(event_hooks=event_hooks, transport=MockTransport(app)) as http:
with httpx.Client(
event_hooks=event_hooks, transport=httpx.MockTransport(app)
) as http:
try:
http.get("http://127.0.0.1:8000/status/400")
except httpx.HTTPStatusError as exc:
@ -73,7 +76,7 @@ async def test_async_event_hooks():
event_hooks = {"request": [on_request], "response": [on_response]}
async with httpx.AsyncClient(
event_hooks=event_hooks, transport=MockTransport(app)
event_hooks=event_hooks, transport=httpx.MockTransport(app)
) as http:
await http.get("http://127.0.0.1:8000/", auth=("username", "password"))
@ -104,7 +107,7 @@ async def test_async_event_hooks_raising_exception():
event_hooks = {"response": [raise_on_4xx_5xx]}
async with httpx.AsyncClient(
event_hooks=event_hooks, transport=MockTransport(app)
event_hooks=event_hooks, transport=httpx.MockTransport(app)
) as http:
try:
await http.get("http://127.0.0.1:8000/status/400")
@ -127,7 +130,9 @@ def test_event_hooks_with_redirect():
event_hooks = {"request": [on_request], "response": [on_response]}
with httpx.Client(event_hooks=event_hooks, transport=MockTransport(app)) as http:
with httpx.Client(
event_hooks=event_hooks, transport=httpx.MockTransport(app)
) as http:
http.get("http://127.0.0.1:8000/redirect", auth=("username", "password"))
assert events == [
@ -166,7 +171,7 @@ async def test_async_event_hooks_with_redirect():
event_hooks = {"request": [on_request], "response": [on_response]}
async with httpx.AsyncClient(
event_hooks=event_hooks, transport=MockTransport(app)
event_hooks=event_hooks, transport=httpx.MockTransport(app)
) as http:
await http.get("http://127.0.0.1:8000/redirect", auth=("username", "password"))

19
tests/client/test_headers.py
View File

@ -3,7 +3,6 @@
import pytest
import httpx
from tests.utils import MockTransport
def echo_headers(request: httpx.Request) -> httpx.Response:
@ -18,7 +17,7 @@ def test_client_header():
url = "http://example.org/echo_headers"
headers = {"Example-Header": "example-value"}
client = httpx.Client(transport=MockTransport(echo_headers), headers=headers)
client = httpx.Client(transport=httpx.MockTransport(echo_headers), headers=headers)
response = client.get(url)
assert response.status_code == 200
@ -38,7 +37,9 @@ def test_header_merge():
url = "http://example.org/echo_headers"
client_headers = {"User-Agent": "python-myclient/0.2.1"}
request_headers = {"X-Auth-Token": "FooBarBazToken"}
client = httpx.Client(transport=MockTransport(echo_headers), headers=client_headers)
client = httpx.Client(
transport=httpx.MockTransport(echo_headers), headers=client_headers
)
response = client.get(url, headers=request_headers)
assert response.status_code == 200
@ -58,7 +59,9 @@ def test_header_merge_conflicting_headers():
url = "http://example.org/echo_headers"
client_headers = {"X-Auth-Token": "FooBar"}
request_headers = {"X-Auth-Token": "BazToken"}
client = httpx.Client(transport=MockTransport(echo_headers), headers=client_headers)
client = httpx.Client(
transport=httpx.MockTransport(echo_headers), headers=client_headers
)
response = client.get(url, headers=request_headers)
assert response.status_code == 200
@ -76,7 +79,7 @@ def test_header_merge_conflicting_headers():
def test_header_update():
url = "http://example.org/echo_headers"
client = httpx.Client(transport=MockTransport(echo_headers))
client = httpx.Client(transport=httpx.MockTransport(echo_headers))
first_response = client.get(url)
client.headers.update(
{"User-Agent": "python-myclient/0.2.1", "Another-Header": "AThing"}
@ -113,7 +116,7 @@ def test_remove_default_header():
"""
url = "http://example.org/echo_headers"
client = httpx.Client(transport=MockTransport(echo_headers))
client = httpx.Client(transport=httpx.MockTransport(echo_headers))
del client.headers["User-Agent"]
response = client.get(url)
@ -143,7 +146,7 @@ def test_host_with_auth_and_port_in_url():
"""
url = "http://username:password@example.org:80/echo_headers"
client = httpx.Client(transport=MockTransport(echo_headers))
client = httpx.Client(transport=httpx.MockTransport(echo_headers))
response = client.get(url)
assert response.status_code == 200
@ -166,7 +169,7 @@ def test_host_with_non_default_port_in_url():
"""
url = "http://username:password@example.org:123/echo_headers"
client = httpx.Client(transport=MockTransport(echo_headers))
client = httpx.Client(transport=httpx.MockTransport(echo_headers))
response = client.get(url)
assert response.status_code == 200

3
tests/client/test_queryparams.py
View File

@ -1,5 +1,4 @@
import httpx
from tests.utils import MockTransport
def hello_world(request: httpx.Request) -> httpx.Response:
@ -28,7 +27,7 @@ def test_client_queryparams_echo():
client_queryparams = "first=str"
request_queryparams = {"second": "dict"}
client = httpx.Client(
transport=MockTransport(hello_world), params=client_queryparams
transport=httpx.MockTransport(hello_world), params=client_queryparams
)
response = client.get(url, params=request_queryparams)

51
tests/client/test_redirects.py
View File

@ -2,7 +2,6 @@ import httpcore
import pytest
import httpx
from tests.utils import MockTransport
def redirects(request: httpx.Request) -> httpx.Response:
@ -116,7 +115,7 @@ def redirects(request: httpx.Request) -> httpx.Response:
def test_redirect_301():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
response = client.post("https://example.org/redirect_301")
assert response.status_code == httpx.codes.OK
assert response.url == "https://example.org/"
@ -124,7 +123,7 @@ def test_redirect_301():
def test_redirect_302():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
response = client.post("https://example.org/redirect_302")
assert response.status_code == httpx.codes.OK
assert response.url == "https://example.org/"
@ -132,7 +131,7 @@ def test_redirect_302():
def test_redirect_303():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
response = client.get("https://example.org/redirect_303")
assert response.status_code == httpx.codes.OK
assert response.url == "https://example.org/"
@ -140,7 +139,7 @@ def test_redirect_303():
def test_next_request():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
request = client.build_request("POST", "https://example.org/redirect_303")
response = client.send(request, allow_redirects=False)
assert response.status_code == httpx.codes.SEE_OTHER
@ -155,7 +154,7 @@ def test_next_request():
@pytest.mark.usefixtures("async_environment")
async def test_async_next_request():
async with httpx.AsyncClient(transport=MockTransport(redirects)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
request = client.build_request("POST", "https://example.org/redirect_303")
response = await client.send(request, allow_redirects=False)
assert response.status_code == httpx.codes.SEE_OTHER
@ -172,7 +171,7 @@ def test_head_redirect():
"""
Contrary to Requests, redirects remain enabled by default for HEAD requests.
"""
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
response = client.head("https://example.org/redirect_302")
assert response.status_code == httpx.codes.OK
assert response.url == "https://example.org/"
@ -182,7 +181,7 @@ def test_head_redirect():
def test_relative_redirect():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
response = client.get("https://example.org/relative_redirect")
assert response.status_code == httpx.codes.OK
assert response.url == "https://example.org/"
@ -191,7 +190,7 @@ def test_relative_redirect():
def test_malformed_redirect():
# https://github.com/encode/httpx/issues/771
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
response = client.get("http://example.org/malformed_redirect")
assert response.status_code == httpx.codes.OK
assert response.url == "https://example.org:443/"
@ -199,13 +198,13 @@ def test_malformed_redirect():
def test_invalid_redirect():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
with pytest.raises(httpx.RemoteProtocolError):
client.get("http://example.org/invalid_redirect")
def test_no_scheme_redirect():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
response = client.get("https://example.org/no_scheme_redirect")
assert response.status_code == httpx.codes.OK
assert response.url == "https://example.org/"
@ -213,7 +212,7 @@ def test_no_scheme_redirect():
def test_fragment_redirect():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
response = client.get("https://example.org/relative_redirect#fragment")
assert response.status_code == httpx.codes.OK
assert response.url == "https://example.org/#fragment"
@ -221,7 +220,7 @@ def test_fragment_redirect():
def test_multiple_redirects():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
response = client.get("https://example.org/multiple_redirects?count=20")
assert response.status_code == httpx.codes.OK
assert response.url == "https://example.org/multiple_redirects"
@ -234,25 +233,25 @@ def test_multiple_redirects():
@pytest.mark.usefixtures("async_environment")
async def test_async_too_many_redirects():
async with httpx.AsyncClient(transport=MockTransport(redirects)) as client:
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
with pytest.raises(httpx.TooManyRedirects):
await client.get("https://example.org/multiple_redirects?count=21")
def test_sync_too_many_redirects():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
with pytest.raises(httpx.TooManyRedirects):
client.get("https://example.org/multiple_redirects?count=21")
def test_redirect_loop():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
with pytest.raises(httpx.TooManyRedirects):
client.get("https://example.org/redirect_loop")
def test_cross_domain_redirect_with_auth_header():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
url = "https://example.com/cross_domain"
headers = {"Authorization": "abc"}
response = client.get(url, headers=headers)
@ -261,7 +260,7 @@ def test_cross_domain_redirect_with_auth_header():
def test_cross_domain_redirect_with_auth():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
url = "https://example.com/cross_domain"
response = client.get(url, auth=("user", "pass"))
assert response.url == "https://example.org/cross_domain_target"
@ -269,7 +268,7 @@ def test_cross_domain_redirect_with_auth():
def test_same_domain_redirect():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
url = "https://example.org/cross_domain"
headers = {"Authorization": "abc"}
response = client.get(url, headers=headers)
@ -281,7 +280,7 @@ def test_body_redirect():
"""
A 308 redirect should preserve the request body.
"""
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
url = "https://example.org/redirect_body"
content = b"Example request body"
response = client.post(url, content=content)
@ -294,7 +293,7 @@ def test_no_body_redirect():
"""
A 303 redirect should remove the request body.
"""
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
url = "https://example.org/redirect_no_body"
content = b"Example request body"
response = client.post(url, content=content)
@ -304,7 +303,7 @@ def test_no_body_redirect():
def test_can_stream_if_no_redirect():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
url = "https://example.org/redirect_301"
with client.stream("GET", url, allow_redirects=False) as response:
assert not response.is_closed
@ -313,7 +312,7 @@ def test_can_stream_if_no_redirect():
def test_cannot_redirect_streaming_body():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
url = "https://example.org/redirect_body"
def streaming_body():
@ -324,7 +323,7 @@ def test_cannot_redirect_streaming_body():
def test_cross_subdomain_redirect():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
url = "https://example.com/cross_subdomain"
response = client.get(url)
assert response.url == "https://www.example.org/cross_subdomain"
@ -364,7 +363,7 @@ def cookie_sessions(request: httpx.Request) -> httpx.Response:
def test_redirect_cookie_behavior():
client = httpx.Client(transport=MockTransport(cookie_sessions))
client = httpx.Client(transport=httpx.MockTransport(cookie_sessions))
# The client is not logged in.
response = client.get("https://example.com/")
@ -393,7 +392,7 @@ def test_redirect_cookie_behavior():
def test_redirect_custom_scheme():
client = httpx.Client(transport=MockTransport(redirects))
client = httpx.Client(transport=httpx.MockTransport(redirects))
with pytest.raises(httpx.UnsupportedProtocol) as e:
client.post("https://example.org/redirect_custom_scheme")
assert str(e.value) == "Scheme 'market' not supported."

9
tests/test_multipart.py
View File

@ -9,7 +9,6 @@ import pytest
import httpx
from httpx._content import encode_request
from httpx._utils import format_form_param
from tests.utils import MockTransport
def echo_request_content(request: httpx.Request) -> httpx.Response:
@ -18,7 +17,7 @@ def echo_request_content(request: httpx.Request) -> httpx.Response:
@pytest.mark.parametrize(("value,output"), (("abc", b"abc"), (b"abc", b"abc")))
def test_multipart(value, output):
client = httpx.Client(transport=MockTransport(echo_request_content))
client = httpx.Client(transport=httpx.MockTransport(echo_request_content))
# Test with a single-value 'data' argument, and a plain file 'files' argument.
data = {"text": value}
@ -44,7 +43,7 @@ def test_multipart(value, output):
@pytest.mark.parametrize(("key"), (b"abc", 1, 2.3, None))
def test_multipart_invalid_key(key):
client = httpx.Client(transport=MockTransport(echo_request_content))
client = httpx.Client(transport=httpx.MockTransport(echo_request_content))
data = {key: "abc"}
files = {"file": io.BytesIO(b"<file content>")}
@ -60,7 +59,7 @@ def test_multipart_invalid_key(key):
@pytest.mark.parametrize(("value"), (1, 2.3, None, [None, "abc"], {None: "abc"}))
def test_multipart_invalid_value(value):
client = httpx.Client(transport=MockTransport(echo_request_content))
client = httpx.Client(transport=httpx.MockTransport(echo_request_content))
data = {"text": value}
files = {"file": io.BytesIO(b"<file content>")}
@ -70,7 +69,7 @@ def test_multipart_invalid_value(value):
def test_multipart_file_tuple():
client = httpx.Client(transport=MockTransport(echo_request_content))
client = httpx.Client(transport=httpx.MockTransport(echo_request_content))
# Test with a list of values 'data' argument,
# and a tuple style 'files' argument.

55
tests/utils.py
View File

@ -1,11 +1,7 @@
import contextlib
import logging
import os
from typing import Callable, List, Optional, Tuple
import httpcore
import httpx
from httpx import _utils
@ -22,54 +18,3 @@ def override_log_level(log_level: str):
finally:
# Reset the logger so we don't have verbose output in all unit tests
logging.getLogger("httpx").handlers = []
class MockTransport(httpcore.SyncHTTPTransport, httpcore.AsyncHTTPTransport):
def __init__(self, handler: Callable) -> None:
self.handler = handler
def request(
self,
method: bytes,
url: Tuple[bytes, bytes, Optional[int], bytes],
headers: List[Tuple[bytes, bytes]] = None,
stream: httpcore.SyncByteStream = None,
ext: dict = None,
) -> Tuple[int, List[Tuple[bytes, bytes]], httpcore.SyncByteStream, dict]:
request = httpx.Request(
method=method,
url=url,
headers=headers,
stream=stream,
)
request.read()
response = self.handler(request)
return (
response.status_code,
response.headers.raw,
response.stream,
response.ext,
)
async def arequest(
self,
method: bytes,
url: Tuple[bytes, bytes, Optional[int], bytes],
headers: List[Tuple[bytes, bytes]] = None,
stream: httpcore.AsyncByteStream = None,
ext: dict = None,
) -> Tuple[int, List[Tuple[bytes, bytes]], httpcore.AsyncByteStream, dict]:
request = httpx.Request(
method=method,
url=url,
headers=headers,
stream=stream,
)
await request.aread()
response = self.handler(request)
return (
response.status_code,
response.headers.raw,
response.stream,
response.ext,
)