模拟副作用只有X次

Question

我有一个芹菜重试任务,我想测试它重试,直到成功.使用mock的side_effect,我可以在一定数量的执行中失败,然后传递None,清除副作用.但是,任务调用的方法不会在该点执行,它只是没有异常.有没有办法清除副作用,仍然有被模拟的方法执行正常？

我可以测试它被称为'x'次(即重复直到成功)然后在一个单独的测试中断言它做了应该做的,但是想知道是否有办法在一次测试中同时做两件事.

tasks.py:

import celery

@celery.task(max_retries=None)
def task():
    print "HERE"
    try:
        do_something("TASK")
    except Exception as exc:
        print exc
        raise task.retry(exc=exc)

def do_something(msg):
    print msg

测试:

import ....

class TaskTests(test.TestCase):

    @mock.patch('tasks.do_something')
    def test_will_retry_until_successful(self, action):
        action.side_effect = [Exception("First"), Exception("Second"), Exception("Third"), None]
        tasks.task.delay()
        self.assert.... [stuff about task]

结果:失败三次然后"成功"但do_something()从未打印. action.call_count等于4.我希望看到最后一个'HERE'后面的空白行将打印出'TASK'.

-------------------- >> begin captured stdout << ---------------------
HERE
First
HERE
Second
HERE
Third
HERE

--------------------- >> end captured stdout << ----------------------

Answer 1

你嘲笑do_something().模拟完全取代了原作; 您的选择是要么具有副作用(从迭代中提高或返回值),要么应用正常的模拟操作(返回一个新的模拟对象).

此外,添加None到side_effect序列不会重置副作用,它只是指示mock返回值None.你可以加入mock.DEFAULT; 在这种情况下,正常的模拟操作适用(就好像模拟被调用而没有副作用):

@mock.patch('tasks.do_something')
def test_will_retry_until_successful(self, action):
    action.side_effect = [Exception("First"), Exception("Second"), Exception("Third"), mock.DEFAULT]
    tasks.task.delay()
    self.assert.... [stuff about task]

如果您认为您的测试必须以调用原始函数结束,则必须存储对原始未修补函数的引用,然后将其side_effect设置为可调用的可调用函数,并在时间到来时调用原始函数:

# reference to original, global to the test module that won't be patched
from tasks import do_something

class TaskTests(test.TestCase):
    @mock.patch('tasks.do_something')
    def test_will_retry_until_successful(self, action):
        exceptions = iter([Exception("First"), Exception("Second"), Exception("Third")])
        def side_effect(*args, **kwargs):
            try:
                raise next(exceptions)
            except StopIteration:
                # raised all exceptions, call original
                return do_something(*args, **kwargs)
        action.side_effect = side_effect
        tasks.task.delay()
        self.assert.... [stuff about task]

但是,我不能预见到你想要这样做的单元测试场景.do_something()不是正在测试的Celery任务的一部分,它是一个外部单元,所以你通常只应该测试它是否被正确调用(使用正确的参数)和正确的次数.