Update device metadata when attaching disks/NICs

This change ensures that we're updating the instance device metadata
when attaching or detaching instance volumes or NICs.

Unlike the other Nova drivers, we're including all Cinder volumes
in the device metadata, not just the tagged ones. The reason is that
this allows Hyper-V guests to properly identify attached disk paths.

Note that the disk mountpoints (e.g. /dev/sdc) are ignored by the
Hyper-V driver. Also, we cannot override SCSI identifiers, unlike
the Libvirt driver, which will use volume IDs as SCSI IDs.

Closes-Bug: #1723431

Depends-On: I7a4c242c80b5b3562bf9513deb89071b5bb5bf28
Depends-On: I6f2c6ecf71d3254d64bc24574619c18184d3d055

Change-Id: I6853b1d167fe675435f3fb8f1fdb0cf0b63bed65
This commit is contained in:
Lucian Petrut 2017-10-09 17:12:56 +03:00
parent 328d087368
commit 8326d2c43c
9 changed files with 637 additions and 157 deletions

View File

@ -18,19 +18,27 @@ Handling of block device information and mapping
Module contains helper methods for dealing with block device information
"""
import itertools
import os
from nova import block_device
from nova import exception
from nova import objects
from nova.virt import block_device as driver_block_device
from nova.virt import configdrive
from nova.virt import driver
from os_win import constants as os_win_const
from os_win import exceptions as os_win_exc
from os_win import utilsfactory
from oslo_log import log as logging
from oslo_serialization import jsonutils
from compute_hyperv.i18n import _
from compute_hyperv.nova import constants
from compute_hyperv.nova import pathutils
from compute_hyperv.nova import volumeops
LOG = logging.getLogger(__name__)
class BlockDeviceInfoManager(object):
@ -48,48 +56,133 @@ class BlockDeviceInfoManager(object):
def __init__(self):
self._volops = volumeops.VolumeOps()
self._pathutils = pathutils.PathUtils()
self._vmutils = utilsfactory.get_vmutils()
@staticmethod
def _get_device_bus(bdm):
def _get_device_bus(ctrl_type, ctrl_addr, ctrl_slot):
"""Determines the device bus and it's hypervisor assigned address.
"""
if bdm['disk_bus'] == constants.CTRL_TYPE_SCSI:
address = ':'.join(['0', '0', str(bdm['drive_addr']),
str(bdm['ctrl_disk_addr'])])
if ctrl_type == constants.CTRL_TYPE_SCSI:
address = ':'.join(map(str, [0, 0, ctrl_addr, ctrl_slot]))
return objects.SCSIDeviceBus(address=address)
elif bdm['disk_bus'] == constants.CTRL_TYPE_IDE:
address = ':'.join([str(bdm['drive_addr']),
str(bdm['ctrl_disk_addr'])])
elif ctrl_type == constants.CTRL_TYPE_IDE:
address = ':'.join(map(str, [ctrl_addr, ctrl_slot]))
return objects.IDEDeviceBus(address=address)
def get_bdm_metadata(self, context, instance, block_device_info):
def _get_vol_bdm_attachment_info(self, bdm):
drv_vol_bdm = driver_block_device.convert_volume(bdm)
if not drv_vol_bdm:
return
connection_info = drv_vol_bdm['connection_info']
if not connection_info:
LOG.warning("Missing connection info for volume %s.",
bdm.volume_id)
return
attachment_info = self._volops.get_disk_attachment_info(
connection_info)
attachment_info['serial'] = connection_info['serial']
return attachment_info
def _get_eph_bdm_attachment_info(self, instance, bdm):
# When attaching ephemeral disks, we're setting this field so that
# we can map them with bdm objects.
connection_info = self.get_bdm_connection_info(bdm)
eph_filename = connection_info.get("eph_filename")
if not eph_filename:
LOG.warning("Missing ephemeral disk filename in "
"BDM connection info. BDM: %s", bdm)
return
eph_path = os.path.join(
self._pathutils.get_instance_dir(instance.name), eph_filename)
if not os.path.exists(eph_path):
LOG.warning("Could not find ephemeral disk %s.", eph_path)
return
return self._vmutils.get_disk_attachment_info(eph_path,
is_physical=False)
def _get_disk_metadata(self, instance, bdm):
attachment_info = None
if bdm.is_volume:
attachment_info = self._get_vol_bdm_attachment_info(bdm)
elif block_device.new_format_is_ephemeral(bdm):
attachment_info = self._get_eph_bdm_attachment_info(
instance, bdm)
if not attachment_info:
LOG.debug("No attachment info retrieved for bdm %s.", bdm)
return
tags = [bdm.tag] if bdm.tag else []
bus = self._get_device_bus(
attachment_info['controller_type'],
attachment_info['controller_addr'],
attachment_info['controller_slot'])
serial = attachment_info.get('serial')
return objects.DiskMetadata(bus=bus,
tags=tags,
serial=serial)
def get_bdm_metadata(self, context, instance):
"""Builds a metadata object for instance devices, that maps the user
provided tag to the hypervisor assigned device address.
"""
# block_device_info does not contain tags information.
bdm_obj_list = objects.BlockDeviceMappingList.get_by_instance_uuid(
bdms = objects.BlockDeviceMappingList.get_by_instance_uuid(
context, instance.uuid)
# create a map between BDM object and its device name.
bdm_obj_map = {bdm.device_name: bdm for bdm in bdm_obj_list}
bdm_metadata = []
for bdm in itertools.chain(block_device_info['block_device_mapping'],
block_device_info['ephemerals'],
[block_device_info['root_disk']]):
# NOTE(claudiub): ephemerals have device_name instead of
# mount_device.
device_name = bdm.get('mount_device') or bdm.get('device_name')
bdm_obj = bdm_obj_map.get(device_name)
if bdm_obj and 'tag' in bdm_obj and bdm_obj.tag:
bus = self._get_device_bus(bdm)
device = objects.DiskMetadata(bus=bus,
tags=[bdm_obj.tag])
bdm_metadata.append(device)
for bdm in bdms:
try:
device_metadata = self._get_disk_metadata(instance, bdm)
if device_metadata:
bdm_metadata.append(device_metadata)
except (exception.DiskNotFound, os_win_exc.DiskNotFound):
LOG.debug("Could not find disk attachment while "
"updating device metadata. It may have been "
"detached. BDM: %s", bdm)
return bdm_metadata
def set_volume_bdm_connection_info(self, context, instance,
connection_info):
# When attaching volumes to already existing instances, the connection
# info passed to the driver is not saved yet within the BDM table.
#
# Nova sets the volume id within the connection info using the
# 'serial' key.
volume_id = connection_info['serial']
bdm = objects.BlockDeviceMapping.get_by_volume_and_instance(
context, volume_id, instance.uuid)
bdm.connection_info = jsonutils.dumps(connection_info)
bdm.save()
@staticmethod
def get_bdm_connection_info(bdm):
# We're using the BDM 'connection_info' field to store ephemeral
# image information so that we can map them. In order to do so,
# we're using this helper.
# The ephemeral bdm object wrapper does not currently expose this
# field.
try:
conn_info = jsonutils.loads(bdm.connection_info)
except TypeError:
conn_info = {}
return conn_info
@staticmethod
def update_bdm_connection_info(bdm, **kwargs):
conn_info = BlockDeviceInfoManager.get_bdm_connection_info(bdm)
conn_info.update(**kwargs)
bdm.connection_info = jsonutils.dumps(conn_info)
bdm.save()
def _initialize_controller_slot_counter(self, instance, vm_gen):
# we have 2 IDE controllers, for a total of 4 slots
free_slots_by_device_type = {

View File

@ -21,6 +21,7 @@ import functools
import platform
import sys
from nova import context as nova_context
from nova import exception
from nova import image
from nova.virt import driver
@ -100,6 +101,8 @@ class HyperVDriver(driver.ComputeDriver):
"supports_migrate_to_same_host": False,
"supports_attach_interface": True,
"supports_device_tagging": True,
"supports_tagged_attach_interface": True,
"supports_tagged_attach_volume": True,
}
def __init__(self, virtapi):
@ -180,13 +183,20 @@ class HyperVDriver(driver.ComputeDriver):
def attach_volume(self, context, connection_info, instance, mountpoint,
disk_bus=None, device_type=None, encryption=None):
return self._volumeops.attach_volume(connection_info,
instance.name)
self._volumeops.attach_volume(context,
connection_info,
instance,
update_device_metadata=True)
def detach_volume(self, connection_info, instance, mountpoint,
encryption=None):
return self._volumeops.detach_volume(connection_info,
instance.name)
context = nova_context.get_admin_context()
# The nova compute manager only updates the device metadata in
# case of tagged devices. We're including untagged devices as well.
self._volumeops.detach_volume(context,
connection_info,
instance,
update_device_metadata=True)
def get_volume_connector(self, instance):
return self._volumeops.get_volume_connector()
@ -367,9 +377,10 @@ class HyperVDriver(driver.ComputeDriver):
self._imagecache.update(context, all_instances)
def attach_interface(self, context, instance, image_meta, vif):
return self._vmops.attach_interface(instance, vif)
self._vmops.attach_interface(context, instance, vif)
def detach_interface(self, context, instance, vif):
# The device metadata gets updated outside the driver.
return self._vmops.detach_interface(instance, vif)
def rescue(self, context, instance, network_info, image_meta,

View File

@ -252,20 +252,19 @@ class VMOps(object):
return vif_metadata
def _save_device_metadata(self, context, instance, block_device_info):
def update_device_metadata(self, context, instance):
"""Builds a metadata object for instance devices, that maps the user
provided tag to the hypervisor assigned device address.
"""
metadata = []
metadata.extend(self._get_vif_metadata(context, instance.uuid))
if block_device_info:
metadata.extend(self._block_dev_man.get_bdm_metadata(
context, instance, block_device_info))
metadata.extend(self._block_dev_man.get_bdm_metadata(context,
instance))
if metadata:
instance.device_metadata = objects.InstanceDeviceMetadata(
devices=metadata)
instance.device_metadata = objects.InstanceDeviceMetadata(
devices=metadata)
instance.save()
def set_boot_order(self, instance_name, vm_gen, block_device_info):
boot_order = self._block_dev_man.get_boot_order(
@ -305,7 +304,7 @@ class VMOps(object):
# This is supported starting from OVS version 2.5
self.plug_vifs(instance, network_info)
self._save_device_metadata(context, instance, block_device_info)
self.update_device_metadata(context, instance)
if configdrive.required_by(instance):
configdrive_path = self._create_config_drive(context,
@ -382,10 +381,10 @@ class VMOps(object):
self.configure_remotefx(instance, vm_gen)
self._vmutils.create_scsi_controller(instance_name)
self._attach_root_device(instance_name, root_device)
self._attach_root_device(context, instance, root_device)
self.attach_ephemerals(instance_name, block_device_info['ephemerals'])
self._volumeops.attach_volumes(
block_device_info['block_device_mapping'], instance_name)
context, block_device_info['block_device_mapping'], instance)
serial_ports = self._get_image_serial_port_settings(image_meta)
self._create_vm_com_port_pipes(instance, serial_ports)
@ -588,13 +587,14 @@ class VMOps(object):
remotefx_max_resolution,
vram_bytes)
def _attach_root_device(self, instance_name, root_dev_info):
def _attach_root_device(self, context, instance, root_dev_info):
if root_dev_info['type'] == constants.VOLUME:
self._volumeops.attach_volume(root_dev_info['connection_info'],
instance_name,
self._volumeops.attach_volume(context,
root_dev_info['connection_info'],
instance,
disk_bus=root_dev_info['disk_bus'])
else:
self._attach_drive(instance_name, root_dev_info['path'],
self._attach_drive(instance.name, root_dev_info['path'],
root_dev_info['drive_addr'],
root_dev_info['ctrl_disk_addr'],
root_dev_info['disk_bus'],
@ -611,6 +611,10 @@ class VMOps(object):
constants.BDI_DEVICE_TYPE_TO_DRIVE_TYPE[
eph['device_type']])
filename = os.path.basename(eph['path'])
self._block_dev_man.update_bdm_connection_info(
eph._bdm_obj, eph_filename=filename)
def _attach_drive(self, instance_name, path, drive_addr, ctrl_disk_addr,
controller_type, drive_type=constants.DISK):
if controller_type == constants.CTRL_TYPE_SCSI:
@ -1073,7 +1077,7 @@ class VMOps(object):
return True
def attach_interface(self, instance, vif):
def attach_interface(self, context, instance, vif):
if not self._check_hotplug_available(instance):
raise exception.InterfaceAttachFailed(instance_uuid=instance.uuid)
@ -1081,6 +1085,8 @@ class VMOps(object):
self._vmutils.create_nic(instance.name, vif['id'], vif['address'])
self._vif_driver.plug(instance, vif)
self.update_device_metadata(context, instance)
def detach_interface(self, instance, vif):
try:
if not self._check_hotplug_available(instance):

View File

@ -72,6 +72,8 @@ class VolumeOps(object):
def __init__(self):
self._volume_api = cinder.API()
self._vmops_prop = None
self._block_dev_man_prop = None
self._vmutils = utilsfactory.get_vmutils()
self._default_root_device = 'vda'
@ -80,30 +82,49 @@ class VolumeOps(object):
constants.STORAGE_PROTOCOL_ISCSI: ISCSIVolumeDriver(),
constants.STORAGE_PROTOCOL_FC: FCVolumeDriver()}
@property
def _vmops(self):
# We have to avoid a circular dependency.
if not self._vmops_prop:
self._vmops_prop = importutils.import_class(
'compute_hyperv.nova.vmops.VMOps')()
return self._vmops_prop
@property
def _block_dev_man(self):
if not self._block_dev_man_prop:
self._block_dev_man_prop = importutils.import_class(
'compute_hyperv.nova.block_device_manager.'
'BlockDeviceInfoManager')()
return self._block_dev_man_prop
def _get_volume_driver(self, connection_info):
driver_type = connection_info.get('driver_volume_type')
if driver_type not in self.volume_drivers:
raise exception.VolumeDriverNotFound(driver_type=driver_type)
return self.volume_drivers[driver_type]
def attach_volumes(self, volumes, instance_name):
def attach_volumes(self, context, volumes, instance):
for vol in volumes:
self.attach_volume(vol['connection_info'], instance_name)
self.attach_volume(context, vol['connection_info'], instance)
def disconnect_volumes(self, block_device_info):
mapping = driver.block_device_info_get_mapping(block_device_info)
for vol in mapping:
self.disconnect_volume(vol['connection_info'])
def attach_volume(self, connection_info, instance_name,
disk_bus=constants.CTRL_TYPE_SCSI):
def attach_volume(self, context, connection_info, instance,
disk_bus=constants.CTRL_TYPE_SCSI,
update_device_metadata=False):
tries_left = CONF.hyperv.volume_attach_retry_count + 1
while tries_left:
try:
self._attach_volume(connection_info,
instance_name,
disk_bus)
self._attach_volume(context,
connection_info,
instance,
disk_bus,
update_device_metadata)
break
except Exception as ex:
tries_left -= 1
@ -113,9 +134,15 @@ class VolumeOps(object):
"to instance %(instance_name)s. ",
{'connection_info': strutils.mask_dict_password(
connection_info),
'instance_name': instance_name})
'instance_name': instance.name})
self.disconnect_volume(connection_info)
# We're requesting a detach as the disk may have
# been attached to the instance but one of the
# post-attach operations failed.
self.detach_volume(context,
connection_info,
instance,
update_device_metadata)
raise exception.VolumeAttachFailed(
volume_id=connection_info['serial'],
reason=ex)
@ -126,22 +153,32 @@ class VolumeOps(object):
"Tries left: %(tries_left)s.",
{'connection_info': strutils.mask_dict_password(
connection_info),
'instance_name': instance_name,
'instance_name': instance.name,
'tries_left': tries_left})
time.sleep(CONF.hyperv.volume_attach_retry_interval)
def _attach_volume(self, connection_info, instance_name,
disk_bus=constants.CTRL_TYPE_SCSI):
def _attach_volume(self, context, connection_info, instance,
disk_bus=constants.CTRL_TYPE_SCSI,
update_device_metadata=False):
LOG.debug(
"Attaching volume: %(connection_info)s to %(instance_name)s",
{'connection_info': strutils.mask_dict_password(connection_info),
'instance_name': instance_name})
'instance_name': instance.name})
volume_driver = self._get_volume_driver(connection_info)
volume_driver.attach_volume(connection_info,
instance_name,
instance.name,
disk_bus)
if update_device_metadata:
# When attaching volumes to already existing instances,
# the connection info passed to the driver is not saved
# yet within the BDM table.
self._block_dev_man.set_volume_bdm_connection_info(
context, instance, connection_info)
self._vmops.update_device_metadata(
context, instance)
qos_specs = connection_info['data'].get('qos_specs') or {}
if qos_specs:
volume_driver.set_disk_qos_specs(connection_info,
@ -151,16 +188,20 @@ class VolumeOps(object):
volume_driver = self._get_volume_driver(connection_info)
volume_driver.disconnect_volume(connection_info)
def detach_volume(self, connection_info, instance_name):
def detach_volume(self, context, connection_info, instance,
update_device_metadata=False):
LOG.debug("Detaching volume: %(connection_info)s "
"from %(instance_name)s",
{'connection_info': strutils.mask_dict_password(
connection_info),
'instance_name': instance_name})
'instance_name': instance.name})
volume_driver = self._get_volume_driver(connection_info)
volume_driver.detach_volume(connection_info, instance_name)
volume_driver.detach_volume(connection_info, instance.name)
volume_driver.disconnect_volume(connection_info)
if update_device_metadata:
self._vmops.update_device_metadata(context, instance)
def fix_instance_volume_disk_paths(self, instance_name, block_device_info):
# Mapping containing the current disk paths for each volume.
actual_disk_mapping = self.get_disk_path_mapping(block_device_info)
@ -319,6 +360,10 @@ class VolumeOps(object):
instance.save(
expected_task_state=[task_states.IMAGE_SNAPSHOT_PENDING])
def get_disk_attachment_info(self, connection_info):
volume_driver = self._get_volume_driver(connection_info)
return volume_driver.get_disk_attachment_info(connection_info)
class BaseVolumeDriver(object):
_is_block_dev = True
@ -456,6 +501,19 @@ class BaseVolumeDriver(object):
def delete_snapshot(self, connection_info, instance, delete_info):
raise NotImplementedError()
def get_disk_attachment_info(self, connection_info):
if self._is_block_dev:
disk_path = None
serial = connection_info['serial']
else:
disk_path = self.get_disk_resource_path(connection_info)
serial = None
return self._vmutils.get_disk_attachment_info(
disk_path,
is_physical=self._is_block_dev,
serial=serial)
class ISCSIVolumeDriver(BaseVolumeDriver):
_is_block_dev = True

View File

@ -62,7 +62,7 @@ def fake_db_instance(**updates):
return db_instance
def fake_instance_obj(context, **updates):
def fake_instance_obj(context='fake-context', **updates):
expected_attrs = updates.pop('expected_attrs', None)
flavor = objects.Flavor(id=1, name='flavor1',
memory_mb=256, vcpus=1,

View File

@ -12,81 +12,276 @@
# License for the specific language governing permissions and limitations
# under the License.
import os
import ddt
import mock
from nova import block_device
from nova import exception
from nova import objects
from nova.virt import block_device as driver_block_device
from os_win import constants as os_win_const
from os_win import exceptions as os_win_exc
from oslo_serialization import jsonutils
from compute_hyperv.nova import block_device_manager
from compute_hyperv.nova import constants
from compute_hyperv.tests.unit import test_base
@ddt.ddt
class BlockDeviceManagerTestCase(test_base.HyperVBaseTestCase):
"""Unit tests for the Hyper-V BlockDeviceInfoManager class."""
_FAKE_CONN_INFO = {
'serial': 'fake_volume_id'
}
_FAKE_ATTACH_INFO = {
'controller_type': constants.CTRL_TYPE_SCSI,
'controller_addr': 0,
'controller_slot': 1
}
def setUp(self):
super(BlockDeviceManagerTestCase, self).setUp()
self._bdman = block_device_manager.BlockDeviceInfoManager()
self._bdman._volops = mock.Mock()
self._bdman._vmutils = mock.Mock()
self._bdman._pathutils = mock.Mock()
def test_get_device_bus_scsi(self):
bdm = {'disk_bus': constants.CTRL_TYPE_SCSI,
'drive_addr': 0, 'ctrl_disk_addr': 2}
self._volops = self._bdman._volops
self._pathutils = self._bdman._pathutils
bus = self._bdman._get_device_bus(bdm)
self.assertEqual('0:0:0:2', bus.address)
@ddt.data(constants.CTRL_TYPE_SCSI, constants.CTRL_TYPE_IDE)
def test_get_device_bus(self, controller_type):
fake_ctrl_addr = self._FAKE_ATTACH_INFO['controller_addr']
fake_ctrl_slot = self._FAKE_ATTACH_INFO['controller_slot']
def test_get_device_bus_ide(self):
bdm = {'disk_bus': constants.CTRL_TYPE_IDE,
'drive_addr': 0, 'ctrl_disk_addr': 1}
bus = self._bdman._get_device_bus(
controller_type, fake_ctrl_addr, fake_ctrl_slot)
bus = self._bdman._get_device_bus(bdm)
self.assertEqual('0:1', bus.address)
if controller_type == constants.CTRL_TYPE_SCSI:
exp_addr = '0:0:%s:%s' % (fake_ctrl_addr, fake_ctrl_slot)
exp_cls = objects.SCSIDeviceBus
else:
exp_addr = '%s:%s' % (fake_ctrl_addr, fake_ctrl_slot)
exp_cls = objects.IDEDeviceBus
@staticmethod
def _bdm_mock(**kwargs):
bdm = mock.MagicMock(**kwargs)
bdm.__contains__.side_effect = (
lambda attr: getattr(bdm, attr, None) is not None)
return bdm
self.assertIsInstance(bus, exp_cls)
self.assertEqual(exp_addr, bus.address)
@mock.patch.object(block_device_manager.objects, 'DiskMetadata')
@ddt.data({},
{'bdm_is_vol': False},
{'conn_info_set': False})
@ddt.unpack
@mock.patch.object(driver_block_device, 'convert_volume')
def test_get_vol_bdm_att_info(self, mock_convert_vol,
bdm_is_vol=True,
conn_info_set=True):
mock_drv_bdm = (dict(connection_info=self._FAKE_CONN_INFO)
if conn_info_set else {})
mock_convert_vol.return_value = (mock_drv_bdm
if bdm_is_vol
else None)
self._volops.get_disk_attachment_info.return_value = (
self._FAKE_ATTACH_INFO.copy())
attach_info = self._bdman._get_vol_bdm_attachment_info(
mock.sentinel.bdm)
mock_convert_vol.assert_called_once_with(
mock.sentinel.bdm)
if bdm_is_vol and conn_info_set:
exp_attach_info = self._FAKE_ATTACH_INFO.copy()
exp_attach_info['serial'] = self._FAKE_CONN_INFO['serial']
self._volops.get_disk_attachment_info.assert_called_once_with(
self._FAKE_CONN_INFO)
else:
exp_attach_info = None
self._volops.get_disk_attachment_info.assert_not_called()
self.assertEqual(exp_attach_info, attach_info)
@ddt.data({},
{'eph_name_set': False},
{'eph_disk_exists': False})
@ddt.unpack
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'get_bdm_connection_info')
@mock.patch('os.path.exists')
def test_get_eph_bdm_attachment_info(self, mock_exists,
mock_get_bdm_conn_info,
eph_name_set=True,
eph_disk_exists=True):
fake_instance_dir = 'fake_instance_dir'
fake_eph_name = 'eph0.vhdx'
mock_instance = mock.Mock()
fake_conn_info = self._FAKE_CONN_INFO.copy()
if eph_name_set:
fake_conn_info['eph_filename'] = fake_eph_name
mock_get_bdm_conn_info.return_value = fake_conn_info
mock_exists.return_value = eph_disk_exists
mock_get_attach_info = self._bdman._vmutils.get_disk_attachment_info
self._pathutils.get_instance_dir.return_value = fake_instance_dir
attach_info = self._bdman._get_eph_bdm_attachment_info(
mock_instance, mock.sentinel.bdm)
if eph_name_set and eph_disk_exists:
exp_attach_info = mock_get_attach_info.return_value
exp_eph_path = os.path.join(fake_instance_dir, fake_eph_name)
mock_exists.assert_called_once_with(exp_eph_path)
mock_get_attach_info.assert_called_once_with(
exp_eph_path,
is_physical=False)
else:
exp_attach_info = None
mock_get_attach_info.assert_not_called()
self.assertEqual(exp_attach_info, attach_info)
mock_get_bdm_conn_info.assert_called_once_with(
mock.sentinel.bdm)
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'_get_vol_bdm_attachment_info')
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'_get_eph_bdm_attachment_info')
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'_get_device_bus')
@mock.patch.object(block_device_manager.objects.BlockDeviceMappingList,
@mock.patch.object(block_device, 'new_format_is_ephemeral')
@mock.patch.object(objects, 'DiskMetadata')
def test_get_disk_metadata(self, mock_diskmetadata_cls,
mock_is_eph,
mock_get_device_bus,
mock_get_vol_attach_info,
mock_get_eph_attach_info,
bdm_is_eph=False,
bdm_is_vol=False,
attach_info_retrieved=True):
mock_instance = mock.Mock()
mock_bdm = mock.Mock()
mock_bdm.is_volume = bdm_is_vol
if attach_info_retrieved:
attach_info = self._FAKE_ATTACH_INFO.copy()
attach_info['serial'] = mock.sentinel.serial
else:
attach_info = None
mock_get_eph_attach_info.return_value = attach_info
mock_get_vol_attach_info.return_value = attach_info
mock_is_eph.return_value = bdm_is_eph
disk_metadata = self._bdman._get_disk_metadata(
mock_instance, mock_bdm)
if (bdm_is_vol or bdm_is_eph) and attach_info_retrieved:
exp_disk_meta = mock_diskmetadata_cls.return_value
mock_get_device_bus.assert_called_once_with(
self._FAKE_ATTACH_INFO['controller_type'],
self._FAKE_ATTACH_INFO['controller_addr'],
self._FAKE_ATTACH_INFO['controller_slot'])
mock_diskmetadata_cls.assert_called_once_with(
bus=mock_get_device_bus.return_value,
tags=[mock_bdm.tag],
serial=mock.sentinel.serial)
else:
exp_disk_meta = None
mock_get_device_bus.assert_not_called()
self.assertEqual(exp_disk_meta, disk_metadata)
if bdm_is_vol:
mock_get_vol_attach_info.assert_called_once_with(mock_bdm)
elif bdm_is_eph:
mock_get_eph_attach_info.assert_called_once_with(mock_instance,
mock_bdm)
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'_get_disk_metadata')
@mock.patch.object(objects.BlockDeviceMappingList,
'get_by_instance_uuid')
def test_get_bdm_metadata(self, mock_get_by_inst_uuid, mock_get_device_bus,
mock_DiskMetadata):
mock_instance = mock.MagicMock()
root_disk = {'mount_device': mock.sentinel.dev0}
ephemeral = {'device_name': mock.sentinel.dev1}
block_device_info = {
'root_disk': root_disk,
'block_device_mapping': [
{'mount_device': mock.sentinel.dev2},
{'mount_device': mock.sentinel.dev3},
],
'ephemerals': [ephemeral],
}
def test_get_bdm_metadata(self, mock_get_bdm_list,
mock_get_disk_meta):
bdms = [mock.Mock()] * 4
disk_meta = mock.Mock()
mock_instance = mock.Mock()
bdm = self._bdm_mock(device_name=mock.sentinel.dev0, tag='taggy')
eph = self._bdm_mock(device_name=mock.sentinel.dev1, tag='ephy')
mock_get_by_inst_uuid.return_value = [
bdm, eph, self._bdm_mock(device_name=mock.sentinel.dev2, tag=None),
]
mock_get_bdm_list.return_value = bdms
mock_get_disk_meta.side_effect = [
None,
exception.DiskNotFound(message='fake_err'),
os_win_exc.DiskNotFound(message='fake_err'),
disk_meta]
bdm_metadata = self._bdman.get_bdm_metadata(mock.sentinel.context,
mock_instance,
block_device_info)
bdm_meta = self._bdman.get_bdm_metadata(mock.sentinel.context,
mock_instance)
mock_get_by_inst_uuid.assert_called_once_with(mock.sentinel.context,
mock_instance.uuid)
mock_get_device_bus.assert_has_calls(
[mock.call(root_disk), mock.call(ephemeral)], any_order=True)
mock_DiskMetadata.assert_has_calls(
[mock.call(bus=mock_get_device_bus.return_value, tags=[bdm.tag]),
mock.call(bus=mock_get_device_bus.return_value, tags=[eph.tag])],
any_order=True)
self.assertEqual([mock_DiskMetadata.return_value] * 2, bdm_metadata)
self.assertEqual([disk_meta], bdm_meta)
mock_get_bdm_list.assert_called_once_with(mock.sentinel.context,
mock_instance.uuid)
mock_get_disk_meta.assert_has_calls(
[mock.call(mock_instance, bdm) for bdm in bdms])
@mock.patch.object(objects.BlockDeviceMapping,
'get_by_volume_and_instance')
def test_set_vol_bdm_conn_info(self, mock_get_bdm):
mock_instance = mock.Mock()
mock_bdm = mock_get_bdm.return_value
self._bdman.set_volume_bdm_connection_info(
mock.sentinel.context, mock_instance, self._FAKE_CONN_INFO)
mock_get_bdm.assert_called_once_with(
mock.sentinel.context,
self._FAKE_CONN_INFO['serial'],
mock_instance.uuid)
self.assertEqual(self._FAKE_CONN_INFO,
jsonutils.loads(mock_bdm.connection_info))
mock_bdm.save.assert_called_once_with()
def test_get_bdm_connection_info(self):
bdm = mock.Mock(connection_info=None)
self.assertEqual({}, self._bdman.get_bdm_connection_info(bdm))
bdm = mock.Mock()
bdm.connection_info = jsonutils.dumps(self._FAKE_CONN_INFO)
self.assertEqual(self._FAKE_CONN_INFO,
self._bdman.get_bdm_connection_info(bdm))
def test_update_bdm_conn_info(self):
connection_info = self._FAKE_CONN_INFO.copy()
mock_bdm = mock.Mock()
mock_bdm.connection_info = jsonutils.dumps(connection_info)
updates = dict(some_key='some_val',
some_other_key='some_other_val')
self._bdman.update_bdm_connection_info(
mock_bdm, **updates)
exp_connection_info = connection_info.copy()
exp_connection_info.update(**updates)
self.assertEqual(exp_connection_info,
jsonutils.loads(mock_bdm.connection_info))
mock_bdm.save.assert_called_once_with()
@mock.patch('nova.virt.configdrive.required_by')
def test_init_controller_slot_counter_gen1_no_configdrive(

View File

@ -207,9 +207,13 @@ class HyperVDriverTestCase(test_base.HyperVBaseTestCase):
mock.sentinel.device_type, mock.sentinel.encryption)
self.driver._volumeops.attach_volume.assert_called_once_with(
mock.sentinel.context,
mock.sentinel.connection_info,
mock_instance.name)
mock_instance,
update_device_metadata=True)
@mock.patch('nova.context.get_admin_context',
lambda: mock.sentinel.admin_context)
def test_detach_volume(self):
mock_instance = fake_instance.fake_instance_obj(self.context)
self.driver.detach_volume(
@ -217,8 +221,10 @@ class HyperVDriverTestCase(test_base.HyperVBaseTestCase):
mock.sentinel.mountpoint, mock.sentinel.encryption)
self.driver._volumeops.detach_volume.assert_called_once_with(
mock.sentinel.admin_context,
mock.sentinel.connection_info,
mock_instance.name)
mock_instance,
update_device_metadata=True)
def test_get_volume_connector(self):
self.driver.get_volume_connector(mock.sentinel.instance)
@ -528,6 +534,15 @@ class HyperVDriverTestCase(test_base.HyperVBaseTestCase):
self.driver._imagecache.update.assert_called_once_with(
mock.sentinel.context, mock.sentinel.all_instances)
def test_attach_interface(self):
mock_instance = fake_instance.fake_instance_obj(self.context)
self.driver.attach_interface(
self.context, mock_instance, mock.sentinel.image_meta,
mock.sentinel.vif)
self.driver._vmops.attach_interface.assert_called_once_with(
self.context, mock_instance, mock.sentinel.vif)
def _check_recreate_image_meta(self, mock_image_meta, image_ref='',
instance_img_ref=''):
system_meta = {'image_base_image_ref': instance_img_ref}

View File

@ -382,20 +382,19 @@ class VMOpsTestCase(test_base.HyperVBaseTestCase):
@mock.patch.object(vmops.objects, 'InstanceDeviceMetadata')
@mock.patch.object(vmops.VMOps, '_get_vif_metadata')
def test_save_device_metadata(self, mock_get_vif_metadata,
mock_InstanceDeviceMetadata):
def test_update_device_metadata(self, mock_get_vif_metadata,
mock_InstanceDeviceMetadata):
mock_instance = mock.MagicMock()
mock_get_vif_metadata.return_value = [mock.sentinel.vif_metadata]
self._vmops._block_dev_man.get_bdm_metadata.return_value = [
mock.sentinel.bdm_metadata]
self._vmops._save_device_metadata(self.context, mock_instance,
mock.sentinel.block_device_info)
self._vmops.update_device_metadata(self.context, mock_instance)
mock_get_vif_metadata.assert_called_once_with(self.context,
mock_instance.uuid)
self._vmops._block_dev_man.get_bdm_metadata.assert_called_once_with(
self.context, mock_instance, mock.sentinel.block_device_info)
self.context, mock_instance)
expected_metadata = [mock.sentinel.vif_metadata,
mock.sentinel.bdm_metadata]
@ -422,7 +421,7 @@ class VMOpsTestCase(test_base.HyperVBaseTestCase):
@mock.patch('compute_hyperv.nova.vmops.VMOps.attach_config_drive')
@mock.patch('compute_hyperv.nova.vmops.VMOps._create_config_drive')
@mock.patch('nova.virt.configdrive.required_by')
@mock.patch('compute_hyperv.nova.vmops.VMOps._save_device_metadata')
@mock.patch('compute_hyperv.nova.vmops.VMOps.update_device_metadata')
@mock.patch('compute_hyperv.nova.vmops.VMOps.create_instance')
@mock.patch('compute_hyperv.nova.vmops.VMOps.get_image_vm_generation')
@mock.patch('compute_hyperv.nova.vmops.VMOps._create_ephemerals')
@ -437,7 +436,7 @@ class VMOpsTestCase(test_base.HyperVBaseTestCase):
mock_delete_disk_files,
mock_create_root_device,
mock_create_ephemerals, mock_get_image_vm_gen,
mock_create_instance, mock_save_device_metadata,
mock_create_instance, mock_update_device_metadata,
mock_configdrive_required,
mock_create_config_drive, mock_attach_config_drive,
mock_set_boot_order,
@ -494,8 +493,8 @@ class VMOpsTestCase(test_base.HyperVBaseTestCase):
block_device_info, fake_vm_gen, mock_image_meta)
mock_plug_vifs.assert_called_once_with(mock_instance,
mock.sentinel.network_info)
mock_save_device_metadata.assert_called_once_with(
self.context, mock_instance, block_device_info)
mock_update_device_metadata.assert_called_once_with(
self.context, mock_instance)
mock_configdrive_required.assert_called_once_with(mock_instance)
if configdrive_required:
mock_create_config_drive.assert_called_once_with(
@ -634,12 +633,13 @@ class VMOpsTestCase(test_base.HyperVBaseTestCase):
mock_create_scsi_ctrl = self._vmops._vmutils.create_scsi_controller
mock_create_scsi_ctrl.assert_called_once_with(mock_instance.name)
mock_attach_root_device.assert_called_once_with(mock_instance.name,
root_device_info)
mock_attach_root_device.assert_called_once_with(
self.context, mock_instance, root_device_info)
mock_attach_ephemerals.assert_called_once_with(mock_instance.name,
block_device_info['ephemerals'])
mock_attach_volumes.assert_called_once_with(
block_device_info['block_device_mapping'], mock_instance.name)
self.context, block_device_info['block_device_mapping'],
mock_instance)
mock_get_port_settings.assert_called_with(mock.sentinel.image_meta)
mock_create_pipes.assert_called_once_with(
@ -822,10 +822,12 @@ class VMOpsTestCase(test_base.HyperVBaseTestCase):
'connection_info': mock.sentinel.CONN_INFO,
'disk_bus': constants.CTRL_TYPE_IDE}
self._vmops._attach_root_device(mock_instance.name, root_device_info)
self._vmops._attach_root_device(self.context,
mock_instance, root_device_info)
mock_attach_volume.assert_called_once_with(
root_device_info['connection_info'], mock_instance.name,
self.context,
root_device_info['connection_info'], mock_instance,
disk_bus=root_device_info['disk_bus'])
@mock.patch.object(vmops.VMOps, '_attach_drive')
@ -838,7 +840,8 @@ class VMOpsTestCase(test_base.HyperVBaseTestCase):
'drive_addr': 0,
'ctrl_disk_addr': 1}
self._vmops._attach_root_device(mock_instance.name, root_device_info)
self._vmops._attach_root_device(
self.context, mock_instance, root_device_info)
mock_attach_drive.assert_called_once_with(
mock_instance.name, root_device_info['path'],
@ -849,28 +852,38 @@ class VMOpsTestCase(test_base.HyperVBaseTestCase):
def test_attach_ephemerals(self, mock_attach_drive):
mock_instance = fake_instance.fake_instance_obj(self.context)
ephemerals = [{'path': mock.sentinel.PATH1,
class FakeBDM(dict):
_bdm_obj = mock.sentinel.bdm_obj
ephemerals = [{'path': os.path.join('eph_dir', 'eph0_path'),
'boot_index': 1,
'disk_bus': constants.CTRL_TYPE_IDE,
'device_type': 'disk',
'drive_addr': 0,
'ctrl_disk_addr': 1},
{'path': mock.sentinel.PATH2,
{'path': os.path.join('eph_dir', 'eph1_path'),
'boot_index': 2,
'disk_bus': constants.CTRL_TYPE_SCSI,
'device_type': 'disk',
'drive_addr': 0,
'ctrl_disk_addr': 0},
{'path': None}]
ephemerals = [FakeBDM(eph) for eph in ephemerals]
self._vmops.attach_ephemerals(mock_instance.name, ephemerals)
mock_attach_drive.assert_has_calls(
[mock.call(mock_instance.name, mock.sentinel.PATH1, 0,
[mock.call(mock_instance.name, ephemerals[0]['path'], 0,
1, constants.CTRL_TYPE_IDE, constants.DISK),
mock.call(mock_instance.name, mock.sentinel.PATH2, 0,
mock.call(mock_instance.name, ephemerals[1]['path'], 0,
0, constants.CTRL_TYPE_SCSI, constants.DISK)
])
mock_update_conn = (
self._vmops._block_dev_man.update_bdm_connection_info)
mock_update_conn.assert_has_calls(
[mock.call(mock.sentinel.bdm_obj,
eph_filename=os.path.basename(eph['path']))
for eph in ephemerals if eph.get('path')])
def test_attach_drive_vm_to_scsi(self):
self._vmops._attach_drive(
@ -1717,25 +1730,31 @@ class VMOpsTestCase(test_base.HyperVBaseTestCase):
self._test_check_hotplug_available(
expected_result=False, windows_version=self._WIN_VERSION_6_3)
@mock.patch.object(vmops.VMOps, 'update_device_metadata')
@mock.patch.object(vmops.VMOps, '_check_hotplug_available')
def test_attach_interface(self, mock_check_hotplug_available):
def test_attach_interface(self, mock_check_hotplug_available,
mock_update_dev_meta):
mock_check_hotplug_available.return_value = True
fake_vm = fake_instance.fake_instance_obj(self.context)
fake_vif = test_virtual_interface.fake_vif
self._vmops.attach_interface(fake_vm, fake_vif)
self._vmops.attach_interface(
mock.sentinel.context, fake_vm, fake_vif)
mock_check_hotplug_available.assert_called_once_with(fake_vm)
self._vmops._vif_driver.plug.assert_called_once_with(
fake_vm, fake_vif)
self._vmops._vmutils.create_nic.assert_called_once_with(
fake_vm.name, fake_vif['id'], fake_vif['address'])
mock_update_dev_meta.assert_called_once_with(
mock.sentinel.context, fake_vm)
@mock.patch.object(vmops.VMOps, '_check_hotplug_available')
def test_attach_interface_failed(self, mock_check_hotplug_available):
mock_check_hotplug_available.return_value = False
self.assertRaises(exception.InterfaceAttachFailed,
self._vmops.attach_interface,
mock.sentinel.context,
mock.MagicMock(), mock.sentinel.fake_vif)
@mock.patch.object(vmops.VMOps, '_check_hotplug_available')

View File

@ -27,10 +27,12 @@ from os_brick.initiator import connector
from os_win import constants as os_win_const
from oslo_utils import units
from compute_hyperv.nova import block_device_manager
import compute_hyperv.nova.conf
from compute_hyperv.nova import constants
from compute_hyperv.nova import vmops
from compute_hyperv.nova import volumeops
from compute_hyperv.tests import fake_instance
from compute_hyperv.tests.unit import test_base
CONF = compute_hyperv.nova.conf.CONF
@ -56,6 +58,7 @@ def get_fake_connection_info(**kwargs):
'serial': mock.sentinel.serial}
@ddt.ddt
class VolumeOpsTestCase(test_base.HyperVBaseTestCase):
"""Unit tests for VolumeOps class."""
@ -87,12 +90,14 @@ class VolumeOpsTestCase(test_base.HyperVBaseTestCase):
block_device_info = get_fake_block_dev_info()
self._volumeops.attach_volumes(
mock.sentinel.context,
block_device_info['block_device_mapping'],
mock.sentinel.instance_name)
mock.sentinel.instance)
mock_attach_volume.assert_called_once_with(
mock.sentinel.context,
block_device_info['block_device_mapping'][0]['connection_info'],
mock.sentinel.instance_name)
mock.sentinel.instance)
def test_fix_instance_volume_disk_paths_empty_bdm(self):
self._volumeops.fix_instance_volume_disk_paths(
@ -152,10 +157,24 @@ class VolumeOpsTestCase(test_base.HyperVBaseTestCase):
fake_volume_driver.disconnect_volume.assert_called_once_with(
block_device_mapping[0]['connection_info'])
@ddt.data({},
{'attach_failed': True},
{'update_device_metadata': True})
@ddt.unpack
@mock.patch('time.sleep')
@mock.patch.object(volumeops.VolumeOps, 'detach_volume')
@mock.patch.object(volumeops.VolumeOps, '_get_volume_driver')
def _test_attach_volume(self, mock_get_volume_driver, mock_sleep,
attach_failed):
@mock.patch.object(vmops.VMOps, 'update_device_metadata')
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'set_volume_bdm_connection_info')
def test_attach_volume(self, mock_set_bdm_conn_info,
mock_update_dev_meta,
mock_get_volume_driver,
mock_detach,
mock_sleep,
attach_failed=False,
update_device_metadata=False):
mock_instance = fake_instance.fake_instance_obj()
fake_conn_info = get_fake_connection_info(
qos_specs=mock.sentinel.qos_specs)
fake_volume_driver = mock_get_volume_driver.return_value
@ -169,40 +188,53 @@ class VolumeOpsTestCase(test_base.HyperVBaseTestCase):
self.assertRaises(exception.VolumeAttachFailed,
self._volumeops.attach_volume,
mock.sentinel.context,
fake_conn_info,
mock.sentinel.inst_name,
mock.sentinel.disk_bus)
mock_instance,
mock.sentinel.disk_bus,
update_device_metadata)
else:
self._volumeops.attach_volume(
mock.sentinel.context,
fake_conn_info,
mock.sentinel.inst_name,
mock.sentinel.disk_bus)
mock_instance,
mock.sentinel.disk_bus,
update_device_metadata)
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_instance.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 update_device_metadata:
mock_set_bdm_conn_info.assert_has_calls(
[mock.call(mock.sentinel.context,
mock_instance,
fake_conn_info)] * expected_try_count)
mock_update_dev_meta.assert_has_calls(
[mock.call(mock.sentinel.context,
mock_instance)] * expected_try_count)
else:
mock_set_bdm_conn_info.assert_not_called()
mock_update_dev_meta.assert_not_called()
if attach_failed:
fake_volume_driver.disconnect_volume.assert_called_once_with(
fake_conn_info)
mock_detach.assert_called_once_with(
mock.sentinel.context,
fake_conn_info,
mock_instance,
update_device_metadata)
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
@ -214,21 +246,34 @@ class VolumeOpsTestCase(test_base.HyperVBaseTestCase):
fake_volume_driver.disconnect_volume.assert_called_once_with(
mock.sentinel.conn_info)
@ddt.data(True, False)
@mock.patch.object(volumeops.VolumeOps, '_get_volume_driver')
def test_detach_volume(self, mock_get_volume_driver):
@mock.patch.object(vmops.VMOps, 'update_device_metadata')
def test_detach_volume(self, update_device_metadata,
mock_update_dev_meta,
mock_get_volume_driver):
mock_instance = fake_instance.fake_instance_obj()
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)
self._volumeops.detach_volume(mock.sentinel.context,
fake_conn_info,
mock_instance,
update_device_metadata)
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_conn_info, mock_instance.name)
fake_volume_driver.disconnect_volume.assert_called_once_with(
fake_conn_info)
if update_device_metadata:
mock_update_dev_meta.assert_called_once_with(
mock.sentinel.context, mock_instance)
else:
mock_update_dev_meta.assert_not_called()
@mock.patch.object(connector, 'get_connector_properties')
def test_get_volume_connector(self, mock_get_connector):
conn = self._volumeops.get_volume_connector()
@ -444,6 +489,19 @@ class VolumeOpsTestCase(test_base.HyperVBaseTestCase):
mock.call(expected_task_state=[
task_states.IMAGE_SNAPSHOT_PENDING])])
@mock.patch.object(volumeops.VolumeOps, '_get_volume_driver')
def test_get_disk_attachment_info(self, mock_get_volume_driver):
fake_conn_info = get_fake_connection_info()
ret_val = self._volumeops.get_disk_attachment_info(fake_conn_info)
mock_vol_driver = mock_get_volume_driver.return_value
mock_vol_driver.get_disk_attachment_info.assert_called_once_with(
fake_conn_info)
self.assertEqual(
mock_vol_driver.get_disk_attachment_info.return_value,
ret_val)
@ddt.ddt
class BaseVolumeDriverTestCase(test_base.HyperVBaseTestCase):
@ -684,6 +742,31 @@ class BaseVolumeDriverTestCase(test_base.HyperVBaseTestCase):
self._base_vol_driver.set_disk_qos_specs(
mock.sentinel.conn_info, mock.sentinel.disk_qos_spes)
@ddt.data(True, False)
@mock.patch.object(volumeops.BaseVolumeDriver,
'get_disk_resource_path')
def test_get_disk_attachment_info(self, is_block_dev,
mock_get_disk_resource_path):
connection_info = get_fake_connection_info()
self._base_vol_driver._is_block_dev = is_block_dev
self._base_vol_driver.get_disk_attachment_info(connection_info)
if is_block_dev:
exp_serial = connection_info['serial']
exp_disk_res_path = None
self.assertFalse(mock_get_disk_resource_path.called)
else:
exp_serial = None
exp_disk_res_path = mock_get_disk_resource_path.return_value
mock_get_disk_resource_path.assert_called_once_with(
connection_info)
self._vmutils.get_disk_attachment_info.assert_called_once_with(
exp_disk_res_path,
is_physical=is_block_dev,
serial=exp_serial)
class ISCSIVolumeDriverTestCase(test_base.HyperVBaseTestCase):
"""Unit tests for Hyper-V BaseVolumeDriver class."""