关于软件测试:RPC接口测试技术websocket-自动化测试实践

WebSocket 是一种在单个 TCP 连贯上进行全双工通信 (Full Duplex 是通信传输的一个术语。通信容许数据在两个方向上同时传输，它在能力上相当于两个单工通信形式的联合。全双工指能够同时（刹时）进行信号的双向传输（A→B 且 B→A）。指 A→B 的同时 B→A，是刹时同步的) 的协定。

WebSocket 通信协议于 2011 年被 IETF 定为规范 RFC 6455，并由 RFC7936 补充标准。WebSocket API (WebSocket API 是一个应用 WebSocket 协定的接口，通过它来建设全双工通道来收发音讯) 也被 W3C 定为规范。

而 HTTP 协定就不反对长久连贯，尽管在 HTTP1.1 中进行了改良，使得有一个 keep-alive，在一个 HTTP 连贯中，能够发送多个 Request，接管多个 Response。

然而在 HTTP 中 Request = Response 永远是成立的，也就是说一个 request 只能有一个 response。而且这个 response 也是被动的，不能被动发动。

websocket 罕用于社交 / 订阅、多玩家游戏、协同办公 / 编辑、股市基金报价、体育实况播放、音视频聊天 / 视频会议 / 在线教育、智能家居与基于地位的利用。

websocket 接口不能应用 requests 间接进行接口的调用，能够依赖第三方库的形式来实现调用，以下内容介绍如何调用第三方库实现 websocket 的接口自动化测试。

实战

应用 python 语言实现 websocket 的接口自动化

环境筹备

1. 装置 pyhton3 环境下载须要的运行库
2. 下载须要的运行库
pip install websocket-client

实战演示

• 连贯 websoket 服务器

import logging
from websocket import create_connection
logger = logging.getLogger(__name__)
url = 'ws://echo.websocket.org/' #一个在线的回环 websocket 接口, 必须以 websocket 的形式连贯后拜访，无奈间接在网页端输出该地址拜访
wss = create_connection(url, timeout=timeout)

• 发送 websocket 音讯

  wss.send('Hello World')
  

• 接管 websocket 音讯

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

• 敞开 websocket 连贯

  wss.close()
  

• websocket 第三方库的调用不反对间接发送除字符串外的其余数据类型，所以在发送申请之前须要将 Python 结构化的格局, 转换为成为字符串类型或者 json 字符串后，再发动 websocket 的接口申请

  # 待发送的数据体格式为：data= {
    "a" : "abcd",
    "b" : 123
    }
  # 发送前须要把数据处理成 json 字符串
  new_data=json.dumps(data,ensure_ascii=False)
  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, dict):
        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 封装 websocket 接口通用操作：

  import logging
  import json
  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 self.load_json(res)
            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, dict):
            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_json = self.wss.recv()
                all_json_recv = self.load_json(recv_json)
                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']

WebSocket 使得客户端和服务器之间的数据交换变得更加简略，容许服务端被动向客户端推送数据。

在 WebSocket API 中，浏览器和服务器只须要实现一次握手，两者之间就间接能够创立持久性的连贯，并进行双向数据传输。