Patching server-side function in python-socket.io and testing client-side call

Question

I'm trying to test a socket.io call. I have a socket.io server mounted as app into FastAPI, which is up and running and communication is working.

Whenever I call foo(), that calls bar() directly in the test it is patched.

Trouble hits, when testing the connection and the socket.io server calls foo()which calls the un-patched bar().

# some_path/tests/test_socketio.py
import pytest

@pytest.mark.anyio
async def test_calling_server(socketio_client):
    foo_response = foo("input_string")
    print(foo_response) # results in "patched_string" !WORKS!
    
    async for client in socket_client():
        await client.emit("message", "Hello world", namespace="/my_events")

Here I'm building a fixture, that provides a socket.io client to test the socket.io server. So the socket.io server is production code, the client is testing only.

# some_path/tests/conftest.py

import pytest
import socketio


@pytest.fixture
async def socketio_client(mock_bar):
   async def _socketio_client():
       client = socketio.AsyncClient()

       @client.event
       def connect():
           pass

       await client.connect(
           "http://localhost:80",
           namespaces=["/my_events"]
       )
       yield client
       await client.disconnect()

   return _socketio_client

I'm patching bar() here:

# ./conftest.py

import pytest
from unittest.mock import patch

@pytest.fixture(scope="function")
async def mock_bar():
    with patch("some_other_path.code.bar") as mock:
        mock.return_value = {"mocked_string"}
        yield mock

# some_other_path/code.py
async def bar(input_bar):
   return(input_bar)

async def foo(input_foo):
   response_bar = bar(input_foo)
   print(response_bar)

The socketio-server instantiates a namespace through the class-based approach.

# socketio_server/server_code.py

import socketio
from some_other_path.code import foo

socketio_server = socketio.AsyncServer(async_mode="asgi")

class MyEvents(socketio.AsyncNamespace):
   def __init__(self):
      super().__init__()

   async def on_connect(self, sid)
      foo("socketio_server_string") # results in "socketio_server_string" !NOT PATCHED!

socketio_server.register_namespace(MyEvents("/my_events"))
socketio_app = ASGIApp(socketio_server)

# Here I'm mounting the socketio_app into FastAPI. Server.io is running and communication is working.

I expect the result from the call to foo() in the on_connect() method to be patched_string", not "socketio_server".

How can I apply the patch on the server side when calling from client side?

Answer 1

Running a separate server for the test within the same process as the testing client solves it:

# conftest.py

@pytest.fixture(scope="function")
async def mock_bar():
    with patch("some_other_path.code.bar") as mock:
        mock.return_value = {"mocked_string"}
        yield mock

@pytest.fixture()
async def socketio_server(mock_bar):
    """Provide a socket.io server."""

    sio = socketio.AsyncServer(async_mode="asgi")
    app = socketio.ASGIApp(sio)

    config = uvicorn.Config(app, host="127.0.0.1", port=8669)
    server = uvicorn.Server(config)

    asyncio.create_task(server.serve())
    await asyncio.sleep(1)
    yield sio
    await server.shutdown()

@pytest.fixture
async def socketio_client():
   async def _socketio_client():
       client = socketio.AsyncClient()

       await client.connect(
           "http://127.0.0.1:8669"
       )
       yield client
       await client.disconnect()

   return _socketio_client

So now the server is getting the patch and the client connects to a different server, now port 8669. And here is the working test:

# test_socketio.py

@pytest.mark.anyio
@pytest.mark.parametrize(
    "mock_bar",
    [bar_return_0, bar_return_1],
    indirect=True,
)
async def test_calling_server(socketio_server, socketio_client):
    """Test the demo socket.io message event."""

    sio = socketio_server

    sio.register_namespace(...)

    async for client in socketio_client():
        await client.emit("demo_message", "Hello World!")

        response = ""

        @client.event()
        async def demo_message(data):

            nonlocal response
            response = data

        # Wait for the response to be set
        await client.sleep(1)

        assert response == "Demo message received from client: Hello World!"

        await client.disconnect()

The documentation shows how to add the Namespaces:

class MyCustomNamespace(socketio.AsyncNamespace):
    def on_connect(self, sid, environ):
        pass

    def on_disconnect(self, sid):
        pass

    async def on_my_event(self, sid, data):
        await self.emit('my_response', data)

sio.register_namespace(MyCustomNamespace('/test'))