用 TestClient 完成 FastAPI WebSocket 测试的完整流程【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi消息推送功能上线前你怎么确认连接能建立、第一条消息不会错序、断开不会把服务搞挂FastAPI WebSocket 测试的答案就一行用TestClient.websocket_connect()模拟真实客户端连进端点收消息、做断言。不需要起真实服务也不需要第三方库。结论先行用和 HTTP 测试同一个 TestClient。它是内置的进程内客户端直接驱动你的应用不走真实网络。测试函数里只写收消息 断言。建连和断开都由with块自动完成。测试函数必须写成同步def。TestClient 在内部驱动异步应用你不需要写await。一个最小可运行的 WebSocket 测试场景换成聊天房间用户进入 77 号房间时服务端推送一条入群通知。端点和测试一共 14 行from fastapi import FastAPI, WebSocket from fastapi.testclient import TestClient app FastAPI() app.websocket(/room/{room_id}) async def join_room(socket: WebSocket, room_id: str): await socket.accept() await socket.send_json({event: joined, room: room_id}) await socket.close() def test_join_notification(): client TestClient(app) with client.websocket_connect(/room/77) as socket: payload socket.receive_json() assert payload {event: joined, room: 77}端点只做三件事接连接、发通知、关连接。测试断言收到的第一条 JSON 等于期望字典。这段代码到底发生了什么你在端点里写await socket.accept()框架完成握手连接才进入可读状态。你进入with client.websocket_connect(/room/77)框架在进程内完成握手不监听端口。应用是 ASGI 应用即一套框架与服务端约定的通信接口。你调用receive_json()框架读走服务端发出的第一帧解码成 Python 字典。assert做值比较pytest 据此记通过或失败。你退出with时框架关闭连接并释放会话。with块就是上下文管理器意思是块结束时代码会自动帮你做清理。它像一次通话拿起建立连接挂断拆除一切。按消息类型选断言写法按消息类型匹配客户端和服务端的方法再对返回值断言文本消息服务端send_text()测试端receive_text()断言assert text xxx。JSON 消息服务端send_json()测试端receive_json()断言assert data {event: joined}。推荐默认用它结构一目了然。二进制消息服务端send_bytes()测试端receive_bytes()断言assert raw b...。同一条连接混用多种类型时按服务端发出的顺序依次读取用对应的receive_*逐个消费。三个容易翻车的点 ⚠️1. 在 async def 测试里用 TestClient错误症状连接无响应或抛出事件循环相关错误。# ❌ 反例 pytest.mark.anyio async def test_room(): client TestClient(app) # 异步上下文里没有可用事件循环正确写法把测试改回同步def。TestClient 只能在同步调用栈里工作。2. 收发顺序和服务端错位错误症状测试卡死或突然抛出WebSocketDisconnect。# ❌ 服务端只发一次测试却收两次 with client.websocket_connect(/room/77) as socket: first socket.receive_json() second socket.receive_json() # 没有第二帧阻塞正确写法把服务端的send_*列表和测试的receive_*列表逐条对齐一一对应。3. 应用靠 lifespan 初始化数据lifespan 是应用启动和停止时执行的回调常用来预置房间、令牌等数据。直接写client TestClient(app)时lifespan 不会运行端点里的数据还是空壳。# ✅ 两层 with 嵌套 with TestClient(app) as client: with client.websocket_connect(/room/77) as socket: payload socket.receive_json()正确写法外层with启动应用内层with管理这条连接两层都要写。跑完后的自查清单测试函数是同步 def函数体里没有 await每次 receive 都能在服务端找到对应的 send依赖 lifespan 时写成了嵌套的 with TestClient(app)服务端 close() 后断连路径有显式断言直接运行 pytest 能一次跑绿想继续深入读官方教程 Testing WebSockets看断连异常类型的再导出源码 fastapi/websockets.py对比异步测试写法 async-tests.md。需要直接可抄的完整示例参考 docs_src/app_testing/tutorial002_py310.py。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考