异步Socket服务测试：如何在单元测试结束前取消asyncio Task？

我来帮你搞定这个异步Socket服务器单元测试里的Pending Task警告问题！这个问题本质上是你的client_loop()任务一直在后台挂着，测试结束时pytest直接终止事件循环，没给它优雅退出的机会。下面是具体的解决方案：

核心思路 #

我们需要在测试的生命周期里主动管理服务器的后台任务，确保每个测试结束前，都能取消任务并等待它完成清理，避免任务处于pending状态被销毁。

具体实现步骤 #

1. 用pytest Fixture管理服务器任务 #

创建一个fixture，专门负责启动和清理服务器任务，这样所有依赖它的测试都会自动处理任务生命周期：

import pytest
import asyncio
from your_server_module import AsyncSocketServer  # 替换成你的服务器类路径

@pytest.fixture
async def socket_server():
    # 初始化并启动服务器
    server = AsyncSocketServer()
    # 启动client_loop任务，保存任务对象以便后续取消
    server_loop_task = asyncio.create_task(server.client_loop())
    
    # 把服务器实例传给测试用例
    yield server
    
    # 测试结束后，取消任务并等待它完成清理
    server_loop_task.cancel()
    try:
        await server_loop_task  # 等待任务处理取消逻辑
    except asyncio.CancelledError:
        # 捕获取消错误是正常流程，不用慌
        pass

2. 在测试用例中使用Fixture #

你的测试直接依赖这个fixture，测试结束后fixture会自动执行清理逻辑：

@pytest.mark.asyncio
async def test_server_handles_request(socket_server):
    # 这里写你的测试逻辑：连接服务器、发送请求、验证响应
    reader, writer = await asyncio.open_connection('127.0.0.1', socket_server.port)
    
    # 发送测试请求
    writer.write(b"Hello Server!")
    await writer.drain()
    
    # 读取响应并验证
    response = await reader.read(1024)
    assert response == b"Hello Client!"
    
    # 关闭连接
    writer.close()
    await writer.wait_closed()

3. 优化服务器的client_loop()方法 #

如果你的client_loop()里有长时间阻塞的操作，最好让它能及时响应取消信号。比如给等待操作加个超时，或者主动处理取消错误：

async def client_loop(self):
    while True:
        try:
            # 用wait_for设置短超时，让任务能定期检查取消信号
            incoming_data = await asyncio.wait_for(self.socket_reader.read(1024), timeout=0.5)
            # 处理请求的逻辑
            await self.process_request(incoming_data)
        except asyncio.TimeoutError:
            # 超时只是循环继续，不影响
            continue
        except asyncio.CancelledError:
            # 收到取消信号，做必要的清理（比如关闭socket）
            self.cleanup()
            raise  # 重新抛出错误，让fixture里的await能捕获到

为什么这能解决问题？ #

• Fixture的yield语法会在测试运行前启动任务，测试结束后自动执行后面的清理代码
• task.cancel()会给任务发送取消信号，await server_loop_task会等待任务处理完清理逻辑再结束，彻底避免任务处于pending状态被销毁

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