经常会遇到了很多人都有这样的问题：python如何做websocket接口如何测试？一直没有时间去完善这里的文章，这次正好去分享下。首先由一个简单的例子去看下python如何测试websocket的接口测试，完成下一个简单的接口测试，然后再去适配复杂的websocket的接口测试即可。
由最简单的接口逐步演进去测试复杂的接口。有简到繁进行。没有websocket接口如何如何呢，自己去创建接口，在之前的文章有分享。拿到之前的一个简单的接口来作为一个入门级的websocket
import uvicorn
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"接受到的消息是: {data}")
if __name__=="__main__":
uvicorn.run(app='views:app', host="127.0.0.1",
port=8081, reload=True, debug=True)这个接口 很简单，就是完成一个链接，获取到链接后，直接返回对应的消息。
一个很简单的接口，那么python如何来实现webscoket协议接口测试呢
需要安装两个依赖包
pip3 install websocket-client
pip3 install websocket安装后，就可以去完成接口的请求。
先创建一个简单的脚本，去测试下接口是否可以调用
from websocket import create_connection
import websocket
url = 'ws://localhost:8081/ws'
wss = create_connection(url)#创建链接
wss.settimeout(10) #设置超时时间
print(wss.getstatus())#状态码是100认为成功
wss.send('hello world')#发送消息
print(wss.recv())#接受小时
wss.shutdown()测试下
这样可以正常请求，那么这样一个简单的接口，可以设计一个简单的用例去测试
其实上面的请求测试，可以改造成一个简单的接口测试用例。导入
import unittest然后定一个一个类，继承
class TestWebsocket(unittest.TestCase):在setup中
def setUp(self) -> None:
self.url = 'ws://localhost:8081/ws'
self.wss = create_connection(self.url) # 创建链接其实就是创建链接
在tearDown中，断开
def tearDown(self) -> None:
self.wss.shutdown()在去新增一个用例
def testone(self):用例中有两个断言，一个是链接是否正确，另一个断言是发送的消息后是否返回对应的值
def testone(self):
status=self.wss.getstatus()
self.assertEqual(status,101,'websocket连接错误')
self.wss.send('hello world')#发送消息
recvdata=self.wss.recv()
self.assertTrue(("hello world" in recvdata),'接口返回错误')这样一个简单的用例就实现完毕了，我们测试下
执行成功，测试用例可以正常测试，这样一个简单的webscoket的接口的测试就完成了。
其实webscoket接口测试流程就是去创建链接，然后传送数据，接受数据。然后根据链接创建和返回数据进行对应的断言就可以了。
简单的接口测试可以这么来做，那么针对复杂的接口如何做呢，我们去拿之前的写的接口。
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"接受到的消息是: {data}")
manager = ConnectionManager()
# 给人们消息
@app.websocket("/ws/{user}/")
async def websocket_many_point(
websocket: WebSocket,
user: str,
):
await manager.connect(user, websocket)
try:
while True:
data = await websocket.receive_json()
senduser = data.get('username')
if senduser:
await manager.send_other_message_json(data, senduser)
else:
await manager.broadcast_json(data)
except WebSocketDisconnect as e:
manager.disconnect(user, websocket)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"接受到的消息是: {data}")
manager = ConnectionManager()
# 给人们消息
@app.websocket("/ws/{user}/")
async def websocket_many_point(
websocket: WebSocket,
user: str,
):
await manager.connect(user, websocket)
try:
while True:
data = await websocket.receive_json()
senduser = data.get('username')
if senduser:
await manager.send_other_message_json(data, senduser)
else:
await manager.broadcast_json(data)
except WebSocketDisconnect as e:
manager.disconnect(user, websocket) 很简单，就是一个给多个人或者单个人发消息，这里的用例可以简单的有四个。
测试发给自己看自己能不能接受到，
一个是群发，看两个能否接受到，
一个是给某个人发，看那个人是否接受到。
A发给B，B可以接受到，C接受不到接着去实现对应的测试用例的代码。
这里面需要注意：数据传输是根据字符串的，我们发送的是json，需要进行json处理。直接使用json模块进行处理即可。
import json
from websocket import create_connection, WebSocketTimeoutException
import unittest
class TestWebSocket(unittest.TestCase):
def setUp(self) -> None:
#初始化链接地址
self.url = 'ws://localhost:8081/ws/1/'
#链接
self.wss = create_connection(self.url)
def tearDown(self) -> None:
#关闭链接
self.wss.shutdown()
def testnouusernmae(self):
#获取链接状态
status = self.wss.getstatus()
#断言状态码
self.assertEqual(status, 101, 'websocket连接错误')
#拼装参数
senddata = {"dict": 122}
#发送
self.wss.send(json.dumps(senddata))
#看群发是否自己能收到
recvdata = self.wss.recv()
#转化下消息
reslut = json.loads(recvdata)
#断言
self.assertEqual(reslut, senddata, '消息发送接受到不一致')
def testorther(self):
#链接第二个
self.urlorther = 'ws://localhost:8081/ws/212/'
#创建链接
self.ortherwss = create_connection(self.urlorther)
#获取状态
status = self.wss.getstatus()
ortherstatus = self.ortherwss.getstatus()
#断言
self.assertEqual(ortherstatus, 101, '其他用户websocket连接错误')
self.assertEqual(status, 101, 'websocket连接错误')
#给212发送数据
senddata = {"dict": 122, "username": '212'}
self.wss.send(json.dumps(senddata))
#212接受数据
recvdata = self.ortherwss.recv()
#断言212接受到的数据
self.assertEqual(json.loads(recvdata), senddata, '212消息接受到不一致')
self.ortherwss.shutdown()
def testsendme(self):
#自己给自己发送
#状态吗
status = self.wss.getstatus()
#断言
self.assertEqual(status, 101, 'websocket连接错误')
senddata = {"dict": 122, "username": '1'}
#发送消息，因为发送的是字符串
self.wss.send(json.dumps(senddata))
#自己接受下
recvdata = self.wss.recv()
reslut = json.loads(recvdata)
self.assertEqual(reslut, senddata, '消息接受到不一致')
def testorthernew(self):
#三个用户对话
self.urlorther = 'ws://localhost:8081/ws/212/'
self.urlthird = 'ws://localhost:8081/ws/211/'
#创建链接
self.thirdwss = create_connection(self.urlthird)
# 设置超时时间，因为要看第三个人是否接受到消息给一个超时时间
self.thirdwss.settimeout(1)
#创建第二个人的链接
self.ortherwss = create_connection(self.urlorther)
#获取状态吗
status = self.wss.getstatus()
ortherstatus = self.ortherwss.getstatus()
thirdstatus=self.thirdwss.getstatus()
#断言
self.assertEqual(ortherstatus, 101, '其他用户websocket连接错误')
self.assertEqual(status, 101, 'websocket连接错误')
self.assertEqual(thirdstatus, 101, '其他用户websocket连接错误')
#拼接数据，发给第二个人
senddata = {"dict": 122, "username": '212'}
self.wss.send(json.dumps(senddata))
#第三个用户获取，一定是超时的
try:
data = self.thirdwss.recv()
self.assertFalse(True, msg="不应该接受到消息")
except WebSocketTimeoutException as e:
self.assertTrue(True, "第三个人不应该收到消息")
#第二个用户获取
recvdata = self.ortherwss.recv()
#断言是否可以正常接受到消息
self.assertEqual(json.loads(recvdata), senddata, '212消息接受到不一致')
self.ortherwss.shutdown()
self.thirdwss.shutdown()其实很简单，四个场景的测试。
执行下没有其问题。这里就实现完毕了。
这样基于python+unittest的webscoket测试就演示完毕了。
其实很简单用两个简单的类库，就完成了整体的测试。
这里只是用了websocket一些简单的方法，里面还有很多方法，可以在实际中灵活选择。
不知道怎么用，最简单的方式就是看库的代码，看对应的方法，都有详细的注释。有的还有对应的例子可以用。
其实webscoket的测试没有那么难，很简单和我们正常测试的http没什么不同，创建链接，发送数据，接受数据，然后进行断言。只是用的类库，和处理不一样。