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