FastAPI WebSocket 测试一个 TestClient 就够了【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapiWebSocket 端点写好了一跑测试就卡住——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 并没有自己实现 TestClientfastapi/testclient.py 全文只有一行from starlette.testclient import TestClient as TestClient # noqa所有 WebSocket 测试能力都来自 Starlette 的会话机制FastAPI 只是把它转手给你用。连接会话里怎么收发消息连接建立后你面对的不是一次请求-响应而是一条保持打开的对话。会话对象上有一组收发方法两端一一对应测试端方法作用服务端对应方法receive_json()收一条 JSON 消息并解码send_json()receive_text()收一条文本消息send_text()receive_bytes()收一条二进制消息send_bytes()send_text(...)向服务端发文本receive_text()send_json(...)向服务端发 JSONreceive_json()send_bytes(...)向服务端发二进制receive_bytes()一个回显型端点这样测先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_codeWebSocket 测试面对的是消息流断言对象是逐条收到的数据没有状态码可查。用 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考