Remove the Hyper-V driver

Nova Hyper-V driver is not tested in OpenStack upstream and no maintianers.
This driver has been marked as deprecated in Antelope release. It has dependency
on the OpenStack Winstacker project which has been retired[1].

As discussed in vPTG[2], removing the HyperV driver, tests, and its config.

[1] https://review.opendev.org/c/openstack/governance/+/886880
[2] https://etherpad.opendev.org/p/nova-caracal-ptg#L301

Change-Id: I568c79bae9b9736a20c367096d748c730ed59f0e
This commit is contained in:
Ghanshyam Mann 2024-01-25 00:45:11 -08:00
parent 6531ed6310
commit b068b04372
45 changed files with 30 additions and 10655 deletions

View File

@ -39,7 +39,7 @@ compute host and image.
.. rubric:: Compute host requirements
The following virt drivers support the config drive: libvirt, Hyper-V and
The following virt drivers support the config drive: libvirt and
VMware. The Bare Metal service also supports the config drive.
- To use config drives with libvirt or VMware, you must first
@ -49,12 +49,6 @@ VMware. The Bare Metal service also supports the config drive.
the same path as the :program:`nova-compute` service, you do not need to set
this flag.
- To use config drives with Hyper-V, you must set the
:oslo.config:option:`mkisofs_cmd` config option to the full path to an
:command:`mkisofs.exe` installation. Additionally, you must set the
:oslo.config:option:`hyperv.qemu_img_cmd` config option to the full path to an
:command:`qemu-img` command installation.
- To use config drives with the Bare Metal service, you do not need to prepare
anything.
@ -81,11 +75,6 @@ options:
- :oslo.config:option:`force_config_drive`
- :oslo.config:option:`config_drive_format`
If using the HyperV compute driver, the following additional options are
supported:
- :oslo.config:option:`hyperv.config_drive_cdrom`
For example, to ensure nova always provides a config drive to instances but
versions ``2018-08-27`` (Rocky) and ``2017-02-22`` (Ocata) are skipped, add the
following to :file:`nova.conf`:

View File

@ -35,7 +35,6 @@ from nova.conf import devices
from nova.conf import ephemeral_storage
from nova.conf import glance
from nova.conf import guestfs
from nova.conf import hyperv
from nova.conf import imagecache
from nova.conf import ironic
from nova.conf import key_manager
@ -84,7 +83,6 @@ devices.register_opts(CONF)
ephemeral_storage.register_opts(CONF)
glance.register_opts(CONF)
guestfs.register_opts(CONF)
hyperv.register_opts(CONF)
mks.register_opts(CONF)
imagecache.register_opts(CONF)
ironic.register_opts(CONF)

View File

@ -40,7 +40,6 @@ Possible values:
* ``fake.FakeDriver``
* ``ironic.IronicDriver``
* ``vmwareapi.VMwareVCDriver``
* ``hyperv.HyperVDriver``
* ``zvm.ZVMDriver``
"""),
cfg.BoolOpt('allow_resize_to_same_host',

View File

@ -44,10 +44,6 @@ Related options:
config drive option
3. the image used to create the instance requires a config drive,
this is defined by ``img_config_drive`` property for that image.
* A compute node running Hyper-V hypervisor can be configured to attach
config drive as a CD drive. To attach the config drive as a CD drive, set the
``[hyperv] config_drive_cdrom`` option to true.
"""),
cfg.BoolOpt('force_config_drive',
default=False,
@ -71,11 +67,6 @@ Related options:
* Use the 'mkisofs_cmd' flag to set the path where you install the
genisoimage program. If genisoimage is in same path as the
nova-compute service, you do not need to set this flag.
* To use a config drive with Hyper-V, you must set the
'mkisofs_cmd' value to the full path to an mkisofs.exe installation.
Additionally, you must set the qemu_img_cmd value in the hyperv
configuration section to the full path to an qemu-img command
installation.
"""),
cfg.StrOpt('mkisofs_cmd',
default='genisoimage',
@ -86,11 +77,6 @@ Use the ``mkisofs_cmd`` flag to set the path where you install the
``genisoimage`` program. If ``genisoimage`` is on the system path, you do not
need to change the default value.
To use a config drive with Hyper-V, you must set the ``mkisofs_cmd`` value to
the full path to an ``mkisofs.exe`` installation. Additionally, you must set
the ``qemu_img_cmd`` value in the hyperv configuration section to the full path
to an ``qemu-img`` command installation.
Possible values:
* Name of the ISO image creator program, in case it is in the same directory
@ -100,9 +86,6 @@ Possible values:
Related options:
* This option is meaningful when config drives are enabled.
* To use config drive with Hyper-V, you must set the ``qemu_img_cmd``
value in the hyperv configuration section to the full path to an ``qemu-img``
command installation.
"""),
]

View File

@ -1,337 +0,0 @@
# Copyright (c) 2016 TUBITAK BILGEM
# 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.
from oslo_config import cfg
hyperv_opt_group = cfg.OptGroup("hyperv",
title='The Hyper-V feature',
help="""
The hyperv feature allows you to configure the Hyper-V hypervisor
driver to be used within an OpenStack deployment.
""")
hyperv_opts = [
cfg.FloatOpt('dynamic_memory_ratio',
default=1.0,
help="""
Dynamic memory ratio
Enables dynamic memory allocation (ballooning) when set to a value
greater than 1. The value expresses the ratio between the total RAM
assigned to an instance and its startup RAM amount. For example a
ratio of 2.0 for an instance with 1024MB of RAM implies 512MB of
RAM allocated at startup.
Possible values:
* 1.0: Disables dynamic memory allocation (Default).
* Float values greater than 1.0: Enables allocation of total implied
RAM divided by this value for startup.
"""),
cfg.BoolOpt('enable_instance_metrics_collection',
default=False,
help="""
Enable instance metrics collection
Enables metrics collections for an instance by using Hyper-V's
metric APIs. Collected data can be retrieved by other apps and
services, e.g.: Ceilometer.
"""),
cfg.StrOpt('instances_path_share',
default="",
help="""
Instances path share
The name of a Windows share mapped to the "instances_path" dir
and used by the resize feature to copy files to the target host.
If left blank, an administrative share (hidden network share) will
be used, looking for the same "instances_path" used locally.
Possible values:
* "": An administrative share will be used (Default).
* Name of a Windows share.
Related options:
* "instances_path": The directory which will be used if this option
here is left blank.
"""),
cfg.BoolOpt('limit_cpu_features',
default=False,
help="""
Limit CPU features
This flag is needed to support live migration to hosts with
different CPU features and checked during instance creation
in order to limit the CPU features used by the instance.
"""),
cfg.IntOpt('mounted_disk_query_retry_count',
default=10,
min=0,
help="""
Mounted disk query retry count
The number of times to retry checking for a mounted disk.
The query runs until the device can be found or the retry
count is reached.
Possible values:
* Positive integer values. Values greater than 1 is recommended
(Default: 10).
Related options:
* Time interval between disk mount retries is declared with
"mounted_disk_query_retry_interval" option.
"""),
cfg.IntOpt('mounted_disk_query_retry_interval',
default=5,
min=0,
help="""
Mounted disk query retry interval
Interval between checks for a mounted disk, in seconds.
Possible values:
* Time in seconds (Default: 5).
Related options:
* This option is meaningful when the mounted_disk_query_retry_count
is greater than 1.
* The retry loop runs with mounted_disk_query_retry_count and
mounted_disk_query_retry_interval configuration options.
"""),
cfg.IntOpt('power_state_check_timeframe',
default=60,
min=0,
help="""
Power state check timeframe
The timeframe to be checked for instance power state changes.
This option is used to fetch the state of the instance from Hyper-V
through the WMI interface, within the specified timeframe.
Possible values:
* Timeframe in seconds (Default: 60).
"""),
cfg.IntOpt('power_state_event_polling_interval',
default=2,
min=0,
help="""
Power state event polling interval
Instance power state change event polling frequency. Sets the
listener interval for power state events to the given value.
This option enhances the internal lifecycle notifications of
instances that reboot themselves. It is unlikely that an operator
has to change this value.
Possible values:
* Time in seconds (Default: 2).
"""),
cfg.StrOpt('qemu_img_cmd',
default="qemu-img.exe",
help=r"""
qemu-img command
qemu-img is required for some of the image related operations
like converting between different image types. You can get it
from here: (http://qemu.weilnetz.de/) or you can install the
Cloudbase OpenStack Hyper-V Compute Driver
(https://cloudbase.it/openstack-hyperv-driver/) which automatically
sets the proper path for this config option. You can either give the
full path of qemu-img.exe or set its path in the PATH environment
variable and leave this option to the default value.
Possible values:
* Name of the qemu-img executable, in case it is in the same
directory as the nova-compute service or its path is in the
PATH environment variable (Default).
* Path of qemu-img command (DRIVELETTER:\PATH\TO\QEMU-IMG\COMMAND).
Related options:
* If the config_drive_cdrom option is False, qemu-img will be used to
convert the ISO to a VHD, otherwise the config drive will
remain an ISO. To use config drive with Hyper-V, you must
set the ``mkisofs_cmd`` value to the full path to an ``mkisofs.exe``
installation.
"""),
cfg.StrOpt('vswitch_name',
help="""
External virtual switch name
The Hyper-V Virtual Switch is a software-based layer-2 Ethernet
network switch that is available with the installation of the
Hyper-V server role. The switch includes programmatically managed
and extensible capabilities to connect virtual machines to both
virtual networks and the physical network. In addition, Hyper-V
Virtual Switch provides policy enforcement for security, isolation,
and service levels. The vSwitch represented by this config option
must be an external one (not internal or private).
Possible values:
* If not provided, the first of a list of available vswitches
is used. This list is queried using WQL.
* Virtual switch name.
"""),
cfg.IntOpt('wait_soft_reboot_seconds',
default=60,
min=0,
help="""
Wait soft reboot seconds
Number of seconds to wait for instance to shut down after soft
reboot request is made. We fall back to hard reboot if instance
does not shutdown within this window.
Possible values:
* Time in seconds (Default: 60).
"""),
cfg.BoolOpt('config_drive_cdrom',
default=False,
help="""
Mount config drive as a CD drive.
OpenStack can be configured to write instance metadata to a config drive, which
is then attached to the instance before it boots. The config drive can be
attached as a disk drive (default) or as a CD drive.
Related options:
* This option is meaningful with ``force_config_drive`` option set to ``True``
or when the REST API call to create an instance will have
``--config-drive=True`` flag.
* ``config_drive_format`` option must be set to ``iso9660`` in order to use
CD drive as the config drive image.
* To use config drive with Hyper-V, you must set the
``mkisofs_cmd`` value to the full path to an ``mkisofs.exe`` installation.
Additionally, you must set the ``qemu_img_cmd`` value to the full path
to an ``qemu-img`` command installation.
* You can configure the Compute service to always create a configuration
drive by setting the ``force_config_drive`` option to ``True``.
"""),
cfg.BoolOpt('config_drive_inject_password',
default=False,
help="""
Inject password to config drive.
When enabled, the admin password will be available from the config drive image.
Related options:
* This option is meaningful when used with other options that enable
config drive usage with Hyper-V, such as ``force_config_drive``.
"""),
cfg.IntOpt('volume_attach_retry_count',
default=10,
min=0,
help="""
Volume attach retry count
The number of times to retry attaching a volume. Volume attachment
is retried until success or the given retry count is reached.
Possible values:
* Positive integer values (Default: 10).
Related options:
* Time interval between attachment attempts is declared with
volume_attach_retry_interval option.
"""),
cfg.IntOpt('volume_attach_retry_interval',
default=5,
min=0,
help="""
Volume attach retry interval
Interval between volume attachment attempts, in seconds.
Possible values:
* Time in seconds (Default: 5).
Related options:
* This options is meaningful when volume_attach_retry_count
is greater than 1.
* The retry loop runs with volume_attach_retry_count and
volume_attach_retry_interval configuration options.
"""),
cfg.BoolOpt('enable_remotefx',
default=False,
help="""
Enable RemoteFX feature
This requires at least one DirectX 11 capable graphics adapter for
Windows / Hyper-V Server 2012 R2 or newer and RDS-Virtualization
feature has to be enabled.
Instances with RemoteFX can be requested with the following flavor
extra specs:
**os:resolution**. Guest VM screen resolution size. Acceptable values::
1024x768, 1280x1024, 1600x1200, 1920x1200, 2560x1600, 3840x2160
``3840x2160`` is only available on Windows / Hyper-V Server 2016.
**os:monitors**. Guest VM number of monitors. Acceptable values::
[1, 4] - Windows / Hyper-V Server 2012 R2
[1, 8] - Windows / Hyper-V Server 2016
**os:vram**. Guest VM VRAM amount. Only available on
Windows / Hyper-V Server 2016. Acceptable values::
64, 128, 256, 512, 1024
"""),
cfg.BoolOpt('use_multipath_io',
default=False,
help="""
Use multipath connections when attaching iSCSI or FC disks.
This requires the Multipath IO Windows feature to be enabled. MPIO must be
configured to claim such devices.
"""),
cfg.ListOpt('iscsi_initiator_list',
default=[],
help="""
List of iSCSI initiators that will be used for establishing iSCSI sessions.
If none are specified, the Microsoft iSCSI initiator service will choose the
initiator.
""")
]
def register_opts(conf):
conf.register_group(hyperv_opt_group)
conf.register_opts(hyperv_opts, group=hyperv_opt_group)
def list_opts():
return {hyperv_opt_group: hyperv_opts}

View File

@ -1,20 +0,0 @@
# 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 unittest
try:
import os_win # noqa: F401
except ImportError:
raise unittest.SkipTest(
"The 'os-win' dependency is not installed."
)

View File

@ -1,40 +0,0 @@
# 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.
from unittest import mock
from os_win import utilsfactory
from nova import test
class HyperVBaseTestCase(test.NoDBTestCase):
def setUp(self):
super(HyperVBaseTestCase, self).setUp()
self._mock_wmi = mock.MagicMock()
wmi_patcher = mock.patch('builtins.wmi', create=True,
new=self._mock_wmi)
platform_patcher = mock.patch('sys.platform', 'win32')
utilsfactory_patcher = mock.patch.object(utilsfactory, '_get_class')
platform_patcher.start()
wmi_patcher.start()
utilsfactory_patcher.start()
self.addCleanup(wmi_patcher.stop)
self.addCleanup(platform_patcher.stop)
self.addCleanup(utilsfactory_patcher.stop)

View File

@ -1,438 +0,0 @@
# Copyright (c) 2016 Cloudbase Solutions Srl
#
# 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.
from os_win import constants as os_win_const
from unittest import mock
from nova import exception
from nova.tests.unit.virt.hyperv import test_base
from nova.virt.hyperv import block_device_manager
from nova.virt.hyperv import constants
class BlockDeviceManagerTestCase(test_base.HyperVBaseTestCase):
"""Unit tests for the Hyper-V BlockDeviceInfoManager class."""
def setUp(self):
super(BlockDeviceManagerTestCase, self).setUp()
self._bdman = block_device_manager.BlockDeviceInfoManager()
def test_get_device_bus_scsi(self):
bdm = {'disk_bus': constants.CTRL_TYPE_SCSI,
'drive_addr': 0, 'ctrl_disk_addr': 2}
bus = self._bdman._get_device_bus(bdm)
self.assertEqual('0:0:0:2', bus.address)
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(bdm)
self.assertEqual('0:1', bus.address)
@staticmethod
def _bdm_mock(**kwargs):
bdm = mock.MagicMock(**kwargs)
bdm.__contains__.side_effect = (
lambda attr: getattr(bdm, attr, None) is not None)
return bdm
@mock.patch.object(block_device_manager.objects, 'DiskMetadata')
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'_get_device_bus')
@mock.patch.object(block_device_manager.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],
}
bdm = self._bdm_mock(device_name=mock.sentinel.dev0, tag='taggy',
volume_id=mock.sentinel.uuid1)
eph = self._bdm_mock(device_name=mock.sentinel.dev1, tag='ephy',
volume_id=mock.sentinel.uuid2)
mock_get_by_inst_uuid.return_value = [
bdm, eph, self._bdm_mock(device_name=mock.sentinel.dev2, tag=None),
]
bdm_metadata = self._bdman.get_bdm_metadata(mock.sentinel.context,
mock_instance,
block_device_info)
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,
serial=bdm.volume_id, tags=[bdm.tag]),
mock.call(bus=mock_get_device_bus.return_value,
serial=eph.volume_id, tags=[eph.tag])],
any_order=True)
self.assertEqual([mock_DiskMetadata.return_value] * 2, bdm_metadata)
@mock.patch('nova.virt.configdrive.required_by')
def test_init_controller_slot_counter_gen1_no_configdrive(
self, mock_cfg_drive_req):
mock_cfg_drive_req.return_value = False
slot_map = self._bdman._initialize_controller_slot_counter(
mock.sentinel.FAKE_INSTANCE, constants.VM_GEN_1)
self.assertEqual(slot_map[constants.CTRL_TYPE_IDE][0],
os_win_const.IDE_CONTROLLER_SLOTS_NUMBER)
self.assertEqual(slot_map[constants.CTRL_TYPE_IDE][1],
os_win_const.IDE_CONTROLLER_SLOTS_NUMBER)
self.assertEqual(slot_map[constants.CTRL_TYPE_SCSI][0],
os_win_const.SCSI_CONTROLLER_SLOTS_NUMBER)
@mock.patch('nova.virt.configdrive.required_by')
def test_init_controller_slot_counter_gen1(self, mock_cfg_drive_req):
slot_map = self._bdman._initialize_controller_slot_counter(
mock.sentinel.FAKE_INSTANCE, constants.VM_GEN_1)
self.assertEqual(slot_map[constants.CTRL_TYPE_IDE][1],
os_win_const.IDE_CONTROLLER_SLOTS_NUMBER - 1)
@mock.patch.object(block_device_manager.configdrive, 'required_by')
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'_initialize_controller_slot_counter')
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'_check_and_update_root_device')
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'_check_and_update_ephemerals')
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'_check_and_update_volumes')
def _check_validate_and_update_bdi(self, mock_check_and_update_vol,
mock_check_and_update_eph,
mock_check_and_update_root,
mock_init_ctrl_cntr,
mock_required_by, available_slots=1):
mock_required_by.return_value = True
slot_map = {constants.CTRL_TYPE_SCSI: [available_slots]}
mock_init_ctrl_cntr.return_value = slot_map
if available_slots:
self._bdman.validate_and_update_bdi(mock.sentinel.FAKE_INSTANCE,
mock.sentinel.IMAGE_META,
constants.VM_GEN_2,
mock.sentinel.BLOCK_DEV_INFO)
else:
self.assertRaises(exception.InvalidBDMFormat,
self._bdman.validate_and_update_bdi,
mock.sentinel.FAKE_INSTANCE,
mock.sentinel.IMAGE_META,
constants.VM_GEN_2,
mock.sentinel.BLOCK_DEV_INFO)
mock_init_ctrl_cntr.assert_called_once_with(
mock.sentinel.FAKE_INSTANCE, constants.VM_GEN_2)
mock_check_and_update_root.assert_called_once_with(
constants.VM_GEN_2, mock.sentinel.IMAGE_META,
mock.sentinel.BLOCK_DEV_INFO, slot_map)
mock_check_and_update_eph.assert_called_once_with(
constants.VM_GEN_2, mock.sentinel.BLOCK_DEV_INFO, slot_map)
mock_check_and_update_vol.assert_called_once_with(
constants.VM_GEN_2, mock.sentinel.BLOCK_DEV_INFO, slot_map)
mock_required_by.assert_called_once_with(mock.sentinel.FAKE_INSTANCE)
def test_validate_and_update_bdi(self):
self._check_validate_and_update_bdi()
def test_validate_and_update_bdi_insufficient_slots(self):
self._check_validate_and_update_bdi(available_slots=0)
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'_get_available_controller_slot')
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'is_boot_from_volume')
def _test_check_and_update_root_device(self, mock_is_boot_from_vol,
mock_get_avail_ctrl_slot,
disk_format,
vm_gen=constants.VM_GEN_1,
boot_from_volume=False):
image_meta = mock.MagicMock(disk_format=disk_format)
bdi = {'root_device': '/dev/sda',
'block_device_mapping': [
{'mount_device': '/dev/sda',
'connection_info': mock.sentinel.FAKE_CONN_INFO}]}
mock_is_boot_from_vol.return_value = boot_from_volume
mock_get_avail_ctrl_slot.return_value = (0, 0)
self._bdman._check_and_update_root_device(vm_gen, image_meta, bdi,
mock.sentinel.SLOT_MAP)
root_disk = bdi['root_disk']
if boot_from_volume:
self.assertEqual(root_disk['type'], constants.VOLUME)
self.assertIsNone(root_disk['path'])
self.assertEqual(root_disk['connection_info'],
mock.sentinel.FAKE_CONN_INFO)
else:
image_type = self._bdman._TYPE_FOR_DISK_FORMAT.get(
image_meta.disk_format)
self.assertEqual(root_disk['type'], image_type)
self.assertIsNone(root_disk['path'])
self.assertIsNone(root_disk['connection_info'])
disk_bus = (constants.CTRL_TYPE_IDE if
vm_gen == constants.VM_GEN_1 else constants.CTRL_TYPE_SCSI)
self.assertEqual(root_disk['disk_bus'], disk_bus)
self.assertEqual(root_disk['drive_addr'], 0)
self.assertEqual(root_disk['ctrl_disk_addr'], 0)
self.assertEqual(root_disk['boot_index'], 0)
self.assertEqual(root_disk['mount_device'], bdi['root_device'])
mock_get_avail_ctrl_slot.assert_called_once_with(
root_disk['disk_bus'], mock.sentinel.SLOT_MAP)
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'is_boot_from_volume', return_value=False)
def test_check_and_update_root_device_exception(self, mock_is_boot_vol):
bdi = {}
image_meta = mock.MagicMock(disk_format=mock.sentinel.fake_format)
self.assertRaises(exception.InvalidImageFormat,
self._bdman._check_and_update_root_device,
constants.VM_GEN_1, image_meta, bdi,
mock.sentinel.SLOT_MAP)
def test_check_and_update_root_device_gen1(self):
self._test_check_and_update_root_device(disk_format='vhd')
def test_check_and_update_root_device_gen1_vhdx(self):
self._test_check_and_update_root_device(disk_format='vhdx')
def test_check_and_update_root_device_gen1_iso(self):
self._test_check_and_update_root_device(disk_format='iso')
def test_check_and_update_root_device_gen2(self):
self._test_check_and_update_root_device(disk_format='vhd',
vm_gen=constants.VM_GEN_2)
def test_check_and_update_root_device_boot_from_vol_gen1(self):
self._test_check_and_update_root_device(disk_format='vhd',
boot_from_volume=True)
def test_check_and_update_root_device_boot_from_vol_gen2(self):
self._test_check_and_update_root_device(disk_format='vhd',
vm_gen=constants.VM_GEN_2,
boot_from_volume=True)
@mock.patch('nova.virt.configdrive.required_by', return_value=True)
def _test_get_available_controller_slot(self, mock_config_drive_req,
bus=constants.CTRL_TYPE_IDE,
fail=False):
slot_map = self._bdman._initialize_controller_slot_counter(
mock.sentinel.FAKE_VM, constants.VM_GEN_1)
if fail:
slot_map[constants.CTRL_TYPE_IDE][0] = 0
slot_map[constants.CTRL_TYPE_IDE][1] = 0
self.assertRaises(exception.InvalidBDMFormat,
self._bdman._get_available_controller_slot,
constants.CTRL_TYPE_IDE,
slot_map)
else:
(disk_addr,
ctrl_disk_addr) = self._bdman._get_available_controller_slot(
bus, slot_map)
self.assertEqual(0, disk_addr)
self.assertEqual(0, ctrl_disk_addr)
def test_get_available_controller_slot(self):
self._test_get_available_controller_slot()
def test_get_available_controller_slot_scsi_ctrl(self):
self._test_get_available_controller_slot(bus=constants.CTRL_TYPE_SCSI)
def test_get_available_controller_slot_exception(self):
self._test_get_available_controller_slot(fail=True)
def test_is_boot_from_volume_true(self):
vol = {'mount_device': self._bdman._DEFAULT_ROOT_DEVICE}
block_device_info = {'block_device_mapping': [vol]}
ret = self._bdman.is_boot_from_volume(block_device_info)
self.assertTrue(ret)
def test_is_boot_from_volume_false(self):
block_device_info = {'block_device_mapping': []}
ret = self._bdman.is_boot_from_volume(block_device_info)
self.assertFalse(ret)
def test_get_root_device_bdm(self):
mount_device = '/dev/sda'
bdm1 = {'mount_device': None}
bdm2 = {'mount_device': mount_device}
bdi = {'block_device_mapping': [bdm1, bdm2]}
ret = self._bdman._get_root_device_bdm(bdi, mount_device)
self.assertEqual(bdm2, ret)
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'_check_and_update_bdm')
def test_check_and_update_ephemerals(self, mock_check_and_update_bdm):
fake_ephemerals = [mock.sentinel.eph1, mock.sentinel.eph2,
mock.sentinel.eph3]
fake_bdi = {'ephemerals': fake_ephemerals}
expected_calls = []
for eph in fake_ephemerals:
expected_calls.append(mock.call(mock.sentinel.fake_slot_map,
mock.sentinel.fake_vm_gen,
eph))
self._bdman._check_and_update_ephemerals(mock.sentinel.fake_vm_gen,
fake_bdi,
mock.sentinel.fake_slot_map)
mock_check_and_update_bdm.assert_has_calls(expected_calls)
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'_check_and_update_bdm')
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'_get_root_device_bdm')
def test_check_and_update_volumes(self, mock_get_root_dev_bdm,
mock_check_and_update_bdm):
fake_vol1 = {'mount_device': '/dev/sda'}
fake_vol2 = {'mount_device': '/dev/sdb'}
fake_volumes = [fake_vol1, fake_vol2]
fake_bdi = {'block_device_mapping': fake_volumes,
'root_disk': {'mount_device': '/dev/sda'}}
mock_get_root_dev_bdm.return_value = fake_vol1
self._bdman._check_and_update_volumes(mock.sentinel.fake_vm_gen,
fake_bdi,
mock.sentinel.fake_slot_map)
mock_get_root_dev_bdm.assert_called_once_with(fake_bdi, '/dev/sda')
mock_check_and_update_bdm.assert_called_once_with(
mock.sentinel.fake_slot_map, mock.sentinel.fake_vm_gen, fake_vol2)
self.assertNotIn(fake_vol1, fake_bdi)
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'_get_available_controller_slot')
def test_check_and_update_bdm_with_defaults(self, mock_get_ctrl_slot):
mock_get_ctrl_slot.return_value = ((mock.sentinel.DRIVE_ADDR,
mock.sentinel.CTRL_DISK_ADDR))
bdm = {'device_type': None,
'disk_bus': None,
'boot_index': None}
self._bdman._check_and_update_bdm(mock.sentinel.FAKE_SLOT_MAP,
constants.VM_GEN_1, bdm)
mock_get_ctrl_slot.assert_called_once_with(
bdm['disk_bus'], mock.sentinel.FAKE_SLOT_MAP)
self.assertEqual(mock.sentinel.DRIVE_ADDR, bdm['drive_addr'])
self.assertEqual(mock.sentinel.CTRL_DISK_ADDR, bdm['ctrl_disk_addr'])
self.assertEqual('disk', bdm['device_type'])
self.assertEqual(self._bdman._DEFAULT_BUS, bdm['disk_bus'])
self.assertIsNone(bdm['boot_index'])
def test_check_and_update_bdm_exception_device_type(self):
bdm = {'device_type': 'cdrom',
'disk_bus': 'IDE'}
self.assertRaises(exception.InvalidDiskInfo,
self._bdman._check_and_update_bdm,
mock.sentinel.FAKE_SLOT_MAP, constants.VM_GEN_1, bdm)
def test_check_and_update_bdm_exception_disk_bus(self):
bdm = {'device_type': 'disk',
'disk_bus': 'fake_bus'}
self.assertRaises(exception.InvalidDiskInfo,
self._bdman._check_and_update_bdm,
mock.sentinel.FAKE_SLOT_MAP, constants.VM_GEN_1, bdm)
def test_sort_by_boot_order(self):
original = [{'boot_index': 2}, {'boot_index': None}, {'boot_index': 1}]
expected = [original[2], original[0], original[1]]
self._bdman._sort_by_boot_order(original)
self.assertEqual(expected, original)
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'_get_boot_order_gen1')
def test_get_boot_order_gen1_vm(self, mock_get_boot_order):
self._bdman.get_boot_order(constants.VM_GEN_1,
mock.sentinel.BLOCK_DEV_INFO)
mock_get_boot_order.assert_called_once_with(
mock.sentinel.BLOCK_DEV_INFO)
@mock.patch.object(block_device_manager.BlockDeviceInfoManager,
'_get_boot_order_gen2')
def test_get_boot_order_gen2_vm(self, mock_get_boot_order):
self._bdman.get_boot_order(constants.VM_GEN_2,
mock.sentinel.BLOCK_DEV_INFO)
mock_get_boot_order.assert_called_once_with(
mock.sentinel.BLOCK_DEV_INFO)
def test_get_boot_order_gen1_iso(self):
fake_bdi = {'root_disk': {'type': 'iso'}}
expected = [os_win_const.BOOT_DEVICE_CDROM,
os_win_const.BOOT_DEVICE_HARDDISK,
os_win_const.BOOT_DEVICE_NETWORK,
os_win_const.BOOT_DEVICE_FLOPPY]
res = self._bdman._get_boot_order_gen1(fake_bdi)
self.assertEqual(expected, res)
def test_get_boot_order_gen1_vhd(self):
fake_bdi = {'root_disk': {'type': 'vhd'}}
expected = [os_win_const.BOOT_DEVICE_HARDDISK,
os_win_const.BOOT_DEVICE_CDROM,
os_win_const.BOOT_DEVICE_NETWORK,
os_win_const.BOOT_DEVICE_FLOPPY]
res = self._bdman._get_boot_order_gen1(fake_bdi)
self.assertEqual(expected, res)
@mock.patch('nova.virt.hyperv.volumeops.VolumeOps.get_disk_resource_path')
def test_get_boot_order_gen2(self, mock_get_disk_path):
fake_root_disk = {'boot_index': 0,
'path': mock.sentinel.FAKE_ROOT_PATH}
fake_eph1 = {'boot_index': 2,
'path': mock.sentinel.FAKE_EPH_PATH1}
fake_eph2 = {'boot_index': 3,
'path': mock.sentinel.FAKE_EPH_PATH2}
fake_bdm = {'boot_index': 1,
'connection_info': mock.sentinel.FAKE_CONN_INFO}
fake_bdi = {'root_disk': fake_root_disk,
'ephemerals': [fake_eph1,
fake_eph2],
'block_device_mapping': [fake_bdm]}
mock_get_disk_path.return_value = fake_bdm['connection_info']
expected_res = [mock.sentinel.FAKE_ROOT_PATH,
mock.sentinel.FAKE_CONN_INFO,
mock.sentinel.FAKE_EPH_PATH1,
mock.sentinel.FAKE_EPH_PATH2]
res = self._bdman._get_boot_order_gen2(fake_bdi)
self.assertEqual(expected_res, res)

View File

@ -1,493 +0,0 @@
# 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 the Hyper-V Driver.
"""
import platform
import sys
from unittest import mock
from os_win import exceptions as os_win_exc
from nova import exception
from nova import safe_utils
from nova.tests.unit import fake_instance
from nova.tests.unit.virt.hyperv import test_base
from nova.virt import driver as base_driver
from nova.virt.hyperv import driver
class HyperVDriverTestCase(test_base.HyperVBaseTestCase):
FAKE_WIN_2008R2_VERSION = '6.0.0'
@mock.patch.object(driver.HyperVDriver, '_check_minimum_windows_version')
def setUp(self, mock_check_minimum_windows_version):
super(HyperVDriverTestCase, self).setUp()
self.context = 'context'
self.driver = driver.HyperVDriver(mock.sentinel.virtapi)
self.driver._hostops = mock.MagicMock()
self.driver._volumeops = mock.MagicMock()
self.driver._vmops = mock.MagicMock()
self.driver._snapshotops = mock.MagicMock()
self.driver._livemigrationops = mock.MagicMock()
self.driver._migrationops = mock.MagicMock()
self.driver._rdpconsoleops = mock.MagicMock()
self.driver._serialconsoleops = mock.MagicMock()
self.driver._imagecache = mock.MagicMock()
@mock.patch.object(driver.LOG, 'warning')
@mock.patch.object(driver.utilsfactory, 'get_hostutils')
def test_check_minimum_windows_version(self, mock_get_hostutils,
mock_warning):
mock_hostutils = mock_get_hostutils.return_value
mock_hostutils.check_min_windows_version.return_value = False
self.assertRaises(exception.HypervisorTooOld,
self.driver._check_minimum_windows_version)
mock_hostutils.check_min_windows_version.side_effect = [True, False]
self.driver._check_minimum_windows_version()
self.assertTrue(mock_warning.called)
def test_public_api_signatures(self):
# NOTE(claudiub): wrapped functions do not keep the same signature in
# Python 2.7, which causes this test to fail. Instead, we should
# compare the public API signatures of the unwrapped methods.
for attr in driver.HyperVDriver.__dict__:
class_member = getattr(driver.HyperVDriver, attr)
if callable(class_member):
mocked_method = mock.patch.object(
driver.HyperVDriver, attr,
safe_utils.get_wrapped_function(class_member))
mocked_method.start()
self.addCleanup(mocked_method.stop)
self.assertPublicAPISignatures(base_driver.ComputeDriver,
driver.HyperVDriver)
def test_converted_exception(self):
self.driver._vmops.get_info.side_effect = (
os_win_exc.OSWinException)
self.assertRaises(exception.NovaException,
self.driver.get_info, mock.sentinel.instance)
self.driver._vmops.get_info.side_effect = os_win_exc.HyperVException
self.assertRaises(exception.NovaException,
self.driver.get_info, mock.sentinel.instance)
self.driver._vmops.get_info.side_effect = (
os_win_exc.HyperVVMNotFoundException(vm_name='foofoo'))
self.assertRaises(exception.InstanceNotFound,
self.driver.get_info, mock.sentinel.instance)
def test_assert_original_traceback_maintained(self):
def bar(self):
foo = "foofoo"
raise os_win_exc.HyperVVMNotFoundException(vm_name=foo)
self.driver._vmops.get_info.side_effect = bar
try:
self.driver.get_info(mock.sentinel.instance)
self.fail("Test expected exception, but it was not raised.")
except exception.InstanceNotFound:
# exception has been raised as expected.
_, _, trace = sys.exc_info()
while trace.tb_next:
# iterate until the original exception source, bar.
trace = trace.tb_next
# original frame will contain the 'foo' variable.
self.assertEqual('foofoo', trace.tb_frame.f_locals['foo'])
@mock.patch.object(driver.eventhandler, 'InstanceEventHandler')
def test_init_host(self, mock_InstanceEventHandler):
self.driver.init_host(mock.sentinel.host)
mock_start_console_handlers = (
self.driver._serialconsoleops.start_console_handlers)
mock_start_console_handlers.assert_called_once_with()
mock_InstanceEventHandler.assert_called_once_with(
state_change_callback=self.driver.emit_event)
fake_event_handler = mock_InstanceEventHandler.return_value
fake_event_handler.start_listener.assert_called_once_with()
def test_list_instance_uuids(self):
self.driver.list_instance_uuids()
self.driver._vmops.list_instance_uuids.assert_called_once_with()
def test_list_instances(self):
self.driver.list_instances()
self.driver._vmops.list_instances.assert_called_once_with()
def test_spawn(self):
self.driver.spawn(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.image_meta, mock.sentinel.injected_files,
mock.sentinel.admin_password, mock.sentinel.allocations,
mock.sentinel.network_info, mock.sentinel.block_device_info)
self.driver._vmops.spawn.assert_called_once_with(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.image_meta, mock.sentinel.injected_files,
mock.sentinel.admin_password, mock.sentinel.network_info,
mock.sentinel.block_device_info)
def test_reboot(self):
self.driver.reboot(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.network_info, mock.sentinel.reboot_type,
mock.sentinel.block_device_info, mock.sentinel.bad_vol_callback)
self.driver._vmops.reboot.assert_called_once_with(
mock.sentinel.instance, mock.sentinel.network_info,
mock.sentinel.reboot_type)
def test_destroy(self):
self.driver.destroy(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.network_info, mock.sentinel.block_device_info,
mock.sentinel.destroy_disks)
self.driver._vmops.destroy.assert_called_once_with(
mock.sentinel.instance, mock.sentinel.network_info,
mock.sentinel.block_device_info, mock.sentinel.destroy_disks)
def test_cleanup(self):
self.driver.cleanup(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.network_info, mock.sentinel.block_device_info,
mock.sentinel.destroy_disks, mock.sentinel.migrate_data,
mock.sentinel.destroy_vifs)
self.driver._vmops.unplug_vifs.assert_called_once_with(
mock.sentinel.instance, mock.sentinel.network_info)
def test_get_info(self):
self.driver.get_info(mock.sentinel.instance)
self.driver._vmops.get_info.assert_called_once_with(
mock.sentinel.instance)
def test_attach_volume(self):
mock_instance = fake_instance.fake_instance_obj(self.context)
self.driver.attach_volume(
mock.sentinel.context, mock.sentinel.connection_info,
mock_instance, mock.sentinel.mountpoint, mock.sentinel.disk_bus,
mock.sentinel.device_type, mock.sentinel.encryption)
self.driver._volumeops.attach_volume.assert_called_once_with(
mock.sentinel.connection_info,
mock_instance.name)
def test_detach_volume(self):
mock_instance = fake_instance.fake_instance_obj(self.context)
self.driver.detach_volume(
mock.sentinel.context, mock.sentinel.connection_info,
mock_instance, mock.sentinel.mountpoint, mock.sentinel.encryption)
self.driver._volumeops.detach_volume.assert_called_once_with(
mock.sentinel.connection_info,
mock_instance.name)
def test_get_volume_connector(self):
self.driver.get_volume_connector(mock.sentinel.instance)
self.driver._volumeops.get_volume_connector.assert_called_once_with()
def test_get_available_resource(self):
self.driver.get_available_resource(mock.sentinel.nodename)
self.driver._hostops.get_available_resource.assert_called_once_with()
def test_get_available_nodes(self):
response = self.driver.get_available_nodes(mock.sentinel.refresh)
self.assertEqual([platform.node()], response)
def test_host_power_action(self):
self.driver.host_power_action(mock.sentinel.action)
self.driver._hostops.host_power_action.assert_called_once_with(
mock.sentinel.action)
def test_snapshot(self):
self.driver.snapshot(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.image_id, mock.sentinel.update_task_state)
self.driver._snapshotops.snapshot.assert_called_once_with(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.image_id, mock.sentinel.update_task_state)
def test_pause(self):
self.driver.pause(mock.sentinel.instance)
self.driver._vmops.pause.assert_called_once_with(
mock.sentinel.instance)
def test_unpause(self):
self.driver.unpause(mock.sentinel.instance)
self.driver._vmops.unpause.assert_called_once_with(
mock.sentinel.instance)
def test_suspend(self):
self.driver.suspend(mock.sentinel.context, mock.sentinel.instance)
self.driver._vmops.suspend.assert_called_once_with(
mock.sentinel.instance)
def test_resume(self):
self.driver.resume(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.network_info, mock.sentinel.block_device_info)
self.driver._vmops.resume.assert_called_once_with(
mock.sentinel.instance)
def test_power_off(self):
self.driver.power_off(
mock.sentinel.instance, mock.sentinel.timeout,
mock.sentinel.retry_interval)
self.driver._vmops.power_off.assert_called_once_with(
mock.sentinel.instance, mock.sentinel.timeout,
mock.sentinel.retry_interval)
def test_power_on(self):
self.driver.power_on(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.network_info, mock.sentinel.block_device_info)
self.driver._vmops.power_on.assert_called_once_with(
mock.sentinel.instance, mock.sentinel.block_device_info,
mock.sentinel.network_info)
def test_resume_state_on_host_boot(self):
self.driver.resume_state_on_host_boot(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.network_info, mock.sentinel.block_device_info)
self.driver._vmops.resume_state_on_host_boot.assert_called_once_with(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.network_info, mock.sentinel.block_device_info)
def test_live_migration(self):
self.driver.live_migration(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.dest, mock.sentinel.post_method,
mock.sentinel.recover_method, mock.sentinel.block_migration,
mock.sentinel.migrate_data)
self.driver._livemigrationops.live_migration.assert_called_once_with(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.dest, mock.sentinel.post_method,
mock.sentinel.recover_method, mock.sentinel.block_migration,
mock.sentinel.migrate_data)
@mock.patch.object(driver.HyperVDriver, 'destroy')
def test_rollback_live_migration_at_destination(self, mock_destroy):
self.driver.rollback_live_migration_at_destination(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.network_info, mock.sentinel.block_device_info,
mock.sentinel.destroy_disks, mock.sentinel.migrate_data)
mock_destroy.assert_called_once_with(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.network_info, mock.sentinel.block_device_info,
destroy_disks=mock.sentinel.destroy_disks)
def test_pre_live_migration(self):
migrate_data = self.driver.pre_live_migration(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.block_device_info, mock.sentinel.network_info,
mock.sentinel.disk_info, mock.sentinel.migrate_data)
self.assertEqual(mock.sentinel.migrate_data, migrate_data)
pre_live_migration = self.driver._livemigrationops.pre_live_migration
pre_live_migration.assert_called_once_with(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.block_device_info, mock.sentinel.network_info)
def test_post_live_migration(self):
self.driver.post_live_migration(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.block_device_info, mock.sentinel.migrate_data)
post_live_migration = self.driver._livemigrationops.post_live_migration
post_live_migration.assert_called_once_with(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.block_device_info,
mock.sentinel.migrate_data)
def test_post_live_migration_at_destination(self):
self.driver.post_live_migration_at_destination(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.network_info, mock.sentinel.block_migration,
mock.sentinel.block_device_info)
mtd = self.driver._livemigrationops.post_live_migration_at_destination
mtd.assert_called_once_with(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.network_info, mock.sentinel.block_migration)
def test_check_can_live_migrate_destination(self):
self.driver.check_can_live_migrate_destination(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.src_compute_info, mock.sentinel.dst_compute_info,
mock.sentinel.block_migration, mock.sentinel.disk_over_commit)
mtd = self.driver._livemigrationops.check_can_live_migrate_destination
mtd.assert_called_once_with(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.src_compute_info, mock.sentinel.dst_compute_info,
mock.sentinel.block_migration, mock.sentinel.disk_over_commit)
def test_cleanup_live_migration_destination_check(self):
self.driver.cleanup_live_migration_destination_check(
mock.sentinel.context, mock.sentinel.dest_check_data)
_livemigrops = self.driver._livemigrationops
method = _livemigrops.cleanup_live_migration_destination_check
method.assert_called_once_with(
mock.sentinel.context, mock.sentinel.dest_check_data)
def test_check_can_live_migrate_source(self):
self.driver.check_can_live_migrate_source(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.dest_check_data, mock.sentinel.block_device_info)
method = self.driver._livemigrationops.check_can_live_migrate_source
method.assert_called_once_with(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.dest_check_data)
def test_plug_vifs(self):
self.driver.plug_vifs(
mock.sentinel.instance, mock.sentinel.network_info)
self.driver._vmops.plug_vifs.assert_called_once_with(
mock.sentinel.instance, mock.sentinel.network_info)
def test_unplug_vifs(self):
self.driver.unplug_vifs(
mock.sentinel.instance, mock.sentinel.network_info)
self.driver._vmops.unplug_vifs.assert_called_once_with(
mock.sentinel.instance, mock.sentinel.network_info)
def test_migrate_disk_and_power_off(self):
self.driver.migrate_disk_and_power_off(
mock.sentinel.context, mock.sentinel.instance, mock.sentinel.dest,
mock.sentinel.flavor, mock.sentinel.network_info,
mock.sentinel.block_device_info, mock.sentinel.timeout,
mock.sentinel.retry_interval)
migr_power_off = self.driver._migrationops.migrate_disk_and_power_off
migr_power_off.assert_called_once_with(
mock.sentinel.context, mock.sentinel.instance, mock.sentinel.dest,
mock.sentinel.flavor, mock.sentinel.network_info,
mock.sentinel.block_device_info, mock.sentinel.timeout,
mock.sentinel.retry_interval)
def test_confirm_migration(self):
self.driver.confirm_migration(
mock.sentinel.context,
mock.sentinel.migration, mock.sentinel.instance,
mock.sentinel.network_info)
self.driver._migrationops.confirm_migration.assert_called_once_with(
mock.sentinel.context,
mock.sentinel.migration, mock.sentinel.instance,
mock.sentinel.network_info)
def test_finish_revert_migration(self):
self.driver.finish_revert_migration(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.network_info, mock.sentinel.migration,
mock.sentinel.block_device_info, mock.sentinel.power_on)
finish_revert_migr = self.driver._migrationops.finish_revert_migration
finish_revert_migr.assert_called_once_with(
mock.sentinel.context, mock.sentinel.instance,
mock.sentinel.network_info, mock.sentinel.block_device_info,
mock.sentinel.power_on)
def test_finish_migration(self):
self.driver.finish_migration(
mock.sentinel.context, mock.sentinel.migration,
mock.sentinel.instance, mock.sentinel.disk_info,
mock.sentinel.network_info, mock.sentinel.image_meta,
mock.sentinel.resize_instance, mock.sentinel.allocations,
mock.sentinel.block_device_info,
mock.sentinel.power_on)
self.driver._migrationops.finish_migration