test_retry.py 文件源码

def test_retryable_aborts_on_unexpected_exception(
            self, mock_time, mock_exc_to_code):
        mock_time.return_value = 0
        mock_exc_to_code.side_effect = lambda e: e.code

        mock_call = mock.Mock()
        mock_call.side_effect = [CustomException('', _FAKE_STATUS_CODE_2),
                                 mock.DEFAULT]
        mock_call.return_value = 1729

        retry_options = RetryOptions(
            [_FAKE_STATUS_CODE_1],
            BackoffSettings(0, 0, 0, 0, 0, 0, 1))

        my_callable = retry.retryable(mock_call, retry_options)

        try:
            my_callable(None)
            self.fail('Should not have been reached')
        except errors.RetryError as exc:
            self.assertIsInstance(exc.cause, CustomException)

        self.assertEqual(1, mock_call.call_count)