如何Mock类方法内部的异步Redis客户端方法？

我需要Mock被注入到Role类中的redis_client.fields_allowed方法，以此独立测试目标类。

目标类代码（app.auth.roles.py） #

from app.redis.client import redis_client
from dataclasses import dataclass
from typing import Union, Callable

role_count = 0


@dataclass
class Role:
    id: int
    name: str
    byte: int

    @staticmethod
    def new(name: str, *roles: Role) -> Role:
        global role_count
        role_count += 1
        byte: int = 1 << role_count
        for role in roles:
            byte |= role.byte
        return Role(id=role_count, name=name, byte=byte)

    def __or__(self, other: Union[Role, int]) -> int:
        if isinstance(other, Role):
            return self.byte | other.byte
        return self.byte | other

    def __and__(self, other: Union[Role, int]) -> int:
        if isinstance(other, Role):
            return self.byte & other.byte
        return self.byte & other

    def fields(self, *field_ids: int) -> Callable[[int], Role]:
        def func(user_id: int) -> Role:
            # 需要Mock的方法
            if redis_client.fields_allowed(user_id, self.id, list(field_ids)): 
                return self
            return Role(id=0, name="None", byte=0)

        return func

初始测试代码（app.auth.test_auth.py） #

已提前创建access_token fixture：

from unittest.mock import patch
import app.auth.roles as roles

@patch.object(roles.Role, "fields.func.redis_client.fields_allowed")
def test_field_ownership(access_token, fields_allowed):
    fields_allowed.return_value = True
    access_token_str = access_token
    # 其他测试逻辑

运行测试时报错：fixture 'fields_allowed' not found。要求在不修改Role类、不使用依赖注入的前提下完成Mock。

更新1：调整patch位置后的问题 #

将patch装饰器放在fixture参数之前：

@patch("app.auth.roles.redis_client.fields_allowed")
def test_valid_field_ownership(fields_allowed, access_token):
    fields_allowed.return_value = True
    access_token_str = access_token
    # 其他测试逻辑

此时Mock看似生效，但调用时未返回预期的True，而是打印出协程对象：

# Role类中修改后的fields方法
def fields(self, *field_ids: int) -> Callable[[int], Role]:
    def func(user_id: int) -> Role:
        if v := redis_client.fields_allowed(user_id, self.id, list(field_ids)):
            print(v) # 输出：<coroutine object AsyncMockMixin._execute_mock_call at 0x7f8cb7e5e340>
            return self
        else:
            raise error.NotEnoughClaims(
                status_code=401, message="Not enough claims"
            )

    return func

更新2：找到问题根源及解决方案 #

问题在于redis_client.fields_allowed是异步可等待对象（awaitable），因此需要：

1. 将相关函数改为异步（包括func内部调用方法时添加await，以及测试函数改为异步）
2. 使用pytest-asyncio插件来运行异步测试

内容的提问来源于stack exchange，提问作者Sami Al-Subhi