一个客户端,两种用法:用 TestClient 快速搞定 FastAPI WebSocket 测试 📅 发布时间:2026/9/14 13:00:14 👁 浏览次数: 一个客户端两种用法用 TestClient 快速搞定 FastAPI WebSocket 测试【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi在 FastAPI 里一个TestClient就能同时搞定普通 HTTP 接口和 WebSocket 测试。本文从一个最小可跑组合出发带你看清连接、收消息、断言的完整链路并提前绕开常见的坑和边界限制。为什么 WebSocket 测试和普通接口不一样写 HTTP 测试时你的思维模型是一笔一结调一次client.get()拿到一个response对象然后检查status_code和response.json()。WebSocket 测试则是另一回事——更像一通电话拨通之后连接一直挂着双方可以来回说很多句话最后再挂断。断言对象也完全不同对比维度HTTP 测试WebSocket 测试交互模型一次请求一个响应一场长连接多条消息断言对象状态码 响应体会话里逐条收到的消息状态码有没有好消息是FastAPI 没有为 WebSocket 另造一个测试客户端HTTP 用的TestClient直接复用即可。翻看 fastapi/testclient.py 会发现它只有一行本质上是把 Starlette 的TestClient再导出了一下——所以 WebSocket 测试的全部能力都建立在这套现成的会话机制之上。最小可跑组合一个端点 一个测试函数先看被测端点来自官方教程 docs_src/app_testing/tutorial002_py310.pyfrom fastapi import FastAPI from fastapi.websockets import WebSocket app FastAPI() app.websocket(/ws) async def websocket(websocket: WebSocket): await websocket.accept() await websocket.send_json({msg: Hello WebSocket}) await websocket.close()服务端做了三件事测试端恰好各有一个对应动作服务端动作测试端对应动作accept()接受握手with client.websocket_connect(/ws)进入会话send_json({...})发一条消息receive_json()取到这条消息close()关闭连接会话结束再收就会触发断连测试函数就这么短from fastapi.testclient import TestClient def test_websocket(): client TestClient(app) with client.websocket_connect(/ws) as websocket: data websocket.receive_json() assert data {msg: Hello WebSocket}注意测试函数是普通同步def端点虽是async def但TestClient内部替你驱动异步应用测试代码里从头到尾没有await。运行方式和任何 pytest 测试一样uv run pytest 会话内的收发工具箱with client.websocket_connect(...)打开的会话对象上有六个收发方法可以随意组合方法方向断言示例receive_text()客户端收文本assert text hireceive_json()客户端收 JSONassert data {msg: ...}receive_bytes()客户端收二进制assert raw bpingsend_text(...)客户端发文本配合服务端receive_text()send_json(...)客户端发 JSON配合服务端receive_json()send_bytes(...)客户端发二进制配合服务端receive_bytes()receive_*负责听send_*负责说两者配合就能模拟一场完整的对话。比如一个回显型端点def test_websocket_echo(): client TestClient(app) with client.websocket_connect(/ws) as websocket: websocket.send_text(Hello, server) data websocket.receive_text() assert data Hello, server先send_text再receive_text先说后听——顺序错了测试就挂起。真实浏览器里一条 WebSocket 消息到达后呈现的样子可以参考官方教程中的这张图⚠️ 收尾的三个坑坑一时序纪律。WebSocket 是消息流不是一问一答。服务端每send_*一次测试端必须按同样顺序receive_*一次收发送错位置测试要么阻塞等一条永远不会来的消息要么断言拿到错的消息而失败。坑二断连异常。服务端执行close()之后测试端继续调用receive_*()会抛出WebSocketDisconnect在fastapi/websockets.py中从 Starlette 再导出。要专门验证这条断连路径可以用pytest.raises接住import pytest from fastapi.websockets import WebSocketDisconnect def test_websocket_disconnect(): client TestClient(app) with client.websocket_connect(/ws) as websocket: websocket.receive_json() with pytest.raises(WebSocketDisconnect): websocket.receive_json()坑三要不要手动关闭不用。退出with块时连接自动释放相当于电话打完了自动挂断——你只需要保证块内把该收的消息收完。lifespan 登场时双层 with 嵌套如果被测应用靠lifespan在启动时初始化状态比如预置一份items字典而这些状态只有在应用真正跑起来之后才存在那就多一层外层 withdef test_websocket_with_lifespan(): with TestClient(app) as client: with client.websocket_connect(/ws) as websocket: data websocket.receive_json() assert data {msg: Hello WebSocket}规则很直白外层with TestClient(app) as client:负责把应用启动起来lifespan 在这里执行并负责最后关闭它内层with client.websocket_connect(...)负责管理这一次连接会话。两层各管各的缺一不可。边界意识什么时候不能再用 TestClientTestClient的魔法发生在同步调用栈里它在你的一次同步调用中驱动整个异步 ASGI 应用。因此它只适用于同步测试函数。一旦测试函数本身写成async def例如pytest.mark.anyio标记的异步测试函数体内就不能再用TestClient了。此时 HTTP 请求可以改用httpx.AsyncClient配合ASGITransport直连应用写法见 docs_src/async_tests/app_a_py310/test_main.py但这条路线并不覆盖 WebSocket——异步场景下的 WebSocket 测试需要单独设计策略不能照搬本文的同步会话写法。收尾整条链路回顾一遍用TestClient包住应用 →websocket_connect()拨通长连接 → 在会话内按服务端发信顺序逐条receive_*/send_*→ 退出with自动挂断 → 有lifespan时再套一层外层with。五个动作走完一个 FastAPI WebSocket 端点的单元测试就闭环了。延伸阅读WebSocket 端点编写教程被测对象本身怎么写官方 WebSocket 测试文档本文知识点的出处应用级测试示例目录从tutorial001到tutorial004含 lifespan 场景异步测试方案httpx.AsyncClientASGITransport的完整用法TestClient 再导出源码一行代码看清 FastAPI 与 Starlette 的关系【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考