FastAPI WebSocket 测试：3 行建连会话，消息断言只需 1 个 assert

FastAPI WebSocket 测试3 行建连会话消息断言只需 1 个 assert【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapiWebSocket 端点没有状态码也没有响应体HTTP 那套断言习惯在这里全部失效。本文用一个回显服务做例子讲清楚 FastAPI WebSocket 测试中如何用TestClient的websocket_connect建立长会话、如何对逐条消息发送与断言、以及断连和 lifespan 该怎么处理。看完你可以直接拿走一份能跑的测试模板。为什么 HTTP 的 assert 套不进 WebSocket测 HTTP 接口时你断言的都是那一次请求拿回来的response对象status_code、headers、json()。WebSocket 不是——一条连接上会流过多条消息服务端可以先发一条再等你发消息之间没有固定次序。所以测试对象从「一个响应」变成了「一条消息流」测试工具得给你一个能保持连接的会话而不是一次性请求。换句话说你需要的不是更强的 HTTP 客户端而是一个会话模拟器。HTTP 接口测试WebSocket 测试断言对象response状态码、body会话内逐条收到的消息连接形态请求-响应自动结束with打开的长会话入口调用client.get(/path)client.websocket_connect(/path)断连信号无服务端断开体现为WebSocketDisconnect表格右边的websocket_connect就是 FastAPI WebSocket 测试的核心入口它返回一个上下文管理器——那个会话本身。60 秒写一个能跑的多轮回显测试先写被测端点/realtime/echo客户端每发一条文本就回一条带echo:前缀的回复连接一直保持from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.testclient import TestClient app FastAPI() app.websocket(/realtime/echo) async def echo_endpoint(websocket: WebSocket): await websocket.accept() try: while True: text await websocket.receive_text() await websocket.send_text(fecho: {text}) except WebSocketDisconnect: passaccept()负责完成握手while True循环是常驻连接的主体WebSocketDisconnect是客户端挂掉时的出口三者缺一不可。对应的测试函数只有 8 行做了两轮收发模拟一段真实对话def test_echo_roundtrip(): client TestClient(app) with client.websocket_connect(/realtime/echo) as websocket: websocket.send_text(ping) assert websocket.receive_text() echo: ping websocket.send_text(pong) assert websocket.receive_text() echo: pong关键在每对send_text/receive_text的配对上服务端回一次测试就读一次这个一对一就是 WebSocket 消息断言的核心规则。另外注意测试函数是同步的def——TestClient内部会驱动异步端点测试体里不需要await如果消息内容是字典把send_text/receive_text换成send_json/receive_json即可编解码是自动的。在项目目录里直接跑uv run pytest -v跑通之后浏览器里的回显长这样测试断言的内容就是这里出现的消息with 语句背后发生了什么一次连接的时序with client.websocket_connect(/realtime/echo)这一行实际推进了四步握手客户端发出 HTTP Upgrade 请求服务端回 101连接切换为 WebSocket 协议accept处理函数执行await websocket.accept()只有在这之后客户端会话才算真正进入连接状态服务端漏掉这一步时测试会卡在with这一行不动帧send_text/receive_text各自对应一个消息帧测试端和服务端读写的同一条队列谁先读、读几次完全由代码顺序决定关闭退出with块时客户端自动发 close 帧服务端的receive_text()随即抛出WebSocketDisconnect处理函数走 except 分支结束。把「测试代码是客户端处理函数是必须逐帧应答的服务端」记下来整条时序就不会乱。整个过程中没有引入任何新的客户端类fastapi/testclient.py 只有一行再导出WebSocketDisconnect出自 fastapi/websockets.py同样再导出自 Starlette。所以这里学的是 ASGI 层的会话机制换到别的 ASGI 框架里照样成立。两个实战模式嵌套 lifespan、捕获断连lifespan 资源要多包一层 with。应用如果用 lifespan 初始化共享数据这些数据要等进入with TestClient(app)之后才构建WebSocket 会话就得再往里嵌一层def test_echo_with_lifespan(): with TestClient(app) as client: # 外层触发 lifespan 启动 with client.websocket_connect(/realtime/echo) as websocket: websocket.send_text(hi) assert websocket.receive_text() echo: hi # 退出外层 with 时 lifespan 才 shutdown外层管「应用活着吗」内层管「连接活着吗」。只写内层的话测试照样绿但 lifespan 根本没跑等于把最容易出问题的部分漏掉了。服务端主动断连要显式捕获。端点若设计成「发一条 ready 通知就关闭」测试端的第二次receive_text()会抛出WebSocketDisconnect这时应该用pytest.raises把它变成一条断言把「服务端会主动挂断」这个行为固化下来def test_server_closes_after_ready(): client TestClient(app) with pytest.raises(WebSocketDisconnect): with client.websocket_connect(/realtime/ping) as websocket: assert websocket.receive_text() ready websocket.receive_text() # 服务端已关闭这里抛异常不捕获时测试会因未处理异常直接变红而且红得没有信息量。 五个高频坑#坑症状解法1服务端漏写accept()测试卡在with行不动处理函数开头补await websocket.accept()2测试函数写成async defTestClient无法在异步上下文驱动改同步defFastAPI 异步测试场景另选客户端策略3收发顺序与服务端错位测试阻塞或断言值对不上严格对照服务端代码逐条配对 send/receive4服务端主动关闭但没捕获断连WebSocketDisconnect向上冒泡用pytest.raises包住对应的 receive 调用5用了 lifespan 资源却没包with TestClient(app)数据为空KeyError外面多嵌套一层with第 1、3 条占掉日常排障的九成遇到「测试卡住不报错」先查这两处。下一步现在你可以给项目里任何一个 WebSocket 端点写出可运行的端到端断言3 行建会话每轮对话一个 assert。想继续深入仓库里有两条线可跟官方章节含会话用法的完整上下文docs/en/docs/advanced/testing-websockets.md应用级测试含 lifespan 示例与配套可运行源码在docs_src/app_testing/目录本文的回显应用与它们是同一套结构。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考