RPC接口测试技术-websocket 自动化测试实践

简介: RPC接口测试技术-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 定为标准。

而 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

实战演示

  • 连接 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)
  • 发送 websocket 消息
wss.send('Hello World')
  • 接收 websocket 消息
res = wss.recv()
logger.info(res)
  • 关闭 websocket 连接
wss.close()
  • websocket 第三方库的调用不支持直接发送除字符串外的其他数据类型,所以在发送请求之前需要将 Python 结构化的格式,转换为成为字符串类型或者 json 字符串后,再发起 websocket 的接口请求
#待发送的数据体格式为:
data= {
    "a" : "abcd",
    "b" : 123
    }
# 发送前需要把数据处理成 json 字符串
new_data=json.dumps(data,ensure_ascii=False)
wss.send(new_data)
  • 接收的数据体的处理:如果接口定义为 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
  • 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 接口自动化测试用例:

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 使得客户端和服务器之间的数据交换变得更加简单,允许服务端主动向客户端推送数据。

更多技术文章

相关文章
|
12天前
|
运维 监控 Devops
DevOps文化下的自动化运维实践
本文将探讨在DevOps文化背景下,自动化运维的重要性及其实现方式。通过分析自动化运维的优势和挑战,文章提供了具体的实施策略和案例,旨在帮助读者理解如何在DevOps实践中融入自动化运维,以提高软件开发和部署的效率与质量。
|
1天前
BOSHIDA DC/AC电源模块:推动工业自动化技术的发展
BOSHIDA DC/AC电源模块:推动工业自动化技术的发展
BOSHIDA DC/AC电源模块:推动工业自动化技术的发展
|
1天前
|
机器学习/深度学习 人工智能 测试技术
自动化测试框架的演进与实践
【6月更文挑战第23天】在软件工程领域,自动化测试框架的发展不断推动着质量保证的效率和效果。本文将探讨自动化测试框架从简单脚本到复杂集成系统的演变过程,并分析当前流行的框架如Selenium、Appium以及新兴的AI驱动测试工具。我们将通过具体案例,展示如何在现代软件开发实践中有效应用这些框架以提升测试覆盖率和准确性。
|
4天前
|
机器学习/深度学习 人工智能 测试技术
探索自动化测试的前沿技术与实践
自动化测试作为提升软件开发效率和质量的关键工具,正经历着前所未有的变革。随着人工智能、机器学习、云计算等技术的融合与创新,自动化测试不断突破传统界限,展现出更智能、更高效、更灵活的发展趋势。本文将深入探讨自动化测试领域的最新技术进展,分析其在现代软件开发中的应用,并讨论如何有效整合这些技术以最大化测试效率和准确性。
|
6天前
|
数据可视化 前端开发 Java
自动化测试框架的选择与实践: Selenium vs. TestComplete
【6月更文挑战第18天】在软件开发的海洋中,自动化测试是一艘能够确保产品质量和效率的坚固船只。本文将深入探讨两种流行的自动化测试框架——Selenium和TestComplete,从它们的优势、局限性到适用场景进行对比分析。我们将通过实际案例来揭示如何根据项目需求选择最合适的测试工具,并提供一些实用的实施建议。文章旨在为读者提供清晰的指导,帮助他们在自动化测试的旅程中做出明智的决定。
12 3
|
5天前
|
机器学习/深度学习 人工智能 机器人
现代软件测试中的自动化工具与技术
随着软件开发复杂性的增加,自动化测试在现代软件开发中变得至关重要。本文探讨了当前流行的自动化测试工具和技术,以及它们如何提高软件质量、加快发布速度和降低成本。
18 1
|
6天前
|
机器学习/深度学习 人工智能 运维
现代企业中的自动化运维技术与挑战
随着信息技术的迅猛发展,现代企业在面对日益复杂的技术架构和运营需求时,越来越多地依赖自动化运维技术来提升效率和稳定性。本文探讨了自动化运维技术的实施现状、关键挑战以及未来的发展方向,旨在为企业决策者和技术从业者提供深入的思考和实践指导。
22 1
|
8天前
|
机器学习/深度学习 人工智能 数据处理
AI技术来支持开发过程中的自动化和智能化
AI技术来支持开发过程中的自动化和智能化
154 2
|
8天前
|
机器学习/深度学习 人工智能 Devops
探索软件测试自动化的未来:技术挑战与机遇
随着软件开发周期的不断缩短和复杂性的增加,软件测试自动化在确保质量和效率方面扮演着越来越重要的角色。本文将深入探讨软件测试自动化的发展现状、面临的技术挑战以及未来的机遇。
12 2
|
4天前
|
机器学习/深度学习 人工智能 算法
智能家居技术的未来:从自动化到自主
本文探讨了智能家居技术的发展趋势,从早期的自动化设备到现今的智能系统,再到未来可能实现的家居自主化。分析了当前技术的限制、未来的发展潜力以及实现家居自主化所面临的挑战和机遇。文章旨在为读者提供一个关于智能家居技术演进的视角,并探讨这一领域未来的发展方向。

热门文章

最新文章