软件测试-python篇websocket

websocket.create_connection()方法的详解

1. 代码如下所示：

#!/usr/bin/python
# -*- coding: utf-8 -*-

#@Users: LiMu
#@Files：create_connects.py
#@Times: 2022/5/31 
#@Software：PyCharm


import msgpack
import websocket
from lz4 import frame
from socket_connect import asse_parames


class create_ws(object):
    def __init__(self):
        #建立socket连接
        self.wss = websocket.create_connection("ws://10.1.0.85:20210/client")

        #需要传入的参数，以下参数已经过处理，要根据实际项目需求传参
        self.logins = asse_parames.recv_parames().get("login_premise")
        self.fronts = asse_parames.recv_parames().get("front_premise")
        self.business = asse_parames.recv_parames().get("business_premise")

    #发送数据给服务器，以下我们是分别请求登录接口、前置接口、业务接口，所以需要分别发3个接口的参数
    def send_msg(self):
        #服务器对接受数据有约束，直接msgpack格式的数据，需要用send_binary()方法发送，正常情况用send()发送。
        self.wss.send_binary(msgpack.packb(self.logins))
        if self.fronts.get("SlotFormation") != "off":
            self.wss.send_binary(msgpack.packb(self.fronts))
        self.wss.send_binary(msgpack.packb(self.business))

    #接受服务器返回的数据
    def recv_respone(self):
        while True:
            #对服务器返回的数据进行解压处理,不同项目服务器返回有差异,根据以下情况处理。
            #（1）服务器未对返回数据做任何处理，使用json.loads()处理返回结果。
            #（2）服务器对返回数据进行zlib压缩处理，使用zlib.decompress处理返回结果。
            #（3）服务器对返回数据进行frame压缩处理，使用frame.decompress()处理返回结果。
            recving = frame.decompress(self.wss.recv())
            respones = msgpack.loads(recving,strict_map_key=False)

            #过滤服务器主动推送的数据，保留需要的数据
            if respones.get("RequestId") == self.logins.get("RequestId"):
                print("\nLoginPremise：{}".format(self.logins))
                print("LoginRespones：{}\n".format(respones))
                continue
            elif respones.get("RequestId") == self.fronts.get("RequestId"):
                print("FrontsPremise：{}".format(self.fronts))
                print("FrontsRespones：{}\n".format(respones))
                continue
            elif respones.get("RequestId") == self.business.get("RequestId"):
                print("BusinessPremise：{}".format(self.business))
                return respones
    #关闭连接
    def close_ws(self):
        self.wss.close()

if __name__ == '__main__':
    create_line = create_ws()
    create_line.send_msg()
    responses = create_line.recv_respone()
    print("BusinessRespones：{}".format(responses))
    create_line.close_ws()

2. 运行结果如下：