软件测试 | websocket接口测试-Toy模板网

这篇具有很好参考价值的文章主要介绍了软件测试 | 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 的接口自动化测试。

实战

使用 python 语言实现 websocket 的接口自动化

环境准备

安装 pyhton3 环境
下载需要的运行库

pip install websocket-client

实战演示

1.连接 websoket 服务器

import logging
from websocket import create_connection
logger = logging.getLogger(__name__)
url = 'ws://echo.websocket.org/' #一个在线的回环websocket接口,必须以websocket的方式连
接后访问，无法直接在网页端输入该地址访问
wss = create_connection(url, timeout=timeout)

1.发送websoket消息

wss.send('Hello World')

1.接收websocket消息

res = wss.recv()
iogger.info(res)

1.关闭websocket连接

wss.close()

1.websocket 第三方库的调用不支持直接发送除字符串外的其他数据类型，所以在发送请求之前需要将 Python 结构化的格式,转换为成为字符串类型或者 json 字符串后，再发起 websocket 的接口请求

#待发送的数据体格式为：
data= {
 "a" : "abcd",
 "b" : 123
 }
# 发送前需要把数据处理成 json 字符串
new_data=json.dumps(data,ensure_ascii=False)
wss.send(new_data)

1.接收的数据体的处理：如果接口定义为 json 的话，由于数据的传输都是字符串格式的，需要对接收的数据体进行转换操作

# 接收的数据体的格式也为字符串
logger.info(type(res)) # <class 'str'>

对于响应内容进行格式转换处理：

def load_json(base_str):
 if isinstance(base_str, str):
 try:
 res = json.loads(base_str)
 return load_json(res)
 except JSONDecodeError:
 return base_str
 elif isinstance(base_str, list):
 res = []
 for i in base_str:
 res.append(load_json(i))
 return res
 elif isinstance(base_str, dict):
 for key, value in base_str.items():
 base_str[key] = load_json(value)
 return base_str
 return base_str

1.websocket接口自动化测试，二次封装demo展示

web_socket_util.py封装websocket接口通用操作：

import logging
import json
from websocket import create_connection
logger = logging.getLogger(__name__)
class WebsocketUtil():
 def conn(self, uri, timeout=3):
 '''
 连接web服务器
 :param uri: 服务的url
 :param timeout: 超时时间
 :return:
 '''
 self.wss = create_connection(uri, timeout=timeout)
 def send(self, message):
 '''
 发送请求数据体
 :param message: 待发送的数据信息
 :return:
 '''
 if not isinstance(message, str):
 message = json.dumps(message)
 return self.wss.send(message)
 def load_json(self, base_str):
 '''
 进行数据体的处理
 :param base_str: 待处理的数据体
 :return:
 '''
 if isinstance(base_str, str):
 try:
 res = json.loads(base_str)
 return self.load_json(res)
 except JSONDecodeError:
 return base_str
 elif isinstance(base_str, list):
 res = []
 for i in base_str:
 res.append(self.load_json(i))
 return res
 elif isinstance(base_str, dict):
 for key, value in base_str.items():
 base_str[key] = self.load_json(value)
 return base_str
 return base_str
 def recv(self, timeout=3):
 '''
 接收数据体信息，并调用数据体处理方法处理响应体
 :param timeout: 超时时间
 :return:
 '''
 if isinstance(timeout, dict):
 timeout = timeout["timeout"]
 try:
 self.settimeout(timeout)
 recv_json = self.wss.recv()
 all_json_recv = self.load_json(recv_json)
 self._set_response(all_json_recv)
 return all_json_recv
 except WebSocketTimeoutException:
 logger.error(f"已经超过{timeout}秒没有接收数据啦")
def settimeout(self, timeout):
 '''
 设置超时时间
 :param timeout: 超时时间
 :return:
 '''
 self.wss.settimeout(timeout)
 def recv_all(self, timeout=3):
 '''
 接收多个数据体信息，并调用数据体处理方法处理响应体
 :param timeout: 超时时间
 :return:
 '''
 if isinstance(timeout, dict):
 timeout = timeout["timeout"]
 recv_list = []
 while True:
 try:
 self.settimeout(timeout)
 recv_json = self.wss.recv()
 all_json_recv = self.load_json(recv_json)
 recv_list.append(all_json_recv)
 logger.info(f"all::::: {all_json_recv}")
 except WebSocketTimeoutException:
 logger.error(f"已经超过{timeout}秒没有接收数据啦")
 break
 self._set_response(recv_list)
 return recv_list
 def close(self):
 '''
 关闭连接
 :return:
 '''
 return self.wss.close()
 def _set_response(self, response):
 self.response = response
 def _get_response(self) -> list:
 return self.response

test_case.py websocket接口自动化测试用例：文章来源地址https://www.toymoban.com/news/detail-520225.html

class TestWsDemo:
 def setup(self):
 url = 'ws://echo.websocket.org/'
 self.wss = WebsocketUtil()
 self.wss.conn(url)
 def teardown(self):
 self.wss.close()
 def test_demo(self):
 data = {"a": "hello", "b": "world"}
 self.wss.send(data)
 res = self.wss.recv()
 assert 'hello' == res['a']

到了这里，关于软件测试 | websocket接口测试的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！