pytest-httpx插件介绍及使用
简介：
pytest-httpx 是一个专为 pytest 测试框架设计的插件，用于 mock（模拟）使用 httpx 发起的 HTTP 请求。它非常适合在测试中拦截真实的网络请求，以提高测试速度和稳定性。
其中，提供了一个名为 httpx_mock 的 fixture，可以在测试函数中使用它来 mock 请求。
准备：
首先需要安装插件
pip install pytest-httpx
另外还要确认 httpx 和 pytest 框架是否安装，没有的话需要安装！！！
pip install httpx pytest
关键函数：
1、httpx_mock.add_response()
- 用途：添加一个 mock 响应来匹配特定的请求。
-
常用参数：
method: 请求方法（如 "GET", "POST"）。url: 请求的 URL。match_headers: 匹配请求头中的键值对。match_content: 匹配请求体（如 form 表单、纯文本、JSON、XML 等）match_json: 匹配请求体中的 JSON 数据，相比match_content更优，不关注key顺序。match_files: 匹配文件上传设计，支持文件名 + 内容匹配status_code: 返回的状态码，默认是 200。json: 返回的 JSON 数据。text: 返回的文本内容。exc: 抛出的异常。delay: 模拟网络延迟的时间。
表单（form）请求
def test_form_post(httpx_mock):
# 模拟响应
httpx_mock.add_response(
method="POST",
url="https://example.com/login",
match_content=b"username=test&password=123456",
text="Login Success"
)
with httpx.Client() as client:
resp = client.post(
"https://example.com/login",
data={"username": "test", "password": "123456"}
)
assert resp.text == "Login Success"
json 请求
推荐 test_json_post 这个函数的写法！
def test_json_post(httpx_mock):
# 设置 mock：使用 match_json 匹配 JSON 请求体
httpx_mock.add_response(
method="POST",
url="https://example.com/login",
match_json={"username": "test", "password": "123456"},
text="Login Success"
)
with httpx.Client() as client:
resp = client.post(
"https://example.com/login",
json={"username": "test", "password": "123456"} # 使用 json 参数
)
assert resp.text == "Login Success"
def test_json_post_with_match_content(httpx_mock):
# 使用 match_content 匹配原始 JSON 字节内容
httpx_mock.add_response(
method="POST",
url="https://example.com/login",
match_content=b'{"username":"test","password":"123456"}',
text="Login Success"
)
with httpx.Client() as client:
resp = client.post(
"https://example.com/login",
json={"username": "test", "password": "123456"}
)
assert resp.text == "Login Success"
2、httpx_mock.get_request()
- 用途：获取最近一次捕获到的请求对象，可用于验证请求是否符合预期。
- 返回值：最近一次捕获到的
Request对象。
request = httpx_mock.get_request()
assert request.url == "https://example.com/data"
# 验证请求是否发出
assert httpx_mock.get_request(url="https://example.com") is not None
3、httpx_mock.add_callback()
- 用途：添加一个回调函数，该函数将在匹配到请求时被调用，并根据回调函数的返回值生成响应。
- 使用场景：当需要更复杂的逻辑来决定如何响应请求时。
def custom_callback(request):
if request.url == "https://example.com/custom":
return httpx.Response(status_code=200, json={"custom": "response"})
else:
return httpx.Response(status_code=404)
httpx_mock.add_callback(custom_callback)
示例：
# 回调函数
def custom_callback(request):
if request.url == "https://example.com/custom":
return httpx.Response(status_code=200, json={"custom": "response"})
else:
return httpx.Response(status_code=404)
def fetch_data(url: str):
with httpx.Client() as client:
response = client.get(url)
response.raise_for_status()
return response.json()
def test_custom_callback(httpx_mock):
httpx_mock.add_callback(custom_callback)
# 成功响应测试
result = fetch_data("https://example.com/custom")
assert result == {"custom": "response"}
def test_custom_callback2(httpx_mock):
httpx_mock.add_callback(custom_callback)
# 错误响应测试
with pytest.raises(httpx.HTTPStatusError):
fetch_data("https://example.com/other")
4、httpx_mock.reset()
- 用途：重置所有已添加的 mock 响应和回调。
- 使用场景：在多个测试之间清除之前设置的 mock 配置。
大多数情况下，不需要手动调用 httpx_mock.reset()，因为 pytest-httpx 提供的 httpx_mock fixture 会在每个测试函数执行完后自动重置 mock 状态。
httpx_mock.reset()
5、httpx_mock.get_requests()
- 用途：获取所有被捕获的请求列表，这在你需要验证多个请求是否按预期发送时很有用。
def test_full_example(httpx_mock):
httpx_mock.add_response(url="https://example.com/data", text="Success")
with httpx.Client() as client:
resp1 = client.post(
"https://example.com/data",
json={"name": "Alice"},
headers={"Authorization": "Bearer token123"}
)
resp2 = client.get("https://example.com/data?id=1")
assert resp1.text == "Success"
assert resp2.text == "Success"
requests = httpx_mock.get_requests()
# 第一个请求
req1 = requests[0]
assert req1.method == "POST"
assert req1.url == "https://example.com/data"
assert req1.headers["authorization"] == "Bearer token123"
assert req1.json() == {"name": "Alice"}
# 第二个请求
req2 = requests[1]
assert req2.method == "GET"
assert req2.url == "https://example.com/data?id=1"
6、httpx_mock.add_exception()
虽然不是直接通过 httpx_mock 提供的方法，但你可以通过 add_response() 的 exc 参数来实现类似的效果，即模拟抛出异常的情况。
httpx_mock.add_response(exc=httpx.ConnectError("Failed to connect"))
示例1：模拟 httpx.ConnectError
import httpx
import pytest
def test_raises_connect_error(httpx_mock):
# 模拟连接错误
httpx_mock.add_response(
url="https://example.com/api",
exc=httpx.ConnectError("Connection refused")
)
with httpx.Client() as client:
with pytest.raises(httpx.ConnectError) as exc_info:
client.get("https://example.com/api")
# 可选：验证异常信息
assert "Connection refused" in str(exc_info.value)
示例2：模拟 httpx.TimeoutException
def test_raises_timeout(httpx_mock):
httpx_mock.add_response(
url="https://example.com/api",
exc=httpx.TimeoutException("Request timeout")
)
with httpx.Client() as client:
with pytest.raises(httpx.TimeoutException):
client.get("https://example.com/api")
常见异常：
| 异常类 | 说明 |
|---|---|
httpx.ConnectError |
连接失败（如服务器不可达） |
httpx.TimeoutException |
请求超时 |
httpx.ReadTimeout |
读取超时 |
httpx.WriteTimeout |
写入超时 |
httpx.PoolTimeout |
连接池超时 |
httpx.HTTPStatusError |
HTTP 响应状态码非 2xx（如 404、500） |
httpx.RequestError |
所有请求错误的基类 |
场景使用：
场景1：mock 一个 GET 请求
目的：模拟一个简单的 GET 请求并返回预定义的数据。
import pytest
import httpx
def test_basic_get_request(httpx_mock):
httpx_mock.add_response(url="https://example.com/data", json={"key": "value"})
with httpx.Client() as client:
response = client.get("https://example.com/data")
assert response.status_code == 200
assert response.json() == {"key": "value"}
场景2：mock 一个POST请求（JSON Data）
目的：验证 POST 请求发送的 JSON 数据，并根据数据内容返回不同的响应。
@pytest.mark.asyncio
async def test_post_json_match(httpx_mock):
httpx_mock.add_response(
method="POST",
url="https://api.example.com/create",
match_json={"id": 1},
json={"status": "created"}
)
async with httpx.AsyncClient() as client:
response = await client.post("https://api.example.com/create", json={"id": 1})
assert response.status_code == 200
assert response.json() == {"status": "created"}
场景3：mock 一个POST请求（Form Data）
目的：模拟一个表单提交，并验证其内容是否正确。
def test_form_submission(httpx_mock):
httpx_mock.add_response(
method="POST",
url="https://example.com/login",
match_content=b"username=admin&password=secret",
text="Login Successful"
)
with httpx.Client() as client:
response = client.post(
"https://example.com/login",
data={"username": "admin", "password": "secret"}
)
assert response.text == "Login Successful"
场景4：模拟异常情况（如连接错误）
目的：测试当服务器不可达时，代码是否能够正确处理异常。
def test_handle_connect_error(httpx_mock):
httpx_mock.add_response(
url="https://example.com/api",
exc=httpx.ConnectError("Connection refused")
)
with httpx.Client() as client:
with pytest.raises(httpx.ConnectError) as exc_info:
client.get("https://example.com/api")
assert "Connection refused" in str(exc_info.value)
场景5：模拟超时（Timeout）
目的：模拟请求超时的情况，确保代码中对超时有正确的处理逻辑。
def test_timeout_exception(httpx_mock):
httpx_mock.add_response(
url="https://example.com/api",
exc=httpx.TimeoutException("Request timeout")
)
with httpx.Client() as client:
with pytest.raises(httpx.TimeoutException):
client.get("https://example.com/api")
场景 6：模拟延迟响应
目的：模拟网络延迟，确保代码在高延迟情况下仍能正常工作。
import time
def test_delayed_response(httpx_mock):
httpx_mock.add_response(delay=1.0, text="Delayed Response")
start_time = time.time()
with httpx.Client() as client:
response = client.get("https://example.com")
end_time = time.time()
assert response.text == "Delayed Response"
assert end_time - start_time >= 1.0
场景 7：文件上传（Multipart Form Data）
目的：模拟文件上传过程，并验证上传的内容是否正确。
import pytest
import httpx
import tempfile
def test_file_upload(httpx_mock):
# 创建临时文件（自动管理生命周期）
with tempfile.NamedTemporaryFile(mode='w+b', suffix='.log') as tmpfile:
tmpfile.write(b"Hello World!\n")
tmpfile.seek(0) # seek(0) 将指针移回文件开头，确保后续能读取到完整内容。
# 设置mock（匹配基本文件上传）
httpx_mock.add_response(
method="POST",
url="https://example.com/upload",
match_files={"file": (tmpfile.name, b"Hello World!\n")},
text="Upload successful"
)
# 发起请求（使用简化格式）
with httpx.Client() as client:
response = client.post(
"https://example.com/upload",
files={"file": (tmpfile.name, tmpfile)}
)
assert response.text == "Upload successful"
场景 8：多次请求
目的：模拟多个连续的请求，并确保它们按顺序得到相应的响应。
def test_multiple_requests_in_order(httpx_mock):
httpx_mock.add_response(url="https://api.example.com/first", json={"data": "first"})
httpx_mock.add_response(url="https://api.example.com/second", json={"data": "second"})
with httpx.Client() as client:
resp1 = client.get("https://api.example.com/first")
resp2 = client.get("https://api.example.com/second")
assert resp1.json() == {"data": "first"}
assert resp2.json() == {"data": "second"}
场景 9：动态回调函数
目的：根据请求的不同参数，动态地决定返回什么样的响应。
def custom_callback(request):
if request.url.path == "/success":
return httpx.Response(200, json={"status": "ok"})
else:
return httpx.Response(404)
def test_custom_callback(httpx_mock):
httpx_mock.add_callback(custom_callback)
with httpx.Client() as client:
success_resp = client.get("https://example.com/success")
fail_resp = client.get("https://example.com/fail")
assert success_resp.status_code == 200
assert fail_resp.status_code == 404
场景 10：异步请求
目的：使用 AsyncClient 进行异步请求测试。
import pytest
import httpx
@pytest.mark.asyncio
async def test_async_request(httpx_mock):
httpx_mock.add_response(url="https://example.com/async", json={"status": "async ok"})
async with httpx.AsyncClient() as client:
response = await client.get("https://example.com/async")
assert response.status_code == 200
assert response.json() == {"status": "async ok"}
评论区