如何使用httpx-pytest模拟httpx.ConnectError编写测试用例

核心逻辑说明 #

你的异常捕获逻辑本身是生效的：httpx.ConnectError 是 httpx.NetworkError 的子类，目标服务未启动触发的连接拒绝错误，会直接进入NetworkError的捕获分支，返回code=503的GSResponse。
写这类异常场景的测试不要依赖真实的本地端口状态：如果直接写死请求未启动的localhost:8080，一旦本地/CI环境8080端口被其他服务占用，用例会直接失效，稳定性极差。标准做法是通过mock替换HTTP客户端的请求方法，主动抛出对应异常、返回预设响应，就能稳定覆盖所有代码分支。

测试用例实现 #

基于Python标准库自带的unittest.mock加pytest实现即可，不需要额外安装第三方依赖，示例代码如下：

from unittest.mock import Mock
import httpx
import pytest
# 替换成你自己项目的实际导入路径
from your_project.module import GeoServerClient, WorkspaceModel, GSResponse


def test_get_workspace_when_service_down():
    """覆盖目标服务未启动、抛出ConnectError的场景"""
    # 初始化业务客户端
    gs_client = GeoServerClient()
    # 替换http client的get方法，主动抛出连接错误，模拟服务未启动
    mock_request = Mock(side_effect=httpx.ConnectError("Connection refused"))
    gs_client.http_client.get = mock_request

    resp = gs_client.get_workspace("test_ws")

    # 断言返回结果符合预期
    assert isinstance(resp, GSResponse)
    assert resp.code == 503
    assert resp.response == "Geoserver unavailable"
    # 确认请求路径拼接正确
    mock_request.assert_called_once_with("workspaces/test_ws")


def test_get_workspace_when_timeout():
    """覆盖请求超时的场景"""
    gs_client = GeoServerClient()
    mock_request = Mock(side_effect=httpx.TimeoutException("Request timeout"))
    gs_client.http_client.get = mock_request

    resp = gs_client.get_workspace("test_ws")
    assert isinstance(resp, GSResponse)
    assert resp.code == 504
    assert resp.response == "Timeout Error"


def test_get_workspace_success():
    """覆盖接口正常返回200的场景"""
    gs_client = GeoServerClient()
    # 构造200成功响应
    mock_resp = Mock()
    mock_resp.status_code = 200
    mock_resp.json.return_value = {
        "name": "test_ws",
        "href": "http://localhost:8080/geoserver/rest/workspaces/test_ws"
    }
    mock_request = Mock(return_value=mock_resp)
    gs_client.http_client.get = mock_request

    resp = gs_client.get_workspace("test_ws")
    assert isinstance(resp, WorkspaceModel)
    assert resp.name == "test_ws"


def test_get_workspace_non_200():
    """覆盖接口返回非200状态码的场景"""
    gs_client = GeoServerClient()
    # 构造404响应
    mock_resp = Mock()
    mock_resp.status_code = 404
    mock_request = Mock(return_value=mock_resp)
    gs_client.http_client.get = mock_request
    # mock响应解析方法的返回值
    expected_resp = GSResponse.parse_obj({"code": 404, "response": "Workspace not found"})
    gs_client.response_recognise = Mock(return_value=expected_resp)

    resp = gs_client.get_workspace("test_ws")
    assert resp == expected_resp

注意事项 #

• 如果你的http_client是在类初始化时内部构造的，不方便直接替换属性，可以用patch装饰器在测试时替换httpx.Client的构造逻辑，实现效果完全一致。
• 不需要为了测试专门去启停本地服务，mock方式的运行速度比真实请求快几个数量级，也不会有环境污染问题。

内容的提问来源于stack exchange，提问作者krishna lodha