Python Celery单元测试中如何Mock Redis连接及Redis存储的测试数据

核心思路 #

你代码中的Redis实例是在ReferenceData类的__init__方法中初始化绑定到实例属性的，单元测试时不需要启动真实Redis服务，直接用Mock对象替换真实Redis实例即可，同时可以自定义Mock的方法行为来模拟存储/读取数据的逻辑。

具体实现步骤 #

我们以Python标准库unittest.mock为例，两种常用实现方式如下：

方式1：直接替换实例的redis属性（灵活度高，适合单测单个方法） #

import pytest
from unittest.mock import Mock
from your_module import ReferenceData # 替换为你实际的模块导入路径

def test_save_raisr_event_transcription():
    # 1. 初始化任务实例，替换内部redis属性为Mock对象
    task = ReferenceData()
    mock_redis = Mock()
    task.redis = mock_redis

    # 2. 补全测试上下文参数（你贴的代码存在变量未定义问题，测试前需自行补全业务逻辑）
    test_crn = "test_crn_001"
    test_trans = "测试转写内容"
    # 构造业务需要的event入参，补全trans、crn等变量的赋值逻辑

    # 3. 调用待测试方法
    result = task.save_raisr_event(event)

    # 4. 验证逻辑正确性
    assert result == 'transcription saved'
    # 验证redis append方法被正确调用，key和存储值符合预期
    mock_redis.append.assert_called_once_with(f"{test_crn}-transcription", f" {test_trans}")

def test_read_existed_transcription():
    # 模拟Redis中已存储的transcription值场景
    task = ReferenceData()
    mock_redis = Mock()
    # 注意：原生StrictRedis默认返回bytes类型，如果你初始化时加了decode_responses=True则直接返回字符串即可
    mock_redis.get.return_value = b"已存储的历史转写内容"
    task.redis = mock_redis

    # 调用你的读Redis业务逻辑
    stored_val = task.redis.get(f"{test_crn}-transcription")
    assert stored_val == b"已存储的历史转写内容"

方式2：Patch全局Redis类（适合批量测试，不需要逐个修改实例属性） #

import pytest
from unittest.mock import Mock, patch
from your_module import ReferenceData

@patch("your_module.redis.StrictRedis")
def test_patch_redis(mock_strict_redis):
    # 配置全局mock的StrictRedis实例行为
    mock_redis_instance = Mock()
    mock_strict_redis.return_value = mock_redis_instance
    # 模拟真实Redis append方法返回追加后的字符串长度
    mock_redis_instance.append.return_value = 32

    # 初始化任务时，内部的self.redis会自动替换为mock实例
    task = ReferenceData()

    # 后续测试逻辑和方式1一致
    test_crn = "test_crn_002"
    test_trans = "新转写内容"
    task.save_raisr_event(event)
    mock_redis_instance.append.assert_called_once_with(f"{test_crn}-transcription", f" {test_trans}")

提示：你贴的代码存在语法错误和未定义变量，比如get_agents方法结尾缺少右括号，save_raisr_event方法内的trans、crn、transcription都没有定义，实际测试前需要先修正这些问题。

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