WebSocket 协议在2008年诞生,2011年成为国际标准。所有浏览器都已经支持了。
它的最大特点就是,服务器可以主动向客户端推送信息,客户端也可以主动向服务器发送信息,是真正的双向平等对话,属于服务器推送技术的一种。
websocket和http接口特点
- http 接口 获取数据,先请求【客户端先主动,服务端才能响应】
- http协议 --- 请求收到响应后,代表 交互结束
- websocket接口 ,连接建立后, 服务端可以 主动发送数据
- websocket协议 --- 连接建立, 不会自动结束,需要用代码控制结束,属于长连接协议
websocket: 及时性要求高的场景 【微信消息、通知推送、直播间聊天 --- 】
websocket接口和http接口区别
相同点
- 都是基于网络数据交互
- 都需要在请求的时候创建网络连接
- 接口测试要素: 参数传递、响应内容
不同点: - websocket 接口基于一个 websocket url建立连接。而不是每次请求不同的url
- websocket接口地址实际是一个创建连接的地址
- websocket连接建立之后,后续发送数据,基于已创建的连接,用的时候无需再次指定host port...等信息
- websocket 发数据 - 格式根据接口文档而定
jmeter测试
jemter测试websocket的步骤与http一样
,使用jemter websocket插件,jmeter-websocket-samplers-1.2.8.jar,这块就不在赘述,重点说明下用python进行自动化接口测试
python 接口自动化测试
以实例做为说明,有一聊天室
接口文档如下:
- 网页访问地址:
http://127.0.0.1:18004/login
- 系统内置账户包含有:tony,nick,tom,jack,any,gg,mm
- 系统账户密码统一为:1234
- 连接地址示例:
ws://127.0.0.1:18004/webSocket/nick
- 聊天数据发送格式:发送内容为json格式
{
"from": "nick",
"to": "tony",
"text": "hello, i am nick"
}
字段描述:
from: 发送者name
to: 接收者name(0:代表上线消息,-2:代表下线消息,-1代表广播全局消息,其余为用户点对点消息)
text: 发送的文本 - 聊天数据接收格式:接收内容为json模式
{
"from": "nick",
"to": "tony",
"text": "hello, i am nick",
"date": "2021-09-24 20:14:17"
}
字段描述同上
根据此,模拟两人上线对话下线,看聊天室能否正常工作,代码如下:
import json
import pytest
from websocket import create_connection
# Pytest 用例
class Test_Websocket:
def test_chat(self):
# 1、建立连接
ws_tom = create_connection("ws://127.0.0.1:18004/webSocket/tom")
ws_jack = create_connection("ws://127.0.0.1:18004/webSocket/jack")
print("0. tom 和 jack 成功连接")
# 2. tom 发消息给 jack
ws_tom.send('''{
"from": "tom",
"to": "jack",
"text": "hello, i am tom"
}''')
print("1. tom 向 jack 发送消息")
# jack 接收到消息【可能收到 非 用户发送的消息。 可能收到 系统消息】
while True:
result = ws_jack.recv() # 服务器推送过来的数据
dict_result = json.loads(result) # 将 json字符串 转换为 字典类型
to = dict_result['to']
# 判断消息的类型,如果非用户消息,则继续读取下一个消息
if to == '-1' or to == '-2' or to == '0':
continue
msg = dict_result["text"]
assert msg == "hello, i am tom", "消息接收有问题" # 断言
print("2. jack 成功收到 tom 发送的消息")
break
# jack 发送回复消息
ws_jack.send('''{
"from": "jack",
"to": "tom",
"text": "hello, i am jack"
}''')
print("3. jack 发送回复给 tom")
# tom 要能够收到回复
while True:
reply_result = ws_tom.recv() # 服务器推送过来的数据
to = json.loads(reply_result)['to']
# 判断消息的类型,如果非用户消息,则继续读取下一个消息
if to == '-1' or to == '-2' or to == '0':
continue
reply = json.loads(reply_result)["text"]
assert reply == "hello, i am jack", "消息接收有问题" # 断言
print("4. tom 成功收到 jack 的回复")
break
ws_tom.close()
ws_jack.close()
print("5. 关闭websocket 长连接")
if __name__ == '__main__':
# --timeout=10 pytest-timeout插件 针对可能长时间不能结束的测试用例,约束执行的时长。超过时长,没执行结束就是失败
pytest.main(["-s","test_websocket.py::Test_Websocket::test_chat", "--timeout=10", "--html=report.html"])
-------------------输出----------------------
1. tom 向 jack 发送消息
2. jack 成功收到 tom 发送的消息
3. jack 发送回复给 tom
4. tom 成功收到 jack 的回复
5. 关闭websocket 长连接
网友评论