pytest测试框架在websocket接口自动化集成的作用

lsekfe

1、介绍
　　WebSocket是一种在单个TCP通信的协议。WebSocket通信协议于2011年被IETF定为标准RFC 6455，并由RFC7936补充规范。WebSocket API也被W3C定为标准。
　　WebSocket使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。在WebSocket API中，浏览器和服务器只需要完成一次握手，两者之间就直接可以创建持久性的连接，并进行双向数据传输。
　　2、原理




3、与http协议的比较
　　相同之处：
　　·都是基于TCP的应用层协议；
　　· 都使用Request/Response模型进行连接的建立；
　　· 在连接的建立过程中对错误的处理方式相同，在这个阶段WS可能返回和HTTP相同的返回码；
　　· 都可以在网络中传输数据。
　　不同之处:
　　· WS使用HTTP来建立连接，但是定义了一系列新的header域，这些域在HTTP中并不会使用；
　　· WS的连接不能通过中间人来转发，它必须是一个直接连接；
　　· WS连接建立之后，通信双方都可以在任何时刻向另一方发送数据；
　　· WS连接建立之后，数据的传输使用帧来传递，不再需要Request消息；
　　· WS的数据帧有序。
　　4、websocket应用场景
　　websocket的使用场景:
　　社交聊天、弹幕、多玩家游戏、协同编辑、股票基金实时报价、体育实况更新、视频会议/聊天、基于位置的应用、在线教育、智能家居等 需要高实时的场景。
　　websocket接口测试实战
　　以码同学全栈接口项目中的websocket接口为例，项目地址  http://**.***.**.**:****/pinter/


抓包查看，熟悉接口业务
　　1、访问项目首页，打开浏览器的开发者工具，选择Network-->WS。




2、点击页面上websocket案例的开始练习，抓包看到以下数据。
　　绿色箭头表示浏览器向服务器发送的数据，红色箭头表示服务器返回的数据。
　　本次请求是的业务是打开聊天窗口后，匹配一个后台的客服老师。
　　请求参数：
　　·type为match表示这是个匹配类型的动作
　　· from表示客户id
　　· to表示该请求发送给谁
　　· system为服务器系统

1. <font size="3">　{
   
2. 
   
3. 　　"type":"match",
   
4. 
   
5. 　　"from":"发送者",
   
6. 
   
7. 　　"to":"system"
   
8. 
   
9. 　　}</font>

复制代码

　服务器返回数据：
　　·code是业务状态码，0表示成功
　　· from表示该数据是谁发来的，system表示是系统发来的
　　· msg表示系统为客户匹配到的老师id
　　· msgId表示的是消息id
　　· timestamp当前时间戳
　　· to表示该数据是发给谁的，发给客户admin的
　　· type表示请求类型，match表示这个匹配请求

1. <font size="3">{
   
2. 
   
3. 　　"code":"0",   
   
4. 
   
5. 　　"from":"system",   
   
6. 
   
7. 　　"msg":"cf2137234f3943e3898df8fcd14f099a__mtx",   
   
8. 
   
9. 　　"msgId":"4c7ff9b3-ce6b-4891-8c14-41e0f45e0bd0",   
   
10. 
    
11. 　　"timestamp":"1665630374482",   
    
12. 
    
13. 　　"to":"admin",   
    
14. 
    
15. 　　"type":"match"
    
16. 
    
17. 　　}</font>

复制代码

3、聊天窗口发送聊天信息，抓包查看
　　第1步：对应数字图中的数字1，发送数据到服务器，服务器会根据信息进行推送，将数据发给指定的客服老师。
　　·type表示请求类型，normal是正常聊天消息的意思
　　· from表示这个数据是从谁发来的，从客户admin发出来的
　　· to表示这个是发给谁的，发给cf2137234f3943e3898df8fcd14f099a__mtx这个客服老师的
　　· msg是发给客服老师具体的信息

1. <font size="3">　{
   
2. 
   
3. 　　"type":"normal",   
   
4. 
   
5. 　　"from":"admin",   
   
6. 
   
7. 　　"to":"cf2137234f3943e3898df8fcd14f099a__mtx",   
   
8. 
   
9. 　　"msg":"你好"
   
10. 
    
11. 　　}</font>

复制代码

第2步：对应图中的数字2，服务器将消息推送给客服老师成功后，把成功的信息发送给客户，其中核心字段msg为push success表示推送成功，type为receipt表示服务器返回的。

1. <font size="3">{
   
2. 
   
3. 　　"code":"0",
   
4. 
   
5. 　　"from":"system",
   
6. 
   
7. 　　"msg":"push success",
   
8. 
   
9. 　　"msgId":"130f4370-6b2b-4634-b3f1-59a9728eb8f1",
   
10. 
    
11. 　　"timestamp":"1665631227681",
    
12. 
    
13. 　　"to":"admin",
    
14. 
    
15. 　　"type":"receipt"
    
16. 
    
17. 　　}</font>

复制代码

第3步：对应图中的数字3，客服老师在收到服务器推送给自己的客户消息后，返回给客户的信息，也是由服务器转发过来的，而并不是直接返回给客户的。
　　·type表示请求类型，normal是正常聊天消息的意思
　　· from表示这个数据是从谁发来的，从客服老师cf2137234f3943e3898df8fcd14f099a__mtx发出来的
　　· to表示这个是发给谁的，发给admin这个客户的
　　· msg是发给客服具体的信息

1. <font size="3">{
   
2. 
   
3. 　　"code":"0",
   
4. 
   
5. 　　"from":"cf2137234f3943e3898df8fcd14f099a__mtx",
   
6. 
   
7. 　　"msg":"同学，你好，非常高兴为你服务，有什么需要我帮忙的呢?",
   
8. 
   
9. 　　"msgId":"8486956c-e804-42b9-bb9b-d1574f1ca23f",
   
10. 
    
11. 　　"timestamp":"1665631228807",
    
12. 
    
13. 　　"to":"admin",
    
14. 
    
15. 　　"type":"normal"
    
16. 
    
17. 　　}</font>

复制代码

python调用websocket接口





接口框架中适配websocket封装
　　1、封装底层websocket类
　　在config目录下创建一个websocket.yml的文件，写上如下内容:






在client.py中增加如下代码：





2、封装websocket api层
　　在api包下创建websoket_api包，其下创建imserver_api.py ，写上如下代码：






3、编写测试用例层
　　在testcases目录下创建websocketapi包，在其下创建test_websocket_api.py：

1. <font size="3"># !/usr/bin python3                              
   
2. 
   
3. 　　# encoding: utf-8 -*-
   
4. 
   
5. 　　# [url=home.php?mod=space&uid=273702]@file[/url]     : test_websocket_api.py                  
   
6. 
   
7. 　　# [url=home.php?mod=space&uid=267564]@Author[/url]   : 沙陌 Matongxue_2
   
8. 
   
9. 　　# [url=home.php?mod=space&uid=3428]@time[/url]     : 2022-10-19 16:07
   
10. 
    
11. 　　# @Copyright: 北京码同学
    
12. 
    
13. 　　import json
    
14. 
    
15. 　　import pytest
    
16. 
    
17. 　　from api.websoket_api.imserver_api import ImServerApi
    
18. 
    
19. 　　class TestImServerApi:   
    
20. 
    
21. 　　   kfid = '' # 定义客服id，全局变量作为各个测试用例的关联数据        
    
22. 
    
23. 　　   def setup_class(self):      
    
24. 
    
25. 　　      self.im = ImServerApi() # 创建一个websocket协议的接口对象   
    
26. 
    
27. 　　   # 测试客服匹配   
    
28. 
    
29. 　　   def test_match(self):      
    
30. 
    
31. 　　       params = {           
    
32. 
    
33. 　　           "msgId": "111",           
    
34. 
    
35. 　　           "type": "match",           
    
36. 
    
37. 　　           "from": "shamo",           
    
38. 
    
39. 　　           "to": "system"      
    
40. 
    
41. 　　       }      
    
42. 
    
43. 　　       self.im.send()      
    
44. 
    
45. 　　       res = self.im.recv()      
    
46. 
    
47. 　　       res = json.loads(res)# 将其转换成json对象      
    
48. 
    
49. 　　       assert res['code'] == '0'      
    
50. 
    
51. 　　       # 提取msg，msg是匹配到的客服id      
    
52. 
    
53. 　　       self.__class__.kfid = res['msg']   
    
54. 
    
55. 　　   # 测试给客服发送正常消息   
    
56. 
    
57. 　　   def test_message(self):      
    
58. 
    
59. 　　       params={           
    
60. 
    
61. 　　       "msgId": "111",           
    
62. 
    
63. 　　           "type": "normal",           
    
64. 
    
65. 　　           "from": "admin",           
    
66. 
    
67. 　　           "to": f"{self.__class__.kfid}",           
    
68. 
    
69. 　　           "msg": "你好"      
    
70. 
    
71. 　　        }      
    
72. 
    
73. 　　        self.im.send(json.dumps(params))      
    
74. 
    
75. 　　        res = self.im.recv()      
    
76. 
    
77. 　　        res = json.loads(res)# 将其转换成json对象      
    
78. 
    
79. 　　        pytest.assume(res['code'] == '0',f'期望值是0,实际结果是{res["code"]}')　
    
80. 
    
81. 　　        pytest.assume(res['msg'] == 'push success', f'期望值是0,实际结果是{res["msg"]}')      
    
82. 
    
83. 　　        # 再次接收客服发来的数据      
    
84. 
    
85. 　　        res = self.im.recv()      
    
86. 
    
87. 　　        res = json.loads(res)# 将其转换成json对象      
    
88. 
    
89. 　　        pytest.assume(res['code'] == '0',f'期望值是0,实际结果是{res["code"]}')   
    
90. 
    
91. 　　        pytest.assume(res['msg'] == '同学，你好，非常高兴为你服务，有什么需要我帮忙的呢?',                      f'期望值是0,实际结果是{res["msg"]}')   
    
92. 
    
93. 　　
    
94. 
    
95. 　　   # 测试发送数据时消息是空的   
    
96. 
    
97. 　　   def test_message_msgisnull(self):      
    
98. 
    
99. 　　       params={           
    
100. 
     
101. 　　           "msgId": "111",           
     
102. 
     
103. 　　           "type": "normal",           
     
104. 
     
105. 　　           "from": "admin",           
     
106. 
     
107. 　　           "to": f"{self.kfid}",           
     
108. 
     
109. 　　           "msg": ""      
     
110. 
     
111. 　　       }      
     
112. 
     
113. 　　       self.im.send(json.dumps(params))      
     
114. 
     
115. 　　       res = self.im.recv()      
     
116. 
     
117. 　　       res = json.loads(res)# 将其转换成json对象      
     
118. 
     
119. 　　       # 断言系统推送消息时对于消息的判断      
     
120. 
     
121. 　　       pytest.assume(res['code'] == '1',f'期望值是1,实际结果是{res["code"]}')
     
122. 
     
123. 　　       pytest.assume(res['msg'] == '消息内容为空', f'期望值是0,实际结果是{res["msg"]}')</font>

复制代码