python+websocket接口测试

本文主要是介绍python+websocket接口测试，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！

1.连接websocket服务器

import logging
from websocket import creat_connertion
logger = logging.getLogger(__name__)
url = 'ws://echo.websocket.org' #一个在线回环websocket接口，必须以websocket的方式连接后访问，无法直接在网页端输入该地址访问
wss = creat_connertion(url, timeout=timeout)

2.发送websocket消息

wss.send('hello world')

3.接受websocket消息

res = wss.recv()
logger.info(res)

4.关闭websocket连接

wss.close()

websocket第三方库的调用不支持直接发送除字符串外的其他数据类型，所以在发送请求之前需要将python结构化的格式，转换成字符串类型或者json字符串后，再发起websocket的接口请求。

#待发送的数据体格式为：
data= {
      "a" : "abcd",
      "b" : 123
}
#发送前需要把数据处理成json字符串
new_data = json.dumps(data,ensure_ascii=Flase)
wss.send(new_data)

接收的数据体的处理：如果接口定义为json的话，由于数据的传输都是字符串格式的，需要对接收的数据进行转换操作

#接收的数据体的格式也是字符串：
logger.info(type(res)) #<class 'str'>

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

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,str):
        for key, value in base_str.items():
            base_str[key]=load_json(value)
        return base_str
    return base_str

websocket接口自动化测试，二次封装demo展示

web_socket_util.py

import logging
import json
from json import JSONDecodeError

from websocket import create_connection

logger = 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 base_str
            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, str):
            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_list = self.wss.recv()
                all_json_recv = self.load_json(recv_list)
                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

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']

这篇关于python+websocket接口测试的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！