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)) #

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

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

在 WebSocket API 中,浏览器和服务器只需要完成一次握手,两者之间就直接可以创建持久性的连接,并进行双向数据传输。

相关文章
|
11天前
|
Java 测试技术 开发者
初学者入门:掌握单元测试的基础与实践
【10月更文挑战第14天】单元测试是一种软件测试方法,它验证软件中的最小可测试单元——通常是单独的函数或类——是否按预期工作。单元测试的目标是确保每个模块在其自身范围内正确无误地运行。这些测试应该独立于其他模块,并且应该能够反复执行而不受外部环境的影响。
36 2
|
19天前
|
机器学习/深度学习 人工智能 监控
提升软件质量的关键路径:高效测试策略与实践在软件开发的宇宙中,每一行代码都如同星辰般璀璨,而将这些星辰编织成星系的过程,则依赖于严谨而高效的测试策略。本文将引领读者探索软件测试的奥秘,揭示如何通过精心设计的测试方案,不仅提升软件的性能与稳定性,还能加速产品上市的步伐,最终实现质量与效率的双重飞跃。
在软件工程的浩瀚星海中,测试不仅是发现缺陷的放大镜,更是保障软件质量的坚固防线。本文旨在探讨一种高效且创新的软件测试策略框架,它融合了传统方法的精髓与现代技术的突破,旨在为软件开发团队提供一套系统化、可执行性强的测试指引。我们将从测试规划的起点出发,沿着测试设计、执行、反馈再到持续优化的轨迹,逐步展开论述。每一步都强调实用性与前瞻性相结合,确保测试活动能够紧跟软件开发的步伐,及时适应变化,有效应对各种挑战。
|
16天前
|
测试技术 UED
软件测试的艺术与实践
【10月更文挑战第9天】 在数字时代的浪潮中,软件成为了我们生活和工作不可或缺的一部分。然而,高质量的软件背后,是无数测试工程师的默默付出。本文将通过深入浅出的方式,探讨如何进行高效的软件测试,确保软件产品的质量与稳定性。我们将一起揭开软件测试的神秘面纱,从基础理论到实际操作,一步步走进这个充满挑战与创造的世界。
|
2天前
|
机器学习/深度学习 人工智能 自然语言处理
探索软件测试的边界:从基础到高级的实践之旅
【10月更文挑战第21天】 在当今数字化时代,软件已成为我们生活和工作中不可或缺的一部分。随着技术的快速发展,对软件质量的要求也日益提高。本文旨在通过深入浅出的方式,带领读者踏上一场从基础到高级的软件测试实践之旅。我们将探讨软件测试的基本概念、重要性以及如何有效地进行测试规划和执行。通过具体案例分析,揭示常见错误及其解决方案,同时展望未来软件测试领域的发展趋势。无论你是软件开发新手还是经验丰富的测试工程师,这篇文章都将为你提供宝贵的见解和启发。
17 8
|
1天前
|
监控 安全 jenkins
探索软件测试的奥秘:自动化测试框架的搭建与实践
【10月更文挑战第24天】在软件开发的海洋里,测试是确保航行安全的灯塔。本文将带领读者揭开软件测试的神秘面纱,深入探讨如何从零开始搭建一个自动化测试框架,并配以代码示例。我们将一起航行在自动化测试的浪潮之上,体验从理论到实践的转变,最终达到提高测试效率和质量的彼岸。
|
4天前
|
测试技术 C# 数据库
C# 一分钟浅谈:测试驱动开发 (TDD) 实践
【10月更文挑战第18天】测试驱动开发(TDD)是一种软件开发方法论,强调先编写测试代码再编写功能代码,以确保代码质量和可维护性。本文从 TDD 的基本概念入手,详细介绍了其核心步骤——编写测试、运行测试并失败、编写代码使测试通过,以及“红绿重构”循环。文章还探讨了 TDD 的优势,包括提高代码质量、促进设计思考、减少调试时间和文档化。此外,文中分析了常见问题及解决方案,如测试覆盖率不足、测试代码过于复杂、忽视重构和测试依赖过多,并通过一个简单的计算器类的代码案例,展示了 TDD 的实际应用过程。
13 1
|
11天前
|
机器学习/深度学习 人工智能 自然语言处理
探索AI在软件测试中的创新应用与实践###
本文旨在探讨人工智能(AI)技术如何革新软件测试领域,提升测试效率、质量与覆盖范围。通过深入分析AI驱动的自动化测试工具、智能化缺陷预测模型及持续集成/持续部署(CI/CD)流程优化等关键方面,本研究揭示了AI技术在解决传统软件测试痛点中的潜力与价值。文章首先概述了软件测试的重要性和当前面临的挑战,随后详细介绍了AI技术在测试用例生成、执行、结果分析及维护中的应用实例,并展望了未来AI与软件测试深度融合的趋势,强调了技术伦理与质量控制的重要性。本文为软件开发与测试团队提供了关于如何有效利用AI技术提升测试效能的实践指南。 ###
|
20天前
|
测试技术
软件测试中的探索性测试(ET)实践
【10月更文挑战第5天】本文将深入探讨一种与传统脚本化测试不同的测试方法——探索性测试(Exploratory Testing,简称ET)。我们将通过一个实际案例来展示ET的有效性,并分享如何将ET融入日常的软件测试流程中。文章旨在为测试人员提供一种灵活、高效的测试策略,帮助他们更好地发现软件中的缺陷。
|
20天前
|
Web App开发 设计模式 测试技术
自动化测试框架的搭建与实践
【10月更文挑战第5天】本文将引导你理解自动化测试框架的重要性,并通过实际操作案例,展示如何从零开始搭建一个自动化测试框架。文章不仅涵盖理论,还提供具体的代码示例和操作步骤,确保读者能够获得实用技能,提升软件质量保障的效率和效果。
|
2天前
|
测试技术 Python
软件测试的艺术:从理论到实践的探索
【10月更文挑战第23天】在软件开发的世界中,测试是确保产品质量的关键步骤。本文将深入探讨软件测试的核心理念、方法和实践,揭示如何通过精心设计和执行测试来提高软件质量。我们将一起探索测试的不同阶段,包括单元测试、集成测试、系统测试和验收测试,以及它们如何相互补充,共同构建起一个坚实的质量保证体系。文章旨在启发读者思考如何在自己的项目中实施有效的测试策略,从而提升软件的可靠性和性能。