Исправление серверной функции в python-socket.io и тестирование вызова на стороне клиента ⇐ Python

[Anonymous]

Я пытаюсь протестировать вызов 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-сервер создает экземпляр пространства имен с помощью подхода на основе классов.

Код: Выделить всё

# 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".
Как применить патч на стороне сервера при звонке со стороны клиента?

Подробнее здесь: https://stackoverflow.com/questions/791 ... -side-call