# Copyright 2014 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. import mock from os_brick.initiator import connector from oslo_config import cfg from oslo_utils import units from nova import exception from nova import test from nova.tests.unit import fake_block_device from nova.tests.unit.virt.hyperv import test_base from nova.virt.hyperv import constants from nova.virt.hyperv import volumeops CONF = cfg.CONF connection_data = {'volume_id': 'fake_vol_id', 'target_lun': mock.sentinel.fake_lun, 'target_iqn': mock.sentinel.fake_iqn, 'target_portal': mock.sentinel.fake_portal, 'auth_method': 'chap', 'auth_username': mock.sentinel.fake_user, 'auth_password': mock.sentinel.fake_pass} def get_fake_block_dev_info(): return {'block_device_mapping': [ fake_block_device.AnonFakeDbBlockDeviceDict({'source_type': 'volume'})] } def get_fake_connection_info(**kwargs): return {'data': dict(connection_data, **kwargs), 'serial': mock.sentinel.serial} class VolumeOpsTestCase(test_base.HyperVBaseTestCase): """Unit tests for VolumeOps class.""" def setUp(self): super(VolumeOpsTestCase, self).setUp() self._volumeops = volumeops.VolumeOps() self._volumeops._volutils = mock.MagicMock() self._volumeops._vmutils = mock.Mock() def test_get_volume_driver(self): fake_conn_info = {'driver_volume_type': mock.sentinel.fake_driver_type} self._volumeops.volume_drivers[mock.sentinel.fake_driver_type] = ( mock.sentinel.fake_driver) result = self._volumeops._get_volume_driver( connection_info=fake_conn_info) self.assertEqual(mock.sentinel.fake_driver, result) def test_get_volume_driver_exception(self): fake_conn_info = {'driver_volume_type': 'fake_driver'} self.assertRaises(exception.VolumeDriverNotFound, self._volumeops._get_volume_driver, connection_info=fake_conn_info) @mock.patch.object(volumeops.VolumeOps, 'attach_volume') def test_attach_volumes(self, mock_attach_volume): block_device_info = get_fake_block_dev_info() self._volumeops.attach_volumes( block_device_info['block_device_mapping'], mock.sentinel.instance_name) mock_attach_volume.assert_called_once_with( block_device_info['block_device_mapping'][0]['connection_info'], mock.sentinel.instance_name) def test_fix_instance_volume_disk_paths_empty_bdm(self): self._volumeops.fix_instance_volume_disk_paths( mock.sentinel.instance_name, block_device_info={}) self.assertFalse( self._volumeops._vmutils.get_vm_physical_disk_mapping.called) @mock.patch.object(volumeops.VolumeOps, 'get_disk_path_mapping') def test_fix_instance_volume_disk_paths(self, mock_get_disk_path_mapping): block_device_info = get_fake_block_dev_info() mock_disk1 = { 'mounted_disk_path': mock.sentinel.mounted_disk1_path, 'resource_path': mock.sentinel.resource1_path } mock_disk2 = { 'mounted_disk_path': mock.sentinel.mounted_disk2_path, 'resource_path': mock.sentinel.resource2_path } mock_vm_disk_mapping = { mock.sentinel.disk1_serial: mock_disk1, mock.sentinel.disk2_serial: mock_disk2 } # In this case, only the first disk needs to be updated. mock_phys_disk_path_mapping = { mock.sentinel.disk1_serial: mock.sentinel.actual_disk1_path, mock.sentinel.disk2_serial: mock.sentinel.mounted_disk2_path } vmutils = self._volumeops._vmutils vmutils.get_vm_physical_disk_mapping.return_value = ( mock_vm_disk_mapping) mock_get_disk_path_mapping.return_value = mock_phys_disk_path_mapping self._volumeops.fix_instance_volume_disk_paths( mock.sentinel.instance_name, block_device_info) vmutils.get_vm_physical_disk_mapping.assert_called_once_with( mock.sentinel.instance_name) mock_get_disk_path_mapping.assert_called_once_with( block_device_info) vmutils.set_disk_host_res.assert_called_once_with( mock.sentinel.resource1_path, mock.sentinel.actual_disk1_path) @mock.patch.object(volumeops.VolumeOps, '_get_volume_driver') def test_disconnect_volumes(self, mock_get_volume_driver): block_device_info = get_fake_block_dev_info() block_device_mapping = block_device_info['block_device_mapping'] fake_volume_driver = mock_get_volume_driver.return_value self._volumeops.disconnect_volumes(block_device_info) fake_volume_driver.disconnect_volume.assert_called_once_with( block_device_mapping[0]['connection_info']) @mock.patch('time.sleep') @mock.patch.object(volumeops.VolumeOps, '_get_volume_driver') def _test_attach_volume(self, mock_get_volume_driver, mock_sleep, attach_failed): fake_conn_info = get_fake_connection_info( qos_specs=mock.sentinel.qos_specs) fake_volume_driver = mock_get_volume_driver.return_value expected_try_count = 1 if attach_failed: expected_try_count += CONF.hyperv.volume_attach_retry_count fake_volume_driver.set_disk_qos_specs.side_effect = ( test.TestingException) self.assertRaises(exception.VolumeAttachFailed, self._volumeops.attach_volume, fake_conn_info, mock.sentinel.inst_name, mock.sentinel.disk_bus) else: self._volumeops.attach_volume( fake_conn_info, mock.sentinel.inst_name, mock.sentinel.disk_bus) mock_get_volume_driver.assert_any_call( fake_conn_info) fake_volume_driver.attach_volume.assert_has_calls( [mock.call(fake_conn_info, mock.sentinel.inst_name, mock.sentinel.disk_bus)] * expected_try_count) fake_volume_driver.set_disk_qos_specs.assert_has_calls( [mock.call(fake_conn_info, mock.sentinel.qos_specs)] * expected_try_count) if attach_failed: fake_volume_driver.disconnect_volume.assert_called_once_with( fake_conn_info) mock_sleep.assert_has_calls( [mock.call(CONF.hyperv.volume_attach_retry_interval)] * CONF.hyperv.volume_attach_retry_count) else: mock_sleep.assert_not_called() def test_attach_volume(self): self._test_attach_volume(attach_failed=False) def test_attach_volume_exc(self): self._test_attach_volume(attach_failed=True) @mock.patch.object(volumeops.VolumeOps, '_get_volume_driver') def test_disconnect_volume(self, mock_get_volume_driver): fake_volume_driver = mock_get_volume_driver.return_value self._volumeops.disconnect_volume(mock.sentinel.conn_info) mock_get_volume_driver.assert_called_once_with( mock.sentinel.conn_info) fake_volume_driver.disconnect_volume.assert_called_once_with( mock.sentinel.conn_info) @mock.patch.object(volumeops.VolumeOps, '_get_volume_driver') def test_detach_volume(self, mock_get_volume_driver): fake_volume_driver = mock_get_volume_driver.return_value fake_conn_info = {'data': 'fake_conn_info_data'} self._volumeops.detach_volume(fake_conn_info, mock.sentinel.inst_name) mock_get_volume_driver.assert_called_once_with( fake_conn_info) fake_volume_driver.detach_volume.assert_called_once_with( fake_conn_info, mock.sentinel.inst_name) fake_volume_driver.disconnect_volume.assert_called_once_with( fake_conn_info) @mock.patch.object(connector, 'get_connector_properties') def test_get_volume_connector(self, mock_get_connector): conn = self._volumeops.get_volume_connector() mock_get_connector.assert_called_once_with( root_helper=None, my_ip=CONF.my_block_storage_ip, multipath=CONF.hyperv.use_multipath_io, enforce_multipath=True, host=CONF.host) self.assertEqual(mock_get_connector.return_value, conn) @mock.patch.object(volumeops.VolumeOps, '_get_volume_driver') def test_connect_volumes(self, mock_get_volume_driver): block_device_info = get_fake_block_dev_info() self._volumeops.connect_volumes(block_device_info) init_vol_conn = ( mock_get_volume_driver.return_value.connect_volume) init_vol_conn.assert_called_once_with( block_device_info['block_device_mapping'][0]['connection_info']) @mock.patch.object(volumeops.VolumeOps, 'get_disk_resource_path') def test_get_disk_path_mapping(self, mock_get_disk_path): block_device_info = get_fake_block_dev_info() block_device_mapping = block_device_info['block_device_mapping'] fake_conn_info = get_fake_connection_info() block_device_mapping[0]['connection_info'] = fake_conn_info mock_get_disk_path.return_value = mock.sentinel.disk_path resulted_disk_path_mapping = self._volumeops.get_disk_path_mapping( block_device_info) mock_get_disk_path.assert_called_once_with(fake_conn_info) expected_disk_path_mapping = { mock.sentinel.serial: mock.sentinel.disk_path } self.assertEqual(expected_disk_path_mapping, resulted_disk_path_mapping) @mock.patch.object(volumeops.VolumeOps, '_get_volume_driver') def test_get_disk_resource_path(self, mock_get_volume_driver): fake_conn_info = get_fake_connection_info() fake_volume_driver = mock_get_volume_driver.return_value resulted_disk_path = self._volumeops.get_disk_resource_path( fake_conn_info) mock_get_volume_driver.assert_called_once_with(fake_conn_info) get_mounted_disk = fake_volume_driver.get_disk_resource_path get_mounted_disk.assert_called_once_with(fake_conn_info) self.assertEqual(get_mounted_disk.return_value, resulted_disk_path) def test_bytes_per_sec_to_iops(self): no_bytes = 15 * units.Ki expected_iops = 2 resulted_iops = self._volumeops.bytes_per_sec_to_iops(no_bytes) self.assertEqual(expected_iops, resulted_iops) @mock.patch.object(volumeops.LOG, 'warning') def test_validate_qos_specs(self, mock_warning): supported_qos_specs = [mock.sentinel.spec1, mock.sentinel.spec2] requested_qos_specs = {mock.sentinel.spec1: mock.sentinel.val, mock.sentinel.spec3: mock.sentinel.val2} self._volumeops.validate_qos_specs(requested_qos_specs, supported_qos_specs) self.assertTrue(mock_warning.called) class BaseVolumeDriverTestCase(test_base.HyperVBaseTestCase): """Unit tests for Hyper-V BaseVolumeDriver class.""" def setUp(self): super(BaseVolumeDriverTestCase, self).setUp() self._base_vol_driver = volumeops.BaseVolumeDriver() self._base_vol_driver._diskutils = mock.Mock() self._base_vol_driver._vmutils = mock.Mock() self._base_vol_driver._migrutils = mock.Mock() self._base_vol_driver._conn = mock.Mock() self._vmutils = self._base_vol_driver._vmutils self._migrutils = self._base_vol_driver._migrutils self._diskutils = self._base_vol_driver._diskutils self._conn = self._base_vol_driver._conn @mock.patch.object(connector.InitiatorConnector, 'factory') def test_connector(self, mock_conn_factory): self._base_vol_driver._conn = None self._base_vol_driver._protocol = mock.sentinel.protocol self._base_vol_driver._extra_connector_args = dict( fake_conn_arg=mock.sentinel.conn_val) conn = self._base_vol_driver._connector self.assertEqual(mock_conn_factory.return_value, conn) mock_conn_factory.assert_called_once_with( protocol=mock.sentinel.protocol, root_helper=None, use_multipath=CONF.hyperv.use_multipath_io, device_scan_attempts=CONF.hyperv.mounted_disk_query_retry_count, device_scan_interval=( CONF.hyperv.mounted_disk_query_retry_interval), **self._base_vol_driver._extra_connector_args) def test_connect_volume(self): conn_info = get_fake_connection_info() dev_info = self._base_vol_driver.connect_volume(conn_info) expected_dev_info = self._conn.connect_volume.return_value self.assertEqual(expected_dev_info, dev_info) self._conn.connect_volume.assert_called_once_with( conn_info['data']) def test_disconnect_volume(self): conn_info = get_fake_connection_info() self._base_vol_driver.disconnect_volume(conn_info) self._conn.disconnect_volume.assert_called_once_with( conn_info['data']) @mock.patch.object(volumeops.BaseVolumeDriver, '_get_disk_res_path') def _test_get_disk_resource_path_by_conn_info(self, mock_get_disk_res_path, disk_found=True): conn_info = get_fake_connection_info() mock_vol_paths = [mock.sentinel.disk_path] if disk_found else [] self._conn.get_volume_paths.return_value = mock_vol_paths if disk_found: disk_res_path = self._base_vol_driver.get_disk_resource_path( conn_info) self._conn.get_volume_paths.assert_called_once_with( conn_info['data']) self.assertEqual(mock_get_disk_res_path.return_value, disk_res_path) mock_get_disk_res_path.assert_called_once_with( mock.sentinel.disk_path) else: self.assertRaises(exception.DiskNotFound, self._base_vol_driver.get_disk_resource_path, conn_info) def test_get_existing_disk_res_path(self): self._test_get_disk_resource_path_by_conn_info() def test_get_unfound_disk_res_path(self): self._test_get_disk_resource_path_by_conn_info(disk_found=False) def test_get_block_dev_res_path(self): self._base_vol_driver._is_block_dev = True mock_get_dev_number = ( self._diskutils.get_device_number_from_device_name) mock_get_dev_number.return_value = mock.sentinel.dev_number self._vmutils.get_mounted_disk_by_drive_number.return_value = ( mock.sentinel.disk_path) disk_path = self._base_vol_driver._get_disk_res_path( mock.sentinel.dev_name) mock_get_dev_number.assert_called_once_with(mock.sentinel.dev_name) self._vmutils.get_mounted_disk_by_drive_number.assert_called_once_with( mock.sentinel.dev_number) self.assertEqual(mock.sentinel.disk_path, disk_path) def test_get_virt_disk_res_path(self): # For virtual disk images, we expect the resource path to be the # actual image path, as opposed to passthrough disks, in which case we # need the Msvm_DiskDrive resource path when attaching it to a VM. self._base_vol_driver._is_block_dev = False path = self._base_vol_driver._get_disk_res_path( mock.sentinel.disk_path) self.assertEqual(mock.sentinel.disk_path, path) @mock.patch.object(volumeops.BaseVolumeDriver, '_get_disk_res_path') @mock.patch.object(volumeops.BaseVolumeDriver, '_get_disk_ctrl_and_slot') @mock.patch.object(volumeops.BaseVolumeDriver, 'connect_volume') def _test_attach_volume(self, mock_connect_volume, mock_get_disk_ctrl_and_slot, mock_get_disk_res_path, is_block_dev=True): connection_info = get_fake_connection_info() self._base_vol_driver._is_block_dev = is_block_dev mock_connect_volume.return_value = dict(path=mock.sentinel.raw_path) mock_get_disk_res_path.return_value = ( mock.sentinel.disk_path) mock_get_disk_ctrl_and_slot.return_value = ( mock.sentinel.ctrller_path, mock.sentinel.slot) self._base_vol_driver.attach_volume( connection_info=connection_info, instance_name=mock.sentinel.instance_name, disk_bus=mock.sentinel.disk_bus) if is_block_dev: self._vmutils.attach_volume_to_controller.assert_called_once_with( mock.sentinel.instance_name, mock.sentinel.ctrller_path, mock.sentinel.slot, mock.sentinel.disk_path, serial=connection_info['serial']) else: self._vmutils.attach_drive.assert_called_once_with( mock.sentinel.instance_name, mock.sentinel.disk_path, mock.sentinel.ctrller_path, mock.sentinel.slot) mock_get_disk_res_path.assert_called_once_with( mock.sentinel.raw_path) mock_get_disk_ctrl_and_slot.assert_called_once_with( mock.sentinel.instance_name, mock.sentinel.disk_bus) def test_attach_volume_image_file(self): self._test_attach_volume(is_block_dev=False) def test_attach_volume_block_dev(self): self._test_attach_volume(is_block_dev=True) def test_detach_volume_planned_vm(self): self._base_vol_driver.detach_volume(mock.sentinel.connection_info, mock.sentinel.inst_name) self._vmutils.detach_vm_disk.assert_not_called() @mock.patch.object(volumeops.BaseVolumeDriver, 'get_disk_resource_path') def test_detach_volume(self, mock_get_disk_resource_path): self._migrutils.planned_vm_exists.return_value = False connection_info = get_fake_connection_info() self._base_vol_driver.detach_volume(connection_info, mock.sentinel.instance_name) mock_get_disk_resource_path.assert_called_once_with( connection_info) self._vmutils.detach_vm_disk.assert_called_once_with( mock.sentinel.instance_name, mock_get_disk_resource_path.return_value, is_physical=self._base_vol_driver._is_block_dev) def test_get_disk_ctrl_and_slot_ide(self): ctrl, slot = self._base_vol_driver._get_disk_ctrl_and_slot( mock.sentinel.instance_name, disk_bus=constants.CTRL_TYPE_IDE) expected_ctrl = self._vmutils.get_vm_ide_controller.return_value expected_slot = 0 self._vmutils.get_vm_ide_controller.assert_called_once_with( mock.sentinel.instance_name, 0) self.assertEqual(expected_ctrl, ctrl) self.assertEqual(expected_slot, slot) def test_get_disk_ctrl_and_slot_scsi(self): ctrl, slot = self._base_vol_driver._get_disk_ctrl_and_slot( mock.sentinel.instance_name, disk_bus=constants.CTRL_TYPE_SCSI) expected_ctrl = self._vmutils.get_vm_scsi_controller.return_value expected_slot = ( self._vmutils.get_free_controller_slot.return_value) self._vmutils.get_vm_scsi_controller.assert_called_once_with( mock.sentinel.instance_name) self._vmutils.get_free_controller_slot( self._vmutils.get_vm_scsi_controller.return_value) self.assertEqual(expected_ctrl, ctrl) self.assertEqual(expected_slot, slot) def test_set_disk_qos_specs(self): # This base method is a noop, we'll just make sure # it doesn't error out. self._base_vol_driver.set_disk_qos_specs( mock.sentinel.conn_info, mock.sentinel.disk_qos_spes) class ISCSIVolumeDriverTestCase(test_base.HyperVBaseTestCase): """Unit tests for Hyper-V BaseVolumeDriver class.""" def test_extra_conn_args(self): fake_iscsi_initiator = ( 'PCI\\VEN_1077&DEV_2031&SUBSYS_17E8103C&REV_02\\' '4&257301f0&0&0010_0') self.flags(iscsi_initiator_list=[fake_iscsi_initiator], group='hyperv') expected_extra_conn_args = dict( initiator_list=[fake_iscsi_initiator]) vol_driver = volumeops.ISCSIVolumeDriver() self.assertEqual(expected_extra_conn_args, vol_driver._extra_connector_args) class SMBFSVolumeDriverTestCase(test_base.HyperVBaseTestCase): """Unit tests for the Hyper-V SMBFSVolumeDriver class.""" _FAKE_EXPORT_PATH = '//ip/share/' _FAKE_CONN_INFO = get_fake_connection_info(export=_FAKE_EXPORT_PATH) def setUp(self): super(SMBFSVolumeDriverTestCase, self).setUp() self._volume_driver = volumeops.SMBFSVolumeDriver() self._volume_driver._conn = mock.Mock() self._conn = self._volume_driver._conn def test_get_export_path(self): export_path = self._volume_driver._get_export_path( self._FAKE_CONN_INFO) expected_path = self._FAKE_EXPORT_PATH.replace('/', '\\') self.assertEqual(expected_path, export_path) @mock.patch.object(volumeops.BaseVolumeDriver, 'attach_volume') def test_attach_volume(self, mock_attach): # The tested method will just apply a lock before calling # the superclass method. self._volume_driver.attach_volume( self._FAKE_CONN_INFO, mock.sentinel.instance_name, disk_bus=mock.sentinel.disk_bus) mock_attach.assert_called_once_with( self._FAKE_CONN_INFO, mock.sentinel.instance_name, disk_bus=mock.sentinel.disk_bus) @mock.patch.object(volumeops.BaseVolumeDriver, 'detach_volume') def test_detach_volume(self, mock_detach): self._volume_driver.detach_volume( self._FAKE_CONN_INFO, instance_name=mock.sentinel.instance_name) mock_detach.assert_called_once_with( self._FAKE_CONN_INFO, instance_name=mock.sentinel.instance_name) @mock.patch.object(volumeops.VolumeOps, 'bytes_per_sec_to_iops') @mock.patch.object(volumeops.VolumeOps, 'validate_qos_specs') @mock.patch.object(volumeops.BaseVolumeDriver, 'get_disk_resource_path') def test_set_disk_qos_specs(self, mock_get_disk_path, mock_validate_qos_specs, mock_bytes_per_sec_to_iops): fake_total_bytes_sec = 8 fake_total_iops_sec = 1 storage_qos_specs = {'total_bytes_sec': fake_total_bytes_sec} expected_supported_specs = ['total_iops_sec', 'total_bytes_sec'] mock_set_qos_specs = self._volume_driver._vmutils.set_disk_qos_specs mock_bytes_per_sec_to_iops.return_value = fake_total_iops_sec mock_get_disk_path.return_value = mock.sentinel.disk_path self._volume_driver.set_disk_qos_specs(self._FAKE_CONN_INFO, storage_qos_specs) mock_validate_qos_specs.assert_called_once_with( storage_qos_specs, expected_supported_specs) mock_bytes_per_sec_to_iops.assert_called_once_with( fake_total_bytes_sec) mock_get_disk_path.assert_called_once_with(self._FAKE_CONN_INFO) mock_set_qos_specs.assert_called_once_with( mock.sentinel.disk_path, fake_total_iops_sec) class RBDVolumeDriver(test_base.HyperVBaseTestCase): def test_get_vol_driver(self): self._volumeops = volumeops.VolumeOps() self._volumeops._volutils = mock.MagicMock() self._volumeops._vmutils = mock.Mock() connection_info = get_fake_connection_info() connection_info['driver_volume_type'] = 'rbd' drv = self._volumeops._get_volume_driver(connection_info) # Not much to test here. The Hyper-V driver volume attach code # is mostly generic and all the RBD related plumbing is handled # by os-brick. # # We'll just ensure that the RBD driver can be retrieved and that it # has the right fields. self.assertTrue(drv._is_block_dev) self.assertEqual('rbd', drv._protocol) # Hyper-V requires a virtual SCSI disk so we'll ask for a # local attach. self.assertEqual(dict(do_local_attach=True), drv._extra_connector_args)