У меня есть этот кусок кодаPytest: тестирование соединения WebSocket
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
вернуть макет Connection - сделать макет соединения вернуть фиктивные сообщения из заданного списка
- сделать макет установки соединения
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)
Но я не понимаю, как я смогу вернуться предопределенным список сообщений таким образом, а также должен быть лучшим способом сделать это.