Python实现事件循环运行至所有任务阻塞的测试方案（兼容3.9）

需求可行性 #

完全可行。你可以通过控制事件循环的执行节奏，在所有协程进入阻塞状态时暂停循环，此时进行断言或触发外部事件，再恢复循环继续执行，全程无需修改协程内部逻辑。

是否需要自定义事件循环 #

不需要依赖自定义事件循环，利用Python标准库asyncio的原生API就能实现所需的run_until_blocked_or_complete函数。核心思路是反复运行事件循环的短周期迭代，直到检测到所有未完成任务都处于阻塞状态。

Python版本支持 #

Python 3.9完全支持该方案，甚至Python 3.7及以上版本都能正常运行（asyncio.all_tasks()在3.7版本正式成为公开API，替代了之前的asyncio.Task.all_tasks()）。无需升级版本。

实现示例 #

以下是你期望的run_until_blocked_or_complete函数的具体实现，以及完整测试代码：

import asyncio
from asyncio import Event

async def wait_on_event(event: Event):
    print("Waiting on event")
    await event.wait()
    print("Done waiting on event")

def run_until_blocked_or_complete(loop):
    while True:
        # 获取当前所有未完成的任务
        pending_tasks = [task for task in asyncio.all_tasks(loop) if not task.done()]
        if not pending_tasks:
            # 没有待处理任务，直接退出
            break
        
        # 等待0秒，尝试执行所有就绪的任务
        done, _ = loop.run_until_complete(
            asyncio.wait(pending_tasks, timeout=0, return_when=asyncio.FIRST_COMPLETED)
        )
        
        if not done:
            # 没有任务完成，说明所有任务都已阻塞，退出循环
            break

def test_deterministic():
    loop = asyncio.get_event_loop()
    event = Event()
    task = loop.create_task(wait_on_event(event))
    
    run_until_blocked_or_complete(loop)
    # 此时会打印"Waiting on event"，协程阻塞在event.wait()
    
    # 这里可以添加断言，比如检查任务状态是否为pending
    assert not task.done()
    
    event.set()
    run_until_blocked_or_complete(loop)
    # 此时会打印"Done waiting on event"，协程执行完成
    
    assert task.done()

if __name__ == "__main__":
    test_deterministic()

关键逻辑说明 #

• asyncio.wait(..., timeout=0)：让事件循环执行所有当前就绪的任务，但不等待任何阻塞操作完成，立即返回。
• 循环检测：如果某次调用后没有任务完成，说明所有任务都处于等待外部事件的阻塞状态，此时暂停循环，允许你进行测试操作。

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