企业官网建设流程全解析

用 TestClient 完成 FastAPI WebSocket 测试的完整流程

【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi

消息推送功能上线前，你怎么确认连接能建立、第一条消息不会错序、断开不会把服务搞挂？FastAPI WebSocket 测试的答案就一行：用TestClient.websocket_connect()模拟真实客户端连进端点，收消息、做断言。不需要起真实服务，也不需要第三方库。

结论先行

• 用和 HTTP 测试同一个 TestClient。它是内置的进程内客户端，直接驱动你的应用，不走真实网络。
• 测试函数里只写"收消息 + 断言"。建连和断开都由with块自动完成。
• 测试函数必须写成同步def。TestClient 在内部驱动异步应用，你不需要写await。

一个最小可运行的 WebSocket 测试

场景换成聊天房间：用户进入 77 号房间时，服务端推送一条入群通知。端点和测试一共 14 行：

from fastapi import FastAPI, WebSocket from fastapi.testclient import TestClient app = FastAPI() @app.websocket("/room/{room_id}") async def join_room(socket: WebSocket, room_id: str): await socket.accept() await socket.send_json({"event": "joined", "room": room_id}) await socket.close() def test_join_notification(): client = TestClient(app) with client.websocket_connect("/room/77") as socket: payload = socket.receive_json() assert payload == {"event": "joined", "room": "77"}

端点只做三件事：接连接、发通知、关连接。测试断言收到的第一条 JSON 等于期望字典。

这段代码到底发生了什么

• 你在端点里写await socket.accept()，框架完成握手，连接才进入可读状态。
• 你进入with client.websocket_connect("/room/77")，框架在进程内完成握手，不监听端口。应用是 ASGI 应用，即一套框架与服务端约定的通信接口。
• 你调用receive_json()，框架读走服务端发出的第一帧，解码成 Python 字典。
• assert做值比较，pytest 据此记通过或失败。
• 你退出with时，框架关闭连接并释放会话。

with块就是上下文管理器，意思是块结束时代码会自动帮你做清理。它像一次通话，拿起建立连接，挂断拆除一切。

按消息类型选断言写法

按消息类型匹配客户端和服务端的方法，再对返回值断言：

1. 文本消息：服务端send_text()，测试端receive_text()，断言assert text == "xxx"。
2. JSON 消息：服务端send_json()，测试端receive_json()，断言assert data == {"event": "joined"}。推荐默认用它，结构一目了然。
3. 二进制消息：服务端send_bytes()，测试端receive_bytes()，断言assert raw == b"..."。

同一条连接混用多种类型时，按服务端发出的顺序依次读取，用对应的receive_*逐个消费。

三个容易翻车的点 ⚠️

1. 在 async def 测试里用 TestClient

错误症状：连接无响应，或抛出事件循环相关错误。

# ❌ 反例 @pytest.mark.anyio async def test_room(): client = TestClient(app) # 异步上下文里没有可用事件循环

正确写法：把测试改回同步def。TestClient 只能在同步调用栈里工作。

2. 收发顺序和服务端错位

错误症状：测试卡死，或突然抛出WebSocketDisconnect。

# ❌ 服务端只发一次，测试却收两次 with client.websocket_connect("/room/77") as socket: first = socket.receive_json() second = socket.receive_json() # 没有第二帧，阻塞

正确写法：把服务端的send_*列表和测试的receive_*列表逐条对齐，一一对应。

3. 应用靠 lifespan 初始化数据

lifespan 是应用启动和停止时执行的回调，常用来预置房间、令牌等数据。直接写client = TestClient(app)时，lifespan 不会运行，端点里的数据还是空壳。

# ✅ 两层 with 嵌套 with TestClient(app) as client: with client.websocket_connect("/room/77") as socket: payload = socket.receive_json()

正确写法：外层with启动应用，内层with管理这条连接，两层都要写。

跑完后的自查清单

• 测试函数是同步 def，函数体里没有 await
• 每次 receive 都能在服务端找到对应的 send
• 依赖 lifespan 时写成了嵌套的 with TestClient(app)
• 服务端 close() 后，断连路径有显式断言
• 直接运行 pytest 能一次跑绿

想继续深入，读官方教程 Testing WebSockets，看断连异常类型的再导出源码 fastapi/websockets.py，对比异步测试写法 async-tests.md。需要直接可抄的完整示例，参考 docs_src/app_testing/tutorial002_py310.py。

【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi