使用pytest无法Mock get_redis函数的问题排查
问题分析与解决方案
问题背景
测试时代码仍调用真实Redis而非fakeredis,导致测试失败,但另有一个测试可正常运行。相关代码如下:
1. Redis提供函数(app/helpers/providers.py)
from redis import ConnectionPool, Redis redis_pool = None def get_redis() -> Redis: global redis_pool if redis_pool is None: redis_pool = ConnectionPool() return Redis.from_pool(redis_pool)
2. 接口代码(app/endpoints/secrets.py)
from app.helpers.providers import get_redis from fastapi import APIRouter, Depends, HTTPException, Request, status from fastapi.security import HTTPBasic, HTTPBasicCredentials from uuid import UUID from typing import Annotated router = APIRouter() security = HTTPBasic() @router.get("/secrets/{secret_id}") async def get_secret( request: Request, secret_id: UUID, credentials: Annotated[HTTPBasicCredentials, Depends(security)], redis: Redis = Depends(get_redis), ): key = f"secret:{secret_id}" with redis.pipeline() as pipe: pipe.get(key) pipe.delete(key) results = pipe.execute() if not results[0]: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) return results
3. 测试代码
import sys from unittest.mock import patch import fakeredis import pytest from fastapi import HTTPException, status from fastapi.testclient import TestClient @pytest.fixture def fake_redis(): return fakeredis.FakeStrictRedis() @pytest.fixture(autouse=True) def mock_redis_dependencies(fake_redis): with patch('app.endpoints.secrets.get_redis', return_value=fake_redis): yield sys.path.append(".") from app.endpoints.secrets import router def test_secret_not_found(fake_redis): client = TestClient(router) with pytest.raises(HTTPException) as exc: client.get( "/api/secrets/aaaaaaaa-bbbb-4ccc-aaaa-eeeeeeeeeef1", auth=("admin", "admin"), ) assert exc.value.status_code == status.HTTP_404_NOT_FOUND
错误原因
核心问题是模块导入时机与patch生效顺序不匹配:
- 测试文件加载时,
from app.endpoints.secrets import router语句已执行,此时secrets模块会从providers.py导入get_redis函数并保存其引用。 - 后续执行的autouse fixture虽然patch了
app.endpoints.secrets.get_redis,但无法修改已经导入到secrets模块命名空间中的函数引用,导致接口仍使用真实的get_redis实现。
修复方案
方案一:调整导入时机,确保patch生效后再导入模块
将router的导入移到测试函数内部,确保patch生效后再加载secrets模块:
import sys from unittest.mock import patch import fakeredis import pytest from fastapi import HTTPException, status from fastapi.testclient import TestClient @pytest.fixture def fake_redis(): return fakeredis.FakeStrictRedis() @pytest.fixture(autouse=True) def mock_redis_dependencies(fake_redis): with patch('app.endpoints.secrets.get_redis', return_value=fake_redis): yield sys.path.append(".") def test_secret_not_found(fake_redis): # 在patch生效后导入模块 from app.endpoints.secrets import router client = TestClient(router) with pytest.raises(HTTPException) as exc: client.get( "/api/secrets/aaaaaaaa-bbbb-4ccc-aaaa-eeeeeeeeeef1", auth=("admin", "admin"), ) assert exc.value.status_code == status.HTTP_404_NOT_FOUND
方案二:patch原始函数所在模块
直接patchget_redis的原始定义位置(app.helpers.providers.get_redis),这样所有模块引用的该函数都会指向mock版本:
@pytest.fixture(autouse=True) def mock_redis_dependencies(fake_redis): # 修改patch目标为原始函数所在模块 with patch('app.helpers.providers.get_redis', return_value=fake_redis): yield
方案三:使用FastAPI官方推荐的依赖覆盖(更优雅)
FastAPI提供了dependency_overrides机制,专门用于测试时替换依赖,无需使用unittest.mock:
import sys import fakeredis import pytest from fastapi import HTTPException, status from fastapi.testclient import TestClient @pytest.fixture def fake_redis(): return fakeredis.FakeStrictRedis() sys.path.append(".") def test_secret_not_found(fake_redis): from app.endpoints.secrets import router, get_redis client = TestClient(router) # 覆盖依赖 client.app.dependency_overrides[get_redis] = lambda: fake_redis try: with pytest.raises(HTTPException) as exc: client.get( "/api/secrets/aaaaaaaa-bbbb-4ccc-aaaa-eeeeeeeeeef1", auth=("admin", "admin"), ) assert exc.value.status_code == status.HTTP_404_NOT_FOUND finally: # 清理依赖覆盖,避免影响其他测试 client.app.dependency_overrides.clear()
内容的提问来源于stack exchange,提问作者Rodrigo
相关产品推荐
相关产品推荐

