FastAPI WebSocket 测试实战30 秒跑通第一条断言【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi聊天服务上线前谁来保证/ws/orders端点真的能推出一条订单靠浏览器手动点开不算数你得让 FastAPI WebSocket 测试替你盯着。好消息是不用引入新框架用TestClient的websocket_connect打开一个连接会话之后就是收一条、断言一条。30 秒跑通第一条断言下面这段代码包含被测端点和一个测试函数直接放进你的项目from fastapi import FastAPI, WebSocket from fastapi.testclient import TestClient app FastAPI() app.websocket(/ws/orders) async def push_order(websocket: WebSocket): await websocket.accept() await websocket.send_json({order_id: 1001, status: paid}) await websocket.close() def test_push_order(): client TestClient(app) with client.websocket_connect(/ws/orders) as session: data session.receive_json() assert data {order_id: 1001, status: paid}把这段代码丢进项目pytest跑一下绿色就对了。它到底在干什么三件事看懂机制连接会话with client.websocket_connect(url)会向该地址发起真实握手进入with块代表已连上退出块自动断开不用手动清理。消息收发会话对象上调用receive_json()、send_text()等就像操作一个真实客户端每次调用处理一条消息。断言时机每receive到一条就立刻assert断言的是消息内容而不是 HTTP 响应。实现上它没有重新造轮子fastapi/testclient.py 只把 Starlette 的TestClient再导出一次fastapi/websockets.py 也是把WebSocket、WebSocketDisconnect等类型转手给你的。实战演练三个由浅入深的场景断言第一条推送消息先写一条普通 HTTP 测试做对照再测 WebSocket 推送。两者结构一致只是断言对象不同app.get(/orders) async def list_orders(): return [{order_id: 1001, status: paid}] def test_http_orders(): client TestClient(app) response client.get(/orders) assert response.status_code 200 assert response.json() [{order_id: 1001, status: paid}] def test_ws_orders(): client TestClient(app) with client.websocket_connect(/ws/orders) as session: data session.receive_json() assert data[status] paidHTTP 测试拿到response看状态码WebSocket 测试拿到会话收消息。同一套TestClient两套断言目标。模拟客户端与服务端对话echo 型端点是最常见的对话形态客户端先发服务端回。测试端就按这个时序写app.websocket(/ws/chat) async def chat(websocket: WebSocket): await websocket.accept() text await websocket.receive_text() await websocket.send_text(freceived: {text}) await websocket.close() def test_chat_echo(): client TestClient(app) with client.websocket_connect(/ws/chat) as session: session.send_text(ping room-1) reply session.receive_text() assert reply received: ping room-1服务端每send一次测试端就receive一次顺序严格对齐。为 lifespan 多加一层 with应用启动时才初始化数据比如房间成员表而 lifespan 只有进入with TestClient(app)才会触发所以要套两层asynccontextmanager async def lifespan(app: FastAPI): app.state.room {room:1: {members: [alice]}} yield app.websocket(/ws/room/1) async def room_events(websocket: WebSocket): await websocket.accept() await websocket.send_json(websocket.app.state.room[room:1]) await websocket.close() def test_room_events(): with TestClient(app) as client: with client.websocket_connect(/ws/room/1) as session: data session.receive_json() assert data[members] [alice]外层with负责启动 lifespan、保证房间数据存在内层with只管连接本身。少一层数据就是空的。一张表看清收发方法方法作用配套的服务端方法receive_text()收一条文本消息send_text()receive_json()收一条 JSON 并解码send_json()receive_bytes()收一条二进制消息send_bytes()send_text()向服务端发文本receive_text()send_json()向服务端发 JSONreceive_json()send_bytes()向服务端发二进制receive_bytes()记住一条铁律测试端的收发顺序必须和服务端代码里的顺序一致。踩坑清单异步测试函数里用 TestClient→ 测试写成async def后卡住或报事件循环冲突。→ 把测试改回普通同步def让TestClient自己驱动异步应用。HTTP 断言套到 WebSocket 上→ 在会话上找status_code或.json()直接 AttributeError。→ 用receive_json()拿到数据再断言会话没有状态码。消息顺序错位→ 服务端还没发你就receive测试安静地挂死。→ 对照服务端代码逐条排好 send/receive 顺序再写测试。忽略连接关闭→ 服务端close()后你再receive抛出WebSocketDisconnect。→ 要么提前收完所有消息要么用pytest.raises(WebSocketDisconnect)把断连本身当断言对象。收尾一个TestClientwith client.websocket_connect(url)开会话逐条receive后断言。想继续看→ 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），仅供参考