Writing tests

[aaugustin]

The documentation should describe best practices for writing tests for websockets apps.

For testing a server, it's possible to run the server, instantiate a client, and use the client to interact with the server. This only allows writing end-to-end tests.

For testing a server or a client, it must be possible to mock the send() and recv() methods, although I'm not too sure how to mock coroutines.

If you've found a nice approach to this, I'm interested :-)

[cjerdonek]

Something I consider a best practice for async tests is to, where possible, write tests that make no reference to the event loop. This makes for elegant, concise tests. This can be done with vanilla TestCase methods by defining tests with signature async def test_foo(self): and adding a decorator that runs the coroutine in its own loop.

I think something like this should be done for many of websockets' own tests.

[cjerdonek]

Also, the test refactoring I did previously is progress towards this goal (by removing much of the boilerplate), but more needs to be done.

[bradwood]

Any further ideas here guys? I've managed to mock aiohttp okay but having trouble with ws.recv()

Here's my working test with aiohttp:

import pytest
import asyncio

from asynctest import CoroutineMock, MagicMock

from pyskyq import EPG

from .asynccontextmanagermock import AsyncContextManagerMock
from .mock_constants import REMOTE_TCP_MOCK, SERVICE_MOCK

def test_EPG(mocker):

    a = mocker.patch('aiohttp.ClientSession.get', new_callable=AsyncContextManagerMock)
    a.return_value.__aenter__.return_value.json = CoroutineMock(side_effect=SERVICE_MOCK)

    asyncio.set_event_loop(asyncio.new_event_loop())

    epg = EPG('host')

    assert isinstance(epg, EPG)
    assert len(epg._channels) == 2
    assert epg.get_channel(2002).c == "101"
    assert epg.get_channel(2002).t == "BBC One Lon"
    assert epg.get_channel(2002).name == "BBC One Lon"
    assert epg.get_channel('2002').c == "101"
    assert epg.get_channel('2002').t == "BBC One Lon"
    assert epg.get_channel('2002').name == "BBC One Lon"

    with pytest.raises(AttributeError, match='Channel not found. sid = 1234567.'):
        epg.get_channel(1234567)

It also needs this

from asynctest import MagicMock

class AsyncContextManagerMock(MagicMock):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        for key in ('aenter_return', 'aexit_return'):
            setattr(self, key,  kwargs[key] if key in kwargs else MagicMock())

    async def __aenter__(self):
        return self.aenter_return

    async def __aexit__(self, *args):
        return self.aexit_return

But, this is my failing attempt to mock recv().... Any suggestions much appreciated.

import asyncio

import pytest
import websockets
from asynctest import CoroutineMock, MagicMock

from pyskyq.status import Status

from .asynccontextmanagermock import AsyncContextManagerMock
from .mock_constants import WS_STATUS_MOCK

def test_status(mocker):
    a = mocker.patch('websockets.client.WebSocketClientProtocol', new_callable=AsyncContextManagerMock)
    a.return_value.__aenter__.return_value.recv = CoroutineMock(side_effect=WS_STATUS_MOCK)

    stat = Status('some_host')

    stat.create_event_listener()

    assert stat.standby is True

[bradwood]

He @aaugustin --any further thoughts on this? For the record, I have now managed to make a naive implementation of a ws.recv() mock but it is not great.... It can't be make to cancel, ping/pong or timeout easily...

It would be good if you could consider perhaps adding some kind of mocking class or pytest fixture that can deal with this.

FWIW, Here is my (kinda) working mock of ws.recv():

def serve_ws_json_with_detail():
    time.sleep(0.2)
    return WS_STATUS_MOCK

def test_status(mocker):
    a = mocker.patch('websockets.connect', new_callable=AsyncContextManagerMock)
    a.return_value.__aenter__.return_value.recv = \
        CoroutineMock(side_effect=serve_ws_json_with_detail)

    stat = Status('some_host')
    stat.create_event_listener()

    time.sleep(0.5) # allow time for awaiting, etc.
    assert stat.standby is True

    stat.shudown_event_listener()

I'm also beginning to think my own code is a bit ropey here, as if I can dependency-inject the ws object into my Status object or the create_event_listener() call then mocking it out might be easier, although I suspect this is an unrelated issue.

[RemiCardona]

Here are some snippets that could help (note, I'm not yet using any async syntax, 100% plain yield from for now, but it all works on 3.4 and 3.6):

class WebsocketProtocol(unittest.mock.Mock):
    def __await__(self):
        return self
        yield None # pylint: disable=unreachable

This weird piece of code makes both python 3.4 and 3.6 happy : 3.4 doesn't really care (it just wants an __aiter__ method, what Mock provides seem to work), 3.6 wants an __await__ generator method, so let's give it that.

protocol = WebsocketProtocol()
with asynctest.patch('websockets.connect', return_value=protocol) as connect_mock:
    # send
    protocol.send = asynctest.CoroutineMock(return_value=True)

    # recv
    protocol.recv = asynctest.CoroutineMock(return_value=b'data')

    # ping (here comes the hard part...)
    inner = asyncio.Future()
    inner.set_exception(asyncio.TimeoutError())  # simulate a failing wait_for(pong)
    waiter = asyncio.Future()
    waiter.set_result(inner)
    protocol.ping = unittest.mock.Mock(return_value=waiter)

    # test code
    ws = websockets.connect(...)
    self.assertEqual(connect_mock.call_count, 1)

Yes, that's unittest, not asynctest, to mock ping. The latter transforms the Futures into CoroutineMocks and this doesn't work.

HTH

[bradwood]

thanks @RemiCardona -- I'll try this out...

[Marc-Roig]

In case some one wants to use the websockets.connect() in a context manager you just have to

class WebsocketProtocol(unittest.mock.Mock):

    def __await__(self):
        return self
        yield None # pylint: disable=unreachable

    async def __aenter__(self):
        return await self

    async def __aexit__(self, exc_type, exc_value, traceback):
        return

[aaugustin]

I'm still unsure about what would be useful to document here.

I'm skeptical of mocking the entire protocol class, perhaps because I'm generally skeptical of large-scale mocking.

I feel like the best practice is (probably) an I/O sandwich: implement as much logic as you can independently from websockets (or other components doing I/O) so it's easy to test; and then add a few integration tests if you need them.

It isn't hard to run a websockets server in tests. The usual tricks (e.g. bind to port 0) work.

[dangell7]

@aaugustin Can you elaborate on the It isn't hard to run a websockets server in tests. The usual tricks (e.g. bind to port 0) work.?

I'm struggling with this and everyone seems to be discussing the client side. I just want to create a simple server before the tests run.

https://stackoverflow.com/questions/70805272/python-test-mock-websocket-server-for-testing

[aaugustin]

Let's assume you have a way to write tests as coroutines (e.g. https://github.com/aaugustin/websockets/blob/668f320e0547d80afe6529528e1ecc6088955cdc/tests/legacy/utils.py#L10 or equivalent functionality provided by third party libraries).

Then you can write a test as follows:

class TestStuff(...):
    async def test_something(self):
        async with websockets.serve(your_server, "127.0.0.1", 0) as server:
            host, port = server.sockets[0].getsockname()
            ws_uri = f"ws://{host}:{port}/"
            ...  # test whatever you want

Of course you will want to factor this out in a convenient way if you write for than a couple tests.

[summy00]

Let's assume you have a way to write tests as coroutines (e.g.

https://github.com/aaugustin/websockets/blob/668f320e0547d80afe6529528e1ecc6088955cdc/tests/legacy/utils.py#L10

or equivalent functionality provided by third party libraries). Then you can write a test as follows:

class TestStuff(...):
    async def test_something(self):
        with websockets.serve(your_server, "127.0.0.1", 0) as server:
            host, port = server.sockets[0].getsockname()
            ws_uri = f"ws://{host}:{port}/"
            ...  # test whatever you want

Of course you will want to factor this out in a convenient way if you write for than a couple tests.

async with websockets.serve(your_server, "127.0.0.1", 0) as server: you miss the async keyword

[aaugustin]

Fixed. Thank you.

[summy00]

just find a simple way to test client:

import asyncio
import pytest
import websockets
import json
from exception import TracebackException

pytestmark = pytest.mark.asyncio

async def client(websocket):  
    return await websocket.send('123')  # WebSocketClientProtocol.send

from unittest.mock import patch, AsyncMock, MagicMock
async def test_client_simple():
    async_mock = AsyncMock()
    async_mock.send.return_value=''
    await client(async_mock)
    assert async_mock.send.await_args.args[0] == '123'
    return 

[Pithikos]

I use the below fixture to make my life simpler

from unittest.mock import patch, MagicMock

@pytest.fixture
async def mocked_websockets():
    """
    Properly patch websockets.connect to work with any URL
    """
    async def mocked_connect(url):
        async def mocked_recv():
            await asyncio.sleep(0.1)
            return "Hello!!"
        async def mocked_send(message):
            pass
        async def mocked_close():
            pass
        mocked_websocket = MagicMock()
        mocked_websocket.recv = mocked_recv
        mocked_websocket.send = mocked_send
        mocked_websocket.close = mocked_close
        return mocked_websocket

    with patch("websockets.connect", new=mocked_connect):
        yield

And you can edit the mocked functions as needed. Then in testing you just include the fixture:

def test_something(mocked_websockets):
  ..