跳至主要內容

Python httpx 测试实战——异步 HTTP 测试、Mock Server、证书测试

老张2026/4/30大约 6 分钟

Python httpx 测试实战——异步 HTTP 测试、Mock Server、证书测试

适读人群：Python 异步开发者 / 需要测试异步 HTTP 服务的工程师 | 阅读时长：约 15 分钟 | 核心价值：掌握 httpx 的测试能力，特别是异步场景和 Mock Server 技巧

requests 搞不定的那个异步接口

我们团队的新项目用 FastAPI + asyncio 做了一个高并发 API 服务。测试时遇到了问题：

用 requests 库测试同步接口没问题，但当我们想测试真正的并发行为——比如"同时发出 10 个请求，验证限流逻辑是否正确触发"——requests 的同步模型做起来很别扭，需要用 threading 模拟并发，代码又臭又长。

后来换成 httpx，加上 pytest-asyncio，异步测试写起来和同步一样流畅。

httpx 简介与安装

httpx 是一个现代 Python HTTP 客户端，API 设计和 requests 高度兼容，但支持异步 (asyncio)，并且内置了测试用的 ASGITransport 和 WSGITransport，可以直接测试 FastAPI/Flask 应用，不需要启动真实服务器。

pip install httpx pytest-asyncio

同步 httpx（和 requests 的对比）

import httpx

# requests 风格
import requests
response = requests.get("https://api.example.com/users/1")

# httpx 同步风格（几乎一样）
response = httpx.get("https://api.example.com/users/1")
print(response.status_code)
print(response.json())

# 使用 Client（类似 requests.Session）
with httpx.Client(base_url="https://api.example.com", timeout=30) as client:
    response = client.get("/users/1")
    user = response.json()

异步 httpx 基础

import asyncio
import httpx

async def fetch_user(user_id: int) -> dict:
    async with httpx.AsyncClient(base_url="https://api.example.com") as client:
        response = await client.get(f"/users/{user_id}")
        return response.json()

# pytest-asyncio 支持直接测试异步函数
import pytest
import pytest_asyncio

@pytest.mark.asyncio
async def test_fetch_user():
    user = await fetch_user(1)
    assert user["id"] == 1
    assert "email" in user

测试 FastAPI 应用（不启动服务器）

这是 httpx 最强大的特性之一：用 ASGITransport 直接测试 ASGI 应用，不需要启动真实 HTTP 服务器：

# app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

users_db = {}

class UserCreate(BaseModel):
    email: str
    name: str

class UserResponse(BaseModel):
    id: str
    email: str
    name: str

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    user_id = str(len(users_db) + 1)
    users_db[user_id] = {**user.dict(), "id": user_id}
    return users_db[user_id]

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: str):
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    return users_db[user_id]
# tests/conftest.py
import pytest
import httpx
from app import app

@pytest.fixture
def client():
    """同步测试客户端，直接测试 ASGI 应用"""
    with httpx.Client(
        transport=httpx.ASGITransport(app=app),
        base_url="http://testserver"
    ) as client:
        yield client

@pytest_asyncio.fixture
async def async_client():
    """异步测试客户端"""
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app),
        base_url="http://testserver"
    ) as client:
        yield client
# tests/test_user_api.py
import pytest

class TestUserAPI:
    
    def test_create_user_sync(self, client):
        """同步测试"""
        response = client.post("/users", json={
            "email": "test@example.com",
            "name": "测试用户"
        })
        assert response.status_code == 201
        data = response.json()
        assert data["email"] == "test@example.com"
        assert "id" in data
    
    @pytest.mark.asyncio
    async def test_create_user_async(self, async_client):
        """异步测试"""
        response = await async_client.post("/users", json={
            "email": "async@example.com",
            "name": "异步测试用户"
        })
        assert response.status_code == 201
    
    @pytest.mark.asyncio
    async def test_concurrent_requests(self, async_client):
        """测试并发请求（验证限流、竞态条件等）"""
        import asyncio
        
        # 同时发出 10 个创建用户请求
        tasks = [
            async_client.post("/users", json={
                "email": f"user_{i}@example.com",
                "name": f"用户{i}"
            })
            for i in range(10)
        ]
        
        responses = await asyncio.gather(*tasks)
        
        # 验证所有请求都成功
        for response in responses:
            assert response.status_code == 201
        
        # 验证没有 ID 重复（竞态条件检查）
        ids = [r.json()["id"] for r in responses]
        assert len(set(ids)) == 10  # 10 个不同的 ID

Mock Server：使用 respx 库

当测试代码调用外部 API 时（支付平台、短信服务等），需要 Mock 这些外部依赖。respx 是专门为 httpx 设计的 Mock 库：

pip install respx
import httpx
import respx
import pytest

# 被测代码
async def send_sms(phone: str, message: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://sms.external-provider.com/send",
            json={"phone": phone, "message": message},
            headers={"X-API-Key": "real-api-key"}
        )
        return response.json()

# 测试代码
@pytest.mark.asyncio
@respx.mock
async def test_send_sms_success():
    # 设置 Mock：匹配特定 URL 和请求体
    respx.post("https://sms.external-provider.com/send").mock(
        return_value=httpx.Response(
            200,
            json={"success": True, "message_id": "MSG-12345"}
        )
    )
    
    result = await send_sms("13812345678", "您的验证码是：123456")
    
    assert result["success"] is True
    assert result["message_id"] == "MSG-12345"

@pytest.mark.asyncio
@respx.mock
async def test_send_sms_provider_error():
    """测试外部服务报错时的处理"""
    respx.post("https://sms.external-provider.com/send").mock(
        return_value=httpx.Response(
            503,
            json={"error": "Service temporarily unavailable"}
        )
    )
    
    with pytest.raises(Exception) as exc_info:
        await send_sms("13812345678", "测试消息")
    
    assert "503" in str(exc_info.value) or "unavailable" in str(exc_info.value)

@pytest.mark.asyncio
@respx.mock
async def test_send_sms_network_timeout():
    """测试网络超时"""
    respx.post("https://sms.external-provider.com/send").mock(
        side_effect=httpx.ConnectTimeout("Connection timed out")
    )
    
    with pytest.raises(httpx.ConnectTimeout):
        await send_sms("13812345678", "超时测试")

使用 respx 进行更复杂的 Mock

@pytest.mark.asyncio
@respx.mock
async def test_payment_flow():
    """支付流程测试——Mock 多个外部接口"""
    
    # Mock 支付创建接口
    respx.post("https://payment.provider.com/charges").mock(
        return_value=httpx.Response(200, json={
            "charge_id": "ch_test_001",
            "status": "pending",
            "amount": 9999,
        })
    )
    
    # Mock 支付状态查询（第一次返回 pending，第二次返回 succeeded）
    query_route = respx.get(
        url__regex=r"https://payment\.provider\.com/charges/ch_.*"
    )
    query_route.side_effect = [
        httpx.Response(200, json={"status": "pending"}),
        httpx.Response(200, json={"status": "succeeded"}),
    ]
    
    # 执行测试
    from services.payment_service import PaymentService
    service = PaymentService()
    
    result = await service.process_payment(
        amount=9999,
        currency="CNY",
        payment_method_id="pm_test_001"
    )
    
    assert result.success
    assert result.charge_id == "ch_test_001"
    
    # 验证 Mock 被调用了正确的次数
    assert respx.calls.call_count == 3  # 1次创建 + 2次查询

证书测试（HTTPS）

import httpx
import ssl

# 测试自签名证书（开发环境）
client = httpx.Client(verify=False)  # 跳过证书验证，只用于测试

# 使用自定义 CA 证书
ssl_context = ssl.create_default_context(cafile="/path/to/custom-ca.crt")
client = httpx.Client(verify=ssl_context)

# 双向 TLS（mTLS）
client = httpx.Client(
    cert=("/path/to/client.crt", "/path/to/client.key"),
    verify="/path/to/server-ca.crt"
)

# 测试 mTLS 场景
@pytest.mark.asyncio
async def test_mtls_endpoint():
    async with httpx.AsyncClient(
        cert=("tests/fixtures/client.crt", "tests/fixtures/client.key"),
        verify="tests/fixtures/server-ca.crt",
        base_url="https://internal-api.company.com"
    ) as client:
        response = await client.get("/secure-endpoint")
        assert response.status_code == 200

踩坑实录

坑一：pytest-asyncio 的 event_loop scope 警告

现象： 运行异步测试时，pytest 输出大量 DeprecationWarning: There is no current event loop。

解法： 在 pytest.ini 或 conftest.py 中配置 asyncio mode：

# pytest.ini
[pytest]
asyncio_mode = auto  # 自动识别异步测试，不需要每个都加 @pytest.mark.asyncio

或者：

# conftest.py
import pytest

@pytest.fixture(scope="session")
def event_loop_policy():
    """使用同一个 event loop 跑所有测试"""
    import asyncio
    return asyncio.DefaultEventLoopPolicy()

坑二：respx.mock 和 httpx.AsyncClient 实例的生命周期冲突

现象： 使用 @respx.mock 装饰器时，fixture 中创建的 AsyncClient 有时候 Mock 不生效。

原因： @respx.mock 装饰器和 httpx.AsyncClient 的实例化时机需要对齐——@respx.mock 是在测试函数执行时激活的，fixture 中创建的 client 如果在装饰器激活前就绑定了 transport，Mock 可能不生效。

解法： 在测试函数内部创建 AsyncClient，或者使用 respx.mock 作为 context manager：

@pytest.mark.asyncio
async def test_external_api():
    with respx.mock:
        respx.get("https://external.api.com/data").mock(
            return_value=httpx.Response(200, json={"key": "value"})
        )
        async with httpx.AsyncClient() as client:
            response = await client.get("https://external.api.com/data")
            assert response.json() == {"key": "value"}

坑三：ASGITransport 测试时数据库状态残留

现象： 用 ASGITransport 直接测试 FastAPI 应用，多个测试之间的数据库状态会互相影响。

解法： 在每个测试前后清理数据库，或者用事务回滚：

@pytest_asyncio.fixture
async def async_client(async_db_session):
    """每个测试用独立的数据库 session"""
    # 把 FastAPI 的 DB 依赖替换成测试 session
    app.dependency_overrides[get_db] = lambda: async_db_session
    
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app),
        base_url="http://testserver"
    ) as client:
        yield client
    
    app.dependency_overrides.clear()

小结

httpx 相比 requests 的核心优势在测试场景中体现得特别明显：

  1. ASGITransport：直接测试 FastAPI/Starlette 应用，无需启动服务器，测试速度快 10 倍
  2. 异步支持：用 asyncio.gather 轻松测试并发场景
  3. respx Mock：为 httpx 专门设计的 Mock 库，语法简洁，功能完整

如果你的项目已经在用 FastAPI 或其他异步框架，强烈建议用 httpx + respx 做接口测试。

编辑此页
最近更新:
贡献者: xiaozhangge