如何在FastAPI中Mock依赖类RateLimiter？

我有如下FastAPI接口：

from fastapi_limiter.depends import RateLimiter

@documents_router.delete(
    "/delete_document/{folder}/{file}",
    status_code=200,
    dependencies=[
        Depends(RateLimiter(times=1, seconds=10)),
        Depends(RateLimiter(times=2, hours=30)),
    ],
)
async def delete_document(
    folder: str,
    file: str,
    request: Request,
) -> dict:
    return dict(message="hello world!")

测试文件代码如下：

from fastapi_limiter.depends import RateLimiter
@pytest.fixture
async def test_client(mock_acquire):
    from app.main import app

    client = TestClient(app)

    app.dependency_overrides[RateLimiter] = lambda: Mock()

    return client

运行测试时出现如下错误：

self = <fastapi_limiter.depends.RateLimiter object at 0x7fae1d318910>, 

    async def _check(self, key):
        redis = FastAPILimiter.redis
>       pexpire = await redis.evalsha(
            FastAPILimiter.lua_sha, 1, key, str(self.times), str(self.milliseconds)
        )
E       TypeError: object Mock can't be used in 'await' expression

../venv/lib/python3.10/site-packages/fastapi_limiter/depends.py:29: TypeError

我需要正确Mock RateLimiter的实例及其所有方法，请问该如何实现？

你之前的问题在于直接用普通Mock()替代RateLimiter，但RateLimiter内部会调用异步Redis方法，普通Mock不支持异步操作，且没有正确覆盖限流逻辑。以下是两种可行的解决方法：

方法一：Mock底层Redis连接 #

直接MockFastAPILimiter的Redis客户端，跳过实际的限流Lua脚本执行：

from unittest.mock import AsyncMock
from fastapi_limiter import FastAPILimiter
from fastapi.testclient import TestClient
import pytest

@pytest.fixture
async def test_client():
    from app.main import app
    
    # 创建异步Mock的Redis客户端
    mock_redis = AsyncMock()
    # 让evalsha返回模拟值，避免触发限流逻辑
    mock_redis.evalsha.return_value = 0
    FastAPILimiter.redis = mock_redis
    
    client = TestClient(app)
    return client

方法二：替换RateLimiter依赖为异步兼容Mock #

自定义Mock类，让它的__call__方法是异步空实现，直接跳过限流检查：

from fastapi_limiter.depends import RateLimiter
from fastapi.testclient import TestClient
import pytest

@pytest.fixture
async def test_client():
    from app.main import app
    
    class MockRateLimiter:
        def __init__(self, *args, **kwargs):
            pass
        
        async def __call__(self, *args, **kwargs):
            # 不执行任何限流检查，直接通过
            pass
    
    app.dependency_overrides[RateLimiter] = MockRateLimiter
    
    client = TestClient(app)
    return client

说明 #

• 方法一更贴近真实运行逻辑，只是屏蔽了Redis的实际调用，适合需要保留RateLimiter初始化流程的测试场景。
• 方法二直接替换了RateLimiter类，完全隔离限流逻辑，适合不需要验证限流规则的单元测试。

内容的提问来源于stack exchange，提问作者Jayjay Jay