import os.path
import string
+import mock
from oslo_log import log as logging
from cinder.brick import exception
from cinder.brick.initiator import linuxscsi
from cinder import test
+from cinder.tests.unit import utils
LOG = logging.getLogger(__name__)
('tee -a /sys/block/sdc/device/delete')]
self.assertEqual(expected_commands, self.cmds)
+ @mock.patch('cinder.openstack.common.loopingcall.FixedIntervalLoopingCall',
+ new=utils.ZeroIntervalLoopingCall)
def test_wait_for_volume_removal(self):
fake_path = '/dev/disk/by-path/fake-iscsi-iqn-lun-0'
self.stubs.Set(os.path, "exists", lambda x: True)
import socket
import sys
import tempfile
+import time
import eventlet
import mock
from cinder.backup import driver as backup_driver
from cinder.brick.local_dev import lvm as brick_lvm
-from cinder import compute
+from cinder.compute import nova
from cinder import context
from cinder import db
from cinder import exception
self.assertNotIn(iscsi_target, targets)
targets.append(iscsi_target)
+ # FIXME(jdg): What is this actually testing?
+ # We never call the internal _check method?
for _index in xrange(100):
tests_utils.create_volume(self.context, **self.volume_params)
for volume_id in volume_ids:
self.assertIsNone(volume['migration_status'])
self.assertEqual('available', volume['status'])
- @mock.patch.object(compute.nova.API, 'update_server_volume')
+ @mock.patch.object(nova.API, 'update_server_volume')
@mock.patch('cinder.volume.manager.VolumeManager.'
'migrate_volume_completion')
@mock.patch('cinder.db.volume_get')
fake_new_volume['id'],
error=False)
- @mock.patch.object(compute.nova.API, 'update_server_volume')
+ @mock.patch.object(nova.API, 'update_server_volume')
@mock.patch('cinder.volume.manager.VolumeManager.'
'migrate_volume_completion')
@mock.patch('cinder.db.volume_get')
mock.patch.object(volume_rpcapi.VolumeAPI, 'create_volume') as \
mock_create_volume, \
mock.patch.object(self.volume, '_clean_temporary_volume') as \
- clean_temporary_volume:
+ clean_temporary_volume, \
+ mock.patch.object(time, 'sleep'):
# Exception case at the timeout of the volume creation
mock_create_volume.side_effect = fake_create_volume
from cinder import context
from cinder import exception
from cinder import test
+from cinder.tests.unit import utils
from cinder.volume.drivers import xio
from cinder.volume import qos_specs
from cinder.volume import volume_types
self.setup_driver()
self.driver.create_snapshot(SNAPSHOT1)
+ @mock.patch('cinder.openstack.common.loopingcall.FixedIntervalLoopingCall',
+ new=utils.ZeroIntervalLoopingCall)
def test_negative_create_snapshot_invalid_state_recover(self, mock_req):
ctxt = context.get_admin_context()
extra_specs = {"Feature:Pool": "1",
self.setup_driver()
self.driver.create_snapshot(SNAPSHOT1)
+ @mock.patch('cinder.openstack.common.loopingcall.FixedIntervalLoopingCall',
+ new=utils.ZeroIntervalLoopingCall)
def test_negative_create_snapshot_invalid_state_norecover(self, mock_req):
ctxt = context.get_admin_context()
extra_specs = {"Feature:Pool": "1",