企业官网建设流程全解析

TestClient 一条 with 语句，跑通 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()就能发起连接、断言消息。读到这里，你手里就是一个可独立运行的 WebSocket 测试函数，外加对会话收发、断连异常、lifespan 嵌套几个细节的理解。

会话，不是一问一答

HTTP 测试你闭着眼都会写：

client = TestClient(app) response = client.get("/") # 一次请求，拿到 response 就结束

WebSocket 测试长得不一样：连接建立后不会自动断开，服务端随时可能推消息，所以你拿到的不是 response，而是一个会话——一段需要你自己管住开与关的长连接。对应的写法多了一个with（Python 的上下文管理器，代码块退出时自动执行清理）：

with client.websocket_connect("/ws") as websocket: data = websocket.receive_json() # 会话里收一条消息

结论一句话：同一个TestClient就够用，不用引入任何新客户端。这里有个细节，fastapi/testclient.py 全部内容就是对 StarletteTestClient的一行再导出，会话能力全部继承自 Starlette。

握手、收消息、断言、断开：四步跑通

被测端点精简到三行核心逻辑：

@app.websocket("/ws") async def websocket(websocket: WebSocket): await websocket.accept() # 接受握手 await websocket.send_json({"msg": "Hello WebSocket"}) await websocket.close() # 服务端主动关闭

这三行决定了测试怎么读：没accept()之前连接处于半开状态；send_json()推出一条 JSON 消息；close()收尾。测试函数就四行：

def test_websocket(): client = TestClient(app) with client.websocket_connect("/ws") as websocket: data = websocket.receive_json() # 收服务端第一条消息 assert data == {"msg": "Hello WebSocket"}

逐行拆开：TestClient(app)造出客户端；websocket_connect("/ws")进入with时完成真正的 WebSocket 握手并打开会话；receive_json()阻塞等待第一条 JSON 并自动解码；with块退出时会话自动关闭，不用手写清理。跑完你会看到 pytest 直接给一个绿勾。

注意测试函数是普通的def，全程没有await——TestClient会在同步调用栈里替你驱动那些async def端点，测试代码本身保持同步就行。

会话里的收发工具箱，以及顺序陷阱

send_*和receive_*配着就能做"客户端发、服务端应"的对话式测试，比如回显端点：

def test_echo(): client = TestClient(app) with client.websocket_connect("/ws") as websocket: websocket.send_text("Hello, server") assert websocket.receive_text() == "Hello, server"

⚠️ 小心一点：WebSocket 是消息流，不是一问一答。收发顺序必须和服务端严格对齐——服务端send_*几次，你就receive_*几次，错位不是报错而是挂起，测试会一直阻塞等你。

再就是断连。服务端close()之后，测试端再receive_*会抛出WebSocketDisconnect（fastapi.websockets从 Starlette 再导出，测试和应用代码引用的是同一个类型）。想专门验证断连路径时，用pytest.raises(WebSocketDisconnect)包住那一次receive_text()即可。

两个容易翻车的点：异步测试函数 & lifespan 嵌套

坑 1：TestClient不能出现在async def测试函数里。它靠同步调用栈去驱动异步的 ASGI 应用，一旦测试函数本身跑在事件循环上再套一层，就冲突了。同步写def，这个坑不存在。

坑 2：应用靠lifespan初始化资源（预置字典、建连接池之类），那么只有进入with TestClient(app)时 lifespan 才被触发，WebSocket 测试得写成双层嵌套：

def test_websocket_with_lifespan(): with TestClient(app) as client: # 外层：启动/关闭 lifespan with client.websocket_connect("/ws") as websocket: assert websocket.receive_json() == {"msg": "Hello WebSocket"}

外层with负责把应用"启动"起来，内层with管 WebSocket 会话。普通 HTTP 测试遇到 lifespan 也是同一套嵌套规则，WebSocket 只是多套了一层连接会话。

什么时候该跳出 TestClient

如果测试环境本身就是async def（比如@pytest.mark.anyio），HTTP 请求有现成替代：httpx.AsyncClient配ASGITransport直接打 ASGI 应用。但 ⚠️ WebSocket 在异步测试里没有对等的websocket_connect()会话写法，别照搬上面的同步会话，得单独设计——单独建一个同步测试文件，或在真实浏览器/客户端里验证。

想接着挖，仓库里有这三处入口：WebSocket 端点教程、app_testing 示例目录（tutorial001~004 覆盖 HTTP 与 lifespan 场景），以及 fastapi/websockets.py 里会话类型和异常的定义。

给自己写个 echo 端点，用send_text()/receive_text()跑一遍上面的对话式测试——四条语句，两分钟的事。

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