12.6 WebSocket 测试
FastAPI WebSocket测试教程：从入门到精通
本教程详细讲解了如何在FastAPI中设置和测试WebSocket连接，适合新手学习，包括代码示例、测试方法和最佳实践，帮助您轻松掌握FastAPI WebSocket测试。
FastAPI WebSocket测试教程
什么是WebSocket？
WebSocket是一种在单个TCP连接上进行全双工通信的协议，适用于实时应用，如聊天室或在线游戏。在FastAPI中，WebSocket支持基于异步编程，使得处理实时数据更加高效。
在FastAPI中设置WebSocket端点
首先，确保您已安装FastAPI和WebSocket相关依赖。使用以下命令安装：
pip install fastapi uvicorn websockets
然后，创建一个FastAPI应用并定义一个WebSocket端点。以下是基本示例：
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
import asyncio
app = FastAPI()
@app.get("/")
async def get():
html_content = """
<html>
<head>
<title>WebSocket Test</title>
</head>
<body>
<h1>FastAPI WebSocket 测试</h1>
<script>
const ws = new WebSocket('ws://localhost:8000/ws');
ws.onopen = () => {
console.log('WebSocket 连接已打开');
ws.send('Hello from client');
};
ws.onmessage = (event) => {
console.log('接收到消息:', event.data);
};
ws.onerror = (error) => {
console.error('WebSocket 错误:', error);
};
ws.onclose = () => {
console.log('WebSocket 连接已关闭');
};
</script>
</body>
</html>
"""
return HTMLResponse(content=html_content)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept() # 接受WebSocket连接
try:
while True:
data = await websocket.receive_text() # 接收客户端消息
print(f"接收到客户端消息: {data}")
await websocket.send_text(f"服务器回复: {data}") # 发送回复
except Exception as e:
print(f"WebSocket 错误: {e}")
finally:
await websocket.close()
在这个例子中，我们定义了一个WebSocket端点 /ws，用于处理客户端连接和消息交换。前端HTML页面提供了一个简单的WebSocket客户端进行测试。
测试WebSocket连接的方法
测试WebSocket连接是确保应用正常运行的关键部分。以下是两种常见的测试方法：
方法1：使用FastAPI的TestClient进行单元测试
FastAPI提供了TestClient来模拟HTTP请求，但它也可以用于测试WebSocket。首先，安装测试依赖：
pip install pytest
然后，创建一个测试文件，使用asyncio和websockets库来模拟客户端。示例测试代码如下：
import pytest
from fastapi.testclient import TestClient
from your_app import app # 假设你的应用在your_app.py中
client = TestClient(app)
@pytest.mark.asyncio
async def test_websocket():
# 使用TestClient模拟WebSocket连接
with client.websocket_connect("/ws") as websocket:
# 发送消息
websocket.send_text("Hello")
data = websocket.receive_text()
assert data == "服务器回复: Hello"
# 测试断开连接
websocket.close()
assert websocket.closed is True
这个测试模拟了一个WebSocket客户端，发送消息并验证服务器回复。
方法2：使用异步WebSocket客户端库（如aiohttp）
对于更复杂的测试，可以使用aiohttp的WebSocket客户端。首先安装aiohttp：
pip install aiohttp
然后，编写一个异步客户端脚本进行测试：
import asyncio
import aiohttp
async def test_websocket_client():
async with aiohttp.ClientSession() as session:
async with session.ws_connect('http://localhost:8000/ws') as ws:
await ws.send_str("Test message")
msg = await ws.receive()
print(f"收到服务器回复: {msg.data}")
assert msg.data == "服务器回复: Test message"
await ws.close()
if __name__ == "__main__":
asyncio.run(test_websocket_client())
运行此脚本前，确保FastAPI服务器已启动（例如，使用uvicorn your_app:app --reload）。
完整示例：集成测试和最佳实践
为了确保WebSocket功能全面，建议进行集成测试，并遵循以下最佳实践：
- 处理错误和超时：在WebSocket端点中添加异常处理和超时机制，避免连接挂起。
- 使用异步模式：确保所有WebSocket操作是异步的，以提高性能。
- 测试不同场景：包括正常连接、断开重连、大数据传输等。
例如，扩展之前的端点以处理错误：
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
# 模拟一些处理
await asyncio.sleep(0.1) # 模拟异步操作
await websocket.send_text(f"处理后的回复: {data}")
except WebSocketDisconnect:
print("客户端断开连接")
except Exception as e:
print(f"未知错误: {e}")
await websocket.close(code=1011) # 发送关闭代码
常见问题
-
Q: 为什么我的WebSocket测试失败？ A: 检查服务器是否运行在正确的地址和端口，以及客户端是否正确配置。使用日志来调试连接问题。
-
Q: 如何测试WebSocket在大负载下的性能？ A: 可以使用压力测试工具，如
websocat或编写自定义脚本模拟多个客户端。 -
Q: WebSocket和HTTP有什么区别？ A: WebSocket提供持久连接和双向通信，而HTTP是无状态的请求-响应协议。
通过本教程，您应该能够掌握FastAPI中WebSocket的设置和测试。建议多实践，尝试不同的测试用例来巩固知识。