仅模拟副作用 X 次

Question

我有一个 celery 重试任务，我想测试它是否会重试直到成功。使用 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]

结果：失败三次，然后 "succeeds" 但 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()。一个模拟完全取代了原来的;您的选择是产生副作用（从可迭代对象中增加或 return 一个值）或应用正常的模拟操作（return 一个新的模拟对象）。

此外，将 None 添加到 side_effect 序列不会重置副作用，它只是指示 mock 将 return 值改为 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 任务的一部分，它是一个外部单元，因此您通常应该只测试它是否被正确调用（使用正确的参数）以及正确的次数。

Answer 2

如果有人在 stopIterations 发生后只需要 return 值，您只需要 return 迭代器，而不是例外。

responses = [1,2,3,4,5]

def requests_side_effect(*args, **kwargs):
    try:
        return next(responses)
    except StopIteration:
        # raised all exceptions, call original
        return default_request_mock

这将检索前 5 个调用的值，然后 return 默认值。

仅模拟副作用 X 次

Mock side effect only X number of times

python

unit-testing

side-effects

mocking