Python接口测试教程：如何安装和使用WebSocket进行高效通信
更新时间为 2020 年 10 月 16 日 15 时 05 分 55 秒，作者称自己被狗咬了。
这篇文章主要是对用 Python 进行 websocket 接口测试进行介绍。它有助于大家更好地理解和运用 Python。对此感兴趣的朋友可以进行了解。
我们在进行接口测试的时候，除了常见的 http 接口之外，还有一种较为多见的接口，那就是 socket 接口。今天，我们来讲讲如何使用 Python 来进行 websocket 接口测试。
现在大多数使用的是 websocket 。那么，我们首先来安装 websocket 的安装包。
pip install websocket-client
安装完之后，我们就开始我们的websocket之旅了。
我们先来看个炒鸡简单的栗子：
import websocket
ws = websocket.WebSocket()
ws 连接到 "ws://example.com/websocket" ，
http_proxy_host 的值为 "proxy_host_name" 。
http_proxy_port=3128)
创建一个 websocket 连接就是这个栗子。这个模块具备通过 http 代理访问 websocket 的功能。代理服务器能够使用 connect 方法连接到 websocket 端口。默认的 squid 设置为“仅允许连接 HTTPS 端口”。
在websocket里，我们有常用的这几个方法：
on_message方法：
def on_message(ws, message): print(message)
on_message 的作用是接受消息。server 发送的所有消息都能够通过 on_message 这个方法来进行收取。
on_error方法：
def on_error(ws, error): print(error)
这个方法是用于处理错误异常的。如果 socket 的程序出现通信问题，那么这个方法能够捕捉到。
on_open方法：
def on_open(ws):
def run(*args):
for i in range(30):
# send the message, then wait
# isn't closed
ws.send("Hello %d" % i)
time.sleep(1)
time.sleep(1)
ws.close()
打印("线程终止...")
Thread(target=run).start()
on_open 方法的作用是保持连接。像上面这样的一个例子，就是保持连接的过程。它每隔一段时间就会做一件事，会在 30 秒内一直发送 hello，最后停止。
on_close方法：
def on_close(ws):
print("### closed ###")
onclose主要就是关闭socket连接的。
如何创建一个websocket应用：
ws 为 websocket.WebSocketApp，其连接的地址是 "wss://echo.websocket.org"
括号里面是要连接的 socket 的地址，在 WebSocketApp 实例化的方法中还能有其他参数，这些参数就是我们刚才介绍的那些方法。
ws 等于 websocket 的 WebSocketApp，其连接地址为 "ws://echo.websocket.org/"
on_message=on_message,
on_error=on_error,
on_close=on_close)
指定了这些参数之后就可以直接进行调用了，例如：
ws.on_open = on_open
这样就是调用了on_open方法
如果我们期望我们的 socket 保持长时间的连接，始终保持连接状态，那么就能够运用 run_forever 方法。
ws.run_forever()
完整代码：
import websocket
from threading import Thread
import time
import sys
def on_message(ws, message):
print(message)
def on_error(ws, error):
print(error)
def on_close(ws):
print("### closed ###")
def on_open(ws):
def run(*args):
for i in range(3):
# send the message, then wait
# so thread doesn't exit and socket
# isn't closed
ws.send("Hello %d" % i)
time.sleep(1)
time.sleep(1)
ws.close()
print("Thread terminating...")
Thread(target=run).start()
if __name__ == "__main__":
websocket.enableTrace(True)
主机地址为“ws://echo.websocket.org/”。
ws 是 websocket 的 WebSocketApp 并传入了 host 参数
on_message=on_message,
on_error=on_error,
on_close=on_close)
ws.on_open = on_open
ws.run_forever()
如果要发送一条短消息，并且在发送完成后即刻断开连接，那么我们可以运用短连接。
通过 create_connection 函数来创建一个 WebSocket 连接。
创建了一个连接，连接的地址是 "ws://echo.websocket.org/"，并将其赋值给变量 ws 。
输出“正在发送‘Hello, World’……”
ws.send("Hello, World")
print("Sent")
print("Receiving...")
result = ws.recv()
打印("接收到 '%s'" % result)
ws.close()
关于websocket的介绍就到这儿了。
phper