Python实现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
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

发送 websocket 消息

wss.send('Hello World')
1

接收 websocket 消息

res = wss.recv()logger.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'>
1

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

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
1
2
3
4
5
6
7
8
9
10

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']
1
2
3
4

WebSocket 使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。

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

结语

作为过来人，给所有测试员一些忠告~~一名测试员，你真的不该只会点点点~随着体系的改变，对于现在的测试人员来说，不是自动化或者代码有多重要，而是懂自动化，懂代码，能够理解系统的实现，已经变成了必备技能。

自动化测试到底应该学什么？

1、建议是学 selenium，开源的，免费的，你可以下载源码研究，去了解其原理，再者 selenium 测试思路和手工测试类似，学起来比较轻松。

2、不建议学 seleniumIDE 录制，当然你可以通过录制一些然后转换为相应的脚本去学习，等你学好了 selenium，再去学 appium 你会有一种天然的似曾相识。为什么不建议学 QTP/UFT？因为收费/臃肿/现在做桌面软件测试的工作已经不太多了。

3、关于买书，看书能够提高一定的理论知识，但是解决不了实际问题，自动化测试的能力还是要靠一行一行代码敲出来的。

学习安排上！

感谢每一个认真阅读我文章的人，下面这个网盘链接也是我费了几个月时间整理的非常全面的，希望也能帮助到有需要的你！

这些资料，对于想转行做【软件测试】的朋友来说应该是最全面最完整的备战仓库，这个仓库也陪伴我走过了最艰难的路程，希望也能帮助到你！凡事要趁早，特别是技术行业，一定要提升技术功底。希望对大家有所帮助……

如果你不想一个人野蛮生长，找不到系统的资料，问题得不到帮助，坚持几天便放弃的感受的话，可以点击下方小卡片加入我们群，大家可以一起讨论交流，里面会有各种软件测试资料和技术交流，同时我也把上面花几个月整理的资料放里边了，赶快加入吧。

敲字不易，如果此文章对你有帮助的话，点个赞收个藏来个关注，给作者一个鼓励。也方便你下次能够快速查找。