openstack/deb-nova
Code Issues Proposed changes

Hyper-V: Adds RDPConsoleOps unit tests

Browse Source 
The Hyper-V driver oldest tests (test_hypervapi.py) are proving hard to
maintain.

The tests in test_hypervapi.py in particular can also be refactored and
split in separate TestCases, one of each *ops module.

Adds RDPConsoleOps unit tests.
Removes test_hypervapi, as it does not contain any tests anymore.

This is the final commit, it removes the rest of test_hypervapi, its
fakes and stubs.

Change-Id: I8b583875438aac6082be75553942c7fe0b9dfbd7
This commit is contained in:
Claudiu Belu
2014-12-06 15:16:00 +02:00
parent bffdec0f0f
commit 63d65d9b8b
4 changed files with 47 additions and 446 deletions
Download Patch File Download Diff File Expand all files Collapse all files

166
nova/tests/unit/virt/hyperv/db_fakes.py
View File

@@ -1,166 +0,0 @@
# Copyright 2012 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.
"""
Stubouts, mocks and fixtures for the test suite
"""
import uuid
from nova.compute import task_states
from nova.compute import vm_states
from nova import db
from nova import utils
from nova.virt.hyperv import constants
def get_fake_instance_data(name, project_id, user_id):
return {'name': name,
'id': 1,
'uuid': str(uuid.uuid4()),
'project_id': project_id,
'user_id': user_id,
'image_ref': "1",
'kernel_id': "1",
'ramdisk_id': "1",
'mac_address': "de:ad:be:ef:be:ef",
'flavor':
{'name': 'm1.tiny',
'memory_mb': 512,
'vcpus': 1,
'root_gb': 1024,
'flavorid': 1,
'rxtx_factor': 1}
}
def get_fake_image_data(project_id, user_id):
return {'name': 'image1',
'id': 1,
'project_id': project_id,
'user_id': user_id,
'image_ref': "1",
'kernel_id': "1",
'ramdisk_id': "1",
'mac_address': "de:ad:be:ef:be:ef",
'flavor': 'm1.tiny',
'properties': {
"hw_machine_type": constants.IMAGE_PROP_VM_GEN_1}
}
def get_fake_volume_info_data(target_portal, volume_id):
return {
'driver_volume_type': 'iscsi',
'data': {
'volume_id': volume_id,
'target_iqn': 'iqn.2010-10.org.openstack:volume-' + volume_id,
'target_portal': target_portal,
'target_lun': 1,
'auth_method': 'CHAP',
'auth_username': 'fake_username',
'auth_password': 'fake_password',
'target_discovered': False,
},
'mount_device': 'vda',
'delete_on_termination': False
}
def get_fake_block_device_info(target_portal, volume_id):
connection_info = get_fake_volume_info_data(target_portal, volume_id)
return {'block_device_mapping': [{'connection_info': connection_info}],
'root_device_name': 'fake_root_device_name',
'ephemerals': [],
'swap': None
}
def stub_out_db_instance_api(stubs):
"""Stubs out the db API for creating Instances."""
FLAVORS = {
'm1.tiny': dict(memory_mb=512, vcpus=1, root_gb=0, flavorid=1),
'm1.small': dict(memory_mb=2048, vcpus=1, root_gb=20, flavorid=2),
'm1.medium': dict(memory_mb=4096, vcpus=2, root_gb=40, flavorid=3),
'm1.large': dict(memory_mb=8192, vcpus=4, root_gb=80, flavorid=4),
'm1.xlarge': dict(memory_mb=16384, vcpus=8, root_gb=160, flavorid=5)}
class FakeModel(object):
"""Stubs out for model."""
def __init__(self, values):
self.values = values
def get(self, key, default=None):
if key in self.values:
return self.values[key]
else:
return default
def __getattr__(self, name):
return self.values[name]
def __getitem__(self, key):
return self.get(key)
def __setitem__(self, key, value):
self.values[key] = value
def __str__(self):
return str(self.values)
def fake_instance_create(context, values):
"""Stubs out the db.instance_create method."""
if 'flavor' not in values:
return
flavor = values['flavor']
base_options = {
'name': values['name'],
'id': values['id'],
'uuid': str(uuid.uuid4()),
'reservation_id': utils.generate_uid('r'),
'image_ref': values['image_ref'],
'kernel_id': values['kernel_id'],
'ramdisk_id': values['ramdisk_id'],
'vm_state': vm_states.BUILDING,
'task_state': task_states.SCHEDULING,
'user_id': values['user_id'],
'project_id': values['project_id'],
'flavor': flavor,
'memory_mb': flavor['memory_mb'],
'vcpus': flavor['vcpus'],
'mac_addresses': [{'address': values['mac_address']}],
'root_gb': flavor['root_gb'],
'system_metadata': {'image_shutdown_timeout': 0},
}
return FakeModel(base_options)
def fake_flavor_get_all(context, inactive=0, filters=None):
return FLAVORS.values()
def fake_flavor_get_by_name(context, name):
return FLAVORS[name]
def fake_block_device_mapping_get_all_by_instance(context, instance_uuid):
return {}
stubs.Set(db, 'instance_create', fake_instance_create)
stubs.Set(db, 'flavor_get_all', fake_flavor_get_all)
stubs.Set(db, 'flavor_get_by_name', fake_flavor_get_by_name)
stubs.Set(db, 'block_device_mapping_get_all_by_instance',
fake_block_device_mapping_get_all_by_instance)

93
nova/tests/unit/virt/hyperv/fake.py
View File

@@ -1,93 +0,0 @@
# Copyright 2013 Cloudbase Solutions Srl
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import io
import os
class PathUtils(object):
def open(self, path, mode):
return io.BytesIO(b'fake content')
def exists(self, path):
return False
def makedirs(self, path):
pass
def remove(self, path):
pass
def rename(self, src, dest):
pass
def copyfile(self, src, dest):
pass
def copy(self, src, dest):
pass
def move_folder_files(self, src_dir, dest_dir):
pass
def rmtree(self, path):
pass
def get_instances_dir(self, remote_server=None):
return 'C:\\FakeInstancesPath\\'
def get_instance_migr_revert_dir(self, instance_name, create_dir=False,
remove_dir=False):
return os.path.join(self.get_instances_dir(), instance_name, '_revert')
def get_instance_dir(self, instance_name, remote_server=None,
create_dir=True, remove_dir=False):
return os.path.join(self.get_instances_dir(remote_server),
instance_name)
def lookup_root_vhd_path(self, instance_name):
instance_path = self.get_instance_dir(instance_name)
return os.path.join(instance_path, 'root.vhd')
def lookup_configdrive_path(self, instance_name):
instance_path = self.get_instance_dir(instance_name)
return os.path.join(instance_path, 'configdrive.iso')
def lookup_ephemeral_vhd_path(self, instance_name):
instance_path = self.get_instance_dir(instance_name)
if instance_path:
return os.path.join(instance_path, 'ephemeral.vhd')
def get_root_vhd_path(self, instance_name, format_ext):
instance_path = self.get_instance_dir(instance_name)
return os.path.join(instance_path, 'root.' + format_ext)
def get_ephemeral_vhd_path(self, instance_name, format_ext):
instance_path = self.get_instance_dir(instance_name)
return os.path.join(instance_path, 'ephemeral.' + format_ext.lower())
def get_base_vhd_dir(self):
return os.path.join(self.get_instances_dir(), '_base')
def get_export_dir(self, instance_name):
export_dir = os.path.join(self.get_instances_dir(), 'export',
instance_name)
return export_dir
def vhd_exists(self, path):
return False
def get_vm_console_log_paths(self, vm_name, remote_server=None):
return 'fake_vm_log_path'

187
nova/tests/unit/virt/hyperv/test_hypervapi.py
View File

@@ -1,187 +0,0 @@
# Copyright 2012 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.
"""
Test suite for the Hyper-V driver and related APIs.
"""
import time
import uuid
from mox3 import mox
from oslo_config import cfg
from oslo_utils import fileutils
from nova.api.metadata import base as instance_metadata
from nova import context
from nova.image import glance
from nova import objects
from nova import test
from nova.tests.unit import fake_instance
from nova.tests.unit import fake_network
from nova.tests.unit.image import fake as fake_image
from nova.tests.unit.virt.hyperv import db_fakes
from nova.tests.unit.virt.hyperv import fake
from nova import utils
from nova.virt import configdrive
from nova.virt.hyperv import driver as driver_hyperv
from nova.virt.hyperv import hostutils
from nova.virt.hyperv import ioutils
from nova.virt.hyperv import pathutils
from nova.virt.hyperv import rdpconsoleutils
from nova.virt.hyperv import vmutils
from nova.virt import images
CONF = cfg.CONF
CONF.import_opt('vswitch_name', 'nova.virt.hyperv.vif', 'hyperv')
class HyperVAPIBaseTestCase(test.NoDBTestCase):
"""Base unit tests class for Hyper-V driver calls."""
def __init__(self, test_case_name):
self._mox = mox.Mox()
super(HyperVAPIBaseTestCase, self).__init__(test_case_name)
def setUp(self):
super(HyperVAPIBaseTestCase, self).setUp()
self._user_id = 'fake'
self._project_id = 'fake'
self._instance = None
self._image_metadata = None
self._fetched_image = None
self._update_image_raise_exception = False
self._volume_target_portal = 'testtargetportal:3260'
self._volume_id = '0ef5d708-45ab-4129-8c59-d774d2837eb7'
self._context = context.RequestContext(self._user_id, self._project_id)
self._instance_disks = []
self._instance_dvds = []
self._instance_volume_disks = []
self._test_vm_name = None
self._test_instance_dir = 'C:\\FakeInstancesPath\\instance-0000001'
self._check_min_windows_version_satisfied = True
self._setup_stubs()
self.flags(instances_path=r'C:\Hyper-V\test\instances',
network_api_class='nova.network.neutronv2.api.API')
self.flags(force_volumeutils_v1=True, group='hyperv')
self.flags(force_hyperv_utils_v1=True, group='hyperv')
self._conn = driver_hyperv.HyperVDriver(None)
def _setup_stubs(self):
db_fakes.stub_out_db_instance_api(self.stubs)
fake_image.stub_out_image_service(self.stubs)
fake_network.stub_out_nw_api_get_instance_nw_info(self.stubs)
def fake_fetch(context, image_id, target, user, project):
self._fetched_image = target
self.stubs.Set(images, 'fetch', fake_fetch)
def fake_get_remote_image_service(context, name):
class FakeGlanceImageService(object):
def update(self_fake, context, image_id, image_metadata, f):
if self._update_image_raise_exception:
raise vmutils.HyperVException(
"Simulated update failure")
self._image_metadata = image_metadata
return (FakeGlanceImageService(), 1)
self.stubs.Set(glance, 'get_remote_image_service',
fake_get_remote_image_service)
def fake_check_min_windows_version(fake_self, major, minor):
if [major, minor] >= [6, 3]:
return False
return self._check_min_windows_version_satisfied
self.stubs.Set(hostutils.HostUtils, 'check_min_windows_version',
fake_check_min_windows_version)
def fake_sleep(ms):
pass
self.stubs.Set(time, 'sleep', fake_sleep)
class FakeIOThread(object):
def __init__(self, src, dest, max_bytes):
pass
def start(self):
pass
self.stubs.Set(pathutils, 'PathUtils', fake.PathUtils)
self.stubs.Set(ioutils, 'IOThread', FakeIOThread)
self._mox.StubOutWithMock(vmutils.VMUtils, 'get_vm_id')
self._mox.StubOutWithMock(hostutils.HostUtils, 'get_local_ips')
self._mox.StubOutWithMock(rdpconsoleutils.RDPConsoleUtils,
'get_rdp_console_port')
self._mox.StubOutClassWithMocks(instance_metadata, 'InstanceMetadata')
self._mox.StubOutWithMock(instance_metadata.InstanceMetadata,
'metadata_for_config_drive')
# Can't use StubOutClassWithMocks due to __exit__ and __enter__
self._mox.StubOutWithMock(configdrive, 'ConfigDriveBuilder')
self._mox.StubOutWithMock(configdrive.ConfigDriveBuilder, 'make_drive')
self._mox.StubOutWithMock(fileutils, 'delete_if_exists')
self._mox.StubOutWithMock(utils, 'execute')
def tearDown(self):
self._mox.UnsetStubs()
super(HyperVAPIBaseTestCase, self).tearDown()
class HyperVAPITestCase(HyperVAPIBaseTestCase):
"""Unit tests for Hyper-V driver calls."""
def _get_instance_data(self):
instance_name = 'openstack_unit_test_vm_' + str(uuid.uuid4())
return db_fakes.get_fake_instance_data(instance_name,
self._project_id,
self._user_id)
def _get_instance(self):
updates = self._get_instance_data()
expected_attrs = updates.pop('expected_attrs', None)
return objects.Instance._from_db_object(
context, objects.Instance(),
fake_instance.fake_db_instance(**updates),
expected_attrs=expected_attrs)
def test_get_rdp_console(self):
self.flags(my_ip="192.168.1.1")
self._instance = self._get_instance()
fake_port = 9999
fake_vm_id = "fake_vm_id"
m = rdpconsoleutils.RDPConsoleUtils.get_rdp_console_port()
m.AndReturn(fake_port)
m = vmutils.VMUtils.get_vm_id(mox.IsA(str))
m.AndReturn(fake_vm_id)
self._mox.ReplayAll()
connect_info = self._conn.get_rdp_console(self._context,
self._instance)
self._mox.VerifyAll()
self.assertEqual(CONF.my_ip, connect_info.host)
self.assertEqual(fake_port, connect_info.port)
self.assertEqual(fake_vm_id, connect_info.internal_access_path)

47
nova/tests/unit/virt/hyperv/test_rdpconsoleops.py Normal file
View File

@@ -0,0 +1,47 @@
# 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 RDPConsoleOps.
"""
import mock
from nova.tests.unit.virt.hyperv import test_base
from nova.virt.hyperv import rdpconsoleops
class RDPConsoleOpsTestCase(test_base.HyperVBaseTestCase):
def setUp(self):
super(RDPConsoleOpsTestCase, self).setUp()
self.rdpconsoleops = rdpconsoleops.RDPConsoleOps()
self.rdpconsoleops._hostops = mock.MagicMock()
self.rdpconsoleops._vmutils = mock.MagicMock()
self.rdpconsoleops._rdpconsoleutils = mock.MagicMock()
def test_get_rdp_console(self):
mock_get_host_ip = self.rdpconsoleops._hostops.get_host_ip_addr
mock_get_rdp_port = (
self.rdpconsoleops._rdpconsoleutils.get_rdp_console_port)
mock_get_vm_id = self.rdpconsoleops._vmutils.get_vm_id
connect_info = self.rdpconsoleops.get_rdp_console(mock.DEFAULT)
self.assertEqual(mock_get_host_ip.return_value, connect_info.host)
self.assertEqual(mock_get_rdp_port.return_value, connect_info.port)
self.assertEqual(mock_get_vm_id.return_value,
connect_info.internal_access_path)