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 中，浏览器和服务器只需要完成一次握手，两者之间就直接可以创建持久性的连接，并进行双向数据传输。

最后感谢每一个认真阅读我文章的人，看着粉丝一路的上涨和关注，礼尚往来总是要有的，虽然不是什么很值钱的东西，如果你用得到的话可以直接拿走！

软件测试面试文档

我们学习必然是为了找到高薪的工作，下面这些面试题是来自阿里、腾讯、字节等一线互联网大厂最新的面试资料，并且有字节大佬给出了权威的解答，刷完这一套面试资料相信大家都能找到满意的工作。