test_session.py 文件源码

def test_serialized_with_retry_socket_error_conn_reset(self):
        exc = socket.error()
        exc.errno = errno.ECONNRESET
        plugin = 'glance'
        fn = 'download_vhd'
        num_retries = 1
        callback = None
        retry_cb = mock.Mock()
        with mock.patch.object(self.session, 'call_plugin_serialized',
                               spec=True) as call_plugin_serialized:
            call_plugin_serialized.side_effect = exc
            self.assertRaises(
                exception.PluginRetriesExceeded,
                self.session.call_plugin_serialized_with_retry, plugin, fn,
                num_retries, callback, retry_cb)
            call_plugin_serialized.assert_called_with(plugin, fn)
            self.assertEqual(2, call_plugin_serialized.call_count)
            self.assertEqual(2, retry_cb.call_count)