使用Twisted、pytest_twisted与asyncio reactor时遇RuntimeError错误求助

已启用pytest_twisted插件并添加--reactor asyncio参数，基于asyncio反应器的纯Twisted测试运行正常，但以下混合asyncio与Twisted的测试代码触发RuntimeError: await wasn't used with future错误：

async def _sleep():
    await asyncio.sleep(1.0)


@defer.inlineCallbacks
def test_sleep():
    yield defer.ensureDeferred(_sleep())

完整堆栈跟踪：

Traceback (most recent call last):
  File "test/test_simple.py", line 23, in test_sleep
    yield defer.ensureDeferred(_sleep())
  File "/usr/local/lib/python3.10/site-packages/twisted/internet/defer.py", line 1697, in _inlineCallbacks
    result = context.run(gen.send, result)
  File "test/test_cli.py", line 18, in _sleep
    await asyncio.sleep(1.0)
  File "/usr/local/lib/python3.10/asyncio/tasks.py", line 605, in sleep
    return await future
RuntimeError: await wasn't used with future

相关版本：

• Twisted: 22.10.0
• Python: 3.10.11
• pytest: 7.3.1
• pytest_twisted: 1.14.0

问题出在你直接在Twisted的inlineCallbacks函数里用defer.ensureDeferred包装asyncio协程，但没有让asyncio的Future在正确的事件循环上下文里运行。可以通过两种方式修复：

1. 将asyncio协程转为Twisted Deferred并适配上下文
   改用defer.Deferred.fromFuture包装asyncio协程生成的Future，让Twisted能正确对接asyncio的循环：

   import asyncio
   from twisted.internet import defer
   
   async def _sleep():
       await asyncio.sleep(1.0)
   
   @defer.inlineCallbacks
   def test_sleep():
       future = asyncio.ensure_future(_sleep())
       yield defer.Deferred.fromFuture(future)
   

2. 直接用async def定义测试函数
   pytest-twisted原生支持async测试函数，不需要依赖inlineCallbacks，这样可以自然混合asyncio和Twisted操作：

   import asyncio
   from twisted.internet import defer
   
   async def _sleep():
       await asyncio.sleep(1.0)
   
   async def test_sleep():
       await _sleep()
       # 也可以直接await Twisted的Deferred
       d = defer.Deferred()
       d.callback("test_done")
       assert await d == "test_done"
   

核心原因是：defer.ensureDeferred更适合处理Twisted自身的异步生成器，而非asyncio协程。直接用它包装asyncio协程时，无法保证asyncio的Future在绑定的事件循环中被正确调度，因此触发错误。改用Deferred.fromFuture或直接用async测试函数，能让协程在正确的循环环境中执行。

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