含wrapt_timeout_decorator元类继承的类Mock失败原因排查

问题背景 #

我们在测试场景中模拟调用类方法，涉及的两个类FirstService和SecondService均继承了使用wrapt_timeout_decorator库的元类TimeoutManager，所有类方法都会被该装饰器包裹以实现超时控制。但执行测试时发现mock对象的call_count始终为0，测试失败；而移除SecondService类的元类继承后，测试即可正常运行。需要明确元类继承对测试失败的影响机制。

代码示例 #

from wrapt_timeout_decorator import timeout
import inspect
import unittest
from unittest.mock import patch

class TimeoutManager(type):
    """
    用于给外部类的所有方法包裹超时装饰器的元类，用于控制后端超时
    使用方式：
    class YourClass(metaclass=TimeoutManager):
        ...
    """

    def __new__(cls, name, bases, dct):
        """
        重写类创建逻辑
        参数：
        - name (str): 类名
        - bases (tuple): 基类元组
        - dct (dict): 类字典
        返回：
        - type: 创建后的新类
        """

        for key, value in dct.items():
            # 排除魔法方法
            if callable(value) and not key.startswith("__"):
                # 检查方法是否已有@timeout装饰器
                validation: bool = cls.has_timeout_decorator(value)
                if not validation:
                    dct[key] = cls.default_timeout_wrapper()(value)
        return super().__new__(cls, name, bases, dct)

    @classmethod
    def default_timeout_wrapper(cls) -> callable:
        """返回默认超时装饰器"""
        return timeout(
            45,
            use_signals=False,
            timeout_exception=TimeoutError,
            exception_message="超时，请重试。",
        )

    @classmethod
    def has_timeout_decorator(cls, method: callable) -> bool:
        """检查方法是否带有@timeout装饰器"""
        decorators_method: list = cls.get_decorators(method)
        if len(decorators_method) > 0:
            if "timeout" in decorators_method:
                return True
        return False

    @classmethod
    def get_decorators(cls, method: callable) -> list:
        """从方法源码中提取装饰器名称"""
        decorators: list = []
        source_lines, _ = inspect.getsourcelines(method)
        for line in source_lines:
            stripped_line: str = line.strip()
            if stripped_line.startswith("@"):
                decorator_name: str = stripped_line.split("(")[0][1:]
                decorators.append(decorator_name)
        return decorators


class FirstService(metaclass=TimeoutManager):
    """执行推理操作的服务类"""

    @classmethod
    def persist_metadata(cls, resource_info):
        """处理请求并返回响应"""
        return resource_info
    
    @classmethod
    def get_sample_sql(cls, setup_conversation):
        return setup_conversation

# 带有元类的服务类
class SecondService(metaclass=TimeoutManager):
    @staticmethod
    def setup_chat(setup_conversation):
        metadata_response: dict = FirstService().persist_metadata(
            setup_conversation
        )
        sql_code_sample: str = FirstService().get_sample_sql(setup_conversation)
        response: dict = {
            "metadata": metadata_response,
            "sql_code_sample": sql_code_sample,
        }
        return response


class TestBaseServices(unittest.TestCase):

    def test_setup_conversation_chat(self):
        mock_setup_conversation_chat = "HELLO"
        metadata_mock_response: dict = {"response": "value", "is_saved_succes": True}
        sql_code_sample_mock: str = "SELECT * FROM table;"

        with patch.object(
            FirstService, "persist_metadata", return_value=metadata_mock_response
        ) as mock_persist:
            with patch.object(
                FirstService, "get_sample_sql", return_value=sql_code_sample_mock
            ) as mock_get_sql:
                response = SecondService.setup_chat(
                    mock_setup_conversation_chat
                )

                print("Persist Metadata Calls:", mock_persist.call_count)
                print("Get Sample SQL Calls:", mock_get_sql.call_count)

                mock_persist.assert_called_once_with(mock_setup_conversation_chat)
                mock_get_sql.assert_called_once_with(mock_setup_conversation_chat)

                assert "metadata" in response and "sql_code_sample" in response
                assert response["metadata"] == metadata_mock_response
                assert response["sql_code_sample"] == sql_code_sample_mock

if __name__ == "__main__":
    unittest.main()

问题原因分析 #

• 元类装饰时机早于Mock绑定
  元类TimeoutManager会在SecondService类定义的瞬间（程序加载阶段），就将setup_chat方法用timeout装饰器包裹。装饰器内部会保存setup_chat原始方法的引用，而此时测试中的Mock还未生效。当测试执行时，调用的是被装饰后的方法，它内部对FirstService方法的调用指向的是类定义时的原始方法，而非后续被Mock替换的类属性。

• 装饰器的静态引用特性
  wrapt_timeout_decorator的timeout装饰器属于静态包装型装饰器，它在初始化时会固定绑定原始方法的所有内部引用。也就是说，setup_chat里对FirstService().persist_metadata的调用，会被装饰器固化为原始方法的地址，后续通过patch.object替换FirstService类上的方法属性时，装饰器内部的引用不会同步更新，导致Mock无法捕获调用。

• 实例调用类方法的额外干扰
  setup_chat中使用FirstService().persist_metadata()（通过实例调用类方法）的方式，会进一步绕过Mock：因为Mock是替换类层面的方法属性，而实例调用时如果装饰器已经保存了原始方法的引用，就会直接调用原始方法，完全不经过类属性的访问路径。

解决方案 #

• 修改类方法调用方式
  将FirstService().persist_metadata()改为FirstService.persist_metadata()，直接通过类调用类方法。这样调用时会直接访问类上的方法属性，Mock替换后能被正确捕获。

• 调整元类装饰逻辑
  修改TimeoutManager的装饰逻辑，让装饰器在调用时动态获取方法引用，而非在类创建时固定绑定。比如，在装饰器内部通过类属性来访问被调用的方法，避免固化引用。

• 优化元类的装饰检测逻辑
  修改has_timeout_decorator方法，避免对测试环境中的Mock方法进行重复装饰，或者在测试环境中禁用元类的自动装饰逻辑。

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