外部服务模拟：HTTPX/responses/vcrpy录制回放

Python 测试与调试专题 · 可靠测试外部HTTP依赖的多种方案

专题：Python 测试与调试系统学习

关键词：Python, 测试, 调试, responses, pytest-httpx, vcrpy, WireMock, HTTP模拟, 录制回放, Python测试

一、外部服务测试概述

在现代应用开发中，几乎每个项目都会依赖外部HTTP服务——支付网关、天气预报API、第三方OAuth认证、数据库REST接口、消息推送服务等等。这些外部依赖给测试带来了巨大挑战：首先，外部服务可能不稳定或不可用，导致测试结果不可靠；其次，测试环境与生产环境的网络隔离意味着很多外部服务根本无法访问；再者，调用真实外部服务会产生费用（如短信、支付接口）并引入安全风险；最后，很多外部服务存在调用频率限制，频繁测试会导致IP被屏蔽。

为了解决上述挑战，Python生态提供了四大类外部服务模拟方案：第一类是Mock/桩代码方案（如responses、pytest-httpx、unittest.mock.patch），通过拦截代码中的HTTP调用并返回预设响应，在调用侧进行模拟，适用于单元测试和白盒集成测试；第二类是模拟服务器方案（如WireMock、MockServer），运行一个真实的HTTP服务器监听端口，录制并回放响应，适用于端到端测试和黑盒测试；第三类是录制回放方案（如vcrpy、betamax），首次运行时记录真实HTTP交互保存为Cassette文件，后续运行回放记录，兼具真实性和再现性；第四类是沙箱/测试环境方案（如Stripe测试模式、PayPal Sandbox），直接使用第三方提供的测试环境，最真实但最慢且需网络连接。

方案选型需要根据测试金字塔层级权衡。单元测试层应优先使用Mock方案，速度快且不依赖网络，能精确控制边界条件和异常场景。集成测试层可以混合使用录制回放和模拟服务器方案，在确保HTTP交互正确性的同时避免依赖外部服务。端到端测试层可考虑使用模拟服务器方案或真实沙箱环境。实际项目中，通常是多种方案组合使用：单元测试用Mock，集成测试用录制回放，E2E测试用沙箱环境。下面的章节将逐一深入介绍各主流方案的使用方式和最佳实践。

核心原则：外部服务模拟的核心目标是让测试快、稳、准——快指毫秒级响应不依赖网络，稳指不受外部服务波动影响，准指精确模拟各种边界条件（超时、错误码、特殊响应体）。Mock方案在三者中平衡最佳，录制回放次之，模拟服务器最接近真实但开销最大。

二、responses库

responses库是Python生态中最成熟的requests库模拟工具，它通过猴子补丁（Monkey Patching）方式拦截requests库的HTTP调用，在不需要网络连接的情况下返回预设响应数据。其核心工作原理是替换requests库底层的`urllib3`和`http.client`模块的发送方法，使得所有通过requests库发出的HTTP请求都被重定向到responses内部的路由表中进行匹配和响应。

安装responses非常简单，只需执行`pip install responses`。使用responses时需要先创建`responses.RequestsMock`实例并通过`responses.activate`装饰器或上下文管理器启用。最基本的用法是使用`responses.get()`、`responses.post()`等方法注册预期的HTTP方法和URL及其对应的响应。每个注册项包含URL匹配规则、响应状态码、响应体和响应头。当测试代码中发出匹配的HTTP请求时，responses会自动返回预设响应；如果没有匹配规则则会抛出`ConnectionError`异常，帮助开发者及时发现自己遗漏了某个外部调用。

responses除了支持基本的请求模拟之外，还提供了回调函数（Callback）机制，允许动态生成响应而非使用静态预设数据。这在需要根据请求体或请求参数动态响应时非常有用。此外，responses还可以模拟连接超时（`ConnectionError`）、HTTP异常等多种异常场景，让开发者可以全面测试代码的错误处理逻辑。

# 安装
$ pip install responses

# 基础用法：通过装饰器启用
import responses
import requests

@responses.activate
def test_get_user():
    # 注册模拟响应
    responses.get(
        url="https://api.example.com/users/1",
        json={"id": 1, "name": "Alice", "email": "alice@example.com"},
        status=200
    )

    # 被测代码发出请求
    resp = requests.get("https://api.example.com/users/1")

    # 验证结果
    assert resp.status_code == 200
    assert resp.json()["name"] == "Alice"
    assert responses.calls[0].request.url == "https://api.example.com/users/1"

# 模拟POST请求和异常场景
@responses.activate
def test_create_user_failure():
    # 模拟服务器错误
    responses.post(
        url="https://api.example.com/users",
        json={"error": "Internal Server Error"},
        status=500
    )

    resp = requests.post(
        "https://api.example.com/users",
        json={"name": "Bob"}
    )
    assert resp.status_code == 500

@responses.activate
def test_timeout():
    # 模拟连接超时
    responses.get(
        url="https://api.example.com/users/1",
        body=responses.ConnectionError("Connection timed out")
    )

    try:
        requests.get("https://api.example.com/users/1")
    except requests.ConnectionError:
        pass  # 超时被正确处理

# 回调函数：根据请求动态生成响应
def request_callback(request):
    payload = request.json()
    if payload.get("name"):
        return (201, {}, {"id": 999, "name": payload["name"]})
    return (400, {}, {"error": "name is required"})

@responses.activate
def test_callback():
    responses.add_callback(
        responses.POST,
        url="https://api.example.com/users",
        callback=request_callback,
        content_type="application/json"
    )

    # 测试带名字的请求
    resp1 = requests.post("https://api.example.com/users", json={"name": "Charlie"})
    assert resp1.status_code == 201

    # 测试不带名字的请求
    resp2 = requests.post("https://api.example.com/users", json={})
    assert resp2.status_code == 400

三、responses进阶

在实际项目中，HTTP请求往往比简单的GET/POST更复杂，需要匹配特定查询参数、请求头、请求体等内容。responses提供了丰富的匹配能力来应对这些场景。其中，`responses.matchers`模块包含了`query_param_matcher`、`header_matcher`、`json_params_matcher`等多个匹配器函数，可以精确控制哪些请求被哪些预设规则匹配。

当需要模拟同一URL的多次不同响应时，responses支持链式添加多条规则并依次消耗。即第一次请求时使用第一条匹配规则返回第一个响应，第二次请求使用第二条规则。这对于测试重试逻辑、轮询操作等场景非常有用。如果规则数量少于请求次数，responses会抛出`ConnectionError`异常提示遗漏了匹配规则。

在某些测试场景中，我们希望对大部分HTTP请求进行模拟拦截，但放行（Passthrough）特定请求到真实网络。responses提供了`passthrough_prefixes`参数，允许指定URL前缀列表，匹配这些前缀的请求将被放行到真实网络。这在混合测试场景中非常实用——部分请求用Mock验证，部分请求（如健康检查）发往真实服务。

# 匹配查询参数和请求头
from responses import matchers

@responses.activate
def test_match_query_params():
    responses.get(
        url="https://api.example.com/search",
        json={"results": ["item1", "item2"]},
        status=200,
        match=[
            matchers.query_param_matcher({"q": "python", "page": "1"}),
            matchers.header_matcher({"Authorization": "Bearer test-token"})
        ]
    )

    # 精确匹配的请求会成功
    resp = requests.get(
        "https://api.example.com/search",
        params={"q": "python", "page": "1"},
        headers={"Authorization": "Bearer test-token"}
    )
    assert resp.status_code == 200

    # 缺少参数的请求不会匹配（会抛ConnectionError）
    # requests.get("https://api.example.com/search", params={"q": "python"})

# 多次响应：模拟重试场景
@responses.activate
def test_retry_logic():
    # 前两次返回503，第三次返回200
    responses.get(
        url="https://api.example.com/data",
        body='{"error": "Service Unavailable"}',
        status=503
    )
    responses.get(
        url="https://api.example.com/data",
        body='{"error": "Service Unavailable"}',
        status=503
    )
    responses.get(
        url="https://api.example.com/data",
        json={"status": "ok", "data": [1, 2, 3]},
        status=200
    )

    # 模拟带有重试逻辑的代码
    for attempt in range(3):
        resp = requests.get("https://api.example.com/data")
        if resp.status_code == 200:
            break

    assert resp.status_code == 200
    assert resp.json()["status"] == "ok"

# Passthrough放行真实请求
@responses.activate
def test_mixed_mock_and_real():
    # 模拟的请求
    responses.get(
        url="https://api.example.com/data",
        json={"mock": True},
        status=200
    )

    # 对内部状态检查URL放行到真实网络
    # 实际使用时可配置一个.env或settings中的放行列表
    # unittest.mock.patch的方式也可以结合responses使用

    # 模拟请求被拦截
    resp = requests.get("https://api.example.com/data")
    assert resp.json()["mock"] is True

四、pytest-httpx

随着异步编程在Python中的普及，基于asyncio和httpx的HTTP客户端使用越来越广泛。pytest-httpx库正是为httpx（包括其同步和异步客户端）量身定制的测试工具。与responses拦截requests底层不同，pytest-httpx通过httpx的传输层（Transport）拦截机制工作，提供了名为`httpx_mock`的pytest fixture来管理模拟响应。

pytest-httpx的核心API围绕`httpx_mock` fixture展开，使用时直接在测试函数参数中声明该fixture，pytest会自动注入。然后通过`httpx_mock.add_response()`方法注册模拟响应，该方法支持指定URL、HTTP方法、状态码、响应体（文本/JSON/字节）、响应头等。如果不指定URL或方法，该响应会成为兜底响应，匹配所有请求——这在测试中应当谨慎使用，避免掩盖未预期的请求。

pytest-httpx的一个强大特性是其请求断言能力。测试完成后，可以通过`httpx_mock.get_request()`、`httpx_mock.get_requests()`等方法获取所有被拦截的请求对象，进而断言请求的URL、方法、头信息、请求体等内容。此外，pytest-httpx还支持在响应时调用回调函数进行断言，在返回响应前验证请求的完整性。pytest-httpx原生支持异步测试，与pytest-asyncio完美配合，可以测试`async with httpx.AsyncClient() as client`的异步代码路径。

# 安装
$ pip install pytest-httpx

# 基础用法
import httpx
import pytest

async def test_get_user(httpx_mock):
    # 注册模拟响应
    httpx_mock.add_response(
        url="https://api.example.com/users/1",
        json={"id": 1, "name": "Alice"},
        status_code=200
    )

    async with httpx.AsyncClient() as client:
        resp = await client.get("https://api.example.com/users/1")

    assert resp.status_code == 200
    assert resp.json()["name"] == "Alice"

    # 断言发出的请求
    request = httpx_mock.get_request()
    assert request.url.path == "/users/1"
    assert request.method == "GET"

# 匹配URL/方法/头/参数
async def test_match_conditions(httpx_mock):
    # 精确匹配URL和方法
    httpx_mock.add_response(
        method="POST",
        url="https://api.example.com/users",
        json={"id": 2, "name": "Bob"},
        status_code=201,
        match_headers={"Authorization": "Bearer token123"}
    )

    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "https://api.example.com/users",
            json={"name": "Bob"},
            headers={"Authorization": "Bearer token123"}
        )

    assert resp.status_code == 201
    assert resp.json()["name"] == "Bob"

    # 验证请求头
    request = httpx_mock.get_request()
    assert request.headers["authorization"] == "Bearer token123"

# 回调断言与异常模拟
async def test_callback_assert(httpx_mock):
    def assert_request(request):
        # 在返回响应前验证请求
        body = request.read()
        assert b"email" in body, "请求体必须包含email字段"

    httpx_mock.add_response(
        url="https://api.example.com/users",
        method="POST",
        json={"id": 3, "name": "Charlie"},
        status_code=201,
        callback=assert_request
    )

    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "https://api.example.com/users",
            json={"name": "Charlie", "email": "charlie@example.com"}
        )
    assert resp.status_code == 201

async def test_simulate_timeout(httpx_mock):
    # 模拟超时异常
    httpx_mock.add_exception(
        httpx.TimeoutException("Request timed out"),
        url="https://api.example.com/slow"
    )

    import pytest
    with pytest.raises(httpx.TimeoutException):
        async with httpx.AsyncClient() as client:
            await client.get("https://api.example.com/slow")

五、vcrpy录制回放

vcrpy是Python生态中最流行的HTTP交互录制回放库，其名称和设计灵感来源于Ruby社区著名的VCR库。vcrpy的核心思想是：首次运行测试时拦截并记录所有HTTP请求及其响应，保存为Cassette文件（默认YAML格式）；后续运行同一测试时，不再发起真实网络请求，而是使用Cassette中记录的响应数据。这样一来，每个测试在首次运行时获得实时的真实数据，而后续运行既保持了零外部依赖的高速度，又获得了完全一致的响应数据。

vcrpy的工作机制与pip install的方式非常类似，它同样通过猴子补丁拦截底层的HTTP库。vcrpy自动检测当前使用的HTTP库（requests、httpx、urllib3等）并注入对应的钩子，无需开发者额外配置。使用时通过`@vcr.use_cassette`装饰器或`with vcr.use_cassette()`上下文管理器指定Cassette文件路径，该文件保存了完整的HTTP交互记录，包括请求URL、请求方法、请求头、请求体、响应状态码、响应头和响应体。

vcrpy提供了多种录制模式（record_mode）来精确控制何时录制、何时回放。`once`模式（默认）表示如果Cassette文件不存在则录制，存在则回放；`new_episodes`在第一次回放之外的额外请求会触发录制；`all`模式每次都重新录制；`none`模式只回放不录制，适合CI环境；`once_no_write`回放但不更新Cassette文件。匹配规则（match_on）也是vcrpy的关键配置，默认基于`method`、`scheme`、`host`、`port`、`path`、`query`六个维度进行匹配，自定义匹配规则可以根据实际需求调整敏感度。

# 安装
$ pip install vcrpy

# 基础用法：使用装饰器
import vcr
import requests

@vcr.use_cassette("fixtures/cassettes/test_get_user.yaml")
def test_get_user():
    # 首次运行：发送真实HTTP请求并录制
    # 后续运行：使用录制的响应，不发送真实请求
    resp = requests.get("https://api.example.com/users/1")
    assert resp.status_code == 200
    assert resp.json()["name"] == "Alice"

# 使用上下文管理器
def test_get_user_context():
    with vcr.use_cassette("fixtures/cassettes/get_user.yaml"):
        resp = requests.get("https://api.example.com/users/1")
        assert resp.status_code == 200

# 生成的YAML cassette文件示例
"""
interactions:
- request:
    method: GET
    uri: https://api.example.com/users/1
    headers:
      Accept: ['*/*']
      Accept-Encoding: ['gzip, deflate']
      User-Agent: ['python-requests/2.31.0']
    body: null
  response:
    status:
      code: 200
      message: OK
    headers:
      content-type: ['application/json']
      x-request-id: ['abc-123']
    body: '{"id": 1, "name": "Alice", "email": "alice@example.com"}'
    http_version: '1.1'
"""

# 自定义录制模式
@vcr.use_cassette(
    "fixtures/cassettes/login.yaml",
    record_mode="new_episodes",  # 新请求触发录制，已有请求回放
    match_on=["method", "host", "path"]  # 仅匹配方法、主机和路径
)
def test_login():
    resp = requests.post(
        "https://api.example.com/login",
        json={"username": "admin", "password": "secret"}
    )
    assert resp.status_code == 200

# 录制模式对比演示
@vcr.use_cassette("fixtures/cassettes/once_mode.yaml", record_mode="once")
def test_record_once():
    """once模式：有cassette则回放，无则录制。默认行为。"""
    pass

@vcr.use_cassette("fixtures/cassettes/all_mode.yaml", record_mode="all")
def test_record_all():
    """all模式：每次都重新录制，覆盖已有cassette。适合数据更新频繁的场景。"""
    pass

@vcr.use_cassette("fixtures/cassettes/none_mode.yaml", record_mode="none")
def test_replay_only():
    """none模式：只回放不录制。Cassette不存在时会报错。适合CI环境确保不产生网络依赖。"""
    pass

六、vcrpy进阶

在实际项目中，HTTP交互中经常包含敏感信息——API密钥、认证令牌、个人身份信息等。将这些信息原封不动地保存到Cassette文件中并提交到版本控制系统是一个严重的安全隐患。vcrpy通过`filter_headers`、`filter_query_parameters`、`filter_post_data_parameters`等配置参数解决了这个问题，允许在录制时自动屏蔽请求和响应中的敏感字段。例如设置`filter_headers=["authorization", "x-api-key"]`后，录制的Cassette中将使用`[FILTERED]`替换实际值。

vcrpy的匹配规则（match_on）是一个高度灵活的扩展点。除了默认的六维度匹配外，开发者在callback匹配模式中可以实现完全自定义的匹配逻辑，比如根据请求体中的特定字段进行匹配。匹配规则的精准度需要在实际场景中权衡——过于宽松可能导致不同请求使用错误的Cassette记录，过于严格则可能每次都需要重新录制。合理做法是从宽松规则开始，在发现不匹配问题时逐步收紧。

pytest-vcr库为vcrpy提供了pytest集成支持，安装后可以通过`@pytest.mark.vcr`装饰器或conftest.py中的全局配置来使用vcrpy，无需在每个测试中重复配置Cassette路径和过滤器。配合pytest的fixture机制，还可以在conftest.py中定义`vcr_config` fixture来统一管理所有vcrpy测试的配置参数。此外，pytest-vcr会自动为每个测试生成Cassette文件名，格式为`测试文件名.测试函数名.yaml`，大大简化了文件管理。

# 敏感信息过滤
import vcr

# 配置过滤器
my_vcr = vcr.VCR(
    filter_headers=["authorization", "x-api-key", "cookie", "set-cookie"],
    filter_query_parameters=["api_key", "token", "secret"],
    filter_post_data_parameters=["password", "credit_card", "ssn"],
    # 也可以对响应体中的敏感字段做过滤
    before_record_response=lambda response: response
)

@my_vcr.use_cassette("fixtures/cassettes/auth.yaml")
def test_auth_api():
    """令牌和密码字段在cassette中会被替换为[FILTERED]"""
    resp = requests.post(
        "https://api.example.com/auth",
        json={"username": "admin", "password": "super-secret-123"},
        headers={"Authorization": "Bearer eyJhbGciOiJIUzI1NiJ9.xxx"}
    )
    assert resp.status_code == 200

# pytest-vcr集成
# 安装：pip install pytest-vcr

# conftest.py - 全局vcr配置
"""
import pytest

@pytest.fixture(scope="module")
def vcr_config():
    return {
        "filter_headers": ["authorization", "x-api-key"],
        "filter_query_parameters": ["token"],
        "record_mode": "once",
        "match_on": ["method", "host", "path", "query"],
    }
"""

# test_api.py - 使用pytest-vcr
"""
import pytest
import requests

@pytest.mark.vcr
def test_get_user():
    # pytest-vcr自动创建和管理cassette文件
    # 文件名格式：test_api.test_get_user.yaml
    resp = requests.get("https://api.example.com/users/1")
    assert resp.status_code == 200

@pytest.mark.vcr
def test_list_users():
    # 每个测试有独立的cassette文件
    resp = requests.get("https://api.example.com/users", params={"page": 1})
    assert resp.status_code == 200
    assert len(resp.json()["users"]) > 0
"""

# 自定义匹配规则与编码处理
import vcr

def custom_body_matcher(r1, r2):
    """自定义匹配器：仅比较请求体中的"action"字段"""
    import json
    try:
        body1 = json.loads(r1.body)
        body2 = json.loads(r2.body)
        return body1.get("action") == body2.get("action")
    except (ValueError, TypeError):
        return r1.body == r2.body

my_vcr = vcr.VCR(
    match_on=[
        "method",
        "host",
        "path",
        custom_body_matcher  # 自定义匹配函数
    ],
    # 在录制时使用原始响应而非压缩
    before_record_request=lambda req: req,
    decode_compressed_response=True  # 自动解压gzip响应体
)

@my_vcr.use_cassette("fixtures/cassettes/custom_match.yaml")
def test_custom_match():
    resp = requests.post(
        "https://api.example.com/rpc",
        json={"action": "get_user", "user_id": 42, "internal_token": "xxx"}
    )
    assert resp.status_code == 200

七、WireMock模拟服务器

WireMock是一个运行在真实HTTP端口上的模拟服务器，最初源自Java生态，后来通过WireMock的独立运行模式（Standalone）和Python客户端（wiremock-python）实现了跨语言使用。与基于代码拦截的Mock方案不同，WireMock启动一个真正的HTTP服务器监听指定端口（默认8080），应用代码无需任何修改即可直接发送HTTP请求到WireMock。这使得WireMock特别适合端到端测试（E2E）、微服务集成测试以及前端开发中的后端依赖模拟。

WireMock的核心概念是Stub映射（Stub Mapping）。每个Stub定义了请求匹配规则和对应的响应定义。请求匹配支持URL精确匹配、URL路径正则匹配、请求头匹配、请求体JSON/XML匹配等多种方式。响应定义支持静态JSON/XML响应、基于Handlebars模板的动态响应、延迟模拟、故障注入（返回500/503或连接超时）等高级功能。Stub可以通过WireMock提供的REST API动态注册和管理，也可以通过JSON文件持久化存储。

WireMock还提供请求验证功能——即使Stub已经匹配并返回了响应，WireMock会记录每个接收到的请求，开发者可以在测试结束时通过REST API查询哪些请求被接收、匹配了哪些Stub、以及是否存在未匹配的请求。这对于验证被测试系统是否按预期发送了正确数量的请求、正确的请求参数非常有价值。WireMock的录制代理模式（Proxying）可以在真实API前做代理，录制所有经过的HTTP交互并自动生成Stub映射定义，这个功能可以大幅降低编写Stub的工作量。

# 下载并运行WireMock独立服务器
$ wget https://repo1.maven.org/maven2/org/wiremock/wiremock-standalone/3.5.2/wiremock-standalone-3.5.2.jar
$ java -jar wiremock-standalone-3.5.2.jar --port 8080
# WireMock启动后监听localhost:8080

# 使用requests向WireMock发送请求
import requests

# 首先通过WireMock的API注册一个stub
stub_config = {
    "request": {
        "method": "GET",
        "urlPattern": "/api/users/([0-9]+)"
    },
    "response": {
        "status": 200,
        "jsonBody": {"id": 1, "name": "Alice"},
        "headers": {"Content-Type": "application/json"}
    }
}
requests.post("http://localhost:8080/__admin/mappings", json=stub_config)

# 现在向WireMock发送请求，它会返回预设响应
resp = requests.get("http://localhost:8080/api/users/42")
print(resp.json())  # 输出: {"id": 1, "name": "Alice"}

# 使用wiremock-python库（更Pythonic的方式）
$ pip install wiremock-python

"""
from wiremock import WireMockServer, Mappings, MappingBuilder
from wiremock.constants import Response, Request

# 启动管理WireMock服务器
wm = WireMockServer(port=8080, host="localhost")
wm.start()

# 定义stub
mapping = (
    MappingBuilder("POST", "/api/orders")
    .with_json_body({"product_id": "123", "quantity": 1})
    .will_return(
        Response(status=201, json_body={
            "order_id": "ORD-001",
            "status": "created",
            "total": 99.99
        })
    )
    .with_delay(100)  # 模拟100ms延迟
)

# 注册stub
wm.register_mapping(mapping.build())

# 被测代码无需任何修改，只需指向WireMock的地址
import requests
resp = requests.post(
    "http://localhost:8080/api/orders",
    json={"product_id": "123", "quantity": 1}
)
assert resp.status_code == 201

# 验证请求
requests_ = wm.count_requests_matching(
    Request(method="POST", url_path="/api/orders")
)
assert requests_["count"] == 1

wm.stop()
"""

# WireMock录制代理模式
# 启动时指定代理目标：
$ java -jar wiremock-standalone-3.5.2.jar --proxy-all="https://api.example.com" --record-mappings

# 所有发往localhost:8080的请求都会被代理到api.example.com
# 同时自动生成stub映射JSON文件保存在mappings/目录下
"""
import requests

# 发送真实请求（被代理到api.example.com且录制为stub）
resp = requests.get("http://localhost:8080/users/1")
print(resp.json())

# WireMock会在mappings/目录下自动生成类似这样的文件：
# mappings/get_users_1-xxxxxxxx.json
# {
#   "request": {"method": "GET", "url": "/users/1"},
#   "response": {"status": 200, "jsonBody": {"id": 1, "name": "Alice"}, ...}
# }

# 此后停止代理，这些请求依然可以独立回复
"""

八、方案对比与选型

在选择外部服务模拟方案时，需要从多个维度评估各方案的适用性：测试速度、真实性、易用性、调试能力、团队协作支持等。responses和pytest-httpx作为基于代码拦截的Mock方案，测试速度最快（毫秒级），资源消耗最小，且可以精确控制各种边界条件和异常场景，非常适合单元测试和白盒集成测试。其缺点是修改了底层的网络调用路径，如果被测试代码使用了非标准HTTP库或原生socket通信，则无法生效。

vcrpy录制回放方案在真实性和速度之间取得了较好的平衡。首次运行时使用真实HTTP交互获得了高度的真实性，后续运行则完全脱离网络进行回放保证了速度。vcrpy特别适合那些需要验证与外部服务交互正确性、但不必每次运行都触发真实网络调用的场景。Cassette文件的版本控制还带来了团队协作上的优势——所有团队成员共享同一套Cassette记录，测试结果一致。但Cassette文件会随接口变更而过时，需要定期重新录制，且Cassette文件较大时会影响仓库大小。

WireMock作为独立模拟服务器方案提供了最高的真实性和最彻底的隔离性。应用代码不需要任何修改，只需更改目标地址即可。WireMock的录制代理、请求验证、延迟模拟、故障注入等高级功能使其在端到端测试和微服务测试中不可替代。但测试速度相对较慢（需要启动Java进程），资源占用较大，测试环境配置也相对复杂。WireMock更适合需要真实HTTP协议栈的集成测试和合同测试场景。

# 四方案对比表（代码形式）
"""
+-------------------+----------------+----------------+----------------+----------------+
|     维度          |   responses    | pytest-httpx   |    vcrpy       |   WireMock     |
+-------------------+----------------+----------------+----------------+----------------+
| 测试速度          | 极快(<1ms)     | 极快(<1ms)     | 快(1-10ms)     | 中等(1-50ms)   |
| 真实度            | 低(Mock层)     | 低(Mock层)     | 高(真实数据)   | 高(真实HTTP)   |
| 协议模拟          | HTTP层         | HTTP层         | HTTP层         | TCP/IP层       |
| 异常模拟能力      | 极强           | 极强           | 中等(依赖录制) | 极强           |
| 敏感信息过滤      | 不适用         | 不适用         | 原生支持       | 需手动处理     |
| 版本控制友好      | 否(代码内)     | 否(代码内)     | 是(YAML文件)   | 是(JSON文件)   |
| 跨语言支持        | Python only    | Python only    | Python only    | 全语言         |
| 异步支持          | 否             | 原生支持       | 部分支持       | 完全支持       |
| 资源占用          | 无             | 无             | 无             | Java进程       |
| 配置复杂度        | 低             | 低             | 中             | 高             |
+-------------------+----------------+----------------+----------------+----------------+
"""

# 混合使用策略示例
# conftest.py - 根据测试层级自动切换方案
"""
import pytest
import os

# 单元测试：使用pytest-httpx（默认速度最快）
# 集成测试：使用pytest-vcr（兼顾真实性和速度）

@pytest.fixture(autouse=True)
def _http_test_strategy(request):
    # 如果设置了INTEGRATION_TEST环境变量，使用vcrpy录制回放
    if os.environ.get("INTEGRATION_TEST"):
        # 跳过不需要录制回放的纯单元测试
        if "httpx_mock" in request.fixturenames:
            request.applymarker(pytest.mark.skip)
    # CI环境使用none模式，确保不产生网络依赖
    if os.environ.get("CI"):
        request.applymarker(
            pytest.mark.vcr(record_mode="none")
        )
"""

# 选型决策流程图
"""
选型决策:
1. 被测代码使用requests库？
   ├─ 是 → 使用responses（最简单直接）
   └─ 否 → 进入2
2. 被测代码使用httpx库（同步或异步）？
   ├─ 是 → 使用pytest-httpx（原生异步支持）
   └─ 否 → 进入3
3. 需要录制真实HTTP交互？
   ├─ 是 → 使用vcrpy（自动检测HTTP库）
   └─ 否 → 进入4
4. 需要真实HTTP服务器/跨语言支持？
   ├─ 是 → 使用WireMock
   └─ 否 → 使用unittest.mock.patch（兜底方案）
"""

# 推荐组合方案
"""
项目类型          | 推荐方案
------------------+-------------------------------
小型脚本          | responses 或 pytest-httpx
REST API服务      | pytest-httpx(单元) + vcrpy(集成)
微服务项目        | pytest-httpx(单元) + WireMock(E2E)
数据分析流水线    | vcrpy(全量录制回放)
第三方SDK开发     | vcrpy(集成) + WireMock(合同测试)
全栈项目          | pytest-httpx + vcrpy + WireMock三层
"""

九、实战案例

理论结合实践是最有效的学习方式。本节通过三个完整的实战案例展示如何将前述技术方案应用到真实项目中。第一个案例是外部支付API的单元测试，通过pytest-httpx模拟支付网关在不同场景下的响应，验证订单服务在支付成功、支付失败、支付超时等情况下的处理逻辑。第二个案例是天气预报API的集成测试，使用vcrpy录制真实的天气API响应，在确保数据格式正确的同时避免频繁调用有频率限制的免费API。第三个案例是第三方OAuth登录测试，综合使用Mock和录制回放来模拟OAuth流程中的多个步骤。

每个案例都遵循相同的模式：先定义被测函数的输入输出契约，再根据契约编写测试用例覆盖正常路径和所有异常路径，最后通过断言验证函数的正确行为。测试用例的组织遵循AAA模式（Arrange-Act-Assert），将测试分为准备（Mock注册）、执行（调用被测函数）、验证（断言结果和请求）三个阶段。

这些案例不仅仅展示了技术本身的使用方法，更重要的是展示了测试设计的思维方式——如何识别外部依赖、如何选择模拟策略、如何设计覆盖各种场景的测试用例。在实际项目中，花费在测试设计上的时间往往能十倍地减少调试和排查问题的时间。

# 案例一：外部支付API测试（pytest-httpx）
# order_service.py
"""
import httpx

class PaymentService:
    def __init__(self, api_base_url: str):
        self.client = httpx.AsyncClient(base_url=api_base_url)

    async def charge(self, user_id: str, amount: float) -> dict:
        resp = await self.client.post(
            "/payments/charge",
            json={
                "user_id": user_id,
                "amount": amount,
                "currency": "CNY",
                "source": "balance"
            }
        )
        if resp.status_code == 200:
            return resp.json()
        elif resp.status_code == 402:
            raise InsufficientBalanceError("余额不足")
        elif resp.status_code >= 500:
            raise PaymentServiceError("支付服务异常")
        raise PaymentError(f"未知错误: {resp.status_code}")
"""

# test_payment.py
"""
import pytest
import httpx
from order_service import PaymentService, PaymentError

class TestPaymentService:
    async def test_charge_success(self, httpx_mock):
        httpx_mock.add_response(
            method="POST",
            url="https://pay.example.com/payments/charge",
            json={
                "transaction_id": "txn_001",
                "status": "success",
                "amount": 99.99,
                "balance_remaining": 900.01
            },
            status_code=200
        )

        service = PaymentService("https://pay.example.com")
        result = await service.charge("user_123", 99.99)

        assert result["transaction_id"] == "txn_001"
        assert result["status"] == "success"

    async def test_charge_insufficient_balance(self, httpx_mock):
        httpx_mock.add_response(
            method="POST",
            url="https://pay.example.com/payments/charge",
            json={"error": "余额不足"},
            status_code=402
        )

        service = PaymentService("https://pay.example.com")
        with pytest.raises(PaymentError, match="余额不足"):
            await service.charge("user_123", 99999.0)

    async def test_charge_retry_on_timeout(self, httpx_mock):
        # 第一次超时，第二次成功
        httpx_mock.add_exception(
            httpx.TimeoutException("timeout"),
            method="POST",
            url="https://pay.example.com/payments/charge"
        )
        httpx_mock.add_response(
            method="POST",
            url="https://pay.example.com/payments/charge",
            json={"transaction_id": "txn_002", "status": "success"},
            status_code=200
        )

        service = PaymentService("https://pay.example.com")
        result = await service.charge_with_retry("user_123", 50.0)

        assert result["transaction_id"] == "txn_002"
"""

# 案例二：天气预报API集成测试（vcrpy）
# weather_service.py
"""
import requests

class WeatherService:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.weather.com/v1"

    def get_forecast(self, city: str, days: int = 3) -> dict:
        resp = requests.get(
            f"{self.base_url}/forecast",
            params={
                "city": city,
                "days": days,
                "api_key": self.api_key
            },
            timeout=10
        )
        resp.raise_for_status()
        data = resp.json()
        return {
            "city": data["location"]["name"],
            "country": data["location"]["country"],
            "forecast": [
                {
                    "date": day["date"],
                    "high": day["temp_max"],
                    "low": day["temp_min"],
                    "condition": day["weather"]["description"]
                }
                for day in data["forecast"]
            ]
        }
"""

# test_weather.py - 使用vcrpy录制回放
"""
import vcr
import pytest
from weather_service import WeatherService

# 配置vcr，过滤API key
weather_vcr = vcr.VCR(
    filter_query_parameters=["api_key"],
    filter_headers=["x-api-key"],
    record_mode="once",
    cassette_library_dir="fixtures/cassettes/weather"
)

@weather_vcr.use_cassette()
def test_get_forecast_beijing():
    service = WeatherService(api_key="test-key-123")
    result = service.get_forecast("Beijing", days=3)

    # 验证数据结构（不依赖具体数值，因为数据会变化）
    assert result["city"] == "Beijing"
    assert len(result["forecast"]) == 3
    for day in result["forecast"]:
        assert "date" in day
        assert "high" in day
        assert "low" in day
        assert "condition" in day
        # 温度应该是数值
        assert isinstance(day["high"], (int, float))
        assert day["high"] >= day["low"]

@weather_vcr.use_cassette()
def test_get_forecast_invalid_city():
    service = WeatherService(api_key="test-key-123")
    with pytest.raises(Exception):
        service.get_forecast("NonExistentCity123")
"""

# 案例三：第三方OAuth登录测试
# auth_service.py
"""
import httpx

class OAuthService:
    def __init__(self, client_id: str, client_secret: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = "https://oauth.example.com/token"
        self.user_url = "https://oauth.example.com/userinfo"

    async def authenticate(self, code: str) -> dict:
        # 第一步：用授权码换取access token
        async with httpx.AsyncClient() as client:
            token_resp = await client.post(
                self.token_url,
                data={
                    "grant_type": "authorization_code",
                    "code": code,
                    "client_id": self.client_id,
                    "client_secret": self.client_secret
                },
                headers={"Accept": "application/json"}
            )
            token_resp.raise_for_status()
            token_data = token_resp.json()

        # 第二步：用access token获取用户信息
        access_token = token_data["access_token"]
        async with httpx.AsyncClient() as client:
            user_resp = await client.get(
                self.user_url,
                headers={"Authorization": f"Bearer {access_token}"}
            )
            user_resp.raise_for_status()
            user_data = user_resp.json()

        return {
            "user_id": user_data["id"],
            "name": user_data["name"],
            "email": user_data["email"],
            "token_type": token_data["token_type"]
        }
"""

# test_auth.py - 综合使用模拟方案
"""
import pytest
from auth_service import OAuthService

class TestOAuthService:
    async def test_authenticate_success(self, httpx_mock):
        # 模拟第一步：token换取
        httpx_mock.add_response(
            method="POST",
            url="https://oauth.example.com/token",
            json={
                "access_token": "mock-access-token-abc",
                "token_type": "Bearer",
                "expires_in": 3600
            },
            status_code=200
        )
        # 模拟第二步：用户信息获取
        httpx_mock.add_response(
            method="GET",
            url="https://oauth.example.com/userinfo",
            json={
                "id": "user_456",
                "name": "测试用户",
                "email": "test@example.com"
            },
            status_code=200
        )

        service = OAuthService("client-1", "secret-1")
        result = await service.authenticate("auth-code-xyz")

        assert result["user_id"] == "user_456"
        assert result["name"] == "测试用户"
        assert result["email"] == "test@example.com"

    async def test_authenticate_token_expired(self, httpx_mock):
        # 模拟token过期
        httpx_mock.add_response(
            method="POST",
            url="https://oauth.example.com/token",
            json={"error": "invalid_grant", "error_description": "授权码已过期"},
            status_code=400
        )

        service = OAuthService("client-1", "secret-1")
        with pytest.raises(Exception):
            await service.authenticate("expired-code")

    async def test_authenticate_refresh_flow(self, httpx_mock):
        # token换取成功
        httpx_mock.add_response(
            method="POST",
            url="https://oauth.example.com/token",
            json={"access_token": "token-1", "token_type": "Bearer"},
            status_code=200
        )
        # 用户信息第一次失败(token无效)，第二次成功(刷新后重试)
        httpx_mock.add_response(
            method="GET",
            url="https://oauth.example.com/userinfo",
            json={"error": "invalid_token"},
            status_code=401
        )
        # 刷新token
        httpx_mock.add_response(
            method="POST",
            url="https://oauth.example.com/token",
            json={"access_token": "token-2", "token_type": "Bearer"},
            status_code=200
        )
        # 重试用户信息
        httpx_mock.add_response(
            method="GET",
            url="https://oauth.example.com/userinfo",
            json={"id": "user_789", "name": "Refreshed User", "email": "r@example.com"},
            status_code=200
        )

        service = OAuthService("client-1", "secret-1")
        result = await service.authenticate_with_refresh("some-code")

        assert result["user_id"] == "user_789"
"""