如何触发Socket.IO事件以完成AsyncClient的connect方法单元测试

解决方案 #

问题1：如何编写对应单元测试 #

首先先修正源码的笔误：原AsyncClient类存在变量名不一致问题，connect事件回调中赋值的self.connection_established和connect方法中判断的self.conn_established命名不统一，先统一改为self.conn_established，否则逻辑永远无法退出循环。

单元测试可以按照如下思路实现：

• 继续用AsyncMock mock io.connect方法
• 将side_effect改为异步函数，在其中设置io.connected = True，再手动触发connect事件
• mock asyncio.sleep避免测试时产生真实的1秒等待
• 添加必要的断言验证逻辑符合预期

完整测试代码如下：

import pytest
import asyncio
from unittest import mock
import socketio

# 你自己定义的endpoint变量要提前声明，这里假设值为"ws://localhost:5000"
endpoint = "ws://localhost:5000"

# 修正后的AsyncClient源码
class AsyncClient:
    def __init__(self):
        self.conn_established = False
        self.io = socketio.AsyncClient()

        @self.io.event
        def connect():
            # 修正变量名和上层判断一致
            self.conn_established = True

    async def connect(self):
        while not self.io.connected:
            await self.io.connect(endpoint)
        
        while not self.conn_established:
            await asyncio.sleep(1) 

class AsyncMock(mock.MagicMock):
    async def __call__(self, *args, **kwargs):
        return super().__call__(*args, **kwargs)

@pytest.fixture
def async_client():
    yield AsyncClient()

@pytest.mark.asyncio
async def test_connect(async_client):
    # mock asyncio.sleep避免真实等待
    mock_sleep = AsyncMock(spec=asyncio.sleep)
    with mock.patch("asyncio.sleep", new=mock_sleep):
        async def mock_successful_conn(*args, **kwargs):
            async_client.io.connected = True
            # 异步调用trigger_event触发connect事件
            await async_client.io.trigger_event("connect")

        # mock io.connect方法
        async_client.io.connect = AsyncMock(
            spec=async_client.io.connect, 
            side_effect=mock_successful_conn
        )

        # 执行待测方法
        await async_client.connect()

        # 断言验证
        async_client.io.connect.assert_awaited_once_with(endpoint)
        assert async_client.conn_established is True
        # 验证事件触发正常，没有进入等待循环
        mock_sleep.assert_not_awaited()

问题2：测试场景下是否可以手动触发Socket.IO的事件 #

可以，python-socketio的AsyncClient原生提供异步方法trigger_event(event_name: str, *args)，可以直接调用触发指定事件的所有已注册回调，注意该方法是异步的，必须使用await调用才能正常执行。

如果不想依赖socketio的内置方法，也可以直接手动修改conn_established属性值，但是用trigger_event更贴近真实运行逻辑，测试可信度更高。

内容的提问来源于stack exchange，提问作者dekauliya