Matt Riedemann
29dab997b4
Hyper-V supports UEFI SecureBoot since the 2012 R2 version for Windows guests and this has been extended to Linux guests as well with the upcoming release. This blueprint implements UEFI SecureBoot for Linux guests. DocImpact: The nova flavor extra specs docs needs to be updated to include 'os:secure_boot' and its possible values. The image metadata property docs needs to be updated to include "os_secure_boot" property and its possible values. Co-Authored-By: Claudiu Belu <cbelu@cloudbasesolutions.com> Implements: blueprint hyper-v-uefi-secureboot Change-Id: I1ea96930018d997820df2b7b4640fe1f241ee8d6
539 lines
25 KiB
Python
539 lines
25 KiB
Python
# Copyright 2014 IBM Corp.
|
|
#
|
|
# 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 os
|
|
|
|
import mock
|
|
from os_win import exceptions as os_win_exc
|
|
from oslo_utils import units
|
|
|
|
from nova import exception
|
|
from nova import objects
|
|
from nova import test
|
|
from nova.tests.unit import fake_instance
|
|
from nova.tests.unit.virt.hyperv import test_base
|
|
from nova.virt.hyperv import constants
|
|
from nova.virt.hyperv import migrationops
|
|
|
|
|
|
class MigrationOpsTestCase(test_base.HyperVBaseTestCase):
|
|
"""Unit tests for the Hyper-V MigrationOps class."""
|
|
|
|
_FAKE_DISK = 'fake_disk'
|
|
_FAKE_TIMEOUT = 10
|
|
_FAKE_RETRY_INTERVAL = 5
|
|
|
|
def setUp(self):
|
|
super(MigrationOpsTestCase, self).setUp()
|
|
self.context = 'fake-context'
|
|
|
|
self._migrationops = migrationops.MigrationOps()
|
|
self._migrationops._hostutils = mock.MagicMock()
|
|
self._migrationops._vmops = mock.MagicMock()
|
|
self._migrationops._vmutils = mock.MagicMock()
|
|
self._migrationops._pathutils = mock.Mock()
|
|
self._migrationops._vhdutils = mock.MagicMock()
|
|
self._migrationops._pathutils = mock.MagicMock()
|
|
self._migrationops._volumeops = mock.MagicMock()
|
|
self._migrationops._imagecache = mock.MagicMock()
|
|
self._migrationops._block_dev_man = mock.MagicMock()
|
|
|
|
def _check_migrate_disk_files(self, shared_storage=False):
|
|
instance_path = 'fake/instance/path'
|
|
dest_instance_path = 'remote/instance/path'
|
|
self._migrationops._pathutils.get_instance_dir.side_effect = (
|
|
instance_path, dest_instance_path)
|
|
get_revert_dir = (
|
|
self._migrationops._pathutils.get_instance_migr_revert_dir)
|
|
check_shared_storage = (
|
|
self._migrationops._pathutils.check_dirs_shared_storage)
|
|
check_shared_storage.return_value = shared_storage
|
|
self._migrationops._pathutils.exists.return_value = True
|
|
|
|
fake_disk_files = [os.path.join(instance_path, disk_name)
|
|
for disk_name in
|
|
['root.vhd', 'configdrive.vhd', 'configdrive.iso',
|
|
'eph0.vhd', 'eph1.vhdx']]
|
|
|
|
expected_get_dir = [mock.call(mock.sentinel.instance_name),
|
|
mock.call(mock.sentinel.instance_name,
|
|
mock.sentinel.dest_path)]
|
|
expected_move_calls = [mock.call(instance_path,
|
|
get_revert_dir.return_value)]
|
|
|
|
self._migrationops._migrate_disk_files(
|
|
instance_name=mock.sentinel.instance_name,
|
|
disk_files=fake_disk_files,
|
|
dest=mock.sentinel.dest_path)
|
|
|
|
self._migrationops._pathutils.exists.assert_called_once_with(
|
|
dest_instance_path)
|
|
check_shared_storage.assert_called_once_with(
|
|
instance_path, dest_instance_path)
|
|
get_revert_dir.assert_called_with(mock.sentinel.instance_name,
|
|
remove_dir=True, create_dir=True)
|
|
if shared_storage:
|
|
fake_dest_path = '%s_tmp' % instance_path
|
|
expected_move_calls.append(mock.call(fake_dest_path,
|
|
instance_path))
|
|
self._migrationops._pathutils.rmtree.assert_called_once_with(
|
|
fake_dest_path)
|
|
else:
|
|
fake_dest_path = dest_instance_path
|
|
|
|
self._migrationops._pathutils.makedirs.assert_called_once_with(
|
|
fake_dest_path)
|
|
check_remove_dir = self._migrationops._pathutils.check_remove_dir
|
|
check_remove_dir.assert_called_once_with(fake_dest_path)
|
|
|
|
self._migrationops._pathutils.get_instance_dir.assert_has_calls(
|
|
expected_get_dir)
|
|
self._migrationops._pathutils.copy.assert_has_calls(
|
|
mock.call(fake_disk_file, fake_dest_path)
|
|
for fake_disk_file in fake_disk_files)
|
|
self.assertEqual(len(fake_disk_files),
|
|
self._migrationops._pathutils.copy.call_count)
|
|
self._migrationops._pathutils.move_folder_files.assert_has_calls(
|
|
expected_move_calls)
|
|
|
|
def test_migrate_disk_files(self):
|
|
self._check_migrate_disk_files()
|
|
|
|
def test_migrate_disk_files_same_host(self):
|
|
self._check_migrate_disk_files(shared_storage=True)
|
|
|
|
@mock.patch.object(migrationops.MigrationOps,
|
|
'_cleanup_failed_disk_migration')
|
|
def test_migrate_disk_files_exception(self, mock_cleanup):
|
|
instance_path = 'fake/instance/path'
|
|
fake_dest_path = '%s_tmp' % instance_path
|
|
self._migrationops._pathutils.get_instance_dir.return_value = (
|
|
instance_path)
|
|
get_revert_dir = (
|
|
self._migrationops._pathutils.get_instance_migr_revert_dir)
|
|
self._migrationops._hostutils.get_local_ips.return_value = [
|
|
mock.sentinel.dest_path]
|
|
self._migrationops._pathutils.copy.side_effect = IOError(
|
|
"Expected exception.")
|
|
|
|
self.assertRaises(IOError, self._migrationops._migrate_disk_files,
|
|
instance_name=mock.sentinel.instance_name,
|
|
disk_files=[self._FAKE_DISK],
|
|
dest=mock.sentinel.dest_path)
|
|
mock_cleanup.assert_called_once_with(instance_path,
|
|
get_revert_dir.return_value,
|
|
fake_dest_path)
|
|
|
|
def test_cleanup_failed_disk_migration(self):
|
|
self._migrationops._pathutils.exists.return_value = True
|
|
|
|
self._migrationops._cleanup_failed_disk_migration(
|
|
instance_path=mock.sentinel.instance_path,
|
|
revert_path=mock.sentinel.revert_path,
|
|
dest_path=mock.sentinel.dest_path)
|
|
|
|
expected = [mock.call(mock.sentinel.dest_path),
|
|
mock.call(mock.sentinel.revert_path)]
|
|
self._migrationops._pathutils.exists.assert_has_calls(expected)
|
|
move_folder_files = self._migrationops._pathutils.move_folder_files
|
|
move_folder_files.assert_called_once_with(
|
|
mock.sentinel.revert_path, mock.sentinel.instance_path)
|
|
self._migrationops._pathutils.rmtree.assert_has_calls([
|
|
mock.call(mock.sentinel.dest_path),
|
|
mock.call(mock.sentinel.revert_path)])
|
|
|
|
def test_check_target_flavor(self):
|
|
mock_instance = fake_instance.fake_instance_obj(self.context)
|
|
mock_instance.flavor.root_gb = 1
|
|
mock_flavor = mock.MagicMock(root_gb=0)
|
|
self.assertRaises(exception.InstanceFaultRollback,
|
|
self._migrationops._check_target_flavor,
|
|
mock_instance, mock_flavor)
|
|
|
|
def test_check_and_attach_config_drive(self):
|
|
mock_instance = fake_instance.fake_instance_obj(
|
|
self.context, expected_attrs=['system_metadata'])
|
|
mock_instance.config_drive = 'True'
|
|
|
|
self._migrationops._check_and_attach_config_drive(
|
|
mock_instance, mock.sentinel.vm_gen)
|
|
|
|
self._migrationops._vmops.attach_config_drive.assert_called_once_with(
|
|
mock_instance,
|
|
self._migrationops._pathutils.lookup_configdrive_path.return_value,
|
|
mock.sentinel.vm_gen)
|
|
|
|
def test_check_and_attach_config_drive_unknown_path(self):
|
|
instance = fake_instance.fake_instance_obj(
|
|
self.context, expected_attrs=['system_metadata'])
|
|
instance.config_drive = 'True'
|
|
self._migrationops._pathutils.lookup_configdrive_path.return_value = (
|
|
None)
|
|
self.assertRaises(exception.ConfigDriveNotFound,
|
|
self._migrationops._check_and_attach_config_drive,
|
|
instance,
|
|
mock.sentinel.FAKE_VM_GEN)
|
|
|
|
@mock.patch.object(migrationops.MigrationOps, '_migrate_disk_files')
|
|
@mock.patch.object(migrationops.MigrationOps, '_check_target_flavor')
|
|
def test_migrate_disk_and_power_off(self, mock_check_flavor,
|
|
mock_migrate_disk_files):
|
|
instance = fake_instance.fake_instance_obj(self.context)
|
|
flavor = mock.MagicMock()
|
|
network_info = mock.MagicMock()
|
|
|
|
disk_files = [mock.MagicMock()]
|
|
volume_drives = [mock.MagicMock()]
|
|
|
|
mock_get_vm_st_path = self._migrationops._vmutils.get_vm_storage_paths
|
|
mock_get_vm_st_path.return_value = (disk_files, volume_drives)
|
|
|
|
self._migrationops.migrate_disk_and_power_off(
|
|
self.context, instance, mock.sentinel.FAKE_DEST, flavor,
|
|
network_info, None, self._FAKE_TIMEOUT, self._FAKE_RETRY_INTERVAL)
|
|
|
|
mock_check_flavor.assert_called_once_with(instance, flavor)
|
|
self._migrationops._vmops.power_off.assert_called_once_with(
|
|
instance, self._FAKE_TIMEOUT, self._FAKE_RETRY_INTERVAL)
|
|
mock_get_vm_st_path.assert_called_once_with(instance.name)
|
|
mock_migrate_disk_files.assert_called_once_with(
|
|
instance.name, disk_files, mock.sentinel.FAKE_DEST)
|
|
self._migrationops._vmops.destroy.assert_called_once_with(
|
|
instance, destroy_disks=False)
|
|
|
|
def test_confirm_migration(self):
|
|
mock_instance = fake_instance.fake_instance_obj(self.context)
|
|
self._migrationops.confirm_migration(
|
|
migration=mock.sentinel.migration, instance=mock_instance,
|
|
network_info=mock.sentinel.network_info)
|
|
get_instance_migr_revert_dir = (
|
|
self._migrationops._pathutils.get_instance_migr_revert_dir)
|
|
get_instance_migr_revert_dir.assert_called_with(mock_instance.name,
|
|
remove_dir=True)
|
|
|
|
def test_revert_migration_files(self):
|
|
instance_path = (
|
|
self._migrationops._pathutils.get_instance_dir.return_value)
|
|
get_revert_dir = (
|
|
self._migrationops._pathutils.get_instance_migr_revert_dir)
|
|
|
|
self._migrationops._revert_migration_files(
|
|
instance_name=mock.sentinel.instance_name)
|
|
|
|
self._migrationops._pathutils.get_instance_dir.assert_called_once_with(
|
|
mock.sentinel.instance_name, create_dir=False, remove_dir=True)
|
|
get_revert_dir.assert_called_with(mock.sentinel.instance_name)
|
|
self._migrationops._pathutils.rename.assert_called_once_with(
|
|
get_revert_dir.return_value, instance_path)
|
|
|
|
@mock.patch.object(migrationops.MigrationOps,
|
|
'_check_and_attach_config_drive')
|
|
@mock.patch.object(migrationops.MigrationOps, '_revert_migration_files')
|
|
@mock.patch.object(migrationops.MigrationOps, '_check_ephemeral_disks')
|
|
@mock.patch.object(objects.ImageMeta, "from_instance")
|
|
def _check_finish_revert_migration(self, mock_image,
|
|
mock_check_eph_disks,
|
|
mock_revert_migration_files,
|
|
mock_check_attach_config_drive,
|
|
disk_type=constants.DISK):
|
|
mock_image.return_value = objects.ImageMeta.from_dict({})
|
|
mock_instance = fake_instance.fake_instance_obj(self.context)
|
|
root_device = {'type': disk_type}
|
|
block_device_info = {'root_disk': root_device, 'ephemerals': []}
|
|
|
|
self._migrationops.finish_revert_migration(
|
|
context=self.context, instance=mock_instance,
|
|
network_info=mock.sentinel.network_info,
|
|
block_device_info=block_device_info,
|
|
power_on=True)
|
|
|
|
mock_revert_migration_files.assert_called_once_with(
|
|
mock_instance.name)
|
|
if root_device['type'] == constants.DISK:
|
|
lookup_root_vhd = (
|
|
self._migrationops._pathutils.lookup_root_vhd_path)
|
|
lookup_root_vhd.assert_called_once_with(mock_instance.name)
|
|
self.assertEqual(lookup_root_vhd.return_value,
|
|
root_device['path'])
|
|
|
|
get_image_vm_gen = self._migrationops._vmops.get_image_vm_generation
|
|
get_image_vm_gen.assert_called_once_with(
|
|
mock_instance.uuid, test.MatchType(objects.ImageMeta))
|
|
self._migrationops._vmops.create_instance.assert_called_once_with(
|
|
mock_instance, mock.sentinel.network_info, root_device,
|
|
block_device_info, get_image_vm_gen.return_value,
|
|
mock_image.return_value)
|
|
mock_check_attach_config_drive.assert_called_once_with(
|
|
mock_instance, get_image_vm_gen.return_value)
|
|
self._migrationops._vmops.power_on.assert_called_once_with(
|
|
mock_instance)
|
|
|
|
def test_finish_revert_migration_boot_from_volume(self):
|
|
self._check_finish_revert_migration(disk_type=constants.VOLUME)
|
|
|
|
def test_finish_revert_migration_boot_from_disk(self):
|
|
self._check_finish_revert_migration(disk_type=constants.DISK)
|
|
|
|
@mock.patch.object(objects.ImageMeta, "from_instance")
|
|
def test_finish_revert_migration_no_root_vhd(self, mock_image):
|
|
mock_instance = fake_instance.fake_instance_obj(self.context)
|
|
self._migrationops._pathutils.lookup_root_vhd_path.return_value = None
|
|
bdi = {'root_disk': {'type': constants.DISK},
|
|
'ephemerals': []}
|
|
|
|
self.assertRaises(
|
|
exception.DiskNotFound,
|
|
self._migrationops.finish_revert_migration, self.context,
|
|
mock_instance, mock.sentinel.network_info, bdi, True)
|
|
|
|
def test_merge_base_vhd(self):
|
|
fake_diff_vhd_path = 'fake/diff/path'
|
|
fake_base_vhd_path = 'fake/base/path'
|
|
base_vhd_copy_path = os.path.join(
|
|
os.path.dirname(fake_diff_vhd_path),
|
|
os.path.basename(fake_base_vhd_path))
|
|
|
|
self._migrationops._merge_base_vhd(diff_vhd_path=fake_diff_vhd_path,
|
|
base_vhd_path=fake_base_vhd_path)
|
|
|
|
self._migrationops._pathutils.copyfile.assert_called_once_with(
|
|
fake_base_vhd_path, base_vhd_copy_path)
|
|
recon_parent_vhd = self._migrationops._vhdutils.reconnect_parent_vhd
|
|
recon_parent_vhd.assert_called_once_with(fake_diff_vhd_path,
|
|
base_vhd_copy_path)
|
|
self._migrationops._vhdutils.merge_vhd.assert_called_once_with(
|
|
fake_diff_vhd_path)
|
|
self._migrationops._pathutils.rename.assert_called_once_with(
|
|
base_vhd_copy_path, fake_diff_vhd_path)
|
|
|
|
def test_merge_base_vhd_exception(self):
|
|
fake_diff_vhd_path = 'fake/diff/path'
|
|
fake_base_vhd_path = 'fake/base/path'
|
|
base_vhd_copy_path = os.path.join(
|
|
os.path.dirname(fake_diff_vhd_path),
|
|
os.path.basename(fake_base_vhd_path))
|
|
|
|
self._migrationops._vhdutils.reconnect_parent_vhd.side_effect = (
|
|
os_win_exc.HyperVException)
|
|
self._migrationops._pathutils.exists.return_value = True
|
|
|
|
self.assertRaises(os_win_exc.HyperVException,
|
|
self._migrationops._merge_base_vhd,
|
|
fake_diff_vhd_path, fake_base_vhd_path)
|
|
self._migrationops._pathutils.exists.assert_called_once_with(
|
|
base_vhd_copy_path)
|
|
self._migrationops._pathutils.remove.assert_called_once_with(
|
|
base_vhd_copy_path)
|
|
|
|
@mock.patch.object(migrationops.MigrationOps, '_resize_vhd')
|
|
def test_check_resize_vhd(self, mock_resize_vhd):
|
|
self._migrationops._check_resize_vhd(
|
|
vhd_path=mock.sentinel.vhd_path, vhd_info={'VirtualSize': 1},
|
|
new_size=2)
|
|
mock_resize_vhd.assert_called_once_with(mock.sentinel.vhd_path, 2)
|
|
|
|
def test_check_resize_vhd_exception(self):
|
|
self.assertRaises(exception.CannotResizeDisk,
|
|
self._migrationops._check_resize_vhd,
|
|
mock.sentinel.vhd_path,
|
|
{'VirtualSize': 1}, 0)
|
|
|
|
@mock.patch.object(migrationops.MigrationOps, '_merge_base_vhd')
|
|
def test_resize_vhd(self, mock_merge_base_vhd):
|
|
fake_vhd_path = 'fake/path.vhd'
|
|
new_vhd_size = 2
|
|
self._migrationops._resize_vhd(vhd_path=fake_vhd_path,
|
|
new_size=new_vhd_size)
|
|
|
|
get_vhd_parent_path = self._migrationops._vhdutils.get_vhd_parent_path
|
|
get_vhd_parent_path.assert_called_once_with(fake_vhd_path)
|
|
mock_merge_base_vhd.assert_called_once_with(
|
|
fake_vhd_path,
|
|
self._migrationops._vhdutils.get_vhd_parent_path.return_value)
|
|
self._migrationops._vhdutils.resize_vhd.assert_called_once_with(
|
|
fake_vhd_path, new_vhd_size)
|
|
|
|
def test_check_base_disk(self):
|
|
mock_instance = fake_instance.fake_instance_obj(self.context)
|
|
fake_src_vhd_path = 'fake/src/path'
|
|
fake_base_vhd = 'fake/vhd'
|
|
get_cached_image = self._migrationops._imagecache.get_cached_image
|
|
get_cached_image.return_value = fake_base_vhd
|
|
|
|
self._migrationops._check_base_disk(
|
|
context=self.context, instance=mock_instance,
|
|
diff_vhd_path=mock.sentinel.diff_vhd_path,
|
|
src_base_disk_path=fake_src_vhd_path)
|
|
|
|
get_cached_image.assert_called_once_with(self.context, mock_instance)
|
|
recon_parent_vhd = self._migrationops._vhdutils.reconnect_parent_vhd
|
|
recon_parent_vhd.assert_called_once_with(
|
|
mock.sentinel.diff_vhd_path, fake_base_vhd)
|
|
|
|
@mock.patch.object(migrationops.MigrationOps,
|
|
'_check_and_attach_config_drive')
|
|
@mock.patch.object(migrationops.MigrationOps, '_check_base_disk')
|
|
@mock.patch.object(migrationops.MigrationOps, '_check_resize_vhd')
|
|
@mock.patch.object(migrationops.MigrationOps, '_check_ephemeral_disks')
|
|
def _check_finish_migration(self, mock_check_eph_disks,
|
|
mock_check_resize_vhd,
|
|
mock_check_base_disk,
|
|
mock_check_attach_config_drive,
|
|
disk_type=constants.DISK):
|
|
mock_instance = fake_instance.fake_instance_obj(self.context)
|
|
mock_instance.flavor.ephemeral_gb = 1
|
|
root_device = {'type': disk_type}
|
|
block_device_info = {'root_disk': root_device, 'ephemerals': []}
|
|
|
|
lookup_root_vhd = self._migrationops._pathutils.lookup_root_vhd_path
|
|
get_vhd_info = self._migrationops._vhdutils.get_vhd_info
|
|
mock_vhd_info = get_vhd_info.return_value
|
|
|
|
expected_check_resize = []
|
|
expected_get_info = []
|
|
|
|
self._migrationops.finish_migration(
|
|
context=self.context, migration=mock.sentinel.migration,
|
|
instance=mock_instance, disk_info=mock.sentinel.disk_info,
|
|
network_info=mock.sentinel.network_info,
|
|
image_meta=mock.sentinel.image_meta, resize_instance=True,
|
|
block_device_info=block_device_info)
|
|
|
|
if root_device['type'] == constants.DISK:
|
|
root_device_path = lookup_root_vhd.return_value
|
|
lookup_root_vhd.assert_called_with(mock_instance.name)
|
|
expected_get_info = [mock.call(root_device_path)]
|
|
mock_vhd_info.get.assert_called_once_with("ParentPath")
|
|
mock_check_base_disk.assert_called_once_with(
|
|
self.context, mock_instance, root_device_path,
|
|
mock_vhd_info.get.return_value)
|
|
expected_check_resize.append(
|
|
mock.call(root_device_path, mock_vhd_info,
|
|
mock_instance.flavor.root_gb * units.Gi))
|
|
|
|
ephemerals = block_device_info['ephemerals']
|
|
mock_check_eph_disks.assert_called_once_with(
|
|
mock_instance, ephemerals, True)
|
|
|
|
mock_check_resize_vhd.assert_has_calls(expected_check_resize)
|
|
self._migrationops._vhdutils.get_vhd_info.assert_has_calls(
|
|
expected_get_info)
|
|
get_image_vm_gen = self._migrationops._vmops.get_image_vm_generation
|
|
get_image_vm_gen.assert_called_once_with(mock_instance.uuid,
|
|
mock.sentinel.image_meta)
|
|
self._migrationops._vmops.create_instance.assert_called_once_with(
|
|
mock_instance, mock.sentinel.network_info, root_device,
|
|
block_device_info, get_image_vm_gen.return_value,
|
|
mock.sentinel.image_meta)
|
|
mock_check_attach_config_drive.assert_called_once_with(
|
|
mock_instance, get_image_vm_gen.return_value)
|
|
self._migrationops._vmops.power_on.assert_called_once_with(
|
|
mock_instance)
|
|
|
|
def test_finish_migration(self):
|
|
self._check_finish_migration(disk_type=constants.DISK)
|
|
|
|
def test_finish_migration_boot_from_volume(self):
|
|
self._check_finish_migration(disk_type=constants.VOLUME)
|
|
|
|
def test_finish_migration_no_root(self):
|
|
mock_instance = fake_instance.fake_instance_obj(self.context)
|
|
self._migrationops._pathutils.lookup_root_vhd_path.return_value = None
|
|
bdi = {'root_disk': {'type': constants.DISK},
|
|
'ephemerals': []}
|
|
|
|
self.assertRaises(exception.DiskNotFound,
|
|
self._migrationops.finish_migration,
|
|
self.context, mock.sentinel.migration,
|
|
mock_instance, mock.sentinel.disk_info,
|
|
mock.sentinel.network_info,
|
|
mock.sentinel.image_meta, True, bdi, True)
|
|
|
|
@mock.patch.object(migrationops.MigrationOps, '_check_resize_vhd')
|
|
@mock.patch.object(migrationops.LOG, 'warning')
|
|
def test_check_ephemeral_disks_multiple_eph_warn(self, mock_warn,
|
|
mock_check_resize_vhd):
|
|
mock_instance = fake_instance.fake_instance_obj(self.context)
|
|
mock_instance.ephemeral_gb = 3
|
|
mock_ephemerals = [{'size': 1}, {'size': 1}]
|
|
|
|
self._migrationops._check_ephemeral_disks(mock_instance,
|
|
mock_ephemerals,
|
|
True)
|
|
|
|
mock_warn.assert_called_once_with(
|
|
"Cannot resize multiple ephemeral disks for instance.",
|
|
instance=mock_instance)
|
|
|
|
def test_check_ephemeral_disks_exception(self):
|
|
mock_instance = fake_instance.fake_instance_obj(self.context)
|
|
mock_ephemerals = [dict()]
|
|
|
|
lookup_eph_path = (
|
|
self._migrationops._pathutils.lookup_ephemeral_vhd_path)
|
|
lookup_eph_path.return_value = None
|
|
|
|
self.assertRaises(exception.DiskNotFound,
|
|
self._migrationops._check_ephemeral_disks,
|
|
mock_instance, mock_ephemerals)
|
|
|
|
@mock.patch.object(migrationops.MigrationOps, '_check_resize_vhd')
|
|
def _test_check_ephemeral_disks(self, mock_check_resize_vhd,
|
|
existing_eph_path=None, new_eph_size=42):
|
|
mock_instance = fake_instance.fake_instance_obj(self.context)
|
|
mock_instance.ephemeral_gb = new_eph_size
|
|
eph = {}
|
|
mock_ephemerals = [eph]
|
|
|
|
mock_pathutils = self._migrationops._pathutils
|
|
lookup_eph_path = mock_pathutils.lookup_ephemeral_vhd_path
|
|
lookup_eph_path.return_value = existing_eph_path
|
|
mock_get_eph_vhd_path = mock_pathutils.get_ephemeral_vhd_path
|
|
mock_get_eph_vhd_path.return_value = mock.sentinel.get_path
|
|
|
|
mock_vhdutils = self._migrationops._vhdutils
|
|
mock_get_vhd_format = mock_vhdutils.get_best_supported_vhd_format
|
|
mock_get_vhd_format.return_value = mock.sentinel.vhd_format
|
|
|
|
self._migrationops._check_ephemeral_disks(mock_instance,
|
|
mock_ephemerals,
|
|
True)
|
|
|
|
self.assertEqual(mock_instance.ephemeral_gb, eph['size'])
|
|
if not existing_eph_path:
|
|
mock_vmops = self._migrationops._vmops
|
|
mock_vmops.create_ephemeral_disk.assert_called_once_with(
|
|
mock_instance.name, eph)
|
|
self.assertEqual(mock.sentinel.vhd_format, eph['format'])
|
|
self.assertEqual(mock.sentinel.get_path, eph['path'])
|
|
elif new_eph_size:
|
|
mock_check_resize_vhd.assert_called_once_with(
|
|
existing_eph_path,
|
|
self._migrationops._vhdutils.get_vhd_info.return_value,
|
|
mock_instance.ephemeral_gb * units.Gi)
|
|
self.assertEqual(existing_eph_path, eph['path'])
|
|
else:
|
|
self._migrationops._pathutils.remove.assert_called_once_with(
|
|
existing_eph_path)
|
|
|
|
def test_check_ephemeral_disks_create(self):
|
|
self._test_check_ephemeral_disks()
|
|
|
|
def test_check_ephemeral_disks_resize(self):
|
|
self._test_check_ephemeral_disks(existing_eph_path=mock.sentinel.path)
|
|
|
|
def test_check_ephemeral_disks_remove(self):
|
|
self._test_check_ephemeral_disks(existing_eph_path=mock.sentinel.path,
|
|
new_eph_size=0)
|