企业官网建设流程全解析

FastAPI WebSocket 测试：一个 TestClient 就够了

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

WebSocket 端点写好了，一跑测试就卡住——receive_json()永远等不到数据，断言全红。多数情况不是框架的问题，而是你的测试没把那条保持打开的对话管起来。用 FastAPI 的 TestClient 测 WebSocket，核心就一件事：先建连接会话，再逐条收消息、逐条断言。

最小可跑示例：端点与测试函数放一起

仓库里的最小示例在 docs_src/app_testing/tutorial002_py310.py，端点和测试函数写在同一个文件里：

from fastapi import FastAPI from fastapi.testclient import TestClient from fastapi.websockets import WebSocket app = FastAPI() @app.websocket("/ws") async def websocket(websocket: WebSocket): await websocket.accept() await websocket.send_json({"msg": "Hello WebSocket"}) await websocket.close() def test_websocket(): client = TestClient(app) with client.websocket_connect("/ws") as websocket: data = websocket.receive_json() assert data == {"msg": "Hello WebSocket"}

端点侧三件事各管一段：accept()完成握手，连接才真正通；send_json()以 JSON 编码推一条消息；close()由服务端收尾断开。

测试函数是普通同步def，不需要await。TestClient 内部会替你驱动异步应用，你的测试代码只管同步写。退出with块时，连接自动断开，不用手动清理。

同一个客户端为什么够用：一行再导出的证据

你测 HTTP 用的TestClient，测 WebSocket 照样能跑，不需要引入新的测试客户端。FastAPI 并没有自己实现 TestClient，fastapi/testclient.py 全文只有一行：

from starlette.testclient import TestClient as TestClient # noqa

所有 WebSocket 测试能力都来自 Starlette 的会话机制，FastAPI 只是把它转手给你用。

连接会话里怎么收发消息

连接建立后，你面对的不是一次请求-响应，而是一条保持打开的对话。会话对象上有一组收发方法，两端一一对应：

一个回显型端点这样测：先send_text("hi")，再receive_text()拿回"hi"，断言相等。

收发顺序必须与服务端处理顺序严格一致。服务端每send_*一次，你就该receive_*一次；反过来也一样。顺序错位，测试要么阻塞，要么断言失败。

两个高频坑：断连异常怎么断言、lifespan 为什么要多包一层

坑一：服务端close()后，receive_*会抛WebSocketDisconnect。

服务端调了close()，测试端再receive_*就会触发这个异常。要验证"断连"这条路径，用pytest.raises(WebSocketDisconnect)包住那个 receive 调用即可。WebSocketDisconnect从 fastapi/websockets.py 再导出，应用代码和测试代码引用的是同一个类。

坑二：应用依赖 lifespan 时，要再多一层with。

如果应用靠lifespan初始化状态，只有进入with TestClient(app) as client:时 lifespan 才真正启动。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"}

外层负责拉起和关闭应用，内层负责那条对话的生命周期。

三条边界：同步函数、消息顺序、断言对象

仅适用于同步测试函数。TestClient 靠同步调用栈驱动异步 ASGI 应用；如果你的测试函数本身是async def，就不能在函数体内再用 TestClient，得另想办法。

别把 HTTP 的状态码断言套到 WebSocket 上。HTTP 测试断言response.status_code，WebSocket 测试面对的是消息流，断言对象是逐条收到的数据，没有状态码可查。

用 pytest 把测试跑起来

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

uv run pytest

仓库的回归测试直接导入示例文件里的两个测试函数并调用，印证了这套写法本身就是最小可运行单元。

延伸阅读

• WebSocket 端点怎么写：docs/en/docs/advanced/websockets.md
• 应用级测试更多示例：docs_src/app_testing/
• WebSocket 测试专题文档：docs/en/docs/advanced/testing-websockets.md

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