#  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

from eventlet import timeout as etimeout
import mock
from nova.compute import vm_states
from nova import exception
from nova.objects import flavor as flavor_obj
from nova.tests.unit.objects import test_flavor
from nova.tests.unit.objects import test_virtual_interface
from nova.virt import hardware
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_utils import fileutils
from oslo_utils import units

from hyperv.nova import constants
from hyperv.nova import vmops
from hyperv.nova import vmutils
from hyperv.nova import volumeops
from hyperv.tests import fake_instance
from hyperv.tests.unit import test_base

CONF = cfg.CONF


class VMOpsTestCase(test_base.HyperVBaseTestCase):
    """Unit tests for the Hyper-V VMOps class."""

    _FAKE_TIMEOUT = 2
    FAKE_SIZE = 10
    FAKE_DIR = 'fake_dir'
    FAKE_ROOT_PATH = 'C:\\path\\to\\fake.%s'
    FAKE_CONFIG_DRIVE_ISO = 'configdrive.iso'
    FAKE_CONFIG_DRIVE_VHD = 'configdrive.vhd'
    FAKE_UUID = '4f54fb69-d3a2-45b7-bb9b-b6e6b3d893b3'
    FAKE_LOG = 'fake_log'
    _WIN_VERSION_6_3 = '6.3.0'
    _WIN_VERSION_6_4 = '6.4.0'

    ISO9660 = 'iso9660'
    _FAKE_CONFIGDRIVE_PATH = 'C:/fake_instance_dir/configdrive.vhd'

    def setUp(self):
        super(VMOpsTestCase, self).setUp()
        self.context = 'fake-context'

        self._vmops = vmops.VMOps()
        self._vmops._vmutils = mock.MagicMock()
        self._vmops._vhdutils = mock.MagicMock()
        self._vmops._pathutils = mock.MagicMock()
        self._vmops._hostutils = mock.MagicMock()
        self._vmops._serial_console_ops = mock.MagicMock()

    def test_get_vif_driver_cached(self):
        self._vmops._vif_driver_cache = mock.MagicMock()
        self._vmops._vif_driver_cache.get.return_value = mock.sentinel.VIF_DRV

        self._vmops._get_vif_driver(mock.sentinel.VIF_TYPE)
        self._vmops._vif_driver_cache.get.assert_called_with(
            mock.sentinel.VIF_TYPE)

    @mock.patch('hyperv.nova.vif.get_vif_driver')
    def test_get_vif_driver_not_cached(self, mock_get_vif_driver):
        mock_get_vif_driver.return_value = mock.sentinel.VIF_DRV

        self._vmops._get_vif_driver(mock.sentinel.VIF_TYPE)
        mock_get_vif_driver.assert_called_once_with(mock.sentinel.VIF_TYPE)
        self.assertEqual(mock.sentinel.VIF_DRV,
                self._vmops._vif_driver_cache[mock.sentinel.VIF_TYPE])

    def test_list_instances(self):
        mock_instance = mock.MagicMock()
        self._vmops._vmutils.list_instances.return_value = [mock_instance]
        response = self._vmops.list_instances()
        self._vmops._vmutils.list_instances.assert_called_once_with()
        self.assertEqual(response, [mock_instance])

    def _test_get_info(self, vm_exists):
        mock_instance = fake_instance.fake_instance_obj(self.context)
        mock_info = mock.MagicMock(spec_set=dict)
        fake_info = {'EnabledState': 2,
                     'MemoryUsage': mock.sentinel.FAKE_MEM_KB,
                     'NumberOfProcessors': mock.sentinel.FAKE_NUM_CPU,
                     'UpTime': mock.sentinel.FAKE_CPU_NS}

        def getitem(key):
            return fake_info[key]
        mock_info.__getitem__.side_effect = getitem

        expected = hardware.InstanceInfo(state=constants.HYPERV_POWER_STATE[2],
                                         max_mem_kb=mock.sentinel.FAKE_MEM_KB,
                                         mem_kb=mock.sentinel.FAKE_MEM_KB,
                                         num_cpu=mock.sentinel.FAKE_NUM_CPU,
                                         cpu_time_ns=mock.sentinel.FAKE_CPU_NS)

        self._vmops._vmutils.vm_exists.return_value = vm_exists
        self._vmops._vmutils.get_vm_summary_info.return_value = mock_info

        if not vm_exists:
            self.assertRaises(exception.InstanceNotFound,
                              self._vmops.get_info, mock_instance)
        else:
            response = self._vmops.get_info(mock_instance)
            self._vmops._vmutils.vm_exists.assert_called_once_with(
                mock_instance.name)
            self._vmops._vmutils.get_vm_summary_info.assert_called_once_with(
                mock_instance.name)
            self.assertEqual(response, expected)

    def test_get_info(self):
        self._test_get_info(vm_exists=True)

    def test_get_info_exception(self):
        self._test_get_info(vm_exists=False)

    def _prepare_create_root_vhd_mocks(self, use_cow_images, vhd_format,
                                       vhd_size):
        mock_instance = fake_instance.fake_instance_obj(self.context)
        mock_instance.root_gb = self.FAKE_SIZE
        self.flags(use_cow_images=use_cow_images)
        self._vmops._vhdutils.get_vhd_info.return_value = {'MaxInternalSize':
                                                           vhd_size * units.Gi}
        self._vmops._vhdutils.get_vhd_format.return_value = vhd_format
        root_vhd_internal_size = mock_instance.root_gb * units.Gi
        get_size = self._vmops._vhdutils.get_internal_vhd_size_by_file_size
        get_size.return_value = root_vhd_internal_size
        self._vmops._pathutils.exists.return_value = True

        return mock_instance

    @mock.patch('hyperv.nova.imagecache.ImageCache.get_cached_image')
    def _test_create_root_vhd_exception(self, mock_get_cached_image,
                                        vhd_format):
        mock_instance = self._prepare_create_root_vhd_mocks(
            use_cow_images=False, vhd_format=vhd_format,
            vhd_size=(self.FAKE_SIZE + 1))
        fake_vhd_path = self.FAKE_ROOT_PATH % vhd_format
        mock_get_cached_image.return_value = fake_vhd_path
        fake_root_path = self._vmops._pathutils.get_root_vhd_path.return_value

        self.assertRaises(vmutils.VHDResizeException,
                          self._vmops._create_root_vhd, self.context,
                          mock_instance)

        self.assertFalse(self._vmops._vhdutils.resize_vhd.called)
        self._vmops._pathutils.exists.assert_called_once_with(
            fake_root_path)
        self._vmops._pathutils.remove.assert_called_once_with(
            fake_root_path)

    @mock.patch('hyperv.nova.imagecache.ImageCache.get_cached_image')
    def _test_create_root_vhd_qcow(self, mock_get_cached_image, vhd_format):
        mock_instance = self._prepare_create_root_vhd_mocks(
            use_cow_images=True, vhd_format=vhd_format,
            vhd_size=(self.FAKE_SIZE - 1))
        fake_vhd_path = self.FAKE_ROOT_PATH % vhd_format
        mock_get_cached_image.return_value = fake_vhd_path

        fake_root_path = self._vmops._pathutils.get_root_vhd_path.return_value
        root_vhd_internal_size = mock_instance.root_gb * units.Gi
        get_size = self._vmops._vhdutils.get_internal_vhd_size_by_file_size

        response = self._vmops._create_root_vhd(context=self.context,
                                                instance=mock_instance)

        self.assertEqual(fake_root_path, response)
        self._vmops._pathutils.get_root_vhd_path.assert_called_with(
            mock_instance.name, vhd_format, False)
        differencing_vhd = self._vmops._vhdutils.create_differencing_vhd
        differencing_vhd.assert_called_with(fake_root_path, fake_vhd_path)
        self._vmops._vhdutils.get_vhd_info.assert_called_once_with(
            fake_vhd_path)

        if vhd_format is constants.DISK_FORMAT_VHD:
            self.assertFalse(get_size.called)
            self.assertFalse(self._vmops._vhdutils.resize_vhd.called)
        else:
            get_size.assert_called_once_with(fake_vhd_path,
                                             root_vhd_internal_size)
            self._vmops._vhdutils.resize_vhd.assert_called_once_with(
                fake_root_path, root_vhd_internal_size, is_file_max_size=False)

    @mock.patch('hyperv.nova.imagecache.ImageCache.get_cached_image')
    def _test_create_root_vhd(self, mock_get_cached_image, vhd_format,
                              is_rescue_vhd=False):
        mock_instance = self._prepare_create_root_vhd_mocks(
            use_cow_images=False, vhd_format=vhd_format,
            vhd_size=(self.FAKE_SIZE - 1))
        fake_vhd_path = self.FAKE_ROOT_PATH % vhd_format
        mock_get_cached_image.return_value = fake_vhd_path
        rescue_image_id = (
            mock.sentinel.rescue_image_id if is_rescue_vhd else None)

        fake_root_path = self._vmops._pathutils.get_root_vhd_path.return_value
        root_vhd_internal_size = mock_instance.root_gb * units.Gi
        get_size = self._vmops._vhdutils.get_internal_vhd_size_by_file_size

        response = self._vmops._create_root_vhd(
            context=self.context,
            instance=mock_instance,
            rescue_image_id=rescue_image_id)

        self.assertEqual(fake_root_path, response)
        mock_get_cached_image.assert_called_once_with(self.context,
                                                      mock_instance,
                                                      rescue_image_id)
        self._vmops._pathutils.get_root_vhd_path.assert_called_with(
            mock_instance.name, vhd_format, is_rescue_vhd)

        self._vmops._pathutils.copyfile.assert_called_once_with(
            fake_vhd_path, fake_root_path)
        get_size.assert_called_once_with(fake_vhd_path, root_vhd_internal_size)
        if is_rescue_vhd:
            self.assertFalse(self._vmops._vhdutils.resize_vhd.called)
        else:
            self._vmops._vhdutils.resize_vhd.assert_called_once_with(
                fake_root_path, root_vhd_internal_size,
                is_file_max_size=False)

    def test_create_root_vhd(self):
        self._test_create_root_vhd(vhd_format=constants.DISK_FORMAT_VHD)

    def test_create_root_vhdx(self):
        self._test_create_root_vhd(vhd_format=constants.DISK_FORMAT_VHDX)

    def test_create_root_vhd_use_cow_images_true(self):
        self._test_create_root_vhd_qcow(vhd_format=constants.DISK_FORMAT_VHD)

    def test_create_root_vhdx_use_cow_images_true(self):
        self._test_create_root_vhd_qcow(vhd_format=constants.DISK_FORMAT_VHDX)

    def test_create_rescue_vhd(self):
        self._test_create_root_vhd(vhd_format=constants.DISK_FORMAT_VHD,
                                   is_rescue_vhd=True)

    def test_create_root_vhdx_size_less_than_internal(self):
        self._test_create_root_vhd_exception(
            vhd_format=constants.DISK_FORMAT_VHD)

    def test_is_resize_needed_exception(self):
        inst = mock.MagicMock()
        self.assertRaises(
            vmutils.VHDResizeException, self._vmops._is_resize_needed,
            mock.sentinel.FAKE_PATH, self.FAKE_SIZE, self.FAKE_SIZE - 1, inst)

    def test_is_resize_needed_true(self):
        inst = mock.MagicMock()
        self.assertTrue(self._vmops._is_resize_needed(
            mock.sentinel.FAKE_PATH, self.FAKE_SIZE, self.FAKE_SIZE + 1, inst))

    def test_is_resize_needed_false(self):
        inst = mock.MagicMock()
        self.assertFalse(self._vmops._is_resize_needed(
            mock.sentinel.FAKE_PATH, self.FAKE_SIZE, self.FAKE_SIZE, inst))

    def test_create_ephemeral_vhd(self):
        mock_instance = fake_instance.fake_instance_obj(self.context)
        mock_instance.ephemeral_gb = self.FAKE_SIZE
        best_supported = self._vmops._vhdutils.get_best_supported_vhd_format
        best_supported.return_value = mock.sentinel.FAKE_FORMAT
        self._vmops._pathutils.get_ephemeral_vhd_path.return_value = (
            mock.sentinel.FAKE_PATH)

        response = self._vmops.create_ephemeral_vhd(instance=mock_instance)

        self._vmops._pathutils.get_ephemeral_vhd_path.assert_called_with(
            mock_instance.name, mock.sentinel.FAKE_FORMAT)
        self._vmops._vhdutils.create_dynamic_vhd.assert_called_with(
            mock.sentinel.FAKE_PATH, mock_instance.ephemeral_gb * units.Gi,
            mock.sentinel.FAKE_FORMAT)
        self.assertEqual(mock.sentinel.FAKE_PATH, response)

    @mock.patch('hyperv.nova.vmops.VMOps.destroy')
    @mock.patch('hyperv.nova.vmops.VMOps.power_on')
    @mock.patch('hyperv.nova.vmops.VMOps.attach_config_drive')
    @mock.patch('hyperv.nova.vmops.VMOps._create_config_drive')
    @mock.patch('nova.virt.configdrive.required_by')
    @mock.patch('hyperv.nova.vmops.VMOps.create_instance')
    @mock.patch('hyperv.nova.vmops.VMOps.get_image_vm_generation')
    @mock.patch('hyperv.nova.vmops.VMOps.create_ephemeral_vhd')
    @mock.patch('hyperv.nova.vmops.VMOps._create_root_vhd')
    @mock.patch('hyperv.nova.volumeops.VolumeOps.'
                'ebs_root_in_block_devices')
    @mock.patch('hyperv.nova.vmops.VMOps._delete_disk_files')
    @mock.patch('hyperv.nova.vif.get_vif_driver')
    def _test_spawn(self, mock_get_vif_driver, mock_delete_disk_files,
                    mock_ebs_root_in_block_devices, mock_create_root_vhd,
                    mock_create_ephemeral_vhd, mock_get_image_vm_gen,
                    mock_create_instance, mock_configdrive_required,
                    mock_create_config_drive, mock_attach_config_drive,
                    mock_power_on, mock_destroy, exists, boot_from_volume,
                    configdrive_required, fail):
        mock_instance = fake_instance.fake_instance_obj(self.context)
        mock_image_meta = mock.MagicMock()

        fake_root_path = mock_create_root_vhd.return_value
        fake_root_path = None if boot_from_volume else fake_root_path
        fake_ephemeral_path = mock_create_ephemeral_vhd.return_value
        fake_vm_gen = mock_get_image_vm_gen.return_value
        fake_config_drive_path = mock_create_config_drive.return_value
        fake_network_info = {'id': mock.sentinel.ID,
                             'address': mock.sentinel.ADDRESS}

        self._vmops._vmutils.vm_exists.return_value = exists
        mock_ebs_root_in_block_devices.return_value = boot_from_volume
        mock_create_root_vhd.return_value = fake_root_path
        mock_configdrive_required.return_value = configdrive_required
        mock_create_instance.side_effect = fail
        if exists:
            self.assertRaises(exception.InstanceExists, self._vmops.spawn,
                              self.context, mock_instance, mock_image_meta,
                              [mock.sentinel.FILE], mock.sentinel.PASSWORD,
                              mock.sentinel.INFO, mock.sentinel.DEV_INFO)
        elif fail is vmutils.HyperVException:
            self.assertRaises(vmutils.HyperVException, self._vmops.spawn,
                              self.context, mock_instance, mock_image_meta,
                              [mock.sentinel.FILE], mock.sentinel.PASSWORD,
                              mock.sentinel.INFO, mock.sentinel.DEV_INFO)
            mock_destroy.assert_called_once_with(mock_instance)
        else:
            self._vmops.spawn(self.context, mock_instance, mock_image_meta,
                              [mock.sentinel.FILE], mock.sentinel.PASSWORD,
                              [fake_network_info], mock.sentinel.DEV_INFO)
            self._vmops._vmutils.vm_exists.assert_called_once_with(
                mock_instance.name)
            mock_delete_disk_files.assert_called_once_with(
                mock_instance.name)
            mock_ebs_root_in_block_devices.assert_called_once_with(
                mock.sentinel.DEV_INFO)
            if not boot_from_volume:
                mock_create_root_vhd.assert_called_once_with(self.context,
                                                             mock_instance)
            mock_create_ephemeral_vhd.assert_called_once_with(mock_instance)
            mock_get_image_vm_gen.assert_called_once_with(fake_root_path,
                                                          mock_image_meta)
            mock_create_instance.assert_called_once_with(
                mock_instance, [fake_network_info], mock.sentinel.DEV_INFO,
                fake_root_path, fake_ephemeral_path, fake_vm_gen,
                mock_image_meta)
            mock_configdrive_required.assert_called_once_with(mock_instance)
            if configdrive_required:
                mock_create_config_drive.assert_called_once_with(
                    mock_instance, [mock.sentinel.FILE],
                    mock.sentinel.PASSWORD,
                    [fake_network_info])
                mock_attach_config_drive.assert_called_once_with(
                    mock_instance, fake_config_drive_path, fake_vm_gen)
            mock_power_on.assert_called_once_with(
                mock_instance, network_info=[fake_network_info])

    def test_spawn(self):
        self._test_spawn(exists=False, boot_from_volume=False,
                         configdrive_required=True, fail=None)

    def test_spawn_instance_exists(self):
        self._test_spawn(exists=True, boot_from_volume=False,
                         configdrive_required=True, fail=None)

    def test_spawn_create_instance_exception(self):
        self._test_spawn(exists=False, boot_from_volume=False,
                         configdrive_required=True,
                         fail=vmutils.HyperVException)

    def test_spawn_not_required(self):
        self._test_spawn(exists=False, boot_from_volume=False,
                         configdrive_required=False, fail=None)

    def test_spawn_root_in_block(self):
        self._test_spawn(exists=False, boot_from_volume=True,
                         configdrive_required=False, fail=None)

    def test_spawn_no_admin_permissions(self):
        self._vmops._vmutils.check_admin_permissions.side_effect = (
            vmutils.HyperVException)
        self.assertRaises(vmutils.HyperVException,
                          self._vmops.spawn,
                          self.context, mock.DEFAULT, mock.DEFAULT,
                          [mock.sentinel.FILE], mock.sentinel.PASSWORD,
                          mock.sentinel.INFO, mock.sentinel.DEV_INFO)

    @mock.patch('hyperv.nova.vif.get_vif_driver')
    @mock.patch.object(vmops.VMOps, '_set_instance_disk_qos_specs')
    @mock.patch.object(volumeops.VolumeOps, 'attach_volumes')
    @mock.patch.object(vmops.VMOps, '_attach_drive')
    @mock.patch.object(vmops.VMOps, '_get_image_serial_port_settings')
    @mock.patch.object(vmops.VMOps, '_create_vm_com_port_pipes')
    @mock.patch.object(vmops.VMOps, '_configure_remotefx')
    def _test_create_instance(self, mock_configure_remotefx, mock_create_pipes,
                              mock_get_port_settings, mock_attach_drive,
                              mock_attach_volumes, mock_set_qos_specs,
                              mock_get_vif_driver,
                              fake_root_path, fake_ephemeral_path,
                              enable_instance_metrics,
                              vm_gen=constants.VM_GEN_1, remotefx=False):
        mock_vif_driver = mock_get_vif_driver()
        self.flags(enable_instance_metrics_collection=enable_instance_metrics,
                   group='hyperv')
        fake_network_info = {'id': mock.sentinel.ID,
                             'address': mock.sentinel.ADDRESS}
        mock_instance = fake_instance.fake_instance_obj(self.context)
        instance_path = os.path.join(CONF.instances_path, mock_instance.name)
        flavor = flavor_obj.Flavor(**test_flavor.fake_flavor)
        if remotefx is True:
            flavor.extra_specs['hyperv:remotefx'] = "1920x1200,2"
        mock_instance.flavor = flavor

        if remotefx is True and vm_gen == constants.VM_GEN_2:
            self.assertRaises(vmutils.HyperVException,
                              self._vmops.create_instance,
                              instance=mock_instance,
                              network_info=[fake_network_info],
                              block_device_info=mock.sentinel.DEV_INFO,
                              root_vhd_path=fake_root_path,
                              eph_vhd_path=fake_ephemeral_path,
                              vm_gen=vm_gen,
                              image_meta=mock.sentinel.image_meta)
        else:
            self._vmops.create_instance(
                    instance=mock_instance,
                    network_info=[fake_network_info],
                    block_device_info=mock.sentinel.DEV_INFO,
                    root_vhd_path=fake_root_path,
                    eph_vhd_path=fake_ephemeral_path,
                    vm_gen=vm_gen,
                    image_meta=mock.sentinel.image_meta)
            if remotefx is True:
                mock_configure_remotefx.assert_called_once_with(
                    mock_instance,
                    flavor.extra_specs['hyperv:remotefx'])

            self._vmops._vmutils.create_vm.assert_called_once_with(
                mock_instance.name, mock_instance.memory_mb,
                mock_instance.vcpus, CONF.hyperv.limit_cpu_features,
                CONF.hyperv.dynamic_memory_ratio, vm_gen, instance_path,
                [mock_instance.uuid])
            expected = []
            ctrl_type = vmops.VM_GENERATIONS_CONTROLLER_TYPES[vm_gen]
            ctrl_disk_addr = 0
            if fake_root_path:
                expected.append(mock.call(mock_instance.name, fake_root_path,
                                          0, ctrl_disk_addr, ctrl_type,
                                          constants.DISK))
                ctrl_disk_addr += 1
            if fake_ephemeral_path:
                expected.append(mock.call(mock_instance.name,
                                          fake_ephemeral_path, 0,
                                          ctrl_disk_addr,
                                          ctrl_type, constants.DISK))
            mock_attach_drive.has_calls(expected)
            mock_create_scsi_ctrl = self._vmops._vmutils.create_scsi_controller
            mock_create_scsi_ctrl.assert_called_once_with(mock_instance.name)

            ebs_root = (
                vm_gen is not constants.VM_GEN_2 and fake_root_path is None)
            mock_attach_volumes.assert_called_once_with(mock.sentinel.DEV_INFO,
                                                        mock_instance.name,
                                                        ebs_root)

            mock_get_port_settings.assert_called_with(mock.sentinel.image_meta)
            mock_create_pipes.assert_called_once_with(
                mock_instance, mock_get_port_settings.return_value)

            self._vmops._vmutils.create_nic.assert_called_once_with(
                mock_instance.name, mock.sentinel.ID, mock.sentinel.ADDRESS)
            mock_vif_driver.plug.assert_called_once_with(mock_instance,
                                                         fake_network_info)
            mock_enable = self._vmops._vmutils.enable_vm_metrics_collection
            if enable_instance_metrics:
                mock_enable.assert_called_once_with(mock_instance.name)
            mock_set_qos_specs.assert_called_once_with(mock_instance)

    def test_create_instance(self):
        fake_ephemeral_path = mock.sentinel.FAKE_EPHEMERAL_PATH
        self._test_create_instance(fake_root_path=mock.sentinel.FAKE_ROOT_PATH,
                                   fake_ephemeral_path=fake_ephemeral_path,
                                   enable_instance_metrics=True)

    def test_create_instance_no_root_path(self):
        fake_ephemeral_path = mock.sentinel.FAKE_EPHEMERAL_PATH
        self._test_create_instance(fake_root_path=None,
                                   fake_ephemeral_path=fake_ephemeral_path,
                                   enable_instance_metrics=True)

    def test_create_instance_no_ephemeral_path(self):
        self._test_create_instance(fake_root_path=mock.sentinel.FAKE_ROOT_PATH,
                                   fake_ephemeral_path=None,
                                   enable_instance_metrics=True)

    def test_create_instance_no_path(self):
        self._test_create_instance(fake_root_path=None,
                                   fake_ephemeral_path=None,
                                   enable_instance_metrics=False)

    def test_create_instance_enable_instance_metrics_false(self):
        fake_ephemeral_path = mock.sentinel.FAKE_EPHEMERAL_PATH
        self._test_create_instance(fake_root_path=mock.sentinel.FAKE_ROOT_PATH,
                                   fake_ephemeral_path=fake_ephemeral_path,
                                   enable_instance_metrics=False)

    def test_create_instance_gen2(self):
        self._test_create_instance(fake_root_path=None,
                                   fake_ephemeral_path=None,
                                   enable_instance_metrics=False,
                                   vm_gen=constants.VM_GEN_2)

    def test_create_instance_with_remote_fx(self):
        self._test_create_instance(fake_root_path=None,
                                   fake_ephemeral_path=None,
                                   enable_instance_metrics=False,
                                   vm_gen=constants.VM_GEN_1,
                                   remotefx=True)

    def test_create_instance_with_remote_fx_gen2(self):
        self._test_create_instance(fake_root_path=None,
                                   fake_ephemeral_path=None,
                                   enable_instance_metrics=False,
                                   vm_gen=constants.VM_GEN_2,
                                   remotefx=True)

    def test_attach_drive_vm_to_scsi(self):
        self._vmops._attach_drive(
            mock.sentinel.FAKE_VM_NAME, mock.sentinel.FAKE_PATH,
            mock.sentinel.FAKE_DRIVE_ADDR, mock.sentinel.FAKE_CTRL_DISK_ADDR,
            constants.CTRL_TYPE_SCSI)

        self._vmops._vmutils.attach_scsi_drive.assert_called_once_with(
            mock.sentinel.FAKE_VM_NAME, mock.sentinel.FAKE_PATH,
            constants.DISK)

    def test_attach_drive_vm_to_ide(self):
        self._vmops._attach_drive(
            mock.sentinel.FAKE_VM_NAME, mock.sentinel.FAKE_PATH,
            mock.sentinel.FAKE_DRIVE_ADDR, mock.sentinel.FAKE_CTRL_DISK_ADDR,
            constants.CTRL_TYPE_IDE)

        self._vmops._vmutils.attach_ide_drive.assert_called_once_with(
            mock.sentinel.FAKE_VM_NAME, mock.sentinel.FAKE_PATH,
            mock.sentinel.FAKE_DRIVE_ADDR, mock.sentinel.FAKE_CTRL_DISK_ADDR,
            constants.DISK)

    def _check_get_image_vm_gen_except(self, image_prop):
        image_meta = {"properties": {constants.IMAGE_PROP_VM_GEN: image_prop}}
        self._vmops._hostutils.get_supported_vm_types.return_value = [
            constants.IMAGE_PROP_VM_GEN_1, constants.IMAGE_PROP_VM_GEN_2]

        self.assertRaises(vmutils.HyperVException,
                          self._vmops.get_image_vm_generation,
                          mock.sentinel.FAKE_PATH,
                          image_meta)

    def test_get_image_vm_generation_default(self):
        image_meta = {"properties": {}}
        self._vmops._hostutils.get_default_vm_generation.return_value = (
            constants.IMAGE_PROP_VM_GEN_1)
        self._vmops._hostutils.get_supported_vm_types.return_value = [
            constants.IMAGE_PROP_VM_GEN_1, constants.IMAGE_PROP_VM_GEN_2]

        response = self._vmops.get_image_vm_generation(mock.sentinel.FAKE_PATH,
                                                       image_meta)

        self.assertEqual(constants.VM_GEN_1, response)

    def test_get_image_vm_generation_gen2(self):
        image_meta = {"properties": {
            constants.IMAGE_PROP_VM_GEN: constants.IMAGE_PROP_VM_GEN_2}}
        self._vmops._hostutils.get_supported_vm_types.return_value = [
            constants.IMAGE_PROP_VM_GEN_1, constants.IMAGE_PROP_VM_GEN_2]
        self._vmops._vhdutils.get_vhd_format.return_value = (
            constants.DISK_FORMAT_VHDX)

        response = self._vmops.get_image_vm_generation(mock.sentinel.FAKE_PATH,
                                                       image_meta)

        self.assertEqual(constants.VM_GEN_2, response)

    def test_get_image_vm_generation_bad_prop(self):
        self._check_get_image_vm_gen_except(mock.sentinel.FAKE_IMAGE_PROP)

    def test_get_image_vm_generation_not_vhdx(self):
        self._vmops._vhdutils.get_vhd_format.return_value = (
            constants.DISK_FORMAT_VHD)
        self._check_get_image_vm_gen_except(constants.IMAGE_PROP_VM_GEN_2)

    @mock.patch('nova.api.metadata.base.InstanceMetadata')
    @mock.patch('nova.virt.configdrive.ConfigDriveBuilder')
    @mock.patch('nova.utils.execute')
    def _test_create_config_drive(self, mock_execute, mock_ConfigDriveBuilder,
                                  mock_InstanceMetadata, config_drive_format,
                                  config_drive_cdrom, side_effect,
                                  rescue=False):
        mock_instance = fake_instance.fake_instance_obj(self.context)
        self.flags(config_drive_format=config_drive_format)
        self.flags(config_drive_cdrom=config_drive_cdrom, group='hyperv')
        self.flags(config_drive_inject_password=True, group='hyperv')
        mock_ConfigDriveBuilder().__enter__().make_drive.side_effect = [
            side_effect]

        path_iso = os.path.join(self.FAKE_DIR, self.FAKE_CONFIG_DRIVE_ISO)
        path_vhd = os.path.join(self.FAKE_DIR, self.FAKE_CONFIG_DRIVE_VHD)

        def fake_get_configdrive_path(instance_name, disk_format,
                                      rescue=False):
            return (path_iso
                    if disk_format == constants.DVD_FORMAT else path_vhd)

        mock_get_configdrive_path = self._vmops._pathutils.get_configdrive_path
        mock_get_configdrive_path.side_effect = fake_get_configdrive_path
        expected_get_configdrive_path_calls = [mock.call(mock_instance.name,
                                                         constants.DVD_FORMAT,
                                                         rescue=rescue)]
        if not config_drive_cdrom:
            expected_call = mock.call(mock_instance.name,
                                      constants.DISK_FORMAT_VHD,
                                      rescue=rescue)
            expected_get_configdrive_path_calls.append(expected_call)

        if config_drive_format != self.ISO9660:
            self.assertRaises(vmutils.UnsupportedConfigDriveFormatException,
                              self._vmops._create_config_drive,
                              mock_instance, [mock.sentinel.FILE],
                              mock.sentinel.PASSWORD,
                              mock.sentinel.NET_INFO,
                              rescue)
        elif side_effect is processutils.ProcessExecutionError:
            self.assertRaises(processutils.ProcessExecutionError,
                              self._vmops._create_config_drive,
                              mock_instance, [mock.sentinel.FILE],
                              mock.sentinel.PASSWORD,
                              mock.sentinel.NET_INFO,
                              rescue)
        else:
            path = self._vmops._create_config_drive(mock_instance,
                                                    [mock.sentinel.FILE],
                                                    mock.sentinel.PASSWORD,
                                                    mock.sentinel.NET_INFO,
                                                    rescue)
            mock_InstanceMetadata.assert_called_once_with(
                mock_instance, content=[mock.sentinel.FILE],
                extra_md={'admin_pass': mock.sentinel.PASSWORD},
                network_info=mock.sentinel.NET_INFO)
            mock_get_configdrive_path.assert_has_calls(
                expected_get_configdrive_path_calls)
            mock_ConfigDriveBuilder.assert_called_with(
                instance_md=mock_InstanceMetadata())
            mock_make_drive = mock_ConfigDriveBuilder().__enter__().make_drive
            mock_make_drive.assert_called_once_with(path_iso)
            if not CONF.hyperv.config_drive_cdrom:
                expected = path_vhd
                mock_execute.assert_called_once_with(
                    CONF.hyperv.qemu_img_cmd,
                    'convert', '-f', 'raw', '-O', 'vpc',
                    path_iso, path_vhd, attempts=1)
                self._vmops._pathutils.remove.assert_called_once_with(
                    os.path.join(self.FAKE_DIR, self.FAKE_CONFIG_DRIVE_ISO))
            else:
                expected = path_iso

            self.assertEqual(expected, path)

    def test_create_config_drive_cdrom(self):
        self._test_create_config_drive(config_drive_format=self.ISO9660,
                                       config_drive_cdrom=True,
                                       side_effect=None)

    def test_create_config_drive_vhd(self):
        self._test_create_config_drive(config_drive_format=self.ISO9660,
                                       config_drive_cdrom=False,
                                       side_effect=None)

    def test_create_rescue_config_drive_vhd(self):
        self._test_create_config_drive(config_drive_format=self.ISO9660,
                                       config_drive_cdrom=False,
                                       side_effect=None,
                                       rescue=True)

    def test_create_config_drive_other_drive_format(self):
        self._test_create_config_drive(config_drive_format=mock.sentinel.OTHER,
                                       config_drive_cdrom=False,
                                       side_effect=None)

    def test_create_config_drive_execution_error(self):
        self._test_create_config_drive(
            config_drive_format=self.ISO9660,
            config_drive_cdrom=False,
            side_effect=processutils.ProcessExecutionError)

    def test_attach_config_drive_exception(self):
        instance = fake_instance.fake_instance_obj(self.context)
        self.assertRaises(exception.InvalidDiskFormat,
                          self._vmops.attach_config_drive,
                          instance, 'C:/fake_instance_dir/configdrive.xxx',
                          constants.VM_GEN_1)

    @mock.patch.object(vmops.VMOps, '_attach_drive')
    def test_attach_config_drive(self, mock_attach_drive):
        instance = fake_instance.fake_instance_obj(self.context)
        self._vmops.attach_config_drive(instance,
                                        self._FAKE_CONFIGDRIVE_PATH,
                                        constants.VM_GEN_1)
        mock_attach_drive.assert_called_once_with(
            instance.name, self._FAKE_CONFIGDRIVE_PATH,
            1, 0, constants.CTRL_TYPE_IDE, constants.DISK)

    @mock.patch.object(vmops.VMOps, '_attach_drive')
    def test_attach_config_drive_gen2(self, mock_attach_drive):
        instance = fake_instance.fake_instance_obj(self.context)
        self._vmops.attach_config_drive(instance,
                                        self._FAKE_CONFIGDRIVE_PATH,
                                        constants.VM_GEN_2)
        mock_attach_drive.assert_called_once_with(
            instance.name, self._FAKE_CONFIGDRIVE_PATH,
            1, 0, constants.CTRL_TYPE_SCSI, constants.DISK)

    def test_detach_config_drive(self):
        is_rescue_configdrive = True
        mock_lookup_configdrive = (