from cinder import exception
from cinder import test
+from cinder.tests.utils import ZeroIntervalLoopingCall
from cinder.volume import configuration as conf
from cinder.volume.drivers.emc.emc_cli_fc import EMCCLIFCDriver
from cinder.volume.drivers.emc.emc_cli_iscsi import EMCCLIISCSIDriver
mock.call('lun', '-list', '-name', 'vol2', '-attachedSnapshot')]
fake_cli.assert_has_calls(expect_cmd)
+ @mock.patch('cinder.openstack.common.loopingcall.FixedIntervalLoopingCall',
+ new=ZeroIntervalLoopingCall)
def test_create_volume_from_snapshot_sync_failed(self):
output_smp = ("""LOGICAL UNIT NUMBER 1
expected = [mock.call(*self.testData.LUN_EXTEND_CMD('failed_vol1', 2))]
fake_cli.assert_has_calls(expected)
+ @mock.patch('cinder.openstack.common.loopingcall.FixedIntervalLoopingCall',
+ new=ZeroIntervalLoopingCall)
def test_extend_volume_failed(self):
commands = [self.testData.LUN_PROPERTY_ALL_CMD('failed_vol1')]
results = [self.testData.LUN_PROPERTY('failed_vol1', size=2)]
def test_retry_fail_and_succeed_mixed(self):
- @srb.retry(count=4, exceptions=(Exception))
+ @srb.retry(count=4, exceptions=(Exception),
+ sleep_mechanism=srb.retry.SLEEP_NONE)
def _try_failing(self):
attempted = self.attempts
self.attempts = self.attempts + 1
# under the License.
#
-
from cinder import context
from cinder import db
+from cinder.openstack.common import loopingcall
def get_test_admin_context():
for key in kwargs:
cgsnap[key] = kwargs[key]
return db.cgsnapshot_create(ctxt, cgsnap)
+
+
+class ZeroIntervalLoopingCall(loopingcall.FixedIntervalLoopingCall):
+ def start(self, interval, **kwargs):
+ kwargs['initial_delay'] = 0
+ return super(ZeroIntervalLoopingCall, self).start(0, **kwargs)