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