Django+vue自动化测试平台（27）-- 封装websocket测试
websocket概述：
WebSocket 是一种在单个 TCP 连接上进行全双工通信(Full Duplex 是通讯传输的一个术语。通信允许数 据在两个方向上同时传输，它在能力上相当于两个单工通信方式的结合。全双工指可以同时（瞬时）进 行信号的双向传输（ A→B 且 B→A ）。指 A→B 的同时 B→A，是瞬时同步的)的协议。
WebSocket 通信协议于 2011 年被 IETF 定为标准 RFC 6455，并由 RFC7936 补充规范。WebSocket API (WebSocket API 是一个使用WebSocket 协议的接口，通过它来建立全双工通道来收发消息) 也被 W3C 定为标准。
WebSocket 使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。 在 WebSocket API 中，浏览器和服务器只需要完成一次握手，两者之间就直接可以创建持久性的连接， 并进行双向数据传输。
而 HTTP 协议就不支持持久连接，虽然在 HTTP1.1 中进行了改进，使得有一个 keep-alive，在一个 HTTP 连接中，可以发送多个 Request，接收多个 Response。
但是在 HTTP 中 Request = Response 永远是成立的，也就是说一个 request 只能有一个response。而且 这个response也是被动的，不能主动发起。
websocket 常用于社交/订阅、多玩家游戏、协同办公/编辑、股市基金报价、体育实况播放、音视频聊 天/视频会议/在线教育、智能家居与基于位置的应用。
websocket 接口不能使用 requests 直接进行接口的调用，可以依赖第三方库的方式来实现调用，以下内 容介绍如何调用第三方库实现 websocket 的接口自动化测试。
实战效果预览：
后端逻辑实现（前端就不展示了，各有各的想法）：
# -*-coding: gbk-*-
import io
import json
import socket
from datetime import datetime
from uuid import uuid4
from django.http import JsonResponse
import websocket
import threading
from django.views import View
# 定义一个dict字典，用来存储websocket的连接对象
websocket_info = {}
# 主类，使用的django的视图类，将其视为http对象
class WebSocketTestView(View):
def create_websocket(self, request):
try:
data = json.loads(request.body)
# 生成websocket唯一标识
uuid = str(uuid4())
def on_data(ws, frame_data, frame_opcode, frame_fin):
# WebSocketApp方法的on_data，没用上，做着玩
print("frame_data：", frame_data)
print("frame_opcode：", frame_opcode)
print("frame_fin：", frame_fin)
# 初始化 WebSocket 连接及相关回调
def on_message(ws, message):
websocket_info[uuid]['history'].append({"message": message, "type": 2,
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
# 异常函数回调
def on_error(ws, error):
websocket_info[uuid]["history"].append({"message": error, "type": 2,
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
# websocket连接关闭回调
def on_close(ws, close_status_code, close_message):
websocket_info.pop(uuid, None)
# 建立连接回调
def on_open(ws):
try:
websocket_info[uuid]['status'] = True
except Exception as e:
print(e)
websocket_info[uuid]['status'] = False
# websocket.WebSocketApp建立连接
# params是url的传参，我用的是params=[{"key": "value"}, {"key": "value"}]
# headers是websocket中传的请求头
ws = websocket.WebSocketApp(f"{data['url']}?"
f"{'&'.join([f'{k}={v}' for param in data['params'] for k, v in param.items()])}",
on_open=on_open,
on_message=on_message,
on_error=on_error,
on_close=on_close,
on_data=on_data,
header=data["headers"])
websocket_info[uuid] = {
'ws': ws,
"status": False,
"history": [],
}
# 启动多线程进行websocket的长久连接
thread = threading.Thread(target=ws.run_forever)
thread.start()
on_open(ws)
if websocket_info[uuid]['status']:
websocket_info[uuid]["history"].append(
{'uuid': uuid, "message": "websocket Connecting", "type": 2, "code": 200,
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
return JsonResponse({'uuid': uuid, "message": "websocket Connecting", "type": 2, "code": 200,
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
else:
websocket_info[uuid]["history"].append(
{"message": "disconnected", "type": 2, "code": 100,
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
return JsonResponse({"message": "disconnected", "type": 2, "code": 100,
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
except Exception as e:
return JsonResponse({'Exception': e, "message": "disconnected", "type": 2, "code": 100,
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
# 发送websocket信息
def send_message(self, request):
try:
data = json.loads(request.body)
uuid = data["uuid"]
if uuid in websocket_info:
if data["type"] == 1:
websocket_info[uuid]['ws'].send(data["message"])
elif data["type"] == 2:
websocket_info[uuid]['ws'].send(json.dumps(data["message"]))
websocket_info[uuid]["history"].append({"message": str(data["message"]), "type": 1,
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
return JsonResponse({'message': f'Message sent: {str(data["message"])}', "code": 200,
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
else:
return JsonResponse({'message': 'WebSocket 不存在', "code": 100,
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
except Exception as e:
return JsonResponse({f'message': {str(e)}, "code": 100,
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
# 获取服务端返回的消息
def get_response(self, request):
data = json.loads(request.body)
uuid = data["uuid"]
if uuid in websocket_info and websocket_info[uuid]['history']:
return JsonResponse({'response': websocket_info[uuid]['history'][::-1], "code": 200,
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
else:
return JsonResponse({'error': '未查询到结果', "code": 100,
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
# 主动断开websocket的连接
def disconnect_websocket(self, request):
data = json.loads(request.body)
uuid = data["uuid"]
if uuid in websocket_info:
websocket_info[uuid]['ws'].close()
websocket_info.pop(uuid, None)
return JsonResponse({'message': 'WebSocket 已断开连接', "code": 200})
else:
return JsonResponse({'message': 'WebSocket 不存在', "code": 100})
总结
1 . websocket.WebSocketApp没有定义返回连接状态的函数结果，只能自己根据实际来做判断
2. 以上代码完成了对websocket测试的基本封装，个人愚见，仅供参考。
3. 更多好的内容，尽在个人主页，欢迎沟通交流！！！！
原文地址：https://blog.csdn.net/ljyljyyeah/article/details/140489427
免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！