仅模拟副作用X次

2024-03-28 22:36:08 发布

您现在位置:Python中文网/ 问答频道 /正文

我有一个芹菜重试任务,我想测试它是否重试,直到成功。使用mock的副作用,我可以在一组执行中失败,然后传递None,清除副作用。然而,任务调用的方法并没有在此时执行,它只是没有异常。有没有一种方法可以消除副作用,并且仍然让被模拟的方法正常执行?

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

任务.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。 我希望看到最后一行的空白行是“任务”的打印。

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

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

Tags: 方法importnonetaskheredefexceptionaction
1条回答
网友
1楼 · 发布于 2024-03-28 22:36:08

你嘲笑我。mock完全替代了原来的mock;您的选择要么产生副作用(从iterable中提升或返回一个值),要么应用普通的mock操作(返回一个新的mock对象)。

此外,将None添加到side_effect序列并不能重置副作用,它只是指示mock返回值None。您可以改为添加^{};在这种情况下,将应用正常的模拟操作(就像调用模拟时没有副作用一样):

@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()不是正在测试的芹菜任务的一部分,它是一个外部单元,因此通常只应测试调用是否正确(使用正确的参数)和正确的次数。

相关问题 更多 >