You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

使用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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.28 05:54:52