修补 python-socket.io 中的服务器端函数并测试客户端调用

Question

我正在尝试测试 socket.io 调用。 我有一个 socket.io 服务器作为应用程序安装到 FastAPI 中，该服务器已启动并正在运行，并且通信正在工作。

每当我调用

foo()

时，都会在测试中直接调用

bar()

，它就会被修补。

测试连接时出现问题，socket.io 服务器调用

foo()

，它调用未修补的

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")

在这里，我正在构建一个固定装置，它提供了一个 socket.io 客户端来测试 socket.io 服务器。因此，socket.io 服务器是生产代码，客户端仅用于测试。

# 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

我在这里修补

bar()

：

# ./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)

socketio-server 通过基于类的方法实例化命名空间。

# 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.

我希望在

foo()

方法中调用

on_connect()

的结果是 patched_string”，而不是“socketio_server”。

客户端调用时如何在服务器端打补丁？

Answer 1

在与测试客户端相同的进程中运行单独的服务器进行测试可以解决这个问题：

# 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

现在服务器正在获取补丁，客户端连接到另一台服务器，现在端口为 8669。 这是工作测试：

# 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()

文档展示了如何添加命名空间：

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'))