FastAPI WebSocket 测试5 个决定测试是挂起还是跑通的细节【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi本地端点跑得好好的浏览器手动连也通一上 CI 却挂起 60 秒被超时杀掉。这类 FastAPI WebSocket 测试失败几乎从来不是框架的问题而是没看懂「连接会话」和服务端消息时序的配对规则。一句话讲完WebSocket 测试不是请求-响应而是一场必须和服务端脚本逐行对上的对话。TestClient websocket_connect 的最小通过组合官方示例 docs_src/app_testing/tutorial002_py310.py 把被测端点和测试放进同一个文件。没有额外依赖这就是全部app.websocket(/ws) async def ws(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()→close()三件套是服务端必须完成的最小动作序列测试函数则是可独立调用的最小运行单元官方回归测试就是直接 import 后调用它用常规的uv run pytest跑起来。websocket_connect 的握手与清理时序进入with client.websocket_connect(/ws)块不是在发一次请求而是在建立一条长连接会话进入时执行真实的 WebSocket 握手服务端必须调用await websocket.accept()完成应答101客户端才算进入已连接状态、开始收到消息退出块时会话自动关闭不用手动清理。HTTP 测试和 WebSocket 测试为什么能用同一个TestClient看 fastapi/testclient.py 就一行from starlette.testclient import TestClient as TestClient。FastAPI 完全没有重写测试客户端——websocket_connect()的会话机制、六组收发方法、WebSocketDisconnect异常类型由 fastapi/websockets.py 再导出能力全部来自 Starlette。所以不用引入任何新测试框架要理解的只有「连接会话」这一种新对象。WebSocket 消息断言收发镜像规则视角放成对话客户端发一条、服务端回一条测试代码就是严格镜像服务端收发顺序的脚本。三种消息类型各一行带过文本测试端websocket.send_text(hi)对应服务端await websocket.receive_text()JSONwebsocket.send_json({...})对应await websocket.receive_json()二进制websocket.send_bytes(b\x01)对应await websocket.receive_bytes()接收方向对称receive_text()、receive_json()、receive_bytes()三选一。坑在顺序错位的两种典型表现测试端receive_*的时机早于服务端发送、或服务端在等一条测试端根本没发的消息测试就挂起不返回——这就是 CI 被超时杀掉的真凶服务端已经close()之后测试端再receive_*则抛WebSocketDisconnect异常而不是优雅返回。想验证断连路径时用pytest.raises(WebSocketDisconnect)去断言反而是正确写法关键是错配必须是你有意设计的不是写串了。lifespan 场景下双层 with 的嵌套写法应用若在 lifespan 里初始化资源比如启动时填充一个字典记住lifespan 只有在进入with TestClient(app) as client:时才触发单纯实例化不触发。于是 WebSocket 测试变成两层——外层管「应用活着」内层管「单条连接」同一测试里还可以先后开多条独立连接def test_ws_with_lifespan(): with TestClient(app) as client: with client.websocket_connect(/ws) as ws1: assert ws1.receive_json() {n: 1} with client.websocket_connect(/ws) as ws2: assert ws2.receive_json() {n: 2}ws1和ws2是两个完全独立的会话各自有独立的收发流和服务端状态测「第一个客户端推送状态给第二个客户端」这类场景就靠这个写法。⚠️ 踩坑清单在async def测试函数里实例化TestClientTestClient 的本质是在同步调用栈里驱动异步 ASGI 应用也就是 FastAPI 同步测试驱动异步应用的全部魔法测试函数一旦写成async def机制即失效要么报错要么挂起。测试函数必须是同步def。把 HTTP 断言习惯套到会话上这里没有response.status_code、没有response.json()。WebSocket 会话的断言对象是一条条收到的消息不存在状态码。指望服务端close()之后receive_*返回 None它抛WebSocketDisconnect。要验证「服务端主动关闭」断言的是异常而不是 None。收发顺序错位测试端和服务端是同一条消息流的两端不是两个独立请求每次错位都是挂起或异常二选一。✅ 跑之前自检 Checklist被测端点在任何收发之前先调用了accept()测试断言了首条消息最容易出 bug 的是第一条测试端receive_*的次数与服务端send_*的次数、顺序一一对应断连路径被覆盖pytest.raises(WebSocketDisconnect)涉及 lifespan 时外层包了with TestClient(app)以触发启动与关闭【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考