2023年7月11日发(作者：)

#⽤Python进⾏websocket接⼝测试,配合locust性能测试，实现长链接压⼒测试⽤Python进⾏websocket接⼝测试我们在做接⼝测试时，除了常见的http接⼝，还有⼀种⽐较多见，就是socket接⼝，今天讲解下怎么⽤Python进⾏websocket接⼝测试。webscoket的属性、⽅法和事件需要准备的测试环境安装python，版本3.0以上安装IDE，pycharm等安装websocket、websocket-client 包3、websocket接⼝测试前，把需要⽀持的安装包都安装好pip install websocket-clientpip install websockets…安装完之后，我们就开始我们的websocket之旅了。先举个简单的例⼦：import websocketws = ket()t("ws:///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 # so thread doesn't exit and socket # isn't closed ("Hello %d" % i) (1)

(1) () print("")

Thread(target=run).start()on_open⽅法是⽤来保持连接的，上⾯这样的⼀个例⼦，就是保持连接的⼀个过程，每隔⼀段时间就会来做⼀件事,他会在30s内⼀直发送hello。最后停⽌。on_close⽅法：def on_close(ws): print("### closed ###")onclose主要就是关闭socket连接的。如何创建⼀个websocket应⽤：ws = ketApp("wss://")括号⾥⾯就是你要连接的socket的地址，在WebSocketApp这个实例化的⽅法⾥⾯还可以有其他参数，这些参数就是我们刚刚介绍的这些⽅法。ws = ketApp("ws:///", on_message=on_message, on_error=on_error, on_close=on_close)指定了这些参数之后就可以直接进⾏调⽤了，例如：_open = on_open这样就是调⽤了on_open⽅法如果我们想让我们的socket保持长连接，⼀直连接着，就可以使⽤run_forever⽅法：_forever()综上所述，完整的应该这么写：import websocketfrom threading import Threadimport timeimport 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 ("Hello %d" % i) (1)

(1) () print("")

Thread(target=run).start()

if __name__ == "__main__":

Trace(True) host = "ws:///" ws = ketApp(host, on_message=on_message, on_error=on_error, on_close=on_close) _open = on_open _forever()如果想要通信⼀条短消息，并在完成后⽴即断开连接，我们可以使⽤短连接：from websocket import create_connectionws = create_connection("ws:///")print("Sending 'Hello, World'...")("Hello, World")print("Sent")print("")result = ()print("Received '%s'" % result)()这⾥更新⼀下，对于websocket 发送如果我们直接发送，很可能因为格式不正确，服务端不能解析，这时可以考虑，导⼊json包对 发送的数据进⾏解析，注意把代码放在对应的位置上。import jsonData={"channel": "xxx", "market": 'xxx', "event": "xxxx"} ((Data).encode('utf8'))json dump编码，loads 解码，或者说反序列化⽐较好，dumps是把python中的数据对象，序列化成json格式的字符串；loads是把json格式的字符串反序列化为python中的数据对象obj = ((‘utf8’))#据说是终极⽅法，获取内容⾃⼰解码,通常情况下，响应消息体是json格式都需要转换为python中的数据对象print(obj)print(obj[‘form’])#如果不处理，取数据是很⿇烦的print(obj[‘form’][‘2’])