如何Mock await asyncio.Future()？解决pytest中asyncio主函数阻塞问题

你的测试挂起是因为await asyncio.Future()会一直阻塞——原生asyncio.Future()创建的是未完成的任务，没有外部触发的话永远不会结束。同时你的mock设置存在两处错误：

1. 你patch了内部模块_asyncio.Future，但代码里实际调用的是asyncio.Future，路径不匹配；
2. mock_future.return_value = ''返回的不是可等待对象，mock_serve.return_value=''也不符合websockets.serve返回的异步上下文管理器类型，这两个mock都无法正常工作。

方案1：正确Mock Future，让它立即完成 #

通过创建一个已完成的Future替换原调用，避免阻塞：

import asyncio
from unittest.mock import patch, AsyncMock
import pytest

async def main(host, port):
    log.debug('starting websockets server...')
    async with websockets.serve(myserver, host, port): 
        await asyncio.Future()  # run forever

@pytest.mark.asyncio
async def test_main():
    # Mock websockets.serve，返回合法的异步上下文管理器
    with patch('websockets.serve', new_callable=AsyncMock) as mock_serve:
        # Mock asyncio.Future，返回已完成的Future
        with patch('asyncio.Future') as mock_future:
            completed_future = asyncio.Future()
            completed_future.set_result(None)  # 设置结果让Future立即完成
            mock_future.return_value = completed_future
            
            await main('', '')
            
            # 验证调用是否符合预期
            mock_serve.assert_called_once_with(myserver, '', '')
            mock_future.assert_called_once()

方案2：主动取消Future终止阻塞 #

通过获取mock的Future对象，主动触发取消来终止await：

import asyncio
from unittest.mock import patch, AsyncMock
import pytest

async def main(host, port):
    log.debug('starting websockets server...')
    async with websockets.serve(myserver, host, port): 
        await asyncio.Future()  # run forever

@pytest.mark.asyncio
async def test_main():
    test_future = asyncio.Future()
    
    with patch('websockets.serve', new_callable=AsyncMock):
        with patch('asyncio.Future', return_value=test_future):
            # 用任务包装main，避免直接await阻塞
            task = asyncio.create_task(main('', ''))
            test_future.cancel()  # 取消Future，触发CancelledError
            
            try:
                await task
            except asyncio.CancelledError:
                pass  # 捕获取消异常，不影响测试结果
            
            # 验证调用
            asyncio.Future.assert_called_once()

方案3：临时超时机制（不推荐） #

通过pytest的超时装饰器强制终止阻塞的测试，适合快速验证场景：

import pytest
from unittest.mock import patch

async def main(host, port):
    log.debug('starting websockets server...')
    async with websockets.serve(myserver, host, port): 
        await asyncio.Future()  # run forever

@pytest.mark.asyncio(timeout=1)  # 1秒后自动终止测试
async def test_main():
    with patch('websockets.serve', new_callable=AsyncMock):
        with patch('asyncio.Future'):
            await main('', '')

• 始终patch代码中实际调用的模块路径：比如代码用asyncio.Future()，就patch'asyncio.Future'，不要直接patch内部私有模块_asyncio.Future；
• websockets.serve返回的是异步上下文管理器，必须用AsyncMock来mock，不能用普通字符串或非异步对象替代。

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