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

Make all the tests from test_headers and test_event_hooks to be async

Browse Source
This commit is contained in:
Kar Petrosyan 2025-02-27 19:46:52 +04:00
parent c3f1ee1203
commit b1d5fca5ea
1 changed files with 3 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

112
tests/client/test_event_hooks.py
View File

@ -13,58 +13,8 @@ def app(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, headers={"server": "testserver"})
def test_event_hooks():
events = []
def on_request(request):
events.append({"event": "request", "headers": dict(request.headers)})
def on_response(response):
events.append({"event": "response", "headers": dict(response.headers)})
event_hooks = {"request": [on_request], "response": [on_response]}
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 == [
{
"event": "request",
"headers": {
"host": "127.0.0.1:8000",
"user-agent": f"python-httpx/{httpx.__version__}",
"accept": "*/*",
"accept-encoding": "gzip, deflate, br, zstd",
"connection": "keep-alive",
"authorization": "Basic dXNlcm5hbWU6cGFzc3dvcmQ=",
},
},
{
"event": "response",
"headers": {"server": "testserver"},
},
]
def test_event_hooks_raising_exception(server):
def raise_on_4xx_5xx(response):
response.raise_for_status()
event_hooks = {"response": [raise_on_4xx_5xx]}
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:
assert exc.response.is_closed
@pytest.mark.anyio
async def test_async_event_hooks():
async def test_event_hooks():
events = []
async def on_request(request):
@ -100,7 +50,7 @@ async def test_async_event_hooks():
@pytest.mark.anyio
async def test_async_event_hooks_raising_exception():
async def test_event_hooks_raising_exception():
async def raise_on_4xx_5xx(response):
response.raise_for_status()
@ -115,64 +65,8 @@ async def test_async_event_hooks_raising_exception():
assert exc.response.is_closed
def test_event_hooks_with_redirect():
"""
A redirect request should trigger additional 'request' and 'response' event hooks.
"""
events = []
def on_request(request):
events.append({"event": "request", "headers": dict(request.headers)})
def on_response(response):
events.append({"event": "response", "headers": dict(response.headers)})
event_hooks = {"request": [on_request], "response": [on_response]}
with httpx.Client(
event_hooks=event_hooks,
transport=httpx.MockTransport(app),
follow_redirects=True,
) as http:
http.get("http://127.0.0.1:8000/redirect", auth=("username", "password"))
assert events == [
{
"event": "request",
"headers": {
"host": "127.0.0.1:8000",
"user-agent": f"python-httpx/{httpx.__version__}",
"accept": "*/*",
"accept-encoding": "gzip, deflate, br, zstd",
"connection": "keep-alive",
"authorization": "Basic dXNlcm5hbWU6cGFzc3dvcmQ=",
},
},
{
"event": "response",
"headers": {"location": "/", "server": "testserver"},
},
{
"event": "request",
"headers": {
"host": "127.0.0.1:8000",
"user-agent": f"python-httpx/{httpx.__version__}",
"accept": "*/*",
"accept-encoding": "gzip, deflate, br, zstd",
"connection": "keep-alive",
"authorization": "Basic dXNlcm5hbWU6cGFzc3dvcmQ=",
},
},
{
"event": "response",
"headers": {"server": "testserver"},
},
]
@pytest.mark.anyio
async def test_async_event_hooks_with_redirect():
async def test_event_hooks_with_redirect():
"""
A redirect request should trigger additional 'request' and 'response' event hooks.
"""