WebSocket
seldom > 3.6.0 支持该功能
有些时间我们需要通过WebSocket实现长连接，很高兴的告诉告诉你seldom支持WebSocket测试了。
WebSocket 生命周期
WebSocket 生命周期中包含几个关键的事件，这些事件允许开发人员在连接的不同阶段执行代码。以下是WebSocket API中定义的主要事件：
open: 当WebSocket连接成功建立时触发。这个事件表明客户端与服务器之间的连接已经打开，可以开始数据传输。message: 当客户端接收到服务器发送的消息时触发。这个事件用于处理从服务器接收到的所有消息。error: 当发生错误，导致WebSocket连接关闭之前或连接无法成功建立时触发。这个事件可以用来处理和响应WebSocket过程中出现的任何异常或错误情况。close: 当连接被关闭时触发，无论是客户端还是服务器端主动关闭连接，或是因为某种原因连接被迫关闭。这个事件表明WebSocket连接已经彻底关闭，可以进行清理和后续处理。
seldom测试WebSocket
在seldom中测试WebSocket非常简单。
- 首先，需要一个WebSocket服务。
通过aiohttp实现websocket_server.py。
# websocket_server.py
from aiohttp import web
import aiohttp
async def websocket_handler(request):
ws = web.WebSocketResponse()
await ws.prepare(request)
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
print("message", msg.data)
if msg.data == 'close':
await ws.close()
else:
await ws.send_str(f"Message text was: {msg.data}")
elif msg.type == aiohttp.WSMsgType.ERROR:
print('ws connection closed with exception %s' %
ws.exception())
print('websocket connection closed')
return ws
app = web.Application()
app.router.add_get('/ws', websocket_handler)
web.run_app(app, port=8765)
- 然后，通过seldom编写WebSocket测试用例。
import seldom
from seldom.logging import log
from seldom.websocket_client import WebSocketClient
class WebSocketTest(seldom.TestCase):
def start(self):
# 创建WebSocket客户端线程
self.client = WebSocketClient("ws://0.0.0.0:8765/ws")
self.client.start()
# 等待客户端连接建立
self.sleep(1) # 这里假设服务器可以在1秒内响应连接
def tearDown(self):
# 发送关闭消息
self.client.send_message("close")
# 停止WebSocket客户端线程
self.client.stop()
self.client.join()
def test_send_and_receive_message(self):
# 发送消息
self.client.send_message("Hello, WebSocket!")
self.client.join(1) # 等待接收消息
self.client.send_message("How are you?")
self.client.join(1) # 等待接收消息
# 验证是否收到消息
log.info(self.client.received_messages)
self.assertEqual(len(self.client.received_messages), 2)
self.assertIn("Hello, WebSocket!", self.client.received_messages[0])
self.assertIn("How are you?", self.client.received_messages[1])
if __name__ == '__main__':
seldom.main(debug=True)
- 运行日志
> python test_websocket.py
test_send_and_receive_message (test_websocket.WebSocketTest.test_send_and_receive_message) ...
2024-04-05 23:36:33 | INFO | case.py | 💤️ sleep: 1s.
2024-04-05 23:36:33 | INFO | websocket_client.py | WebSocket connection opened.
2024-04-05 23:36:36 | INFO | test_websocket.py | ['Message text was: Hello, WebSocket!', 'Message text was: How are you?']
ok
----------------------------------------------------------------------
Ran 1 test in 3.006s
OK
2024-04-05 23:36:36 | SUCCESS | runner.py | A run the test in debug mode without generating HTML report!