企业官网建设流程全解析

FastAPI WebSocket 测试：一个 TestClient 断言所有消息

【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi

WebSocket 端点已经能跑了，但谁来保证它不会悄悄挂掉？聊天、实时推送这类双向通道，消息内容或收发时序一旦出问题，往往要拖到生产环境才暴露。这篇文章只解决一件事：FastAPI WebSocket 测试怎么做——如何复用同一个 TestClient 建立连接会话、接收消息并完成断言。读完就能写出可独立运行的 WebSocket 测试函数，并弄清底层的会话机制。

为什么不用额外测试框架

先看源码，FastAPI 对 StarletteTestClient的处理只有一行再导出：

from starlette.testclient import TestClient as TestClient # noqa

fastapi/testclient.py 的全部内容就是这一行。也就是说 FastAPI 没有重新造测试客户端，而是直接复用 Starlette 的实现；同理 fastapi/websockets.py 也只是把 Starlette 的WebSocket、WebSocketDisconnect、WebSocketState重新导出给业务代码引用。

所以 WebSocket 测试不需要任何新工具，同一个TestClient就能覆盖。区别只在进入方式：HTTP 请求用client.get("/")、client.post(...)这类方法，发一次请求拿一个响应；WebSocket 则要先用with client.websocket_connect("/ws")建立一条长连接会话，再在会话里反复收发消息。

最小可跑示例：从握手到断言一条消息

官方文档 docs/en/docs/advanced/testing-websockets.md 引用的完整示例在 docs_src/app_testing/tutorial002_py310.py。被测端点负责握手、发一条 JSON 消息、再主动关闭连接：

from fastapi import FastAPI from fastapi.testclient import TestClient from fastapi.websockets import WebSocket app = FastAPI() @app.get("/") async def read_main(): return {"msg": "Hello World"} @app.websocket("/ws") async def websocket(websocket: WebSocket): await websocket.accept() await websocket.send_json({"msg": "Hello WebSocket"}) await websocket.close()

端点里三个关键调用，逐个拆开：

• accept()：接受握手，连接才真正建立
• send_json()：按 JSON 编码发一条消息
• close()：服务端主动关闭连接

对应的测试函数（示例文件第 27～31 行）用with语句建立连接会话，并在会话内断言收到的第一条消息：

def test_websocket(): client = TestClient(app) with client.websocket_connect("/ws") as websocket: data = websocket.receive_json() assert data == {"msg": "Hello WebSocket"}

第 30 行的receive_json()：接收首条 JSON 并自动解码。进入with块时会向/ws发起真实的 WebSocket 握手；with 语句是上下文管理器（context manager），退出时会话自动关闭，不用手写清理逻辑。测试函数本身是普通同步函数：TestClient会在内部驱动async def的异步应用，测试代码里完全不需要await。

搬进 pytest：和 HTTP 测试长一个样

同文件里还有一个普通 HTTP 用例，可以对照感受两者的结构一致性：

def test_read_main(): client = TestClient(app) response = client.get("/") assert response.status_code == 200 assert response.json() == {"msg": "Hello World"}

运行方式和其它 pytest 测试完全一样：

uv run pytest

也可以只跑示例所在的测试目录或文件。两个用例都是同步def test_...函数，没有async def、没有await，这是TestClient驱动 ASGI 应用的前提。

仓库里对应的回归测试是 tests/test_tutorial/test_testing/test_tutorial002.py，它直接导入并调用示例里的两个测试函数：

from docs_src.app_testing.tutorial002_py310 import test_read_main, test_websocket def test_main(): test_read_main() def test_ws(): test_websocket()

这说明官方示例本身就是可执行、可断言的最小单元，照这个结构组织自己的 WebSocket 测试没有问题。

会话里的收发工具箱

在with client.websocket_connect(...)打开的会话对象上，可以像真实客户端一样自由收发，常用方法如下：

一个回显（echo）型端点的"客户端发送、服务端应答"完整对话测试，大致长这样：

def test_websocket_echo(): client = TestClient(app) with client.websocket_connect("/ws") as websocket: websocket.send_text("Hello, server") data = websocket.receive_text() assert data == "Hello, server"

这里有个硬性要求：收发顺序必须和服务端的处理顺序严格对齐。服务端每send_*一次，测试端就该receive_*一次，反之亦然；消息读取顺序一旦错位，测试会直接阻塞或断言失败。

两个高频坑

坑 A：服务端close()之后，receive_*会抛WebSocketDisconnect。服务端收尾的常规动作是await websocket.close()，连接关闭后测试端再调receive_*通常就抛出WebSocketDisconnect。这个异常由 fastapi/websockets.py 从starlette.websockets重新导出，应用和测试代码统一从它引用。如果端点要验证不同关闭码、或者要在断连前做清理，用pytest.raises把相应调用包起来，专门断言断连路径。

坑 B：应用依赖 lifespan 时，必须嵌套两层 with。像 docs_src/app_testing/tutorial004_py310.py 那样用lifespan预置items字典的应用，状态只在进入with TestClient(app) as client:时才初始化。此时 WebSocket 测试要写成嵌套形式：

def test_websocket_with_lifespan(): with TestClient(app) as client: with client.websocket_connect("/ws") as websocket: data = websocket.receive_json() assert data == {"msg": "Hello WebSocket"}

外层with负责启动和关闭 lifespan，保证应用处于运行状态；内层with只管连接会话。和普通 HTTP 测试"访问 lifespan 资源要在with TestClient(app)块内"是同一套规则，只是 WebSocket 多套了一层连接。

边界与误区

• 仅限同步测试函数：TestClient靠同步调用栈驱动异步应用，测试函数若是async def（比如 docs_src/async_tests/app_a_py310/test_main.py 里用httpx.AsyncClient与ASGITransport发 HTTP 请求的路子，背景见 docs/en/docs/advanced/async-tests.md），函数体内就不能再用它，WebSocket 策略得单独设计。
• 别混断言对象：HTTP 断言的是response（看status_code、response.json()），WebSocket 没有状态码，断言对象是会话里逐条收到的消息。
• 消息流不等于请求-响应：WebSocket 消息不是一一对应的，测试必须严格模拟真实客户端的消息时序。

延伸

• WebSocket 端点本身怎么写，看 docs/en/docs/advanced/websockets.md——是写测试前的前置知识。
• 应用级测试更多示例，看 docs_src/app_testing 目录下tutorial001到tutorial004——覆盖含 lifespan 的场景，是很好的起步模板。
• 会话生命周期与异常类型的底层定义，看 fastapi/websockets.py——内容再导出自 Starlette，排查异常来源时方便。
• WebSocket 教程与依赖作用域的既有回归用例，看 tests/test_tutorial/test_websockets 目录——组织自己的测试时可作参照。

【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi