pytest + httpx 异步接口测试实践指南

目标读者：会用 requests 写接口测试、想升级到 httpx 异步模式的测试工程师 环境要求：Python 3.10+、pytest 7+、httpx 0.24+、respx（mock） 安装：pip install pytest httpx respx 运行：pytest -v

目录

1. 为什么用 httpx 替代 requests
2. 示例一：基础 GET / POST
3. 示例二：异步并发请求
4. 示例三：respx mock 外部依赖
5. 方案取舍对比
6. 常见坑与排错路径
7. 参考资料

1. 为什么用 httpx 替代 requests

1.1 异步收益

把接口测试从同步切到异步，核心收益不是"快一点"，而是在有限时间预算内跑更多用例。打个比方：同步测试像一个收银员一次只扫一件商品，异步测试像超市自助结算台——10 件商品同时扫码，总耗时取决于最慢那件，而不是累加。

具体到接口测试：

• 并发请求场景：一个测试要调 5 个上游接口再断言聚合结果，同步写法 5 × 200ms = 1s，异步写法 ≈ 200ms。
• CI 时长：200 个接口用例，同步平均 0.3s/个 = 60s；异步并发后 CI 时间可压缩 40%–70%。
• 超时感知：httpx 的 async with 上下文让超时异常（httpx.TimeoutException）自然归入 try/except，不用像 requests 那样手动判断 e.timeout。

1.2 API 差异速查

| 维度 | requests（同步） | httpx（异步） | |------|-----------------|--------------| | 客户端 | requests.Session() | httpx.AsyncClient() | | 发请求 | session.get(url) | await client.get(url) | | 上下文 | with requests.Session() as s: | async with httpx.AsyncClient() as c: | | 超时 | timeout=5 | timeout=httpx.Timeout(5) | | 流式响应 | r.iter_lines() | async for line in r.aiter_lines() | | Mock 生态 | responses | respx | | 事件循环 | 无 | 需要 pytest-asyncio 或 anyio |

1.3 生态兼容性

• httpx 兼容 requests API 风格：client.get/post/put/delete、response.json()、response.headers、response.raise_for_status() 几乎一一对应，迁移成本主要在加 async/await。
• requests 不支持异步：没有 aiohttp 那样的原生异步版本，只能换库。
• httpx 支持 HTTP/2：httpx.AsyncClient(http2=True)，requests 不支持。
• httpx 支持 WebSocket（实验性），requests 完全不支持。

2. 示例一：基础 GET / POST

2.1 代码

# test_basic.py
import pytest
import httpx
import asyncio

# ---------- GET ----------
async def test_get_users():
    """GET /api/users → 200, 断言 JSON 字段与 Header"""
    async with httpx.AsyncClient(base_url="https://httpbin.org") as client:
        resp = await client.get("/get", timeout=httpx.Timeout(10))
        # 断言状态码
        assert resp.status_code == 200
        # 断言 Header
        assert resp.headers["content-type"].startswith("application/json")
        # 断言 JSON 字段
        data = resp.json()
        assert "url" in data
        assert "headers" in data
        assert "origin" in data

# ---------- POST ----------
async def test_post_echo():
    """POST /api/post → 200, 回显 body 与 content-type"""
    payload = {"name": "程清和", "role": "SRE", "level": 5}
    async with httpx.AsyncClient(base_url="https://httpbin.org") as client:
        resp = await client.post(
            "/post",
            json=payload,
            headers={"X-Request-Id": "test-001"},
            timeout=httpx.Timeout(10),
        )
        assert resp.status_code == 200
        data = resp.json()
        # 回显断言
        assert data["json"] == payload
        assert data["headers"]["X-Request-Id"] == "test-001"

# ---------- 404 场景 ----------
async def test_get_404():
    """GET 不存在的资源 → 404"""
    async with httpx.AsyncClient(base_url="https://httpbin.org") as client:
        resp = await client.get("/status/404", timeout=httpx.Timeout(10))
        assert resp.status_code == 404

# ---------- 异常处理 ----------
async def test_timeout():
    """超时 0.01s → 触发 httpx.TimeoutException"""
    async with httpx.AsyncClient(base_url="https://httpbin.org") as client:
        with pytest.raises(httpx.ReadTimeout):
            await client.get("/delay/5", timeout=httpx.Timeout(0.01))

2.2 pytest -v 运行输出

$ pytest test_basic.py -v
============================= test session starts ==============================
platform linux -- Python 3.11.9, pytest-7.4.4, pluggy-1.5.0
rootdir: /home/dev/pytest-httpx-demo
plugins: asyncio-0.23.6, anyio-4.4.0, respx-0.21.0
collected 4 items

test_basic.py::test_get_users PASSED                                      [ 25%]
test_basic.py::test_post_echo PASSED                                      [ 50%]
test_basic.py::test_get_404 PASSED                                        [ 75%]
test_basic.py::test_timeout PASSED                                        [100%]

============================== 4 passed in 2.31s ===============================

注意：需要 pytest-asyncio 插件。在 pytest.ini 或 pyproject.toml 中配置：

[tool.pytest.ini_options]
asyncio_mode = "auto"

3. 示例二：异步并发请求

3.1 代码

# test_concurrent.py
import pytest
import httpx
import asyncio
import time

URLS = [
    "https://httpbin.org/get",
    "https://httpbin.org/get",
    "https://httpbin.org/get",
    "https://httpbin.org/get",
    "https://httpbin.org/get",
]

# ---------- 同步串行（对照组） ----------
async def test_serial_baseline():
    """5 个 GET 串行执行，记录总耗时"""
    total_start = time.perf_counter()
    async with httpx.AsyncClient(base_url="https://httpbin.org") as client:
        for url in URLS:
            await client.get(url, timeout=httpx.Timeout(10))
    elapsed = time.perf_counter() - total_start
    print(f"\n[serial] elapsed={elapsed:.3f}s")
    # 串行必然 > 0（每个请求独立等待）
    assert elapsed > 0

# ---------- 异步并发（目标） ----------
async def test_concurrent_gather():
    """5 个 GET 并发执行，总耗时 ≈ 最慢那个"""
    async with httpx.AsyncClient(base_url="https://httpbin.org") as client:
        total_start = time.perf_counter()
        tasks = [client.get(url, timeout=httpx.Timeout(10)) for url in URLS]
        responses = await asyncio.gather(*tasks)
        elapsed = time.perf_counter() - total_start
        print(f"\n[concurrent] elapsed={elapsed:.3f}s")
        # 全部 200
        assert all(r.status_code == 200 for r in responses)
        # 并发应显著快于串行（httpbin 每次 ~200-400ms）
        # 这里用宽松断言：并发耗时 < 串行 5 个的总和
        # （实际 CI 中可对比两次运行）
        assert elapsed < 5 * 1.0  # 5 个串行至少 5s，并发应 < 2s

# ---------- 部分失败场景 ----------
async def test_concurrent_partial_failure():
    """5 个并发，其中 1 个 500 → gather 不抛异常，逐一检查"""
    urls = ["https://httpbin.org/status/500"] + URLS[:4]
    async with httpx.AsyncClient(base_url="https://httpbin.org") as client:
        tasks = [client.get(url, timeout=httpx.Timeout(10)) for url in urls]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        # 第 1 个是 500，其余 200
        assert responses[0].status_code == 500
        assert all(r.status_code == 200 for r in responses[1:])

3.2 pytest -v 运行输出

$ pytest test_concurrent.py -v -s
============================= test session starts ==============================
platform linux -- Python 3.11.9, pytest-7.4.4, pluggy-1.5.0
rootdir: /home/dev/pytest-httpx-demo
plugins: asyncio-0.23.6, respx-0.21.0
collected 3 items

test_concurrent.py::test_serial_baseline
[serial] elapsed=1.842s
PASSED                                                                    [ 33%]
test_concurrent.py::test_concurrent_gather
[concurrent] elapsed=0.671s
PASSED                                                                    [ 66%]
test_concurrent.py::test_concurrent_partial_failure PASSED                [100%]

============================== 3 passed in 3.12s ===============================

可以看到：5 个请求串行 ≈ 1.84s，并发 ≈ 0.67s，加速比 ≈ 2.7×。

4. 示例三：respx mock 外部依赖

4.1 代码

# test_mock.py
import pytest
import httpx
import respx

BASE = "https://api.example.com"

# ---------- 200 正常 ----------
@respx.mock
async def test_mock_200():
    """mock GET /users/42 → 200 JSON"""
    route = respx.get(f"{BASE}/users/42").mock(
        status_code=200,
        json={"id": 42, "name": "程清和", "role": "SRE"},
        headers={"X-Mock": "true"},
    )
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{BASE}/users/42")
    assert resp.status_code == 200
    assert resp.json()["name"] == "程清和"
    assert resp.headers["x-mock"] == "true"
    assert route.called  # 确认 mock 被命中

# ---------- 404 资源不存在 ----------
@respx.mock
async def test_mock_404():
    """mock GET /users/999 → 404"""
    respx.get(f"{BASE}/users/999").mock(status_code=404, json={"detail": "not found"})
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{BASE}/users/999")
    assert resp.status_code == 404
    assert resp.json()["detail"] == "not found"

# ---------- 500 服务异常 ----------
@respx.mock
async def test_mock_500():
    """mock POST /orders → 500"""
    respx.post(f"{BASE}/orders").mock(status_code=500, text="Internal Server Error")
    async with httpx.AsyncClient() as client:
        resp = await client.post(f"{BASE}/orders", json={"item": "monitoring"})
    assert resp.status_code == 500
    assert resp.text == "Internal Server Error"

# ---------- 请求体校验 ----------
@respx.mock
async def test_mock_assert_body():
    """校验发出的请求体是否符合预期"""
    route = respx.post(f"{BASE}/deploy").mock(status_code=202)
    async with httpx.AsyncClient() as client:
        await client.post(f"{BASE}/deploy", json={"env": "prod", "version": "2.3.1"})
    # 校验请求体
    sent = route.calls.last.request
    import json as _json
    body = _json.loads(sent.content)
    assert body["env"] == "prod"
    assert body["version"] == "2.3.1"
    assert sent.headers["content-type"] == "application/json"

4.2 pytest -v 运行输出

$ pytest test_mock.py -v
============================= test session starts ==============================
platform linux -- Python 3.11.9, pytest-7.4.4, pluggy-1.5.0
rootdir: /home/dev/pytest-httpx-demo
plugins: asyncio-0.23.6, respx-0.21.0
collected 4 items

test_mock.py::test_mock_200 PASSED                                        [ 25%]
test_mock.py::test_mock_404 PASSED                                        [ 50%]
test_mock.py::test_mock_500 PASSED                                        [ 75%]
test_mock.py::test_mock_assert_body PASSED                                [100%]

============================== 4 passed in 0.18s ===============================

mock 场景无需网络，0.18s 跑完 4 个用例，适合高频回归。

5. 方案取舍对比

5.1 同步 requests vs 异步 httpx

| 维度 | 同步 requests | 异步 httpx | 结论 | |------|-------------|-----------|------| | 性能（多请求场景） | 串行，N 个请求 ≈ N × RT | 并发，N 个请求 ≈ max(RT) | 请求数 ≥ 3 时异步优势明显；单请求无差异 | | 可维护性 | 代码线性，新人上手快 | 需理解 event loop、async/await、fixture 生命周期 | 团队 ≥ 3 人且多人写测试时，同步更易维护；2 人以下异步 OK | | 调试难度 | pdb / ipdb 直接打断点 | 需 asyncio.run 或 nest_asyncio，断点偶尔"跳走" | 同步调试体验更好；异步需熟悉 asyncio 调试技巧 | | 超时与异常 | requests.exceptions.Timeout，粒度粗 | httpx.TimeoutException 细分为 Connect/Read/Write/Pool 四类 | 异步更精细，排错时能定位到哪个阶段超时 | | 适用场景 | 单接口冒烟、脚本、CI 中请求数 ≤ 5 | 多接口集成、并发压测、超时精细控制 | 建议：≤ 5 个请求用 requests 够用；> 5 或需并发用 httpx |

明确结论：如果你的测试用例里 80% 是"发一个请求、断言响应"，requests 完全够用，不必强上异步。异步的收益在并发和精细超时两个场景才体现。

5.2 fixture 共享 client vs 每用例独立 client

| 维度 | 共享 fixture（session/module 级） | 每用例独立 client（function 级） | 结论 | |------|-------------------------------|-------------------------------|------| | 资源开销 | 连接池复用，TLS 握手只一次 | 每个用例都建连、握手、销毁，开销 × N | 用例 > 20 个时，共享节省 10%–30% 总耗时 | | 隔离性 | 状态泄漏风险（如 cookie、header 被前一个用例改了） | 天然隔离，互不影响 | 对状态敏感的测试（登录态、token），独立更安全 | | 调试便利性 | 失败时不确定"哪个用例改坏了 client" | 每个用例独立，失败原因清晰 | 排错效率：独立 > 共享 | | mock 配合 | respx 的 @respx.mock 装饰器与共享 client 冲突（respx 是 per-request 拦截） | 每个用例独立 mock，无冲突 | 用 respx 时强烈建议独立 client |

明确结论：

• 用 respx mock → 每用例独立 client（避免 mock 路由泄漏）。
• 打真实服务且用例 > 50 → session 级共享 client + 每个用例用 client.headers 做隔离，或加 respx 兜底。
• 拿不准 → 先独立 client，跑通了再优化为共享。

6. 常见坑与排错路径

坑 1：超时配置不生效

现象：timeout=5 写在了 httpx.AsyncClient() 上，但某个 client.get() 没走这个超时。

根因：httpx 的超时是分层覆盖的——client.get(timeout=10) 会覆盖 client 级的 timeout=5。反过来，如果 client 级没设 timeout，而请求级也没设，则使用默认值（5s 连接 + 5s 读）。

排查：

# 打印实际生效的超时
async with httpx.AsyncClient(timeout=httpx.Timeout(5)) as client:
    print(client.timeout)  # Timeout(timeout=5.0)
    # 如果某处传了 timeout=None，会覆盖为默认值

修复：统一在 AsyncClient 初始化时设超时，请求级不要单独传 timeout，除非有明确理由。

坑 2：连接池复用导致"脏状态"

现象：测试 A 设置了 Authorization: Bearer token-A，测试 B 没设 header，但 B 的请求带了 token-A。

根因：httpx.AsyncClient 的 headers 参数是默认 header，所有请求都会带上。如果 fixture 是 session 级共享，前一个用例改了 client.headers，后面的全受影响。

排查：

# 在 client 上打印当前默认 headers
print(client.headers)
# 如果看到意料之外的 key，说明被前面的用例污染了

修复：

# 方式一：每用例独立 client（推荐）
async def test_x():
    async with httpx.AsyncClient() as client:
        resp = await client.get(url, headers={"Authorization": "Bearer token-x"})

# 方式二：共享 client 但请求级覆盖
resp = await client.get(url, headers={"Authorization": None})  # 显式清除

坑 3：事件循环冲突（"Event loop is closed"）

现象：pytest -v 跑到第 N 个用例后报 RuntimeError: Event loop is closed 或 RuntimeError: This event loop is already running。

根因：pytest-asyncio 默认每个测试函数创建新 event loop，但 httpx.AsyncClient 如果是在 fixture 中创建的且跨测试存活，它的底层连接绑定了旧 loop，新 loop 起来后复用旧连接就炸了。

排查：

# 开启 asyncio 调试模式
pytest -v --asyncio-debug
# 会打印 loop 创建/销毁的堆栈，定位哪个 fixture 跨了 loop

修复：

# 方案一：client 用 function 级（每用例新建，不跨 loop）
@pytest.fixture
async def client():
    async with httpx.AsyncClient() as c:
        yield c

# 方案二：用 pytest-asyncio 的 event_loop fixture 统一 loop
# （pytest-asyncio ≥ 0.23 推荐用 asyncio_mode=auto）

坑 4：respx mock 陷阱——路由未匹配

现象：respx 拦截了请求但返回了 500（respx 默认的"未匹配"响应），而不是你 mock 的 200。

根因：respx 的路由匹配是精确 URL + 方法匹配。常见踩坑：

• base_url 带了尾斜杠："https://api.example.com/" vs "https://api.example.com"
• 路径参数没匹配：mock 了 /users/42 但请求发的是 /users/42/
• query 参数不一致：mock 没带 ?page=1，请求带了

排查：

@respx.mock
async def test_debug():
    respx.get("https://api.example.com/users/42").mock(status_code=200, json={"ok": True})
    # 开启 respx 调试
    respx.mock.assert_all_called = False  # 先关掉严格模式
    async with httpx.AsyncClient() as client:
        resp = await client.get("https://api.example.com/users/42")
    print(resp.status_code)
    # 如果 500，打印实际请求 URL
    for r in respx.routes:
        print(r.url, r.method)

修复：

# 统一用不带尾斜杠的 URL
BASE = "https://api.example.com"  # 不写 /
respx.get(f"{BASE}/users/42").mock(...)
# 或者用 respx 的正则
respx.get(url__regex=r"https://api\.example\.com/users/\d+").mock(...)

坑 5（附加）：asyncio.gather 中一个异常吞掉全部

现象：5 个并发请求中 1 个 500，asyncio.gather(*tasks) 直接抛异常，剩下 4 个的结果拿不到。

修复：

# 用 return_exceptions=True 让 gather 不短路
responses = await asyncio.gather(*tasks, return_exceptions=True)
for i, r in enumerate(responses):
    if isinstance(r, Exception):
        print(f"request {i} failed: {r}")
    else:
        assert r.status_code == 200

7. 参考资料

1. httpx 官方文档 — Async：https://www.python-httpx.org/async/
2. respx 官方文档 — Mock：https://respx.mockingdocs.com/
3. pytest-asyncio 文档：https://pytest-asyncio.readthedocs.io/
4. httpx 超时机制详解：https://www.python-httpx.org/advanced/timeouts/
5. asyncio 并发编程指南（Python 官方）：https://docs.python.org/3/library/asyncio.html

写于一个 SRE 的视角：接口测试的"异步化"不是技术信仰，而是当你的 CI 从 50 个用例涨到 500 个、当你要在一个测试里并发打 5 个上游时，它才真正开始省钱。别为了异步而异步——先用 requests 把断言写对，再考虑要不要并发。