test_retry.py 文件源码

def test_call_retry(self, m_sleep, m_count):
        attempts = 3
        timeout = 10
        deadline = self.now + timeout
        failures = [_EX1()] * (attempts - 1)
        event = mock.sentinel.event
        m_handler = mock.Mock()
        m_handler.side_effect = failures + [None]
        m_sleep.return_value = 1
        m_count.return_value = list(range(1, 5))
        retry = h_retry.Retry(m_handler, timeout=timeout, exceptions=_EX1)

        retry(event)

        m_handler.assert_has_calls([mock.call(event)] * attempts)
        m_sleep.assert_has_calls([
            mock.call(deadline, i + 1, failures[i])
            for i in range(len(failures))])