Python异步方法单元测试：如何验证asyncio.sleep调用？

核心思路 #

要验证当fetchone()返回(0,)时asyncio.sleep()被调用，需做好三个关键环节：

• 正确Mock数据库相关对象（DBHook、cursor），确保返回预期的(0,)结果
• 正确驱动异步生成器执行到触发await asyncio.sleep()的分支
• 断言asyncio.sleep()的调用行为

完整测试用例示例 #

使用pytest+unittest.mock实现，代码如下：

import asyncio
from unittest.mock import Mock, patch, AsyncMock
import pytest
from your_module import YourClass, TriggerEvent  # 替换为实际模块和类名

@pytest.mark.asyncio
async def test_run_calls_sleep_when_fetchone_returns_zero():
    # 初始化测试对象
    test_obj = YourClass(db_conn="test_conn", sql="SELECT count(*) FROM table", sleep_interval=5)
    
    # Mock cursor对象，指定fetchone返回(0,)
    mock_cursor = AsyncMock()
    mock_cursor.fetchone.return_value = (0,)
    
    # Mock异步上下文管理器：让get_cursor返回的对象支持async with
    mock_get_cursor = AsyncMock()
    mock_get_cursor.__aenter__.return_value = mock_cursor
    
    # Mock DBHook类及其实例
    with patch("your_module.DBHook") as mock_db_hook_cls:
        mock_db_hook_instance = Mock()
        mock_db_hook_instance.get_cursor.return_value = mock_get_cursor
        mock_db_hook_cls.return_value = mock_db_hook_instance
        
        # Mock asyncio.sleep，用异常终止无限循环
        with patch("asyncio.sleep") as mock_sleep:
            mock_sleep.side_effect = asyncio.CancelledError
            
            # 驱动异步生成器执行一次迭代
            async_gen = test_obj.run()
            try:
                await async_gen.__anext__()
            except asyncio.CancelledError:
                pass
            
            # 断言验证
            mock_cursor.execute.assert_called_once_with(test_obj.sql)
            mock_sleep.assert_called_once_with(test_obj.sleep_interval)

关键细节说明 #

• Mock异步上下文：get_cursor()返回的是异步上下文，必须用AsyncMock模拟__aenter__方法，才能让async with正确获取到mock的cursor。
• 终止无限循环：由于run()是无限while True，给asyncio.sleep设置side_effect为CancelledError，让第一次执行到await asyncio.sleep()时抛出异常，避免测试无限挂起。
• 执行异步生成器：异步生成器不是可直接await的对象，需调用__anext__()并await来触发一次迭代，这是正确的执行方式。
• Mock路径正确性：patch的路径必须是run方法实际导入DBHook的路径，比如YourClass所在模块直接导入DBHook，就patchyour_module.DBHook，路径错误会导致Mock失效。

常见问题排查 #

• 若出现rows[0]报错，检查mock_cursor.fetchone.return_value是否正确设置为(0,)，确保返回的是可索引的元组。
• 若asyncio.sleep未被触发，确认fetchone返回的是(0,)，没有进入yield分支，同时检查生成器是否执行到了await asyncio.sleep的代码段。

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