给 FastAPI 的 WebSocket 端点加测试到底要写几行代码【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapiFastAPI 内置了 WebSocket 端点测试能力复用同一个TestClient用websocket_connect打开长连接会话再逐条收发并断言即可无需引入任何新测试框架。场景消息丢失如何在 CI 里稳定复现你的聊天端点上线后用户偶尔反馈某条消息没收到本地却很难复现。你需要的是一个在 CI 里每次提交都自动跑的用例只要消息流正确就通过一旦服务端改了收发改逻辑它立刻变红。WebSocket 测试和 HTTP 测试的差异在于没有状态码也没有一次性响应你拿到的是一个长连接和一条消息流断言对象就是会话里逐条收到的消息。机制一句话Starlette 的再导出FastAPI 没有自写测试客户端fastapi/testclient.py 的全部正文只有一行from starlette.testclient import TestClient as TestClient # noqa也就是说websocket_connect提供的是 Starlette 的会话机制FastAPI 这边只是起了个别名。fastapi/websockets.py里的WebSocket、WebSocketDisconnect同样是从 Starlette 再导出的。最小可运行的 FastAPI WebSocket 测试先给测试函数def test_websocket(): client TestClient(app) with client.websocket_connect(/ws) as websocket: data websocket.receive_json() assert data {msg: Hello WebSocket}进入with块即向/ws发起握手receive_json()等待服务端发来的第一条 JSON 消息与断言一致就通过离开代码块时连接自动关闭不用写清理逻辑。函数是普通同步def因为TestClient内部替你驱动了异步应用测试代码本身不需要await。被测端点其实只有三行逻辑出自 docs_src/app_testing/tutorial002_py310.pyfrom fastapi import FastAPI from fastapi.websockets import WebSocket app FastAPI() app.websocket(/ws) async def ws_endpoint(ws: WebSocket): await ws.accept() await ws.send_json({msg: Hello WebSocket}) await ws.close()accept()是握手服务端不调用它客户端会话就不会进入已连接状态随后send_json发消息close是服务端收尾。放进任意测试文件uv run pytest直接运行和其它 pytest 用例没有任何区别。websocket_connect 会话内收发速查会话对象上的三类消息各有一套方法调用会话内方向服务端配对方法send_text()/receive_text()文本双向receive_text()/send_text()send_json()/receive_json()JSON双向receive_json()/send_json()send_bytes()/receive_bytes()二进制双向receive_bytes()/send_bytes()同一个会话里你也可以主动发消息回显端点三条就能测完with client.websocket_connect(/ws) as ws: ws.send_text(ping) assert ws.receive_text() ping服务端每send一次测试端就要receive一次两边必须严格交错。lifespan 嵌套写法与服务端断连如果你的应用靠 lifespan 预置数据lifespan 只在进入with TestClient(app) as client:时才运行所以里面还要再套一层with TestClient(app) as client: with client.websocket_connect(/ws) as ws: data ws.receive_json() assert data {msg: Hello WebSocket}外层负责启动和停掉应用内层管理连接会话。若服务端先调用close()测试端再receive_*会抛出WebSocketDisconnectStarlette 再导出要断言断连路径用pytest.raises包住对应调用即可。同步函数限制与收发顺序测试函数必须是同步的TestClient靠同步调用栈驱动异步 ASGI 应用async def测试函数里不能再用它异步场景下的 WebSocket 要单独设计。收发顺序严格服务端send一次测试端就得receive一次错位会阻塞或断言失败。别用 HTTP 式断言HTTP 测试拿到的是response对象WebSocket 会话里只有消息没有状态码。别漏握手服务端不调accept()客户端会话就完不成连接第一次receive会卡住。FastAPI WebSocket 测试不需要额外框架一个websocket_connect会话加上对收消息的断言就是一条能进 CI 的自动化用例。WebSocket 端点本身怎么写docs/en/docs/advanced/websockets.md应用级测试更多示例含 lifespandocs_src/app_testing/tutorial004_py310.py本示例的回归测试直接导入上面两个测试函数tests/test_tutorial/test_testing/test_tutorial002.py【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
