python调⽤websocket接⼝_python接⼝⾃动化测试之
websocket接⼝
前⾔
前⾯写了⼀篇专门介绍使⽤python去做webservice接⼝⾃动化测试的⽂章，然后有⼩伙伴看完之后反馈说能不能出⼀篇python做websocket接⼝⾃动化的⽂章，所以今天这篇⽂章就专门来和⼤家聊聊这个问题，如何使⽤python来实现websocket类型的接⼝⾃动化。其实不管是http的接⼝，还是webservice、websocket的接⼝，做⾃动化的思路都是⼀样的，并没有什么很⼤的差别，⽆⾮就是⽤到的请求库不⼀样⽽已。
在python中，发送http请求，⽐较有名的是requests这个库；webservice的接⼝可以使⽤suds-py3来请求。那么websocket怎么请求呢？其实在python中也有很多第三⽅库可以⽤来发送websocket请求。⽐如websocket-client, 那么接下来我们先来看看websocket-client这个库的使⽤。
1、环境安装
由于 websocket-client 是第三⽅库，使⽤之前需要先安装，安装直接使⽤ pip 命令安装即可。
pip install websocket-client
安装好 websocket-client 这个模块之后，就可以正式开始使⽤了，关于使⽤ websocket-client 模块创建 socket 客户端有两种⽅式。⼀个是 ate_connection ⽅法，还有⼀个是 websocket.WebSocketApp
2、create_connection
2.1、使⽤⽅法
建⽴连接
使⽤ create_connection ⽅法传⼊ WebSocket 的接⼝地址就可以和接⼝建⽴起连接，返回⼀个连接的客户端对象。
获取连接状态
连接对象有⼀个 getstatus ⽅法，可以⽤来获取连接状态，如果连接成功了，获取到的状态码即为 101。
发⽣请求参数
建⽴起连接之后，如果要发送请求参数，则可以使⽤连接对象的 send ⽅法，注意点，send ⽅法发⽣的请求参数必需是字符串类型。
获取返回结果
连接对象有⼀个 recv ⽅法，专门⽤来获取返回结果的
关闭连接
由于 WebSocket 使⽤的是 tcp 长连接，建⽴连接之后，如果要断开连接，需要调⽤连接对象的 close ⽅法才能断开连接。
websocket和socket2.2、案例代码
import json
from websocket import create_connection
# 1、建⽴连接
ws = create_connection("ws://127.0.0.1:5000/info")
# 2、获取连接状态
print("获取连接状态：", ws.getstatus())
# 3、发送请求参数
ws.send('发送数据 hello musen')
# 4、获取返回结果
result = ws.recv()
print("接收结果：", result)
# 5、关闭连接
ws.close()
关于 create_connection 的使⽤就是这么⼏部操作，使⽤起来也很⽅便，⼀般在测试 WebSocket 接⼝消
息发送功能是否正常，发送完消息就断开连接，⽤这个⽅法还是⽐较⽅便。不过很多情况下 WebSocket 接⼝只是⽤来做持续性监听⼯作，并不涉及到传递请求参数。那么使⽤ WebSocket 中 WebSocketApp 这个类来测试监听功能会更合适。
3、WebSocketApp
3.1、基本使⽤
如果要对某个接⼝进⾏长期的进⾏监听，监听其功能是否异常，那么就可以使⽤ webSocketApp 这个类来创建⼀个长连接进⾏监听。接下来我们⼀起来看看 WebSocketApp 这个类源码的初始化⽅法。
1)、WebSocketApp 初始化⽅法的源码
从上述代码中可以看到 WebSocketApp 进⾏初始化的时候定义了很多的参数。接下来给⼤家介绍⼏个常⽤的参数
url : url 是⽤来接收连接的 WebSocket 的 url 地址的
header：header 是⽤来传⼊建⽴连接的请求头的
on_message：是⽤来指定监听到数据返回是的处理⽅法
on_error：当连接出现错误是，会触发 on_error 指定的⽅法
on_close：当连接关闭是，则会触发 on_close ⽅法
其他的参数 ⼤家可以去看官⽅⽂档，这边就不给⼤家⼀⼀说明了。
2)、WebSocketApp 对象的 run_forever ⽅法
创建了⼀个连接对象之后，需要不断监听返回的数据，则调⽤ run_forever ⽅法，要保持长连接即可，接下来我们来看⼀个使⽤案例。
3.2、案例代码
需求：监测 ws://127.0.0.1:5000/info 这个接⼝接⼝是否正常。
在下⾯的代码中我们定义了三个⽅法
on_message ⽅法：专门⽤来监听和处理返回的数据(这边只是打印了⼀下，没有做特别的处理)
on_error ⽅法：专门⽤来监听是否出现错误(这边只是打印了⼀下错误，没有做特别的处理)
on_close ⽅法：⽤来监听连接是否关闭，当连接关闭时，则会调⽤该⽅法
import websocket
# 定义⼀个⽤来接收监听数据的⽅法
def on_message(ws, message):
print("监听到服务器返回的消息，内容如下：")
print(message)
# 定义⼀个⽤来处理错误的⽅法
def on_error(ws, error):
print("-----连接出现异常，异常信息如下-----")
print(error)
# 定义⼀个⽤来处理关闭连接的⽅法
def on_close(ws):
print("-------连接已关闭------")
if __name__ == "__main__":
ws = websocket.WebSocketApp("ws://127.0.0.1:5000/info",
on_message=on_message,
on_error=on_error,
on_close=on_close,
)
ws.run_forever()
关于 websocket-client 的基本使⽤就给⼤家介绍到这⾥，⼤家有兴趣可以⾃⼰去扩展研究。
4、⾃动化测试案例
接下来给⼤家分享⼀下如何通过 unittest 去编写 WebSocket 的接⼝⽤例。下⾯是我⾃⼰写的⼀个 WebSocket 的接⼝1、接⼝说明
投票接⼝
接⼝地址：ws://127.0.0.1:5000/vote
请求协议：WebSocket
请求参数：
id：选⼿编号(JSON 字符串)
返回结果(JSON)：
status: 处理状态
msg：描述
data：相关数据
Time：当前时间
2、⽤例代码
import json
import unittest
from websocket import create_connection
import websocket
class TestVote(unittest.TestCase):
url = 'ws://127.0.0.1:5000/vote'
@classmethod
def setUpClass(cls):
# 前置⽅法
def test_connect_status(self):
"""测试连接状态"""
# 断⾔连接状态
self.assertEqual(101, status())
def test_send_info(self):
# 第⼀步：准备参数
params = {'id': 1}
expected = {'status': "success"}
# 第⼆步：发送请求
self.ws.send(json.dumps(params))
# 获取结果
result = v()
res = json.loads(result)
# 第三步：断⾔：
print("接收结果：", res)
self.assertEqual(expected['status'], res['status'])
@classmethod
def tearDownClass(cls):
cls.ws.close()
if __name__ == '__main__':
unittest.main()
说明：上⾯针对于 ws://127.0.0.1:5000/vote 这个接⼝写了两条⽤例，⼀条测试连接状态是否正常，⼀条测试数据发送的功能是否异常，关于 websockt 接⼝⾃动化测试就给⼤家介绍到这⾥了。

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。