写点什么

软件测试 | websocket 接口测试

  • 2023-03-05
    北京
  • 本文字数:2714 字

    阅读完需:约 9 分钟

简介

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 的接口自动化

环境准备

  1. 安装 pyhton3 环境

  2. 下载需要的运行库

pip install websocket-client

实战演示

1.连接 websoket 服务器

import loggingfrom websocket import create_connectionlogger = 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 loggingimport jsonfrom websocket import create_connectionlogger = 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']
复制代码

搜索微信公众号:TestingStudio 霍格沃兹的干货都很硬核

用户头像

社区:ceshiren.com 微信:ceshiren2023 2022-08-29 加入

微信公众号:霍格沃兹测试开发 提供性能测试、自动化测试、测试开发等资料、实事更新一线互联网大厂测试岗位内推需求,共享测试行业动态及资讯,更可零距离接触众多业内大佬

评论

发布
暂无评论
软件测试 | websocket接口测试_测试_测吧(北京)科技有限公司_InfoQ写作社区