一文揭秘利用python+unittest实现Webscoket协议接口测试_怎么批量连接websocket做测试

经常会遇到了很多人都有这样的问题：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)
1
2
3
4
5
6
7
8
9
10
11
12

这个接口 很简单，就是完成一个链接，获取到链接后，直接返回对应的消息。

一个很简单的接口，那么python如何来实现webscoket协议接口测试呢，

需要安装两个依赖包

pip3 install websocket-client 
pip3 install websocket
1
2

安装后，就可以去完成接口的请求。

先创建一个简单的脚本，去测试下接口是否可以调用

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()
1
2
3
4
5
6
7
8
9

测试下

这样可以正常请求，那么这样一个简单的接口，可以设计一个简单的用例去测试

其实上面的请求测试，可以改造成一个简单的接口测试用例。导入

import  unittest
1

然后定一个一个类，继承

class TestWebsocket(unittest.TestCase):
1

在setup中

def setUp(self) -> None:
    self.url = 'ws://localhost:8081/ws'
    self.wss = create_connection(self.url)  # 创建链接
1
2
3

其实就是创建链接

在tearDown中，断开

def tearDown(self) -> None:
    self.wss.shutdown()
1
2

在去新增一个用例

def testone(self):
1

用例中有两个断言，一个是链接是否正确，另一个断言是发送的消息后是否返回对应的值

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),'接口返回错误')
1
2
3
4
5
6

这样一个简单的用例就实现完毕了，我们测试下

执行成功，测试用例可以正常测试，这样一个简单的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)   
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

很简单，就是一个给多个人或者单个人发消息，这里的用例可以简单的有四个。

测试发给自己看自己能不能接受到，
一个是群发，看两个能否接受到， 
一个是给某个人发，看那个人是否接受到。
A发给B，B可以接受到，C接受不到
1
2
3
4

接着去实现对应的测试用例的代码。

这里面需要注意：数据传输是根据字符串的，我们发送的是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()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

其实很简单，四个场景的测试。

执行下没有其问题。这里就实现完毕了。

这样基于python+unittest的webscoket测试就演示完毕了。

其实很简单用两个简单的类库，就完成了整体的测试。

这里只是用了websocket一些简单的方法，里面还有很多方法，可以在实际中灵活选择。

不知道怎么用，最简单的方式就是看库的代码，看对应的方法，都有详细的注释。有的还有对应的例子可以用。

其实webscoket的测试没有那么难，很简单和我们正常测试的http没什么不同，创建链接，发送数据，接受数据，然后进行断言。只是用的类库，和处理不一样。

现在我邀请你进入我们的软件测试学习交流群：【746506216】，备注“入群”， 大家可以一起探讨交流软件测试，共同学习软件测试技术、面试等软件测试方方面面，还会有免费直播课，收获更多测试技巧，我们一起进阶Python自动化测试/测试开发，走向高薪之路。

---

资源分享

最后感谢每一个认真阅读我文章的人，看着粉丝一路的上涨和关注，礼尚往来总是要有的，虽然不是什么很值钱的东西，如果你用得到的话可以直接拿走…

​这些资料，对于想从事【软件测试】的朋友来说应该是最全面最完整的备战仓库，这个仓库也陪伴我走过了最艰难的路程，希望也能帮助到你！凡事要趁早，特别是技术行业，一定要提升技术功底。希望对大家有所帮助…….