from cinder.volume.drivers.windows import windows
from cinder.volume.drivers.windows import windows_utils
-
CONF = cfg.CONF
drv.delete_snapshot(snapshot)
- def test_create_export(self):
+ def _test_create_export(self, chap_enabled=False):
drv = self._driver
-
volume = db_fakes.get_fake_volume_info()
-
initiator_name = "%s%s" % (CONF.iscsi_target_prefix, volume['name'])
+ fake_chap_username = 'fake_chap_username'
+ fake_chap_password = 'fake_chap_password'
+
+ self.flags(use_chap_auth=chap_enabled)
+ self.flags(chap_username=fake_chap_username)
+ self.flags(chap_password=fake_chap_password)
+ self.mox.StubOutWithMock(windows_utils.WindowsUtils,
+ 'add_disk_to_target')
self.mox.StubOutWithMock(windows_utils.WindowsUtils,
'create_iscsi_target')
- windows_utils.WindowsUtils.create_iscsi_target(initiator_name)
self.mox.StubOutWithMock(windows_utils.WindowsUtils,
- 'add_disk_to_target')
+ 'set_chap_credentials')
+ self.mox.StubOutWithMock(self._driver,
+ 'remove_export')
+
+ self._driver.remove_export(mox.IgnoreArg(), mox.IgnoreArg())
+ windows_utils.WindowsUtils.create_iscsi_target(initiator_name)
+
+ if chap_enabled:
+ windows_utils.WindowsUtils.set_chap_credentials(
+ mox.IgnoreArg(),
+ fake_chap_username,
+ fake_chap_password)
+
windows_utils.WindowsUtils.add_disk_to_target(volume['name'],
initiator_name)
export_info = drv.create_export(None, volume)
- self.assertEqual(export_info['provider_location'], initiator_name)
+ self.assertEqual(initiator_name, export_info['provider_location'])
+ if chap_enabled:
+ expected_provider_auth = ' '.join(('CHAP',
+ fake_chap_username,
+ fake_chap_password))
+ self.assertEqual(expected_provider_auth,
+ export_info['provider_auth'])
+
+ def test_create_export_chap_disabled(self):
+ self._test_create_export()
+
+ def test_create_export_chap_enabled(self):
+ self._test_create_export(chap_enabled=True)
def test_initialize_connection(self):
drv = self._driver
exception.VolumeBackendAPIException,
self.wutils.is_resize_needed,
mock.sentinel.vhd_path, 1, 2)
+
+ @mock.patch.object(windows_utils.WindowsUtils, '_wmi_obj_set_attr')
+ @mock.patch.object(windows_utils, 'wmi', create=True)
+ def test_set_chap_credentials(self, mock_wmi, mock_set_attr):
+ mock_wt_host = mock.Mock()
+ mock_wt_host_class = self.wutils._conn_wmi.WT_Host
+ mock_wt_host_class.return_value = [mock_wt_host]
+
+ self.wutils.set_chap_credentials(mock.sentinel.target_name,
+ mock.sentinel.chap_username,
+ mock.sentinel.chap_password)
+
+ mock_wt_host_class.assert_called_once_with(
+ HostName=mock.sentinel.target_name)
+
+ mock_set_attr.assert_has_calls([
+ mock.call(mock_wt_host, 'EnableCHAP', True),
+ mock.call(mock_wt_host, 'CHAPUserName',
+ mock.sentinel.chap_username),
+ mock.call(mock_wt_host, 'CHAPSecret',
+ mock.sentinel.chap_password)])
+
+ mock_wt_host.put.assert_called_once_with()
+
+ @mock.patch.object(windows_utils.WindowsUtils, '_wmi_obj_set_attr')
+ @mock.patch.object(windows_utils, 'wmi', create=True)
+ def test_set_chap_credentials_exc(self, mock_wmi, mock_set_attr):
+ mock_wmi.x_wmi = Exception
+ mock_set_attr.side_effect = mock_wmi.x_wmi
+ self.assertRaises(exception.VolumeBackendAPIException,
+ self.wutils.set_chap_credentials,
+ mock.sentinel.target_name,
+ mock.sentinel.chap_username,
+ mock.sentinel.chap_password)
+
+ def test_set_wmi_obj_attr(self):
+ wmi_obj = mock.Mock()
+ wmi_property_method = wmi_obj.wmi_property
+ wmi_property = wmi_obj.wmi_property.return_value
+
+ self.wutils._wmi_obj_set_attr(wmi_obj,
+ mock.sentinel.key,
+ mock.sentinel.value)
+
+ wmi_property_method.assert_called_once_with(mock.sentinel.key)
+ wmi_property.set.assert_called_once_with(mock.sentinel.value)
from cinder.volume.drivers.windows import constants
from cinder.volume.drivers.windows import vhdutils
from cinder.volume.drivers.windows import windows_utils
+from cinder.volume import utils
LOG = logging.getLogger(__name__)
def create_export(self, context, volume):
"""Driver entry point to get the export info for a new volume."""
+ # Since the iSCSI targets are not reused, being deleted when the
+ # volume is detached, we should clean up existing targets before
+ # creating a new one.
+ self.remove_export(context, volume)
+
target_name = "%s%s" % (self.configuration.iscsi_target_prefix,
volume['name'])
+ updates = {'provider_location': target_name}
self.utils.create_iscsi_target(target_name)
+ if self.configuration.use_chap_auth:
+ chap_username = (self.configuration.chap_username or
+ utils.generate_username())
+ chap_password = (self.configuration.chap_password or
+ utils.generate_password())
+
+ self.utils.set_chap_credentials(target_name,
+ chap_username,
+ chap_password)
+
+ updates['provider_auth'] = ' '.join(('CHAP',
+ chap_username,
+ chap_password))
# Get the disk to add
vol_name = volume['name']
self.utils.add_disk_to_target(vol_name, target_name)
- return {'provider_location': target_name}
+ return updates
def remove_export(self, context, volume):
"""Driver entry point to remove an export for a volume.
LOG.error(err_msg)
raise exception.VolumeBackendAPIException(data=err_msg)
+ def set_chap_credentials(self, target_name, chap_username, chap_password):
+ try:
+ wt_host = self._conn_wmi.WT_Host(HostName=target_name)[0]
+ self._wmi_obj_set_attr(wt_host, 'EnableCHAP', True)
+ self._wmi_obj_set_attr(wt_host, 'CHAPUserName', chap_username)
+ self._wmi_obj_set_attr(wt_host, 'CHAPSecret', chap_password)
+ wt_host.put()
+ except wmi.x_wmi as exc:
+ err_msg = (_('Failed to set CHAP credentials on '
+ 'target %(target_name)s. WMI exception: %(wmi_exc)s')
+ % {'target_name': target_name,
+ 'wmi_exc': exc})
+ LOG.error(err_msg)
+ raise exception.VolumeBackendAPIException(data=err_msg)
+
+ @staticmethod
+ def _wmi_obj_set_attr(wmi_obj, key, value):
+ # Due to a bug in the python WMI module, some wmi object attributes
+ # cannot be modified. This method is used as a workaround.
+ wmi_property = getattr(wmi_obj, 'wmi_property')
+ wmi_property(key).set(value)
+
def add_disk_to_target(self, vol_name, target_name):
"""Adds the disk to the target."""
try:
Code Review / openstack-build / cinder-build.git / commitdiff