如何为包含嵌套方法的Twisted addCallback逻辑编写单元测试

核心依赖工具 #

• 使用Twisted原生测试框架 twisted.trial.unittest.TestCase 作为测试基类，原生支持Deferred异步用例
• 使用 twisted.internet.task.Clock 模拟reactor时间，无需真实等待即可验证重试逻辑
• Mock method3 返回值，覆盖所有回调分支

完整测试代码示例 #

首先给出与你提供的逻辑对齐的被测类示意：

from twisted.internet.defer import Deferred

class MyService:
    def __init__(self, reactor):
        self.reactor = reactor
        self.state = None

    def method1(self):
        self.method2()

    def method2(self):
        def method2_success(result):
            if result.code == 200:
                self.state = 'running'
            else:
                self.state = 'failed'
                self.reactor.callLater(10, self.method2)

        def method2_failed(failure):
            self.state = 'failed'
            self.reactor.callLater(10, self.method2)

        self.state = 'in_progress'
        d = self.method3()
        d.addCallbacks(method2_success, method2_failed)
        return d

    def method3(self):
        return some_deffered_object

测试类实现：

from twisted.trial import unittest
from twisted.internet.task import Clock
from twisted.internet.defer import Deferred, fail

# 模拟method3返回的带code属性的结果对象
class MockResult:
    def __init__(self, code):
        self.code = code

class TestMyService(unittest.TestCase):
    def setUp(self):
        # 初始化模拟时钟和被测实例
        self.clock = Clock()
        self.service = MyService(reactor=self.clock)

    def test_method2_success_with_200(self):
        """测试method3返回200成功的场景"""
        # Mock method3返回200成功结果
        mock_res = MockResult(code=200)
        self.service.method3 = lambda: Deferred.callback(mock_res)

        d = self.service.method2()
        # Deferred执行完成后做断言
        def assert_result(_):
            self.assertEqual(self.service.state, 'running')
            # 验证无重试调度生成
            self.assertEqual(len(self.clock.calls), 0)
        d.addCallback(assert_result)
        return d

    def test_method2_success_with_non_200(self):
        """测试method3返回非200状态的场景"""
        mock_res = MockResult(code=400)
        self.service.method3 = lambda: Deferred.callback(mock_res)

        d = self.service.method2()
        def assert_result(_):
            self.assertEqual(self.service.state, 'failed')
            # 验证10秒后重试的调度已正确注册
            self.assertEqual(len(self.clock.calls), 1)
            self.assertEqual(self.clock.calls[0].delay, 10)
            self.assertEqual(self.clock.calls[0].func, self.service.method2)
        d.addCallback(assert_result)
        return d

    def test_method2_failure(self):
        """测试method3调用失败的场景"""
        # Mock method3返回失败结果
        self.service.method3 = lambda: fail(Exception("call failed"))

        d = self.service.method2()
        def assert_result(_):
            self.assertEqual(self.service.state, 'failed')
            # 验证重试调度正常生成
            self.assertEqual(len(self.clock.calls), 1)
            self.assertEqual(self.clock.calls[0].delay, 10)
        # 用addBoth处理失败态的Deferred，避免测试直接报错
        d.addBoth(assert_result)
        return d

    def test_method1_call(self):
        """测试method1正确触发method2调用"""
        call_count = 0
        def mock_method2():
            nonlocal call_count
            call_count +=1
            return Deferred.callback(None)
        self.service.method2 = mock_method2

        self.service.method1()
        self.assertEqual(call_count, 1)

测试运行方式 #

直接执行命令 trial 测试文件名.py 即可，Twisted trial会自动处理异步Deferred的等待、异常捕获和断言校验。

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