Pytest: testing a websocket connection

Question

I have this piece of code

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

I want to write an automated test for it. What I intend to do:

1. Monkeypatch websockets.connect to return a mock connection
2. Make the mock connection return mock messages from a predefined list
3. Make the mock connection set is_closed to True
4. Assert that the websocket connection was closed
5. Assert that all predefined messages are in the Queue

My question: how do I mock the websockets.connection to achieve steps 1-3? I am thinking of a pytest fixture like this

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)

But I don't see how I will be able to return a predefined list of messages this way, and also there must be a better way of doing this.

Answer 1

It is not a full answer but maybe it will help you.

I hit the simlar problem with testing rabbitmq messages emission. And looks like the most common and robust approach here is to create Connection, create Emmiter and Consumer, connect Consumer to the fake Channel and that manually send messages to that channel in the test.

So you creating all objects and just mocking their responses. Here is very similar example (with sockets and not with websockets, but maybe still usefull for you): mocking a socket connection in Python