我有一个函数,它有一个decorator@retry
,如果发生特定异常,它会重试该函数。我想测试这个函数执行的次数是否正确,为此我有以下代码正在运行:
@pytest.mark.asyncio
async def test_redis_failling(mocker):
sleep_mock = mocker.patch.object(retry, '_sleep')
with pytest.raises(ConnectionError):
retry_store_redis()
assert sleep_mock.call_count == 4
@retry(ConnectionError, initial_wait=2.0, attempts=5)
def retry_store_redis():
raise ConnectionError()
但是,如果我将retry_store_redis()
修改为异步函数,则sleep_mock.call_count
的返回值为0。
所以您定义了"重试";作为一种功能。然后定义一个测试,然后定义一些使用@retry的代码。
@retry作为一个装饰器,在导入时被调用。所以操作顺序是
- 声明重试
- 声明测试
- 以retry_store_redis为参数调用retry
- 开始测试
- 修补重试
- 调用您在步骤3中定义的函数
so";重试";被调用一次(在导入时(,mock被调用零次。为了得到你想要的行为,(确保重试实际上是在重新调用底层函数(我会做
@pytest.mark.asyncio
async def test_redis_failling(mocker):
fake_function = MagicMock(side_effect=ConnectionError)
decorated_function = retry(ConnectionError, initial_wait=2.0, attempts=5)(fake_function)
with pytest.raises(ConnectionError):
decorated_function()
assert fake_function.call_count == 4
如果您想在构建时对其进行测试(而不是专门针对装饰器的测试(,则必须模拟装饰函数中的原始函数,这将取决于您如何实现装饰器。默认方式(没有任何库(意味着您必须检查";闭包";属性您可以构建对象以保留对原始函数的引用,这里有一个示例
def wrap(func):
class Wrapper:
def __init__(self, func):
self.func = func
def __call__(self, *args, **kwargs):
return self.func(*args, **kwargs)
return Wrapper(func)
@wrap
def wrapped_func():
return 42
在这种情况下,您可以在wrapped_func.func
处修补封装的函数