使用pytest-asyncio测试时asyncio.start_server无法正常运行的问题

我之前也踩过一模一样的坑！核心问题就是单元测试里的事件循环没有被正确用来驱动服务端的异步任务——服务端的监听、连接处理逻辑根本没机会执行，自然回调不会触发，同步连接也会直接无限挂起。

为什么会出现这种情况？ #

• 单独运行程序时，你大概率会用asyncio.run()或者让事件循环run_forever()，这会持续轮询处理所有异步事件（包括服务端的accept事件）。但单元测试里如果只是手动创建了事件循环，却没让它持续运行，或者没等start_server完成启动就急着发起连接，服务端还没进入监听状态，连接请求直接就卡住了。
• 用同步代码连接异步服务端时，同步socket是阻塞式的，它会一直傻等服务端的accept响应，但此时事件循环根本没在跑，服务端的accept任务完全没被处理——两边直接陷入死锁：客户端等服务端，服务端的事件循环停着啥也干不了。

具体解决方案 #

1. 用异步测试框架管理事件循环（最推荐）

如果用pytest，先装pytest-asyncio，然后写异步测试用例，它会自动帮你打理好事件循环的生命周期：

import asyncio
import pytest

async def handle_client(reader, writer):
    # 你的客户端处理逻辑
    data = await reader.read(100)
    message = data.decode()
    print(f"Received {message!r}")
    writer.close()
    await writer.wait_closed()

async def test_server_callback():
    # 启动服务端并等待它完全就绪
    server = await asyncio.start_server(handle_client, '127.0.0.1', 8888)
    addr = server.sockets[0].getsockname()
    
    # 必须用异步方式连接！别用同步socket自找麻烦
    reader, writer = await asyncio.open_connection(*addr)
    writer.write(b"Hello Test")
    await writer.drain()
    
    # 这里可以加断言，比如用个全局/非局部变量标记回调是否触发
    # 示例：假设你在handle_client里设置了一个flag，这里断言flag为True
    
    writer.close()
    await writer.wait_closed()
    
    # 别忘了清理服务端资源
    server.close()
    await server.wait_closed()

这样测试框架会确保事件循环持续运行，同时处理服务端和客户端的异步任务，回调自然会被触发。

2. 手动管理事件循环（适合不用测试框架的场景）

如果必须手动搞，一定要确保事件循环运行足够久来处理连接事件，比如用run_until_complete配合任务创建：

import asyncio

def test_server_manual():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    
    callback_triggered = False
    async def handle_client(reader, writer):
        nonlocal callback_triggered
        callback_triggered = True  # 标记回调被触发
        writer.close()
        await writer.wait_closed()
    
    # 启动服务端任务，别直接run_until_complete，先异步启动
    server_task = loop.create_task(asyncio.start_server(handle_client, '127.0.0.1', 8889))
    # 等待服务端完全启动并开始监听
    server = loop.run_until_complete(server_task)
    
    # 用异步连接代替同步连接，避免阻塞事件循环线程
    async def connect_and_test():
        reader, writer = await asyncio.open_connection('127.0.0.1', 8889)
        await writer.drain()
        writer.close()
        await writer.wait_closed()
    
    # 运行客户端测试任务
    loop.run_until_complete(connect_and_test())
    
    # 验证回调是否触发
    assert callback_triggered is True
    
    # 清理资源
    server.close()
    loop.run_until_complete(server.wait_closed())
    loop.close()

划重点：绝对不要在运行事件循环的线程里用同步socket连接！同步的connect会直接阻塞线程，事件循环根本没法处理服务端的accept，直接死锁到天荒地老。

3. 几个关键注意事项

• 必须等待start_server返回的协程执行完成，确保服务端已经绑定端口并开始监听。
• 测试里的客户端连接一定要用异步方式（asyncio.open_connection），别碰同步socket。
• 如果非要用同步代码测试，得把事件循环放到单独的线程里跑，主线程再用同步socket连接，但这种方式容易出各种线程同步问题，真心不推荐。

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