Pytest：测试 websocket 连接答案

【问题标题】：Pytest: testing a websocket connectionPytest：测试 websocket 连接
【发布时间】：2017-04-14 13:09:38
【问题描述】：

我有这段代码

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

我想为它编写一个自动化测试。我打算做什么：

Monkeypatch websockets.connect 返回模拟连接
使模拟连接从预定义列表中返回模拟消息
将模拟连接设置为is_closed 到True
断言 websocket 连接已关闭
断言所有预定义的消息都在队列中

我的问题：如何模拟 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)

但我不知道如何以这种方式返回预定义的消息列表，而且必须有更好的方法来做到这一点。

【问题讨论】：

标签： python unit-testing websocket automated-tests python-asyncio

【解决方案1】：

这不是一个完整的答案，但也许会对你有所帮助。

我在测试 rabbitmq 消息发射时遇到了类似的问题。看起来这里最常见和最强大的方法是创建Connection，创建Emmiter 和Consumer，将Consumer 连接到假Channel，然后在测试中手动向该通道发送消息。

因此，您创建所有对象并只是模拟它们的响应。这是一个非常相似的示例（使用套接字而不是 websockets，但可能对您仍然有用）：mocking a socket connection in Python

【讨论】：