我有以下函数来实现指数退避算法:
class Retry():
def exponential_backoff_retry(self, function, *args, n=1):
MAX_TRIES = 8
try:
f = function(*args)
except:
if n > MAX_TRIES:
return None
n += 1
time.sleep((2 ** n) + (random.randint(0, 1000) / 1000.0))
return self.exponential_backoff_retry(function, *args, n)
else:
return f
我想做的是写一个单元测试来确认行为,给定一个传入的函数,例如一个发出API请求的函数,在异常的情况下它会重试很多次。
以下是我目前所做的尝试:
@mock.patch('requests.post')
@mock.patch('utils.retry.Retry.exponential_backoff_retry', side_effect=Exception('whoops'))
def test_exponential_backoff(self, mock_retry, req_post_mock):
req_post_mock.return_value = {"status_code": 202}
with self.assertRaises(Exception):
mock_retry(req_post_mock)
self.assertEqual(req_post_mock.return_value["status_code"], 202)
self.assertEqual(mock_retry.call.count, 8)
任何建议都将不胜感激。
1条答案
按热度按时间pengsaosao1#
假设前
首先,提供的代码似乎有一个小错误(除了指示错误),您应该更改函数参数的顺序,如
def exponential_backoff_retry(self, function,n=1, *args):
而不是def exponential_backoff_retry(self, function, *args, n=1):
,或者如果您更喜欢在递归函数调用return self.exponential_backoff_retry(function, *args, n=n)
中将n作为命名参数关于如何测试的主要答案
您可以创建一个模拟类来保存函数调用计数器
您可以在测试方法中使用helper类,如下所示:
或