测试FastAPI AsyncClient WebSocket报错：'AsyncClient'无'websocket_connect'属性

AsyncClient（来自httpx）仅支持HTTP请求，并不具备WebSocket连接能力，这就是你遇到AttributeError的原因。要测试FastAPI的WebSocket端点，需要使用专门的异步WebSocket客户端库——websockets，配合pytest-asyncio来编写异步测试用例。

步骤1：安装依赖 #

先确保安装所需库：

pip install websockets pytest-asyncio

步骤2：编写WebSocket客户端Fixture（conftest.py） #

在conftest.py中添加WebSocket客户端的fixture，方便测试用例复用连接：

import pytest
import websockets
from your_app import app  # 替换为你的FastAPI应用模块路径
from httpx import AsyncClient

# 保留原有的AsyncClient fixture用于HTTP接口测试
@pytest.fixture(scope="module")
async def async_client():
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

# 添加WebSocket客户端fixture
@pytest.fixture(scope="function")
async def websocket_client():
    # 连接到你的WebSocket端点，替换为实际路径
    async with websockets.connect("ws://localhost:8000/ws") as websocket:
        yield websocket

步骤3：WebSocket路由示例（参考） #

假设你的WebSocket路由文件（如main.py）如下：

from fastapi import FastAPI, WebSocket
from fastapi.websockets import WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Received: {data}")
    except WebSocketDisconnect:
        await websocket.close()

步骤4：编写异步测试用例 #

创建测试文件（如test_websockets.py），使用fixture编写测试：

import pytest

@pytest.mark.asyncio
async def test_websocket_message(websocket_client):
    # 发送测试消息
    await websocket_client.send("Test WebSocket Message")
    # 接收响应
    response = await websocket_client.recv()
    # 断言响应内容
    assert response == "Received: Test WebSocket Message"

关键说明 #

• 测试前确保FastAPI应用已启动（或在测试中通过AsyncClient启动测试服务器，若使用测试服务器，WebSocket地址需对应测试端口）。
• 所有异步测试用例必须标记@pytest.mark.asyncio，pytest才能识别并运行异步代码。

内容的提问来源于stack exchange，提问作者T. de Jong