WebSocket 是一种在单个 TCP 连接上进行全双工通信(Full Duplex 是通讯传输的一个术语。通信允许数据在两个方向上同时传输,它在能力上相当于两个单工通信方式的结合。全双工指可以同时(瞬时)进行信号的双向传输( A→B 且 B→A )。指 A→B 的同时 B→A,是瞬时同步的)的协议。
WebSocket 通信协议于 2011 年被 IETF 定为标准 RFC 6455,并由 RFC7936 补充规范。WebSocket API (WebSocket API 是一个使用WebSocket 协议的接口,通过它来建立全双工通道来收发消息) 也被 W3C 定为标准。
而 HTTP 协议就不支持持久连接,虽然在 HTTP1.1 中进行了改进,使得有一个 keep-alive,在一个 HTTP 连接中,可以发送多个 Request,接收多个 Response。
但是在 HTTP 中 Request = Response 永远是成立的,也就是说一个 request 只能有一个response。而且这个response也是被动的,不能主动发起。
websocket 常用于社交/订阅、多玩家游戏、协同办公/编辑、股市基金报价、体育实况播放、音视频聊天/视频会议/在线教育、智能家居与基于位置的应用。
websocket 接口不能使用 requests 直接进行接口的调用,可以依赖第三方库的方式来实现调用,以下内容介绍如何调用第三方库实现 websocket 的接口自动化测试。
使用 python 语言实现 websocket 的接口自动化
1.安装 pyhton3 环境下载需要的运行库 2.下载需要的运行库 pip install websocket-client
import logging
from websocket import create_connection
logger = logging.getLogger(__name__)
url = 'ws://echo.websocket.org/' #一个在线的回环websocket接口,必须以websocket的方式连接后访问,无法直接在网页端输入该地址访问
wss = create_connection(url, timeout=timeout)
wss.send('Hello World')
res = wss.recv()
logger.info(res)
wss.close()
#待发送的数据体格式为:
data= {
"a" : "abcd",
"b" : 123
}
# 发送前需要把数据处理成 json 字符串
new_data=json.dumps(data,ensure_ascii=False)
wss.send(new_data)
# 接收的数据体的格式也为字符串
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
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 接口自动化测试用例:
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 使得客户端和服务器之间的数据交换变得更加简单,允许服务端主动向客户端推送数据。
在 WebSocket API 中,浏览器和服务器只需要完成一次握手,两者之间就直接可以创建持久性的连接,并进行双向数据传输。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。