当前位置:   article > 正文

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

怎么批量连接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自动化测试/测试开发,走向高薪之路。


资源分享

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

在这里插入图片描述

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

在这里插入图片描述

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/码创造者/article/detail/869466
推荐阅读
相关标签
  

闽ICP备14008679号