Replace tearDown with addCleanup in some cinder unit tests.
Infra team has indicated that tearDown should not be used and should
be replaced with addCleanup in all places.
Remove duplicated self.stubs and self._mox declaration in
cinder/tests/test_windows.py.
Implements blueprint replace-teardown-with-addcleanup
Change-Id: I5865cf0197e1d8229afc706b9b14b746f09b5acd
import string
import time
-import mox
-
from cinder.brick import exception
from cinder.brick.initiator import connector
from cinder.brick.initiator import host_driver
def setUp(self):
super(ConnectorTestCase, self).setUp()
self.cmds = []
- self.stubs.Set(os.path, 'exists', lambda x: True)
def fake_execute(self, *cmd, **kwargs):
self.cmds.append(string.join(cmd))
self.stubs.Set(self.connector._linuxscsi,
'get_name_from_path', lambda x: "/dev/sdb")
- def tearDown(self):
- super(ISCSIConnectorTestCase, self).tearDown()
-
def iscsi_connection(self, volume, location, iqn):
return {
'driver_volume_type': 'iscsi',
"""Test cases for AoE initiator class."""
def setUp(self):
super(AoEConnectorTestCase, self).setUp()
- self.mox = mox.Mox()
self.connector = connector.AoEConnector('sudo')
self.connection_properties = {'target_shelf': 'fake_shelf',
'target_lun': 'fake_lun'}
'FixedIntervalLoopingCall',
FakeFixedIntervalLoopingCall)
- def tearDown(self):
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
- super(AoEConnectorTestCase, self).tearDown()
-
def _mock_path_exists(self, aoe_path, mock_values=[]):
self.mox.StubOutWithMock(os.path, 'exists')
for value in mock_values:
def setUp(self):
super(RemoteFsConnectorTestCase, self).setUp()
- self.mox = mox.Mox()
self.connection_properties = {
'export': self.TEST_DEV,
'name': '9c592d52-ce47-4263-8c21-4ecf3c029cdb'}
'nfs', root_helper='sudo', nfs_mount_point_base='/mnt/test',
nfs_mount_options='vers=3')
- def tearDown(self):
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
- super(RemoteFsConnectorTestCase, self).tearDown()
-
def test_connect_volume(self):
"""Test the basic connect volume case."""
client = self.connector._remotefsclient
client = glance._create_glance_client(self.context, 'fake_host:9292',
False)
self.assertIsInstance(client, MyGlanceStubClient)
-
- def tearDown(self):
- self.stubs.UnsetAll()
- super(TestGlanceImageServiceClient, self).tearDown()
self.volume = self.start_service('volume')
self.scheduler = self.start_service('scheduler')
self._start_api_service()
+ self.addCleanup(self.osapi.stop)
self.api = client.TestOpenStackClient('fake', 'fake', self.auth_url)
- def tearDown(self):
- self.osapi.stop()
- super(_IntegratedTestBase, self).tearDown()
-
def _start_api_service(self):
self.osapi = service.WSGIService("osapi_volume")
self.osapi.start()
# Create a file with some data in it.
self.volume_file = tempfile.NamedTemporaryFile()
+ self.addCleanup(self.volume_file.close)
for i in xrange(0, self.num_chunks):
data = os.urandom(self.chunk_size)
self.checksum.update(data)
self.callstack = []
- def tearDown(self):
- self.volume_file.close()
- super(BackupCephTestCase, self).tearDown()
-
@common_mocks
def test_get_rbd_support(self):
self.assertFalse(hasattr(self.service.rbd, 'RBD_FEATURE_LAYERING'))
class CoraidDriverTestCase(test.TestCase):
def setUp(self):
super(CoraidDriverTestCase, self).setUp()
-
- self.mox = mox.Mox()
-
configuration = mox.MockObject(conf.Configuration)
configuration.append_config_values(mox.IgnoreArg())
configuration.coraid_esm_address = fake_esm_ipaddress
self.driver = coraid.CoraidDriver(configuration=configuration)
self.driver.do_setup({})
- def tearDown(self):
- self.mox.UnsetStubs()
- super(CoraidDriverTestCase, self).tearDown()
-
def mock_volume_types(self, repositories=[]):
if not repositories:
repositories = [fake_repository_name]
class AutoReloginCoraidTestCase(test.TestCase):
def setUp(self):
super(AutoReloginCoraidTestCase, self).setUp()
- self.mox = mox.Mox()
-
self.rest_client = coraid.CoraidRESTClient('https://fake')
self.appliance = coraid.CoraidAppliance(self.rest_client,
'fake_username',
'fake_password',
'fake_group')
- def tearDown(self):
- self.mox.UnsetStubs()
- super(AutoReloginCoraidTestCase, self).tearDown()
-
def _test_auto_relogin_fail(self, state):
self.mox.StubOutWithMock(self.rest_client, 'rpc')
glusterfs.GlusterfsDriver(configuration=self._configuration,
db=FakeDb())
self._driver.shares = {}
-
- def tearDown(self):
- self._mox.UnsetStubs()
- self.stubs.UnsetAll()
- super(GlusterFsDriverTestCase, self).tearDown()
+ self.addCleanup(self._mox.UnsetStubs)
def stub_out_not_replaying(self, obj, attr_name):
attr_to_replace = getattr(obj, attr_name)
super(GPFSDriverTestCase, self).setUp()
self.volumes_path = tempfile.mkdtemp(prefix="gpfs_")
self.images_dir = '%s/images' % self.volumes_path
+ self.addCleanup(self._cleanup, self.images_dir, self.volumes_path)
if not os.path.exists(self.volumes_path):
os.mkdir(self.volumes_path)
self.context.project_id = 'fake'
CONF.gpfs_images_dir = self.images_dir
- def tearDown(self):
- try:
- os.rmdir(self.images_dir)
- os.rmdir(self.volumes_path)
- except OSError:
- pass
- super(GPFSDriverTestCase, self).tearDown()
+ def _cleanup(self, images_dir, volumes_path):
+ try:
+ os.rmdir(images_dir)
+ os.rmdir(volumes_path)
+ except OSError:
+ pass
def test_different(self):
self.assertTrue(gpfs._different((True, False)))
def setUp(self):
super(HVSRESTiSCSIDriverTestCase, self).setUp()
self.tmp_dir = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, self.tmp_dir)
self.fake_conf_file = self.tmp_dir + '/cinder_huawei_conf.xml'
+ self.addCleanup(os.remove, self.fake_conf_file)
+
self.create_fake_conf_file()
self.configuration = mox.MockObject(conf.Configuration)
self.configuration.cinder_huawei_conf_file = self.fake_conf_file
self.driver.do_setup({})
self.driver.common.test_normal = True
- def tearDown(self):
- if os.path.exists(self.fake_conf_file):
- os.remove(self.fake_conf_file)
- shutil.rmtree(self.tmp_dir)
- super(HVSRESTiSCSIDriverTestCase, self).tearDown()
-
def test_log_in_success(self):
deviceid = self.driver.common.login()
self.assertIsNotNone(deviceid)
def setUp(self):
super(HVSRESTFCDriverTestCase, self).setUp()
self.tmp_dir = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, self.tmp_dir)
self.fake_conf_file = self.tmp_dir + '/cinder_huawei_conf.xml'
+ self.addCleanup(os.remove, self.fake_conf_file)
+
self.create_fake_conf_file()
self.configuration = mox.MockObject(conf.Configuration)
self.configuration.cinder_huawei_conf_file = self.fake_conf_file
self.driver.do_setup({})
self.driver.common.test_normal = True
- def tearDown(self):
- if os.path.exists(self.fake_conf_file):
- os.remove(self.fake_conf_file)
- shutil.rmtree(self.tmp_dir)
- super(HVSRESTFCDriverTestCase, self).tearDown()
-
def test_log_in_Success(self):
deviceid = self.driver.common.login()
self.assertIsNotNone(deviceid)
super(HuaweiTISCSIDriverTestCase, self).setUp()
self.tmp_dir = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, self.tmp_dir)
+
self.fake_conf_file = self.tmp_dir + '/cinder_huawei_conf.xml'
+ self.addCleanup(os.remove, self.fake_conf_file)
+
create_fake_conf_file(self.fake_conf_file)
self.configuration = mox.MockObject(conf.Configuration)
self.configuration.cinder_huawei_conf_file = self.fake_conf_file
self.driver = HuaweiVolumeDriver(configuration=self.configuration)
self.driver.do_setup(None)
- def tearDown(self):
- if os.path.exists(self.fake_conf_file):
- os.remove(self.fake_conf_file)
- shutil.rmtree(self.tmp_dir)
- super(HuaweiTISCSIDriverTestCase, self).tearDown()
-
def test_conf_invalid(self):
# Test config file not found
tmp_fonf_file = '/xxx/cinder_huawei_conf.xml'
super(HuaweiTFCDriverTestCase, self).setUp()
self.tmp_dir = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, self.tmp_dir)
+
self.fake_conf_file = self.tmp_dir + '/cinder_huawei_conf.xml'
+ self.addCleanup(os.remove, self.fake_conf_file)
+
create_fake_conf_file(self.fake_conf_file)
modify_conf(self.fake_conf_file, 'Storage/Protocol', 'FC')
self.configuration = mox.MockObject(conf.Configuration)
self.driver = HuaweiVolumeDriver(configuration=self.configuration)
self.driver.do_setup(None)
- def tearDown(self):
- if os.path.exists(self.fake_conf_file):
- os.remove(self.fake_conf_file)
- shutil.rmtree(self.tmp_dir)
- super(HuaweiTFCDriverTestCase, self).tearDown()
-
def test_validate_connector_failed(self):
invalid_connector = {'host': 'testhost'}
self.assertRaises(exception.VolumeBackendAPIException,
def setUp(self):
super(SSHMethodTestCase, self).setUp()
-
self.tmp_dir = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, self.tmp_dir)
+
self.fake_conf_file = self.tmp_dir + '/cinder_huawei_conf.xml'
+ self.addCleanup(os.remove, self.fake_conf_file)
+
create_fake_conf_file(self.fake_conf_file)
self.configuration = mox.MockObject(conf.Configuration)
self.configuration.cinder_huawei_conf_file = self.fake_conf_file
self.driver = HuaweiVolumeDriver(configuration=self.configuration)
self.driver.do_setup(None)
- def tearDown(self):
- if os.path.exists(self.fake_conf_file):
- os.remove(self.fake_conf_file)
- shutil.rmtree(self.tmp_dir)
- super(SSHMethodTestCase, self).tearDown()
-
def test_reach_max_connection_limit(self):
self.stubs.Set(FakeChannel, 'recv', self._fake_recv1)
self.assertRaises(exception.CinderException,
self.context = context.RequestContext('fake', 'fake')
policy.reset()
self.target = {}
-
- def tearDown(self):
- super(PolicyFileTestCase, self).tearDown()
- policy.reset()
+ self.addCleanup(policy.reset)
def test_modified_policy_reloads(self):
with utils.tempdir() as tmpdir:
common_policy.set_brain(common_policy.Brain(rules))
self.context = context.RequestContext('fake', 'fake', roles=['member'])
self.target = {}
-
- def tearDown(self):
- policy.reset()
- super(PolicyTestCase, self).tearDown()
+ self.addCleanup(policy.reset)
def test_enforce_nonexistent_action_throws(self):
action = "example:noexist"
self.context = context.RequestContext('fake', 'fake')
+ self.addCleanup(policy.reset)
+
def _set_brain(self, default_rule):
brain = cinder.openstack.common.policy.Brain(self.rules,
default_rule)
cinder.openstack.common.policy.set_brain(brain)
- def tearDown(self):
- super(DefaultPolicyTestCase, self).tearDown()
- policy.reset()
-
def test_policy_called(self):
self.assertRaises(exception.PolicyNotAuthorized, policy.enforce,
self.context, "example:exist", {})
self.volume_type = db.volume_type_create(
context.get_admin_context(),
dict(name=self.volume_type_name))
+ self.addCleanup(db.volume_type_destroy, context.get_admin_context(),
+ self.volume_type['id'])
self.flags(quota_volumes=2,
quota_snapshots=2,
# conflicts with the test cases here that are setting up their own
# defaults.
db.quota_class_destroy_all_by_name(self.context, 'default')
-
- def tearDown(self):
- db.volume_type_destroy(context.get_admin_context(),
- self.volume_type['id'])
- super(QuotaIntegrationTestCase, self).tearDown()
- cinder.tests.image.fake.FakeImageService_reset()
+ self.addCleanup(cinder.tests.image.fake.FakeImageService_reset)
def _create_volume(self, size=1):
"""Create a test volume."""
super(ScalityDriverTestCase, self).setUp()
self.tempdir = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, self.tempdir)
self.TEST_MOUNT = self.tempdir
self.TEST_VOLPATH = os.path.join(self.TEST_MOUNT,
self._create_fake_mount()
self._create_fake_config()
- self._configure_driver()
+ self.addCleanup(self._remove_fake_config)
- def tearDown(self):
- shutil.rmtree(self.tempdir)
- self._remove_fake_config()
- super(ScalityDriverTestCase, self).tearDown()
+ self._configure_driver()
def test_setup_no_config(self):
"""Missing SOFS configuration shall raise an error."""
vol_tmpdir = tempfile.mkdtemp()
self.flags(volumes_dir=vol_tmpdir,
notification_driver=["test"])
+ self.addCleanup(self._cleanup)
self.volume = importutils.import_object(CONF.volume_manager)
self.context = context.get_admin_context()
self.context.user_id = 'fake'
# keep ordered record of what we execute
self.called = []
- def tearDown(self):
+ def _cleanup(self):
try:
shutil.rmtree(CONF.volumes_dir)
except OSError:
pass
fake_notifier.reset()
- super(BaseVolumeTestCase, self).tearDown()
def fake_get_target(obj, iqn):
return 1
def setUp(self):
super(CopyVolumeToImageTestCase, self).setUp()
self.dst_fd, self.dst_path = tempfile.mkstemp()
+ self.addCleanup(os.unlink, self.dst_path)
+
os.close(self.dst_fd)
self.stubs.Set(self.volume.driver, 'local_path', self.fake_local_path)
self.image_meta = {
'disk_format': 'raw'
}
self.volume_id = 1
+ self.addCleanup(db.volume_destroy, self.context, self.volume_id)
+
self.volume_attrs = {
'id': self.volume_id,
'updated_at': datetime.datetime(1, 1, 1, 1, 1, 1),
'host': 'dummy'
}
- def tearDown(self):
- db.volume_destroy(self.context, self.volume_id)
- os.unlink(self.dst_path)
- super(CopyVolumeToImageTestCase, self).tearDown()
-
def test_copy_volume_to_image_status_available(self):
# creating volume testdata
self.volume_attrs['instance_uuid'] = None
return self.output, None
self.volume.driver.set_execute(_fake_execute)
self.volume.driver.set_initialized()
+ self.addCleanup(self._cleanup)
- def tearDown(self):
+ def _cleanup(self):
try:
shutil.rmtree(CONF.volumes_dir)
except OSError:
pass
- super(DriverTestCase, self).tearDown()
def fake_get_target(obj, iqn):
return 1
self.context = context.get_admin_context()
self.stubs.Set(brick_lvm.LVM, '_vg_exists', lambda x: True)
-
- def tearDown(self):
- super(VolumePolicyTestCase, self).tearDown()
- cinder.policy.reset()
+ self.addCleanup(cinder.policy.reset)
def _set_rules(self, rules):
cinder.common.policy.set_brain(cinder.common.policy.Brain(rules))
vol_extra3=3)
self.vol_type1['extra_specs'] = self.vol_type1_specs
ref = db.volume_type_create(self.context, self.vol_type1)
+ self.addCleanup(db.volume_type_destroy, context.get_admin_context(),
+ self.vol_type1['id'])
self.volume_type1_id = ref.id
for k, v in self.vol_type1_specs.iteritems():
self.vol_type1_specs[k] = str(v)
self.vol_type2_noextra = dict(name="TEST: Volume type without extra")
ref = db.volume_type_create(self.context, self.vol_type2_noextra)
+ self.addCleanup(db.volume_type_destroy, context.get_admin_context(),
+ self.vol_type2_noextra['id'])
self.vol_type2_id = ref.id
- def tearDown(self):
- # Remove the volume type from the database
- db.volume_type_destroy(context.get_admin_context(),
- self.vol_type1['id'])
- db.volume_type_destroy(context.get_admin_context(),
- self.vol_type2_noextra['id'])
- super(VolumeTypeExtraSpecsTestCase, self).tearDown()
-
def test_volume_type_specs_get(self):
expected_specs = self.vol_type1_specs.copy()
actual_specs = db.volume_type_extra_specs_get(
from oslo.config import cfg
-import mox as mox_lib
-from mox import IgnoreArg
-from mox import stubout
+import mox
from cinder import test
def setUp(self):
self.lun_path_tempdir = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, self.lun_path_tempdir)
+
super(TestWindowsDriver, self).setUp()
- self._mox = mox_lib.Mox()
- self.stubs = stubout.StubOutForTesting()
self.flags(
windows_iscsi_lun_path=self.lun_path_tempdir,
)
self._driver = windows.WindowsDriver(configuration=configuration)
self._driver.do_setup({})
- def tearDown(self):
- self._mox.UnsetStubs()
- self.stubs.UnsetAll()
- shutil.rmtree(self.lun_path_tempdir)
- super(TestWindowsDriver, self).tearDown()
-
def _setup_stubs(self):
def fake_wutils__init__(self):
str(volume['name']) + ".vhd")
def test_check_for_setup_errors(self):
- mox = self._mox
drv = self._driver
- self._mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'check_for_setup_error')
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'check_for_setup_error')
windows_utils.WindowsUtils.check_for_setup_error()
- mox.ReplayAll()
+ self.mox.ReplayAll()
drv.check_for_setup_error()
- mox.VerifyAll()
-
def test_create_volume(self):
- mox = self._mox
drv = self._driver
vol = db_fakes.get_fake_volume_info()
self.stubs.Set(drv, 'local_path', self.fake_local_path)
- self._mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'create_volume')
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'create_volume')
windows_utils.WindowsUtils.create_volume(self.fake_local_path(vol),
vol['name'], vol['size'])
- mox.ReplayAll()
+ self.mox.ReplayAll()
drv.create_volume(vol)
- mox.VerifyAll()
-
def test_delete_volume(self):
"""delete_volume simple test case."""
- mox = self._mox
drv = self._driver
vol = db_fakes.get_fake_volume_info()
- mox.StubOutWithMock(drv, 'local_path')
+ self.mox.StubOutWithMock(drv, 'local_path')
drv.local_path(vol).AndReturn(self.fake_local_path(vol))
- self._mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'delete_volume')
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'delete_volume')
windows_utils.WindowsUtils.delete_volume(vol['name'],
self.fake_local_path(vol))
- mox.ReplayAll()
+ self.mox.ReplayAll()
drv.delete_volume(vol)
- mox.VerifyAll()
-
def test_create_snapshot(self):
- mox = self._mox
drv = self._driver
- self._mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'create_snapshot')
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'create_snapshot')
volume = db_fakes.get_fake_volume_info()
snapshot = db_fakes.get_fake_snapshot_info()
windows_utils.WindowsUtils.create_snapshot(volume['name'],
snapshot['name'])
- mox.ReplayAll()
+ self.mox.ReplayAll()
drv.create_snapshot(snapshot)
- mox.VerifyAll()
-
def test_create_volume_from_snapshot(self):
- mox = self._mox
drv = self._driver
snapshot = db_fakes.get_fake_snapshot_info()
volume = db_fakes.get_fake_volume_info()
- self._mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'create_volume_from_snapshot')
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'create_volume_from_snapshot')
windows_utils.WindowsUtils.\
create_volume_from_snapshot(volume['name'], snapshot['name'])
- mox.ReplayAll()
+ self.mox.ReplayAll()
drv.create_volume_from_snapshot(volume, snapshot)
- mox.VerifyAll()
-
def test_delete_snapshot(self):
- mox = self._mox
drv = self._driver
snapshot = db_fakes.get_fake_snapshot_info()
- self._mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'delete_snapshot')
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'delete_snapshot')
windows_utils.WindowsUtils.delete_snapshot(snapshot['name'])
- mox.ReplayAll()
+ self.mox.ReplayAll()
drv.delete_snapshot(snapshot)
- mox.VerifyAll()
-
def test_create_export(self):
- mox = self._mox
drv = self._driver
volume = db_fakes.get_fake_volume_info()
initiator_name = "%s%s" % (CONF.iscsi_target_prefix, volume['name'])
- self._mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'create_iscsi_target')
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'create_iscsi_target')
windows_utils.WindowsUtils.create_iscsi_target(initiator_name,
- IgnoreArg())
- self._mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'add_disk_to_target')
+ mox.IgnoreArg())
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'add_disk_to_target')
windows_utils.WindowsUtils.add_disk_to_target(volume['name'],
initiator_name)
- mox.ReplayAll()
+ self.mox.ReplayAll()
export_info = drv.create_export(None, volume)
- mox.VerifyAll()
-
self.assertEqual(export_info['provider_location'], initiator_name)
def test_initialize_connection(self):
- mox = self._mox
drv = self._driver
volume = db_fakes.get_fake_volume_info()
connector = db_fakes.get_fake_connector_info()
- self._mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'associate_initiator_with_iscsi_target')
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'associate_initiator_with_iscsi_target')
windows_utils.WindowsUtils.associate_initiator_with_iscsi_target(
volume['provider_location'], initiator_name, )
- self._mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'get_host_information')
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'get_host_information')
windows_utils.WindowsUtils.get_host_information(
volume, volume['provider_location'])
- mox.ReplayAll()
+ self.mox.ReplayAll()
drv.initialize_connection(volume, connector)
- mox.VerifyAll()
-
def test_terminate_connection(self):
- mox = self._mox
drv = self._driver
volume = db_fakes.get_fake_volume_info()
initiator_name = "%s%s" % (CONF.iscsi_target_prefix, volume['name'])
connector = db_fakes.get_fake_connector_info()
- self._mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'delete_iscsi_target')
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'delete_iscsi_target')
windows_utils.WindowsUtils.delete_iscsi_target(
initiator_name, volume['provider_location'])
- mox.ReplayAll()
+ self.mox.ReplayAll()
drv.terminate_connection(volume, connector)
- mox.VerifyAll()
-
def test_ensure_export(self):
- mox = self._mox
drv = self._driver
volume = db_fakes.get_fake_volume_info()
initiator_name = "%s%s" % (CONF.iscsi_target_prefix, volume['name'])
- self._mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'create_iscsi_target')
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'create_iscsi_target')
windows_utils.WindowsUtils.create_iscsi_target(initiator_name, True)
- self._mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'add_disk_to_target')
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'add_disk_to_target')
windows_utils.WindowsUtils.add_disk_to_target(volume['name'],
initiator_name)
- mox.ReplayAll()
+ self.mox.ReplayAll()
drv.ensure_export(None, volume)
- mox.VerifyAll()
-
def test_remove_export(self):
- mox = self._mox
drv = self._driver
volume = db_fakes.get_fake_volume_info()
target_name = volume['provider_location']
- self._mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'remove_iscsi_target')
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'remove_iscsi_target')
windows_utils.WindowsUtils.remove_iscsi_target(target_name)
- mox.ReplayAll()
+ self.mox.ReplayAll()
drv.remove_export(None, volume)
- mox.VerifyAll()
-
def test_copy_image_to_volume(self):
"""resize_image common case usage."""
- mox = self._mox
drv = self._driver
volume = db_fakes.get_fake_volume_info()
self.stubs.Set(drv, 'local_path', self.fake_local_path)
- mox.StubOutWithMock(image_utils, 'fetch_to_vhd')
+ self.mox.StubOutWithMock(image_utils, 'fetch_to_vhd')
image_utils.fetch_to_vhd(None, None, None,
self.fake_local_path(volume),
- mox_lib.IgnoreArg())
+ mox.IgnoreArg())
- mox.ReplayAll()
+ self.mox.ReplayAll()
drv.copy_image_to_volume(None, volume, None, None)
- mox.VerifyAll()
-
def test_copy_volume_to_image(self):
- mox = self._mox
drv = self._driver
vol = db_fakes.get_fake_volume_info()
self.stubs.Set(drv, 'local_path', self.fake_local_path)
- mox.StubOutWithMock(image_utils, 'upload_volume')
+ self.mox.StubOutWithMock(image_utils, 'upload_volume')
temp_vhd_path = os.path.join(CONF.image_conversion_dir,
str(image_meta['id']) + ".vhd")
image_utils.upload_volume(None, None, image_meta, temp_vhd_path, 'vpc')
- self._mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'copy_vhd_disk')
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'copy_vhd_disk')
windows_utils.WindowsUtils.copy_vhd_disk(self.fake_local_path(vol),
temp_vhd_path)
- mox.ReplayAll()
+ self.mox.ReplayAll()
drv.copy_volume_to_image(None, vol, None, image_meta)
- mox.VerifyAll()
-
def test_create_cloned_volume(self):
- mox = self._mox
drv = self._driver
volume = db_fakes.get_fake_volume_info()
volume_cloned = db_fakes.get_fake_volume_info_cloned()
- self._mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'create_volume')
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'create_volume')
- windows_utils.WindowsUtils.create_volume(IgnoreArg(), IgnoreArg(),
- IgnoreArg())
+ windows_utils.WindowsUtils.create_volume(mox.IgnoreArg(),
+ mox.IgnoreArg(),
+ mox.IgnoreArg())
- self._mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'copy_vhd_disk')
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'copy_vhd_disk')
windows_utils.WindowsUtils.copy_vhd_disk(self.fake_local_path(
volume_cloned), self.fake_local_path(volume))
- mox.ReplayAll()
+ self.mox.ReplayAll()
drv.create_cloned_volume(volume, volume_cloned)
- mox.VerifyAll()
-
def test_extend_volume(self):
- mox = self._mox
drv = self._driver
volume = db_fakes.get_fake_volume_info()
TEST_VOLUME_ADDITIONAL_SIZE_MB = 1024
TEST_VOLUME_ADDITIONAL_SIZE_GB = 1
- self._mox.StubOutWithMock(windows_utils.WindowsUtils, 'extend')
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils, 'extend')
windows_utils.WindowsUtils.extend(volume['name'],
TEST_VOLUME_ADDITIONAL_SIZE_MB)
new_size = volume['size'] + TEST_VOLUME_ADDITIONAL_SIZE_GB
- mox.ReplayAll()
+ self.mox.ReplayAll()
drv.extend_volume(volume, new_size)
-
- mox.VerifyAll()