openstack
/
cinder
Code Issues Proposed changes

Replace tearDown with addCleanup - Part 2 

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
This commit is contained in:
Julia Varlamova 2014-02-17 17:34:09 +04:00
parent eb7d21080a
commit 85ec1aca23
15 changed files with 112 additions and 243 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
cinder/tests/brick/test_brick_connector.py
View File

@ -16,8 +16,6 @@ import os.path
import string
import time
import mox
from cinder.brick import exception
from cinder.brick.initiator import connector
from cinder.brick.initiator import host_driver
@ -34,7 +32,6 @@ class ConnectorTestCase(test.TestCase):
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))
@ -120,9 +117,6 @@ class ISCSIConnectorTestCase(ConnectorTestCase):
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',
@ -471,7 +465,6 @@ class AoEConnectorTestCase(ConnectorTestCase):
"""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'}
@ -479,11 +472,6 @@ class AoEConnectorTestCase(ConnectorTestCase):
'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:
@ -574,7 +562,6 @@ class RemoteFsConnectorTestCase(ConnectorTestCase):
def setUp(self):
super(RemoteFsConnectorTestCase, self).setUp()
self.mox = mox.Mox()
self.connection_properties = {
'export': self.TEST_DEV,
'name': '9c592d52-ce47-4263-8c21-4ecf3c029cdb'}
@ -582,11 +569,6 @@ class RemoteFsConnectorTestCase(ConnectorTestCase):
'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

4
cinder/tests/image/test_glance.py
View File

@ -652,7 +652,3 @@ class TestGlanceImageServiceClient(test.TestCase):
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()

5
cinder/tests/integrated/integrated_helpers.py
View File

@ -66,13 +66,10 @@ class _IntegratedTestBase(test.TestCase):
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()

5
cinder/tests/test_backup_ceph.py
View File

@ -173,6 +173,7 @@ class BackupCephTestCase(test.TestCase):
# 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)
@ -195,10 +196,6 @@ class BackupCephTestCase(test.TestCase):
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'))

13
cinder/tests/test_coraid.py
View File

@ -231,9 +231,6 @@ class FakeRpc(object):
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
@ -253,10 +250,6 @@ class CoraidDriverTestCase(test.TestCase):
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]
@ -677,18 +670,12 @@ class CoraidDriverIntegrationalTestCase(CoraidDriverLoginSuccessTestCase):
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')

6
cinder/tests/test_glusterfs.py
View File

@ -95,11 +95,7 @@ class GlusterFsDriverTestCase(test.TestCase):
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)

14
cinder/tests/test_gpfs.py
View File

@ -59,6 +59,7 @@ class GPFSDriverTestCase(test.TestCase):
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)
@ -80,13 +81,12 @@ class GPFSDriverTestCase(test.TestCase):
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)))

18
cinder/tests/test_huawei_hvs.py
View File

@ -504,7 +504,10 @@ class HVSRESTiSCSIDriverTestCase(test.TestCase):
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
@ -517,12 +520,6 @@ class HVSRESTiSCSIDriverTestCase(test.TestCase):
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)
@ -699,7 +696,10 @@ class HVSRESTFCDriverTestCase(test.TestCase):
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
@ -711,12 +711,6 @@ class HVSRESTFCDriverTestCase(test.TestCase):
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)

31
cinder/tests/test_huawei_t_dorado.py
View File

@ -1029,7 +1029,11 @@ class HuaweiTISCSIDriverTestCase(test.TestCase):
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
@ -1046,12 +1050,6 @@ class HuaweiTISCSIDriverTestCase(test.TestCase):
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'
@ -1424,7 +1422,11 @@ class HuaweiTFCDriverTestCase(test.TestCase):
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)
@ -1442,12 +1444,6 @@ class HuaweiTFCDriverTestCase(test.TestCase):
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,
@ -1658,9 +1654,12 @@ class SSHMethodTestCase(test.TestCase):
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
@ -1674,12 +1673,6 @@ class SSHMethodTestCase(test.TestCase):
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,

16
cinder/tests/test_policy.py
View File

@ -41,10 +41,7 @@ class PolicyFileTestCase(test.TestCase):
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:
@ -86,10 +83,7 @@ class PolicyTestCase(test.TestCase):
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"
@ -170,15 +164,13 @@ class DefaultPolicyTestCase(test.TestCase):
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", {})

9
cinder/tests/test_quota.py
View File

@ -44,6 +44,8 @@ class QuotaIntegrationTestCase(test.TestCase):
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,
@ -59,12 +61,7 @@ class QuotaIntegrationTestCase(test.TestCase):
# 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."""

8
cinder/tests/test_scality.py
View File

@ -119,6 +119,7 @@ class ScalityDriverTestCase(test.TestCase):
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,
@ -137,12 +138,9 @@ class ScalityDriverTestCase(test.TestCase):
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."""

22
cinder/tests/test_volume.py
View File

@ -96,6 +96,7 @@ class BaseVolumeTestCase(test.TestCase):
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'
@ -116,13 +117,12 @@ class BaseVolumeTestCase(test.TestCase):
# 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
@ -2453,6 +2453,8 @@ class CopyVolumeToImageTestCase(BaseVolumeTestCase):
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 = {
@ -2461,6 +2463,8 @@ class CopyVolumeToImageTestCase(BaseVolumeTestCase):
'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),
@ -2470,11 +2474,6 @@ class CopyVolumeToImageTestCase(BaseVolumeTestCase):
'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
@ -2646,13 +2645,13 @@ class DriverTestCase(test.TestCase):
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
@ -3185,10 +3184,7 @@ class VolumePolicyTestCase(test.TestCase):
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))

12
cinder/tests/test_volume_types_extra_specs.py
View File

@ -32,22 +32,18 @@ class VolumeTypeExtraSpecsTestCase(test.TestCase):
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(

174
cinder/tests/test_windows.py
View File

@ -24,9 +24,7 @@ import tempfile
from oslo.config import cfg
import mox as mox_lib
from mox import IgnoreArg
from mox import stubout
import mox
from cinder import test
@ -48,9 +46,9 @@ class TestWindowsDriver(test.TestCase):
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,
)
@ -61,12 +59,6 @@ class TestWindowsDriver(test.TestCase):
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):
@ -78,62 +70,52 @@ class TestWindowsDriver(test.TestCase):
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()
@ -142,73 +124,61 @@ class TestWindowsDriver(test.TestCase):
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()
@ -216,103 +186,88 @@ class TestWindowsDriver(test.TestCase):
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()
@ -321,51 +276,46 @@ class TestWindowsDriver(test.TestCase):
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()
@ -373,15 +323,13 @@ class TestWindowsDriver(test.TestCase):
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()