pytest-async同步对象失效：异步循环未等待，原unittest正常

1. 异步测试标记缺失：pytest需要通过@pytest.mark.asyncio装饰器识别异步测试函数，若没加这个装饰器，测试会直接返回协程对象而不实际执行，导致限流器逻辑完全没触发，测试快速跑完。
2. 事件循环管理差异：IsolatedAsyncioTestCase会为每个测试自动创建独立事件循环并全程管理，但pytest-asyncio默认可能复用循环或初始化方式不同；如果限流器绑定了和测试不匹配的循环（比如手动创建的循环），限流逻辑就会失效。
3. 限流器作用域/初始化问题：如果限流器在全局或模块级别初始化，不同测试共享同一实例会导致状态混乱；或者在非异步环境初始化限流器，导致它没绑定到测试用的事件循环。
4. 依赖版本或插件冲突：旧版本的pytest-asyncio可能存在异步任务等待的Bug，或者和uvloop这类第三方事件循环插件不兼容，导致事件循环没正确等待限流任务完成。

1. 给异步测试加正确标记 #

所有异步测试函数必须添加@pytest.mark.asyncio装饰器，确保pytest能正确识别并运行异步逻辑：

import pytest
import asyncio

@pytest.mark.asyncio
async def test_rate_limiter(rate_limiter):
    start = asyncio.get_event_loop().time()
    # 批量执行带限流的异步任务
    await asyncio.gather(*[execute_async(rate_limiter) for _ in range(5)])
    elapsed = asyncio.get_event_loop().time() - start
    # 根据你的限流规则调整断言阈值
    assert elapsed > 2

2. 配置独立事件循环 #

在项目根目录的conftest.py中添加fixture，让每个测试都用独立的事件循环，和IsolatedAsyncioTestCase的行为对齐：

import pytest
import asyncio

@pytest.fixture(scope="function")
def event_loop():
    """为每个测试创建独立的事件循环"""
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

3. 用异步fixture初始化限流器 #

使用function级别的异步fixture创建限流器，确保每个测试有独立的实例，且绑定到当前测试的事件循环：

@pytest.fixture(scope="function")
async def rate_limiter_a():
    return RateLimiterA(max_concurrent=2)

@pytest.fixture(scope="function")
async def rate_limiter_b():
    return RateLimiterB(rate=2, period=1)  # 每秒2次的速率限制

避免在全局或模块级别初始化限流器，防止测试间状态污染。

4. 更新依赖并排查冲突 #

• 升级到最新稳定版的依赖：

pip install --upgrade pytest-asyncio aiolimiter

• 如果用了uvloop这类第三方循环，可临时禁用确认是否是兼容性问题：

# 在conftest.py中强制使用默认asyncio循环
import asyncio
import pytest

@pytest.fixture(scope="function")
def event_loop():
    asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

5. 检查限流器调用逻辑 #

确保execute_async中正确await限流器的异步acquire方法，并做好资源释放：

async def execute_async(limiter):
    await limiter.acquire()  # 必须await异步的acquire方法
    try:
        # 你的业务逻辑代码
        await asyncio.sleep(0.1)
    finally:
        # Semaphore需要手动释放，aiolimiter的AsyncLimiter会自动处理
        if isinstance(limiter, RateLimiterA):
            limiter.release()

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