# Copyright 2012 Pedro Navarro Perez # Copyright 2015 Cloudbase Solutions SRL # All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. """ Unit tests for Windows Server 2012 OpenStack Cinder volume driver """ import os from unittest import mock import ddt from oslo_utils import fileutils from oslo_utils import timeutils from oslo_utils import units from cinder import context from cinder import exception from cinder.image import image_utils from cinder.tests.unit import fake_constants as fake from cinder.tests.unit import fake_snapshot from cinder.tests.unit import fake_volume from cinder.tests.unit import test from cinder.tests.unit import utils as test_utils from cinder.tests.unit.windows import db_fakes from cinder.volume import configuration as conf from cinder.volume.drivers.windows import iscsi as windows_iscsi @ddt.ddt class TestWindowsISCSIDriver(test.TestCase): @mock.patch.object(windows_iscsi, 'utilsfactory') def setUp(self, mock_utilsfactory): super(TestWindowsISCSIDriver, self).setUp() self.configuration = conf.Configuration(None) self.configuration.append_config_values(windows_iscsi.windows_opts) self.flags(windows_iscsi_lun_path='fake_iscsi_lun_path') self.flags(image_conversion_dir='fake_image_conversion_dir') self._driver = windows_iscsi.WindowsISCSIDriver( configuration=self.configuration) self._context = context.get_admin_context() self.updated_at = timeutils.utcnow() @mock.patch.object(fileutils, 'ensure_tree') def test_do_setup(self, mock_ensure_tree): self._driver.do_setup(mock.sentinel.context) mock_ensure_tree.assert_has_calls( [mock.call('fake_iscsi_lun_path'), mock.call('fake_image_conversion_dir')]) @mock.patch.object(windows_iscsi.WindowsISCSIDriver, '_get_portals') def test_check_for_setup_error(self, mock_get_portals): self._driver.check_for_setup_error() mock_get_portals.assert_called_once_with() @ddt.data(True, False) def test_get_portals(self, portals_available=True): iscsi_port = mock.sentinel.iscsi_port available_ips = ['fake_ip0', 'fake_ip1', 'fake_unrequested_ip'] requested_ips = available_ips[:-1] + ['fake_inexistent_ips'] available_portals = ([":".join([ip_addr, str(iscsi_port)]) for ip_addr in available_ips] if portals_available else []) self._driver.configuration = mock.Mock() self._driver.configuration.target_port = iscsi_port self._driver.configuration.target_ip_address = requested_ips[0] self._driver.configuration.iscsi_secondary_ip_addresses = ( requested_ips[1:]) self._driver._tgt_utils.get_portal_locations.return_value = ( available_portals) if portals_available: portals = self._driver._get_portals() self.assertEqual(set(available_portals[:-1]), set(portals)) else: self.assertRaises(exception.VolumeDriverException, self._driver._get_portals) self._driver._tgt_utils.get_portal_locations.assert_called_once_with( available_only=True, fail_if_none_found=True) @ddt.data(True, False) @mock.patch.object(windows_iscsi.WindowsISCSIDriver, '_get_portals') @mock.patch.object(windows_iscsi.WindowsISCSIDriver, '_get_target_name') def test_get_host_information(self, multipath, mock_get_target_name, mock_get_portals): tgt_utils = self._driver._tgt_utils fake_auth_meth = 'CHAP' fake_chap_username = 'fake_chap_username' fake_chap_password = 'fake_chap_password' fake_target_iqn = 'fake_target_iqn' fake_host_info = {'target_iqn': 'fake_target_iqn', 'fake_prop': 'fake_value'} fake_provider_auth = "%s %s %s" % (fake_auth_meth, fake_chap_username, fake_chap_password) fake_portals = [mock.sentinel.portal_location0, mock.sentinel.portal_location1] volume = fake_volume.fake_volume_obj(mock.sentinel.context, provider_auth=fake_provider_auth) mock_get_target_name.return_value = mock.sentinel.target_name mock_get_portals.return_value = fake_portals tgt_utils.get_target_information.return_value = fake_host_info expected_host_info = dict(fake_host_info, auth_method=fake_auth_meth, auth_username=fake_chap_username, auth_password=fake_chap_password, target_discovered=False, target_portal=fake_portals[0], target_lun=0, volume_id=volume.id) if multipath: expected_host_info['target_portals'] = fake_portals expected_host_info['target_iqns'] = [fake_target_iqn] * 2 expected_host_info['target_luns'] = [0] * 2 host_info = self._driver._get_host_information(volume, multipath) self.assertEqual(expected_host_info, host_info) mock_get_target_name.assert_called_once_with(volume) mock_get_portals.assert_called_once_with() tgt_utils.get_target_information.assert_called_once_with( mock.sentinel.target_name) @mock.patch.object(windows_iscsi.WindowsISCSIDriver, '_get_host_information') def test_initialize_connection(self, mock_get_host_info): tgt_utils = self._driver._tgt_utils volume = fake_volume.fake_volume_obj(mock.sentinel.fake_context) fake_initiator = db_fakes.get_fake_connector_info() fake_initiator['multipath'] = mock.sentinel.multipath fake_host_info = {'fake_host_prop': 'fake_value'} mock_get_host_info.return_value = fake_host_info expected_conn_info = {'driver_volume_type': 'iscsi', 'data': fake_host_info} conn_info = self._driver.initialize_connection(volume, fake_initiator) self.assertEqual(expected_conn_info, conn_info) mock_get_host_info.assert_called_once_with( volume, mock.sentinel.multipath) mock_associate = tgt_utils.associate_initiator_with_iscsi_target mock_associate.assert_called_once_with( fake_initiator['initiator'], volume.provider_location) def test_terminate_connection(self): volume = fake_volume.fake_volume_obj(mock.sentinel.fake_context) fake_initiator = db_fakes.get_fake_connector_info() self._driver.terminate_connection(volume, fake_initiator) self._driver._tgt_utils.deassociate_initiator.assert_called_once_with( fake_initiator['initiator'], volume.provider_location) @mock.patch.object(windows_iscsi.WindowsISCSIDriver, 'local_path') def test_create_volume(self, mock_local_path): volume = fake_volume.fake_volume_obj(mock.sentinel.fake_context) self._driver.create_volume(volume) mock_local_path.assert_called_once_with(volume) self._driver._tgt_utils.create_wt_disk.assert_called_once_with( mock_local_path.return_value, volume.name, size_mb=volume.size * 1024) def test_local_path(self): volume = fake_volume.fake_volume_obj(mock.sentinel.fake_context) fake_lun_path = 'fake_lun_path' self.flags(windows_iscsi_lun_path=fake_lun_path) disk_format = 'vhd' mock_get_fmt = self._driver._tgt_utils.get_supported_disk_format mock_get_fmt.return_value = disk_format disk_path = self._driver.local_path(volume) expected_fname = "%s.%s" % (volume.name, disk_format) expected_disk_path = os.path.join(fake_lun_path, expected_fname) self.assertEqual(expected_disk_path, disk_path) mock_get_fmt.assert_called_once_with() @mock.patch.object(windows_iscsi.WindowsISCSIDriver, 'local_path') @mock.patch.object(fileutils, 'delete_if_exists') def test_delete_volume(self, mock_delete_if_exists, mock_local_path): volume = fake_volume.fake_volume_obj(mock.sentinel.fake_context) self._driver.delete_volume(volume) mock_local_path.assert_called_once_with(volume) self._driver._tgt_utils.remove_wt_disk.assert_called_once_with( volume.name) mock_delete_if_exists.assert_called_once_with( mock_local_path.return_value) def test_create_snapshot(self): volume = fake_volume.fake_volume_obj(context.get_admin_context()) snapshot = fake_snapshot.fake_snapshot_obj(context.get_admin_context(), volume_id=volume.id) snapshot.volume = volume self._driver.create_snapshot(snapshot) self._driver._tgt_utils.create_snapshot.assert_called_once_with( snapshot.volume_name, snapshot.name) @mock.patch.object(windows_iscsi.WindowsISCSIDriver, 'local_path') def test_create_volume_from_snapshot(self, mock_local_path): volume = fake_volume.fake_volume_obj(context.get_admin_context()) snapshot = fake_snapshot.fake_snapshot_obj(context.get_admin_context()) snapshot.volume = volume self._driver.create_volume_from_snapshot(volume, snapshot) self._driver._tgt_utils.export_snapshot.assert_called_once_with( snapshot.name, mock_local_path.return_value) self._driver._tgt_utils.import_wt_disk.assert_called_once_with( mock_local_path.return_value, volume.name) def test_delete_snapshot(self): snapshot = fake_snapshot.fake_snapshot_obj(context.get_admin_context()) self._driver.delete_snapshot(snapshot) self._driver._tgt_utils.delete_snapshot.assert_called_once_with( snapshot.name) def test_get_target_name(self): volume = fake_volume.fake_volume_obj(mock.sentinel.fake_context) expected_target_name = "%s%s" % ( self._driver.configuration.target_prefix, volume.name) target_name = self._driver._get_target_name(volume) self.assertEqual(expected_target_name, target_name) @mock.patch.object(windows_iscsi.WindowsISCSIDriver, '_get_target_name') @mock.patch.object(windows_iscsi.volume_utils, 'generate_username') @mock.patch.object(windows_iscsi.volume_utils, 'generate_password') def test_create_export(self, mock_generate_password, mock_generate_username, mock_get_target_name): tgt_utils = self._driver._tgt_utils volume = fake_volume.fake_volume_obj(mock.sentinel.fake_context) self._driver.configuration.chap_username = None self._driver.configuration.chap_password = None self._driver.configuration.use_chap_auth = True fake_chap_username = 'fake_chap_username' fake_chap_password = 'fake_chap_password' mock_get_target_name.return_value = mock.sentinel.target_name mock_generate_username.return_value = fake_chap_username mock_generate_password.return_value = fake_chap_password tgt_utils.iscsi_target_exists.return_value = False vol_updates = self._driver.create_export(mock.sentinel.context, volume, mock.sentinel.connector) mock_get_target_name.assert_called_once_with(volume) tgt_utils.iscsi_target_exists.assert_called_once_with( mock.sentinel.target_name) tgt_utils.set_chap_credentials.assert_called_once_with( mock.sentinel.target_name, fake_chap_username, fake_chap_password) tgt_utils.add_disk_to_target.assert_called_once_with( volume.name, mock.sentinel.target_name) expected_provider_auth = ' '.join(('CHAP', fake_chap_username, fake_chap_password)) expected_vol_updates = dict( provider_location=mock.sentinel.target_name, provider_auth=expected_provider_auth) self.assertEqual(expected_vol_updates, vol_updates) @mock.patch.object(windows_iscsi.WindowsISCSIDriver, '_get_target_name') def test_remove_export(self, mock_get_target_name): volume = fake_volume.fake_volume_obj(mock.sentinel.fake_context) self._driver.remove_export(mock.sentinel.context, volume) mock_get_target_name.assert_called_once_with(volume) self._driver._tgt_utils.delete_iscsi_target.assert_called_once_with( mock_get_target_name.return_value) @mock.patch.object(windows_iscsi.WindowsISCSIDriver, 'local_path') @mock.patch.object(image_utils, 'temporary_file') @mock.patch.object(image_utils, 'fetch_to_vhd') @mock.patch('os.unlink') def test_copy_image_to_volume(self, mock_unlink, mock_fetch_to_vhd, mock_tmp_file, mock_local_path): tgt_utils = self._driver._tgt_utils volume = fake_volume.fake_volume_obj(mock.sentinel.fake_context) mock_tmp_file.return_value.__enter__.return_value = ( mock.sentinel.tmp_vhd_path) mock_local_path.return_value = mock.sentinel.vol_vhd_path self._driver.copy_image_to_volume(mock.sentinel.context, volume, mock.sentinel.image_service, mock.sentinel.image_id) mock_local_path.assert_called_once_with(volume) mock_tmp_file.assert_called_once_with(suffix='.vhd') image_utils.fetch_to_vhd.assert_called_once_with( mock.sentinel.context, mock.sentinel.image_service, mock.sentinel.image_id, mock.sentinel.tmp_vhd_path, self._driver.configuration.volume_dd_blocksize) mock_unlink.assert_called_once_with(mock.sentinel.vol_vhd_path) self._driver._vhdutils.convert_vhd.assert_called_once_with( mock.sentinel.tmp_vhd_path, mock.sentinel.vol_vhd_path, tgt_utils.get_supported_vhd_type.return_value) self._driver._vhdutils.resize_vhd.assert_called_once_with( mock.sentinel.vol_vhd_path, volume.size * units.Gi, is_file_max_size=False) tgt_utils.change_wt_disk_status.assert_has_calls( [mock.call(volume.name, enabled=False), mock.call(volume.name, enabled=True)]) @mock.patch.object(windows_iscsi.uuidutils, 'generate_uuid') def test_temporary_snapshot(self, mock_generate_uuid): tgt_utils = self._driver._tgt_utils mock_generate_uuid.return_value = mock.sentinel.snap_uuid expected_snap_name = '%s-tmp-snapshot-%s' % ( mock.sentinel.volume_name, mock.sentinel.snap_uuid) with self._driver._temporary_snapshot( mock.sentinel.volume_name) as snap_name: self.assertEqual(expected_snap_name, snap_name) tgt_utils.create_snapshot.assert_called_once_with( mock.sentinel.volume_name, expected_snap_name) tgt_utils.delete_snapshot.assert_called_once_with( expected_snap_name) @mock.patch.object(windows_iscsi.WindowsISCSIDriver, '_temporary_snapshot') @mock.patch.object(image_utils, 'upload_volume') @mock.patch.object(fileutils, 'delete_if_exists') def test_copy_volume_to_image(self, mock_delete_if_exists, mock_upload_volume, mock_tmp_snap): tgt_utils = self._driver._tgt_utils disk_format = 'vhd' fake_image_meta = db_fakes.get_fake_image_meta() fake_volume = test_utils.create_volume( self._context, volume_type_id=fake.VOLUME_TYPE_ID, updated_at=self.updated_at) extra_specs = { 'image_service:store_id': 'fake-store' } test_utils.create_volume_type(self._context.elevated(), id=fake.VOLUME_TYPE_ID, name="test_type", extra_specs=extra_specs) fake_img_conv_dir = 'fake_img_conv_dir' self.flags(image_conversion_dir=fake_img_conv_dir) tgt_utils.get_supported_disk_format.return_value = disk_format mock_tmp_snap.return_value.__enter__.return_value = ( mock.sentinel.tmp_snap_name) expected_tmp_vhd_path = os.path.join( fake_img_conv_dir, fake_image_meta['id'] + '.' + disk_format) self._driver.copy_volume_to_image( mock.sentinel.context, fake_volume, mock.sentinel.image_service, fake_image_meta) mock_tmp_snap.assert_called_once_with(fake_volume.name) tgt_utils.export_snapshot.assert_called_once_with( mock.sentinel.tmp_snap_name, expected_tmp_vhd_path) mock_upload_volume.assert_called_once_with( mock.sentinel.context, mock.sentinel.image_service, fake_image_meta, expected_tmp_vhd_path, volume_format='vhd', store_id='fake-store', base_image_ref=None, compress=True, run_as_root=True) mock_delete_if_exists.assert_called_once_with( expected_tmp_vhd_path) @mock.patch.object(windows_iscsi.WindowsISCSIDriver, '_temporary_snapshot') @mock.patch.object(windows_iscsi.WindowsISCSIDriver, 'local_path') def test_create_cloned_volume(self, mock_local_path, mock_tmp_snap): tgt_utils = self._driver._tgt_utils volume = fake_volume.fake_volume_obj(mock.sentinel.fake_context) src_volume = fake_volume.fake_volume_obj(mock.sentinel.fake_context) mock_tmp_snap.return_value.__enter__.return_value = ( mock.sentinel.tmp_snap_name) mock_local_path.return_value = mock.sentinel.vol_vhd_path self._driver.create_cloned_volume(volume, src_volume) mock_tmp_snap.assert_called_once_with(src_volume.name) tgt_utils.export_snapshot.assert_called_once_with( mock.sentinel.tmp_snap_name, mock.sentinel.vol_vhd_path) self._driver._vhdutils.resize_vhd.assert_called_once_with( mock.sentinel.vol_vhd_path, volume.size * units.Gi, is_file_max_size=False) tgt_utils.import_wt_disk.assert_called_once_with( mock.sentinel.vol_vhd_path, volume.name) @mock.patch('os.path.splitdrive') def test_get_capacity_info(self, mock_splitdrive): mock_splitdrive.return_value = (mock.sentinel.drive, mock.sentinel.path_tail) fake_size_gb = 2 fake_free_space_gb = 1 self._driver._hostutils.get_volume_info.return_value = ( fake_size_gb * units.Gi, fake_free_space_gb * units.Gi) total_gb, free_gb = self._driver._get_capacity_info() self.assertEqual(fake_size_gb, total_gb) self.assertEqual(fake_free_space_gb, free_gb) self._driver._hostutils.get_volume_info.assert_called_once_with( mock.sentinel.drive) mock_splitdrive.assert_called_once_with('fake_iscsi_lun_path') @mock.patch.object(windows_iscsi.WindowsISCSIDriver, '_get_capacity_info') def test_update_volume_stats(self, mock_get_capacity_info): mock_get_capacity_info.return_value = ( mock.sentinel.size_gb, mock.sentinel.free_space_gb) self.flags(volume_backend_name='volume_backend_name') self.flags(reserved_percentage=10) expected_volume_stats = dict( volume_backend_name='volume_backend_name', vendor_name='Microsoft', driver_version=self._driver.VERSION, storage_protocol='iSCSI', total_capacity_gb=mock.sentinel.size_gb, free_capacity_gb=mock.sentinel.free_space_gb, reserved_percentage=10, QoS_support=False) self._driver._update_volume_stats() self.assertEqual(expected_volume_stats, self._driver._stats) def test_extend_volume(self): volume = fake_volume.fake_volume_obj(mock.sentinel.fake_context) new_size_gb = 2 expected_additional_sz_mb = 1024 self._driver.extend_volume(volume, new_size_gb) self._driver._tgt_utils.extend_wt_disk.assert_called_once_with( volume.name, expected_additional_sz_mb)