Pytest：测试WebSocket连接

Question

我有这样一段代码

class RTMClient: 
    ... 
    #not important code 
    ...  
    async def connect(self, queue: asyncio.Queue): 
     """ 
     Connect to the websocket stream and iterate over the messages 
     dumping them in the Queue. 
     """ 
     ws_url=''#url aquisition amended to readability 
     try: 
      self._ws = await websockets.connect(ws_url) 
      while not self.is_closed: 
       msg = await self._ws.recv() 
       if msg is None: 
        break 
       await queue.put(json.loads(msg)) 

     except asyncio.CancelledError: 
      pass 
     finally: 
      self._closed.set() 
      self._ws = None 

我想要写的自动测试它。 我打算做：

1. 猴补丁websockets.connect返回一个模拟连接
2. 使模拟连接预设列表返回假消息
3. 使模拟连接设置is_closed到True
4. 断言websocket连接已关闭
5. 断言所有预定义的消息都在队列中

我的问题：我如何模拟websockets.connection实现步骤1-3？ 我想到一个pytest夹具这样

from websockets import WebSocketClientProtocol() 
@pytest.fixture 
def patch_websockets_connect(monkeypatch): 
    async def mock_ws_connect(*args, **kwargs): 
     mock_connection = WebSocketClientProtocol() 
     mock_connection.is_closed = False 
     return mock_connection 

    monkeypatch.setattr('target_module.websockets.connect', mock_ws_connect) 

但我不看我怎么就能返回消息的预定义列表这种方式，也必须有这样做的更好的方法。

Answer 1

这不是一个完整的答案，但它可能会帮助你。

我碰到了测试rabbitmq消息发射的模拟问题。并且看起来像这里最常见和可靠的方法是创建Connection，创建Emmiter和Consumer，将Consumer连接到假的Channel，并且在测试中手动将消息发送到该通道。

所以你创建所有的对象，只是嘲笑他们的回应。这里是非常相似的例子（与套接字，而不是与websockets，但可能仍然有用）：mocking a socket connection in Python