尧图网站建设 
FastAPI WebSocket 测试TestClient 会话式断言的三个关键写法【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi给 FastAPI 项目补测试时HTTP 路由总是最先被覆盖WebSocket 端点却常被搁置。其实 FastAPI 的 WebSocket 测试不需要任何新工具同一个 TestClient 配合 websocket_connect 就能打开一条长连接会话对逐条收到的消息做断言。下文先给最小可运行示例再拆解会话机制、断连处理与 lifespan 等关键写法。FastAPI WebSocket 测试最小示例先跑通端点三行测试五行拼在一起就是一个能进 CI 的最小用例from fastapi import FastAPI from fastapi.testclient import TestClient from fastapi.websockets import WebSocket app FastAPI() app.websocket(/ws) async def ws_endpoint(websocket: WebSocket): await websocket.accept() await websocket.send_json({event: hello, seq: 1}) await websocket.close() def test_ws_first_frame(): client TestClient(app) with client.websocket_connect(/ws) as ws: payload ws.receive_json() assert payload {event: hello, seq: 1}服务端三动作有严格先后accept() 完成握手send_json() 推送一帧close() 主动收尾测试端只有 accept 之后才算真正接通。websocket_connect 返回会话对象with 块就是连接的生命周期退出时自动断开不需要清理代码。测试函数保持普通同步写法TestClient 会在内部驱动异步应用测试代码里不用写 await。直接用 pytest 跑即可无需任何插件。为什么 WebSocket 测试是会话而不是请求 把 HTTP 想成写信发出一封、回一封彼此独立邮差还会附一张回执状态码。WebSocket 则是打电话拨通之后线路一直开着双方轮流说话直到一方挂断这通电话才结束。websocket_connect 相当于拨号接通返回的会话对象就是这条开着线包在 with 里等于块开始时接通、块结束时挂断连接不会泄漏。顺带说一句实现来源TestClient 在 fastapi/testclient.py 里只有一行导入实质是 Starlette 实现的再导出所以会话能力包括 websocket_connect全部继承自 Starlette。websocket_connect 会话内如何断言会话对象上的收发方法与真实客户端一一对应断言思路就是发一帧、收一帧、比一次测试端方法对端配对方法断言时机receive_text()send_text(...)与预期字符串比对receive_json()send_json(...)与预期 dict 比对receive_bytes()send_bytes(...)与预期字节串比对send_text(...)receive_text()测试端主动发起一轮对话send_json(...)receive_json()发送结构化请求数据send_bytes(...)receive_bytes()发送二进制负载下面用一个消息计数端点演示对话式断言服务端每收到一帧文本就回一帧带累计次数的 JSON。def test_counter_dialog(): client TestClient(app) with client.websocket_connect(/ws) as ws: for tick in range(3): ws.send_text(ftick {tick}) reply ws.receive_json() assert reply[count] tick 1三轮循环里 send 与 receive 交替出现测试因此同时验证了服务端读一帧、算一次、答一帧的完整逻辑。时序纪律读写次序与断连断言次序敏感WebSocket 是消息流不是一对一的请求-响应。服务端每推一帧测试端就要有一帧在等次序错位时测试多半不是报错而是安静地卡在某次 receive 上。断连即异常服务端执行 close() 之后测试端继续调用任何 receive_* 都会抛出 WebSocketDisconnect。可断言的断连路径预期服务端会挂断时用 pytest.raises(WebSocketDisconnect) 包住那一次 receive把被挂断变成一条明确断言而不是靠超时间接暴露。lifespan 下嵌套 TestClient 的写法应用若用 lifespan 预置状态参考 tutorial004 的 lifespan 示例只有进入外层 with 时应用才算启动。这时 FastAPI WebSocket 测试需要双层上下文def test_ws_inside_lifespan(): with TestClient(app) as client: with client.websocket_connect(/ws) as ws: greeting ws.receive_text() assert greeting ready外层 with 负责应用的启动与关闭lifespan 的初始化与清理都发生在这一层内层 with 只管这一条 WebSocket 连接。没有 lifespan 的应用省略外层即可这也是开头示例的写法。避坑清单四个误区对照 误区在 async def 测试函数里构造 TestClient。正确做法TestClient 是同步驱动器异步测试函数里用不了它改走 httpx.AsyncClient ASGITransport 直接驱动 ASGI 应用仓库 docs_src/async_tests 目录下的示例就是这个路线。误区在 WebSocket 会话上找 status_code。正确做法会话没有状态码断言对象永远是逐条收到的消息。误区想先把两帧都读完再一起比对。正确做法读写次序要严格跟着服务端的处理次序走收一帧、断一帧。误区认为服务端 close 之后会话只是没数据了。正确做法close 之后 receive_* 会抛 WebSocketDisconnect这是可以用 pytest.raises 验证的正常路径。延伸阅读WebSocket 端点编写教程从 accept 到循环收发的聊天室端点本文所有测试对象的服务端写法都出自这里。WebSocket 符号定义WebSocket、WebSocketDisconnect、WebSocketState 三个符号自 Starlette 再导出是断连断言中异常的来源。test_websockets 回归测试目录WebSocket 教程配套的测试用例其中包含依赖注入与 WebSocket 组合场景的断言写法。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 FastAPI WebSocket 测试：TestClient 会话式断言的三个关键写法
FastAPI WebSocket 测试TestClient 会话式断言的三个关键写法【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi给 FastAPI 项目补测试时HTTP 路由总是最先被覆盖WebSocket 端点却常被搁置。其实 FastAPI 的 WebSocket 测试不需要任何新工具同一个 TestClient 配合 websocket_connect 就能打开一条长连接会话对逐条收到的消息做断言。下文先给最小可运行示例再拆解会话机制、断连处理与 lifespan 等关键写法。FastAPI WebSocket 测试最小示例先跑通端点三行测试五行拼在一起就是一个能进 CI 的最小用例from fastapi import FastAPI from fastapi.testclient import TestClient from fastapi.websockets import WebSocket app FastAPI() app.websocket(/ws) async def ws_endpoint(websocket: WebSocket): await websocket.accept() await websocket.send_json({event: hello, seq: 1}) await websocket.close() def test_ws_first_frame(): client TestClient(app) with client.websocket_connect(/ws) as ws: payload ws.receive_json() assert payload {event: hello, seq: 1}服务端三动作有严格先后accept() 完成握手send_json() 推送一帧close() 主动收尾测试端只有 accept 之后才算真正接通。websocket_connect 返回会话对象with 块就是连接的生命周期退出时自动断开不需要清理代码。测试函数保持普通同步写法TestClient 会在内部驱动异步应用测试代码里不用写 await。直接用 pytest 跑即可无需任何插件。为什么 WebSocket 测试是会话而不是请求 把 HTTP 想成写信发出一封、回一封彼此独立邮差还会附一张回执状态码。WebSocket 则是打电话拨通之后线路一直开着双方轮流说话直到一方挂断这通电话才结束。websocket_connect 相当于拨号接通返回的会话对象就是这条开着线包在 with 里等于块开始时接通、块结束时挂断连接不会泄漏。顺带说一句实现来源TestClient 在 fastapi/testclient.py 里只有一行导入实质是 Starlette 实现的再导出所以会话能力包括 websocket_connect全部继承自 Starlette。websocket_connect 会话内如何断言会话对象上的收发方法与真实客户端一一对应断言思路就是发一帧、收一帧、比一次测试端方法对端配对方法断言时机receive_text()send_text(...)与预期字符串比对receive_json()send_json(...)与预期 dict 比对receive_bytes()send_bytes(...)与预期字节串比对send_text(...)receive_text()测试端主动发起一轮对话send_json(...)receive_json()发送结构化请求数据send_bytes(...)receive_bytes()发送二进制负载下面用一个消息计数端点演示对话式断言服务端每收到一帧文本就回一帧带累计次数的 JSON。def test_counter_dialog(): client TestClient(app) with client.websocket_connect(/ws) as ws: for tick in range(3): ws.send_text(ftick {tick}) reply ws.receive_json() assert reply[count] tick 1三轮循环里 send 与 receive 交替出现测试因此同时验证了服务端读一帧、算一次、答一帧的完整逻辑。时序纪律读写次序与断连断言次序敏感WebSocket 是消息流不是一对一的请求-响应。服务端每推一帧测试端就要有一帧在等次序错位时测试多半不是报错而是安静地卡在某次 receive 上。断连即异常服务端执行 close() 之后测试端继续调用任何 receive_* 都会抛出 WebSocketDisconnect。可断言的断连路径预期服务端会挂断时用 pytest.raises(WebSocketDisconnect) 包住那一次 receive把被挂断变成一条明确断言而不是靠超时间接暴露。lifespan 下嵌套 TestClient 的写法应用若用 lifespan 预置状态参考 tutorial004 的 lifespan 示例只有进入外层 with 时应用才算启动。这时 FastAPI WebSocket 测试需要双层上下文def test_ws_inside_lifespan(): with TestClient(app) as client: with client.websocket_connect(/ws) as ws: greeting ws.receive_text() assert greeting ready外层 with 负责应用的启动与关闭lifespan 的初始化与清理都发生在这一层内层 with 只管这一条 WebSocket 连接。没有 lifespan 的应用省略外层即可这也是开头示例的写法。避坑清单四个误区对照 误区在 async def 测试函数里构造 TestClient。正确做法TestClient 是同步驱动器异步测试函数里用不了它改走 httpx.AsyncClient ASGITransport 直接驱动 ASGI 应用仓库 docs_src/async_tests 目录下的示例就是这个路线。误区在 WebSocket 会话上找 status_code。正确做法会话没有状态码断言对象永远是逐条收到的消息。误区想先把两帧都读完再一起比对。正确做法读写次序要严格跟着服务端的处理次序走收一帧、断一帧。误区认为服务端 close 之后会话只是没数据了。正确做法close 之后 receive_* 会抛 WebSocketDisconnect这是可以用 pytest.raises 验证的正常路径。延伸阅读WebSocket 端点编写教程从 accept 到循环收发的聊天室端点本文所有测试对象的服务端写法都出自这里。WebSocket 符号定义WebSocket、WebSocketDisconnect、WebSocketState 三个符号自 Starlette 再导出是断连断言中异常的来源。test_websockets 回归测试目录WebSocket 教程配套的测试用例其中包含依赖注入与 WebSocket 组合场景的断言写法。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考