TestClient 一条 with 语句,跑通 FastAPI WebSocket 测试

TestClient 一条 with 语句,跑通 FastAPI WebSocket 测试 TestClient 一条 with 语句跑通 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()就能发起连接、断言消息。读到这里你手里就是一个可独立运行的 WebSocket 测试函数外加对会话收发、断连异常、lifespan 嵌套几个细节的理解。会话不是一问一答HTTP 测试你闭着眼都会写client TestClient(app) response client.get(/) # 一次请求拿到 response 就结束WebSocket 测试长得不一样连接建立后不会自动断开服务端随时可能推消息所以你拿到的不是 response而是一个会话——一段需要你自己管住开与关的长连接。对应的写法多了一个withPython 的上下文管理器代码块退出时自动执行清理with client.websocket_connect(/ws) as websocket: data websocket.receive_json() # 会话里收一条消息结论一句话同一个TestClient就够用不用引入任何新客户端。这里有个细节fastapi/testclient.py 全部内容就是对 StarletteTestClient的一行再导出会话能力全部继承自 Starlette。握手、收消息、断言、断开四步跑通被测端点精简到三行核心逻辑app.websocket(/ws) async def websocket(websocket: WebSocket): await websocket.accept() # 接受握手 await websocket.send_json({msg: Hello WebSocket}) await websocket.close() # 服务端主动关闭这三行决定了测试怎么读没accept()之前连接处于半开状态send_json()推出一条 JSON 消息close()收尾。测试函数就四行def test_websocket(): client TestClient(app) with client.websocket_connect(/ws) as websocket: data websocket.receive_json() # 收服务端第一条消息 assert data {msg: Hello WebSocket}逐行拆开TestClient(app)造出客户端websocket_connect(/ws)进入with时完成真正的 WebSocket 握手并打开会话receive_json()阻塞等待第一条 JSON 并自动解码with块退出时会话自动关闭不用手写清理。跑完你会看到 pytest 直接给一个绿勾。注意测试函数是普通的def全程没有await——TestClient会在同步调用栈里替你驱动那些async def端点测试代码本身保持同步就行。会话里的收发工具箱以及顺序陷阱方法用途websocket.receive_text()收一条文本消息websocket.receive_json()收一条 JSON 消息并自动解码websocket.receive_bytes()收一条二进制消息websocket.send_text(...)发文本给服务端websocket.send_json(...)发 JSON 给服务端websocket.send_bytes(...)发二进制给服务端send_*和receive_*配着就能做客户端发、服务端应的对话式测试比如回显端点def test_echo(): client TestClient(app) with client.websocket_connect(/ws) as websocket: websocket.send_text(Hello, server) assert websocket.receive_text() Hello, server⚠️ 小心一点WebSocket 是消息流不是一问一答。收发顺序必须和服务端严格对齐——服务端send_*几次你就receive_*几次错位不是报错而是挂起测试会一直阻塞等你。再就是断连。服务端close()之后测试端再receive_*会抛出WebSocketDisconnectfastapi.websockets从 Starlette 再导出测试和应用代码引用的是同一个类型。想专门验证断连路径时用pytest.raises(WebSocketDisconnect)包住那一次receive_text()即可。两个容易翻车的点异步测试函数 lifespan 嵌套坑 1TestClient不能出现在async def测试函数里。它靠同步调用栈去驱动异步的 ASGI 应用一旦测试函数本身跑在事件循环上再套一层就冲突了。同步写def这个坑不存在。坑 2应用靠lifespan初始化资源预置字典、建连接池之类那么只有进入with TestClient(app)时 lifespan 才被触发WebSocket 测试得写成双层嵌套def test_websocket_with_lifespan(): with TestClient(app) as client: # 外层启动/关闭 lifespan with client.websocket_connect(/ws) as websocket: assert websocket.receive_json() {msg: Hello WebSocket}外层with负责把应用启动起来内层with管 WebSocket 会话。普通 HTTP 测试遇到 lifespan 也是同一套嵌套规则WebSocket 只是多套了一层连接会话。什么时候该跳出 TestClient如果测试环境本身就是async def比如pytest.mark.anyioHTTP 请求有现成替代httpx.AsyncClient配ASGITransport直接打 ASGI 应用。但 ⚠️ WebSocket 在异步测试里没有对等的websocket_connect()会话写法别照搬上面的同步会话得单独设计——单独建一个同步测试文件或在真实浏览器/客户端里验证。想接着挖仓库里有这三处入口WebSocket 端点教程、app_testing 示例目录tutorial001~004 覆盖 HTTP 与 lifespan 场景以及 fastapi/websockets.py 里会话类型和异常的定义。给自己写个 echo 端点用send_text()/receive_text()跑一遍上面的对话式测试——四条语句两分钟的事。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考