在pytest Fixture中运行asyncio服务器：连接拒绝错误及作用域不匹配问题的解决求助

你遇到的核心问题是错误地将异步函数start提交到了线程池——ThreadPoolExecutor是用来运行同步代码的，而你的ServerEndpoint.start是异步方法，直接丢进去只会得到一个未执行的协程对象，服务器根本没有真正启动，自然会出现连接被拒绝的错误。

另外，结合你调整事件循环作用域的尝试，我们需要配合正确的异步服务器启动逻辑一起调整，下面是具体的修复步骤：

1. 修复服务器启动逻辑：移除线程池，用asyncio并发运行服务器 #

既然你的服务器是基于asyncio的，完全不需要线程池来启动多个服务器——asyncio本身就支持在同一个事件循环中并发运行多个任务。我们可以用asyncio.gather和任务对象来管理服务器的生命周期：

@pytest.fixture(autouse=True, scope="module")
async def start_endpoints(
    event_loop,
    endpoint1: ServerEndpoint,
    endpoint2: ServerEndpoint
):
    # 创建并启动两个服务器的异步任务
    task1 = event_loop.create_task(endpoint1.start())
    task2 = event_loop.create_task(endpoint2.start())
    
    # 测试阶段：yield后测试用例开始执行
    yield
    
    # 测试结束后，取消服务器任务并清理
    task1.cancel()
    task2.cancel()
    # 捕获取消错误，避免测试报错
    try:
        await asyncio.gather(task1, task2, return_exceptions=True)
    except asyncio.CancelledError:
        pass

2. 确保自定义module级事件循环正确工作 #

你的自定义event_loop fixture需要确保创建独立的事件循环（尤其是Windows环境下，默认的事件循环可能存在兼容性问题），可以调整为：

@pytest.fixture(scope="module")
def event_loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    yield loop
    loop.close()

3. 可选优化：确保服务器完全启动后再执行测试 #

为了避免测试在服务器还没就绪时就尝试连接，可以在ServerEndpoint中添加一个启动完成的信号，比如用asyncio.Event：

class ServerEndpoint:
    def __init__(self, name, addr, port):
        self.name = name
        self.addr = addr
        self.port = port
        self.started = asyncio.Event()  # 启动完成信号
    
    async def start(self):
        try:
            server = await asyncio.start_server(self.handle_req, self.addr, self.port)
            addr = server.sockets[0].getsockname()
            print(f'{self.name}: serving on {addr}')
            self.started.set()  # 标记服务器已启动就绪
            async with server:
                await server.serve_forever()
        except asyncio.CancelledError:
            print(f'{self.name}: server stopped gracefully')

然后在start_endpoints fixture中等待两个服务器都启动完成：

@pytest.fixture(autouse=True, scope="module")
async def start_endpoints(
    event_loop,
    endpoint1: ServerEndpoint,
    endpoint2: ServerEndpoint
):
    task1 = event_loop.create_task(endpoint1.start())
    task2 = event_loop.create_task(endpoint2.start())
    
    # 等待两个服务器都发出启动完成信号
    await asyncio.gather(endpoint1.started.wait(), endpoint2.started.wait())
    
    yield
    
    task1.cancel()
    task2.cancel()
    await asyncio.gather(task1, task2, return_exceptions=True)

为什么之前的线程池方法不行？ #

当你调用loop.run_in_executor(pool, endpoint1.start)时，endpoint1.start作为异步函数会立即返回一个协程对象，但线程池中的线程没有绑定事件循环，无法调度执行这个协程——也就是说await asyncio.start_server根本没被执行，服务器从未真正启动，这就是连接被拒绝的根本原因。

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