测试使用loop.call_soon_threadsafe的Asyncio函数时断言失败求助

我维护一个继承自asyncio.Protocol的类，负责通过套接字向硬件发送命令。因为这个类属于wxPython GUI模块，必须在非异步线程中调用方法——直接调用self._transport.write在Linux上正常，但在Windows上会崩溃，所以我改用了call_soon_threadsafe来封装调用：

def send_command(self, cmd):
    self.loop.call_soon_threadsafe(
        functools.partial(
            self._transport.write, str(cmd).encode() + b"\n"
        )
    )

为了测试这个方法，我写了如下pytest用例：

@pytest.mark.asyncio
async def test_send_command(self):
    self.sut._transport = Mock()
    self.sut.send_command("ook eek")
    assert self.sut._transport.write.called is True

但运行测试时总是得到断言错误：self.sut._transport.write从未被调用。如果我在send_command里直接调用self._transport.write（不用call_soon_threadsafe），测试就能通过，但这样代码在Windows上会崩溃。我到底忽略了什么？这应该不是很罕见的场景吧？

这个问题的核心在于**call_soon_threadsafe的工作机制**：它只是把任务提交给asyncio事件循环的待执行队列，并不会立刻执行。你的测试在调用send_command后马上断言，此时事件循环还没来得及处理这个提交的任务，所以Mock的write方法自然没有被调用。

解决步骤 #

1. 让事件循环处理待执行任务

在断言前，我们需要给事件循环一个机会去处理call_soon_threadsafe提交的任务。在pytest的async测试中，最简单的方式是await asyncio.sleep(0)——这会让出当前协程的控制权，让事件循环处理队列里的所有待完成任务。

修改后的测试用例：

import asyncio
import pytest

@pytest.mark.asyncio
async def test_send_command(self):
    self.sut._transport = Mock()
    self.sut.send_command("ook eek")
    # 让事件循环处理call_soon_threadsafe提交的任务
    await asyncio.sleep(0)
    assert self.sut._transport.write.called is True

2. 确保SUT使用测试的事件循环

另外要注意：如果你的SUT的loop属性指向的是另一个线程的事件循环（比如wxGUI的主线程循环），那测试用的事件循环可能无法处理提交的任务。这种情况下，需要在测试的setup阶段，把SUT的loop设置为测试当前的事件循环：

def setup_method(self):
    self.sut = YourProtocolSubclass()
    # 让SUT使用测试的事件循环
    self.sut.loop = asyncio.get_event_loop()

为什么直接调用write测试能通过？ #

当你直接调用self._transport.write时，这是一个同步调用——方法会立刻执行，所以断言的时候write已经被调用过了。而call_soon_threadsafe是异步提交任务，必须等事件循环调度执行后，write才会被调用，所以测试中必须等待事件循环处理完任务再断言。

内容的提问来源于stack exchange，提问作者Sardathrion - against SE abuse