OpenStack Compute (Nova)
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 

28034 lines
1.2 MiB

# Copyright 2010 OpenStack Foundation
# Copyright 2012 University Of Minho
#
# 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 binascii
import collections
import contextlib
import copy
import datetime
import errno
import glob
import io
import os
import random
import re
import shutil
import signal
import sys
import testtools
import threading
import time
import unittest
from castellan import key_manager
import ddt
import eventlet
from eventlet import greenthread
import fixtures
from lxml import etree
import mock
from os_brick import encryptors
from os_brick import exception as brick_exception
from os_brick.initiator import connector
import os_resource_classes as orc
import os_traits as ot
import os_vif
from oslo_concurrency import lockutils
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_serialization import jsonutils
from oslo_service import loopingcall
from oslo_utils import fileutils
from oslo_utils import fixture as utils_fixture
from oslo_utils.fixture import uuidsentinel as uuids
from oslo_utils import strutils
from oslo_utils import units
from oslo_utils import uuidutils
from oslo_utils import versionutils
from nova.api.metadata import base as instance_metadata
from nova.compute import manager
from nova.compute import power_state
from nova.compute import provider_tree
from nova.compute import task_states
from nova.compute import utils as compute_utils
from nova.compute import vm_states
import nova.conf
from nova import context
from nova.db import api as db
from nova.db import constants as db_const
from nova import exception
from nova.network import model as network_model
from nova import objects
from nova.objects import block_device as block_device_obj
from nova.objects import fields
from nova.objects import migrate_data as migrate_data_obj
from nova.objects import virtual_interface as obj_vif
from nova.pci import manager as pci_manager
from nova.pci import utils as pci_utils
import nova.privsep.fs
import nova.privsep.libvirt
from nova.storage import rbd_utils
from nova import test
from nova.tests import fixtures as nova_fixtures
from nova.tests.unit import fake_block_device
from nova.tests.unit import fake_diagnostics
from nova.tests.unit import fake_flavor
from nova.tests.unit import fake_instance
from nova.tests.unit import fake_network
import nova.tests.unit.image.fake as fake_image
from nova.tests.unit import matchers
from nova.tests.unit.objects import test_diagnostics
from nova.tests.unit.objects import test_pci_device
from nova.tests.unit.objects import test_vcpu_model
from nova.tests.unit import utils as test_utils
from nova.tests.unit.virt.libvirt import fake_imagebackend
from nova.tests.unit.virt.libvirt import fake_libvirt_data
from nova.tests.unit.virt.libvirt import fakelibvirt
from nova import utils
from nova import version
from nova.virt import block_device as driver_block_device
from nova.virt import driver
from nova.virt import fake
from nova.virt import hardware
from nova.virt.image import model as imgmodel
from nova.virt.libvirt import blockinfo
from nova.virt.libvirt import config as vconfig
from nova.virt.libvirt import designer
from nova.virt.libvirt import driver as libvirt_driver
from nova.virt.libvirt import event as libvirtevent
from nova.virt.libvirt import guest as libvirt_guest
from nova.virt.libvirt import host
from nova.virt.libvirt.host import SEV_KERNEL_PARAM_FILE
from nova.virt.libvirt import imagebackend
from nova.virt.libvirt import imagecache
from nova.virt.libvirt import migration as libvirt_migrate
from nova.virt.libvirt.storage import dmcrypt
from nova.virt.libvirt.storage import lvm
from nova.virt.libvirt import utils as libvirt_utils
from nova.virt.libvirt import vif as libvirt_vif
from nova.virt.libvirt.volume import fs as fs_drivers
from nova.virt.libvirt.volume import volume as volume_drivers
CONF = nova.conf.CONF
_fake_network_info = fake_network.fake_get_instance_nw_info
# TODO(sean-k-mooney): move the rest of the static data to fake_libvirt_data
# and use it directly instead of creating local references to the common data
_fake_NodeDevXml = fake_libvirt_data._fake_NodeDevXml
_fake_NodeDevXml_parents = fake_libvirt_data._fake_NodeDevXml_parents
_fake_NodeDevXml_children = fake_libvirt_data._fake_NodeDevXml_children
_fake_cpu_info = {
"arch": "test_arch",
"model": "test_model",
"vendor": "test_vendor",
"topology": {
"sockets": 1,
"cores": 8,
"threads": 16
},
"features": ["feature1", "feature2"]
}
_fake_cpu_info_aarch64 = {
"arch": fields.Architecture.AARCH64,
"model": "test_model",
"vendor": "test_vendor",
"topology": {
"sockets": 1,
"cores": 8,
"threads": 16
},
"features": ["feature1", "feature2"]
}
eph_default_ext = utils.get_hash_str(nova.privsep.fs._DEFAULT_FILE_SYSTEM)[:7]
_fake_qemu64_cpu_feature = """
<cpu mode='custom' match='exact'>
<model fallback='forbid'>qemu64</model>
<feature policy='require' name='svm'/>
<feature policy='require' name='lm'/>
<feature policy='require' name='nx'/>
<feature policy='require' name='syscall'/>
<feature policy='require' name='cx16'/>
<feature policy='require' name='pni'/>
<feature policy='require' name='sse2'/>
<feature policy='require' name='sse'/>
<feature policy='require' name='fxsr'/>
<feature policy='require' name='mmx'/>
<feature policy='require' name='clflush'/>
<feature policy='require' name='pse36'/>
<feature policy='require' name='pat'/>
<feature policy='require' name='cmov'/>
<feature policy='require' name='mca'/>
<feature policy='require' name='pge'/>
<feature policy='require' name='mtrr'/>
<feature policy='require' name='sep'/>
<feature policy='require' name='apic'/>
<feature policy='require' name='cx8'/>
<feature policy='require' name='mce'/>
<feature policy='require' name='pae'/>
<feature policy='require' name='msr'/>
<feature policy='require' name='tsc'/>
<feature policy='require' name='pse'/>
<feature policy='require' name='de'/>
<feature policy='require' name='fpu'/>
</cpu>
"""
_fake_sandy_bridge_cpu_feature = """<cpu mode='custom' match='exact'>
<model fallback='forbid'>SandyBridge</model>
<feature policy='require' name='aes'/>
<feature policy='require' name='apic'/>
<feature policy='require' name='avx'/>
<feature policy='require' name='clflush'/>
<feature policy='require' name='cmov'/>
<feature policy='require' name='cx16'/>
<feature policy='require' name='cx8'/>
<feature policy='require' name='de'/>
<feature policy='require' name='fpu'/>
<feature policy='require' name='fxsr'/>
<feature policy='require' name='lahf_lm'/>
<feature policy='require' name='lm'/>
<feature policy='require' name='mca'/>
<feature policy='require' name='mce'/>
<feature policy='require' name='mmx'/>
<feature policy='require' name='msr'/>
<feature policy='require' name='mtrr'/>
<feature policy='require' name='nx'/>
<feature policy='require' name='pae'/>
<feature policy='require' name='pat'/>
<feature policy='require' name='pclmuldq'/>
<feature policy='require' name='pge'/>
<feature policy='require' name='pni'/>
<feature policy='require' name='popcnt'/>
<feature policy='require' name='pse'/>
<feature policy='require' name='pse36'/>
<feature policy='require' name='rdtscp'/>
<feature policy='require' name='sep'/>
<feature policy='require' name='sse'/>
<feature policy='require' name='sse2'/>
<feature policy='require' name='sse4.1'/>
<feature policy='require' name='sse4.2'/>
<feature policy='require' name='ssse3'/>
<feature policy='require' name='syscall'/>
<feature policy='require' name='tsc'/>
<feature policy='require' name='tsc-deadline'/>
<feature policy='require' name='x2apic'/>
<feature policy='require' name='xsave'/>
</cpu>
"""
_fake_broadwell_cpu_feature = """
<cpu mode='custom' match='exact'>
<model fallback='forbid'>Broadwell-noTSX</model>
<vendor>Intel</vendor>
<feature policy='require' name='smap'/>
<feature policy='require' name='adx'/>
<feature policy='require' name='rdseed'/>
<feature policy='require' name='invpcid'/>
<feature policy='require' name='erms'/>
<feature policy='require' name='bmi2'/>
<feature policy='require' name='smep'/>
<feature policy='require' name='avx2'/>
<feature policy='require' name='bmi1'/>
<feature policy='require' name='fsgsbase'/>
<feature policy='require' name='3dnowprefetch'/>
<feature policy='require' name='lahf_lm'/>
<feature policy='require' name='lm'/>
<feature policy='require' name='rdtscp'/>
<feature policy='require' name='nx'/>
<feature policy='require' name='syscall'/>
<feature policy='require' name='avx'/>
<feature policy='require' name='xsave'/>
<feature policy='require' name='aes'/>
<feature policy='require' name='tsc-deadline'/>
<feature policy='require' name='popcnt'/>
<feature policy='require' name='movbe'/>
<feature policy='require' name='x2apic'/>
<feature policy='require' name='sse4.2'/>
<feature policy='require' name='sse4.1'/>
<feature policy='require' name='pcid'/>
<feature policy='require' name='cx16'/>
<feature policy='require' name='fma'/>
<feature policy='require' name='ssse3'/>
<feature policy='require' name='pclmuldq'/>
<feature policy='require' name='pni'/>
<feature policy='require' name='sse2'/>
<feature policy='require' name='sse'/>
<feature policy='require' name='fxsr'/>
<feature policy='require' name='mmx'/>
<feature policy='require' name='clflush'/>
<feature policy='require' name='pse36'/>
<feature policy='require' name='pat'/>
<feature policy='require' name='cmov'/>
<feature policy='require' name='mca'/>
<feature policy='require' name='pge'/>
<feature policy='require' name='mtrr'/>
<feature policy='require' name='sep'/>
<feature policy='require' name='apic'/>
<feature policy='require' name='cx8'/>
<feature policy='require' name='mce'/>
<feature policy='require' name='pae'/>
<feature policy='require' name='msr'/>
<feature policy='require' name='tsc'/>
<feature policy='require' name='pse'/>
<feature policy='require' name='de'/>
<feature policy='require' name='fpu'/>
</cpu>
"""
def eph_name(size):
return ('ephemeral_%(size)s_%(ext)s' %
{'size': size, 'ext': eph_default_ext})
def fake_disk_info_byname(instance, type='qcow2'):
"""Return instance_disk_info corresponding accurately to the properties of
the given Instance object. The info is returned as an OrderedDict of
name->disk_info for each disk.
:param instance: The instance we're generating fake disk_info for.
:param type: libvirt's disk type.
:return: disk_info
:rtype: collections.OrderedDict
"""
instance_dir = os.path.join(CONF.instances_path, instance.uuid)
def instance_path(name):
return os.path.join(instance_dir, name)
disk_info = collections.OrderedDict()
# root disk
if (instance.image_ref is not None and
instance.image_ref != uuids.fake_volume_backed_image_ref):
cache_name = imagecache.get_cache_fname(instance.image_ref)
disk_info['disk'] = {
'type': type,
'path': instance_path('disk'),
'virt_disk_size': instance.flavor.root_gb * units.Gi,
'backing_file': cache_name,
'disk_size': instance.flavor.root_gb * units.Gi,
'over_committed_disk_size': 0}
swap_mb = instance.flavor.swap
if swap_mb > 0:
disk_info['disk.swap'] = {
'type': type,
'path': instance_path('disk.swap'),
'virt_disk_size': swap_mb * units.Mi,
'backing_file': 'swap_%s' % swap_mb,
'disk_size': swap_mb * units.Mi,
'over_committed_disk_size': 0}
eph_gb = instance.flavor.ephemeral_gb
if eph_gb > 0:
disk_info['disk.local'] = {
'type': type,
'path': instance_path('disk.local'),
'virt_disk_size': eph_gb * units.Gi,
'backing_file': eph_name(eph_gb),
'disk_size': eph_gb * units.Gi,
'over_committed_disk_size': 0}
if instance.config_drive:
disk_info['disk.config'] = {
'type': 'raw',
'path': instance_path('disk.config'),
'virt_disk_size': 1024,
'backing_file': '',
'disk_size': 1024,
'over_committed_disk_size': 0}
return disk_info
def fake_diagnostics_object(with_cpus=False, with_disks=False, with_nic=False):
diag_dict = {'config_drive': False,
'driver': 'libvirt',
'hypervisor': 'kvm',
'hypervisor_os': 'linux',
'memory_details': {'maximum': 2048, 'used': 1234},
'state': 'running',
'uptime': 10}
if with_cpus:
diag_dict['cpu_details'] = []
for id, t in enumerate([15340000000, 1640000000,
3040000000, 1420000000]):
diag_dict['cpu_details'].append({'id': id, 'time': t})
if with_disks:
diag_dict['disk_details'] = []
for i in range(2):
diag_dict['disk_details'].append(
{'read_bytes': 688640,
'read_requests': 169,
'write_bytes': 0,
'write_requests': 0,
'errors_count': 1})
if with_nic:
diag_dict['nic_details'] = [
{'mac_address': '52:54:00:a4:38:38',
'rx_drop': 0,
'rx_errors': 0,
'rx_octets': 4408,
'rx_packets': 82,
'tx_drop': 0,
'tx_errors': 0,
'tx_octets': 0,
'tx_packets': 0}]
return fake_diagnostics.fake_diagnostics_obj(**diag_dict)
def fake_disk_info_json(instance, type='qcow2'):
"""Return fake instance_disk_info corresponding accurately to the
properties of the given Instance object.
:param instance: The instance we're generating fake disk_info for.
:param type: libvirt's disk type.
:return: JSON representation of instance_disk_info for all disks.
:rtype: str
"""
disk_info = fake_disk_info_byname(instance, type)
return jsonutils.dumps(disk_info.values())
def get_injection_info(network_info=None, admin_pass=None, files=None):
return libvirt_driver.InjectionInfo(
network_info=network_info, admin_pass=admin_pass, files=files)
def _concurrency(signal, wait, done, target, is_block_dev=False):
signal.send()
wait.wait()
done.send()
class FakeVirtDomain(object):
def __init__(self, fake_xml=None, uuidstr=None, id=None, name=None,
info=None):
if uuidstr is None:
uuidstr = uuids.fake
self.uuidstr = uuidstr
self.id = id
self.domname = name
self._info = info or (
[power_state.RUNNING, 2048 * units.Mi,
1234 * units.Mi, None, None])
if fake_xml:
self._fake_dom_xml = fake_xml
else:
self._fake_dom_xml = """
<domain type='kvm'>
<name>testinstance1</name>
<devices>
<disk type='file'>
<source file='filename'/>
</disk>
</devices>
</domain>
"""
def name(self):
if self.domname is None:
return "fake-domain %s" % self
else:
return self.domname
def ID(self):
return self.id
def info(self):
return self._info
def create(self):
pass
def managedSave(self, *args):
pass
def createWithFlags(self, launch_flags):
pass
def XMLDesc(self, flags):
return self._fake_dom_xml
def UUIDString(self):
return self.uuidstr
def attachDeviceFlags(self, xml, flags):
pass
def attachDevice(self, xml):
pass
def detachDeviceFlags(self, xml, flags):
pass
def snapshotCreateXML(self, xml, flags):
pass
def blockCommit(self, disk, base, top, bandwidth=0, flags=0):
pass
def blockRebase(self, disk, base, bandwidth=0, flags=0):
pass
def blockJobInfo(self, path, flags):
pass
def blockJobAbort(self, path, flags):
pass
def resume(self):
pass
def destroy(self):
pass
def fsFreeze(self, disks=None, flags=0):
pass
def fsThaw(self, disks=None, flags=0):
pass
def isActive(self):
return True
def isPersistent(self):
return True
def undefine(self):
return True
def setMetadata(self, metadata_type, metadata, key, uri, flags=0):
pass
class CacheConcurrencyTestCase(test.NoDBTestCase):
def setUp(self):
super(CacheConcurrencyTestCase, self).setUp()
self.flags(instances_path=self.useFixture(fixtures.TempDir()).path)
# utils.synchronized() will create the lock_path for us if it
# doesn't already exist. It will also delete it when it's done,
# which can cause race conditions with the multiple threads we
# use for tests. So, create the path here so utils.synchronized()
# won't delete it out from under one of the threads.
self.lock_path = os.path.join(CONF.instances_path, 'locks')
fileutils.ensure_tree(self.lock_path)
def fake_exists(fname):
basedir = os.path.join(CONF.instances_path,
CONF.image_cache.subdirectory_name)
if fname == basedir or fname == self.lock_path:
return True
return False
self.stub_out('os.path.exists', fake_exists)
self.stub_out('oslo_concurrency.processutils.execute',
lambda *a, **kw: None)
self.stub_out('nova.virt.disk.api.extend',
lambda image, size, use_cow=False: None)
def _fake_instance(self, uuid):
return objects.Instance(id=1, uuid=uuid)
def test_same_fname_concurrency(self):
# Ensures that the same fname cache runs at a sequentially.
uuid = uuids.fake
backend = imagebackend.Backend(False)
wait1 = eventlet.event.Event()
done1 = eventlet.event.Event()
sig1 = eventlet.event.Event()
thr1 = eventlet.spawn(backend.by_name(self._fake_instance(uuid),
'name').cache,
_concurrency, 'fname', None,
signal=sig1, wait=wait1, done=done1)
eventlet.sleep(0)
# Thread 1 should run before thread 2.
sig1.wait()
wait2 = eventlet.event.Event()
done2 = eventlet.event.Event()
sig2 = eventlet.event.Event()
thr2 = eventlet.spawn(backend.by_name(self._fake_instance(uuid),
'name').cache,
_concurrency, 'fname', None,
signal=sig2, wait=wait2, done=done2)
wait2.send()
eventlet.sleep(0)
try:
self.assertFalse(done2.ready())
finally:
wait1.send()
done1.wait()
eventlet.sleep(0)
self.assertTrue(done2.ready())
# Wait on greenthreads to assert they didn't raise exceptions
# during execution
thr1.wait()
thr2.wait()
def test_different_fname_concurrency(self):
# Ensures that two different fname caches are concurrent.
uuid = uuids.fake
backend = imagebackend.Backend(False)
wait1 = eventlet.event.Event()
done1 = eventlet.event.Event()
sig1 = eventlet.event.Event()
thr1 = eventlet.spawn(backend.by_name(self._fake_instance(uuid),
'name').cache,
_concurrency, 'fname2', None,
signal=sig1, wait=wait1, done=done1)
eventlet.sleep(0)
# Thread 1 should run before thread 2.
sig1.wait()
wait2 = eventlet.event.Event()
done2 = eventlet.event.Event()
sig2 = eventlet.event.Event()
thr2 = eventlet.spawn(backend.by_name(self._fake_instance(uuid),
'name').cache,
_concurrency, 'fname1', None,
signal=sig2, wait=wait2, done=done2)
eventlet.sleep(0)
# Wait for thread 2 to start.
sig2.wait()
wait2.send()
tries = 0
while not done2.ready() and tries < 10:
eventlet.sleep(0)
tries += 1
try:
self.assertTrue(done2.ready())
finally:
wait1.send()
eventlet.sleep(0)
# Wait on greenthreads to assert they didn't raise exceptions
# during execution
thr1.wait()
thr2.wait()
class FakeInvalidVolumeDriver(object):
def __init__(self, *args, **kwargs):
raise brick_exception.InvalidConnectorProtocol('oops!')
class FakeConfigGuestDisk(object):
def __init__(self, *args, **kwargs):
self.source_type = None
self.driver_cache = None
class FakeConfigGuest(object):
def __init__(self, *args, **kwargs):
self.driver_cache = None
class FakeNodeDevice(object):
def __init__(self, fakexml):
self.xml = fakexml
def XMLDesc(self, flags):
return self.xml
def _create_test_instance():
flavor = objects.Flavor(memory_mb=2048,
swap=0,
vcpu_weight=None,
root_gb=10,
id=2,
name=u'm1.small',
ephemeral_gb=20,
rxtx_factor=1.0,
flavorid=u'1',
vcpus=2,
extra_specs={})
return {
'id': 1,
'uuid': uuids.instance,
'memory_kb': '1024000',
'basepath': '/some/path',
'bridge_name': 'br100',
'display_name': "Acme webserver",
'vcpus': 2,
'project_id': 'fake',
'bridge': 'br101',
'image_ref': '155d900f-4e14-4e4c-a73d-069cbf4541e6',
'root_gb': 10,
'ephemeral_gb': 20,
'instance_type_id': '5', # m1.small
'extra_specs': {},
'system_metadata': {
'image_disk_format': 'raw'
},
'flavor': flavor,
'new_flavor': None,
'old_flavor': None,
'pci_devices': objects.PciDeviceList(),
'numa_topology': None,
'config_drive': None,
'vm_mode': None,
'kernel_id': None,
'ramdisk_id': None,
'os_type': 'linux',
'user_id': '838a72b0-0d54-4827-8fd6-fb1227633ceb',
'ephemeral_key_uuid': None,
'vcpu_model': None,
'host': 'fake-host',
'node': 'fake-node',
'task_state': None,
'vm_state': None,
'trusted_certs': None,
'resources': None,
'migration_context': None,
}
@ddt.ddt
class LibvirtConnTestCase(test.NoDBTestCase,
test_diagnostics.DiagnosticsComparisonMixin):
REQUIRES_LOCKING = True
_EPHEMERAL_20_DEFAULT = eph_name(20)
def setUp(self):
super(LibvirtConnTestCase, self).setUp()
self.user_id = 'fake'
self.project_id = 'fake'
self.context = context.get_admin_context()
temp_dir = self.useFixture(fixtures.TempDir()).path
self.flags(instances_path=temp_dir)
self.flags(snapshots_directory=temp_dir, group='libvirt')
self.flags(sysinfo_serial="hardware", group="libvirt")
# normally loaded during nova-compute startup
os_vif.initialize()
self.stub_out('nova.virt.disk.api.extend',
lambda image, size, use_cow=False: None)
self.stub_out('nova.virt.libvirt.imagebackend.Image.'
'resolve_driver_format',
imagebackend.Image._get_driver_format)
self.useFixture(fakelibvirt.FakeLibvirtFixture())
# ensure tests perform the same on all host architectures; this is
# already done by the fakelibvirt fixture but we want to change the
# architecture in some tests
_p = mock.patch('os.uname')
self.mock_uname = _p.start()
self.mock_uname.return_value = fakelibvirt.os_uname(
'Linux', '', '5.4.0-0-generic', '', fields.Architecture.X86_64)
self.addCleanup(_p.stop)
self.test_instance = _create_test_instance()
self.test_image_meta = {
"disk_format": "raw",
}
self.image_service = self.useFixture(nova_fixtures.GlanceFixture(self))
self.device_xml_tmpl = """
<domain type='kvm'>
<devices>
<disk type='block' device='disk'>
<driver name='qemu' type='raw' cache='none'/>
<source dev='{device_path}'/>
<target bus='virtio' dev='vdb'/>
<serial>58a84f6d-3f0c-4e19-a0af-eb657b790657</serial>
<address type='pci' domain='0x0' bus='0x0' slot='0x04' \
function='0x0'/>
</disk>
</devices>
</domain>
"""
def relpath(self, path):
return os.path.relpath(path, CONF.instances_path)
def tearDown(self):
super(LibvirtConnTestCase, self).tearDown()
def test_driver_capabilities(self):
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False)
self.assertTrue(drvr.capabilities['has_imagecache'],
'Driver capabilities for \'has_imagecache\' '
'is invalid')
self.assertTrue(drvr.capabilities['supports_evacuate'],
'Driver capabilities for \'supports_evacuate\' '
'is invalid')
self.assertFalse(drvr.capabilities['supports_migrate_to_same_host'],
'Driver capabilities for '
'\'supports_migrate_to_same_host\' is invalid')
self.assertTrue(drvr.capabilities['supports_attach_interface'],
'Driver capabilities for '
'\'supports_attach_interface\' '
'is invalid')
self.assertTrue(drvr.capabilities['supports_extend_volume'],
'Driver capabilities for '
'\'supports_extend_volume\' '
'is invalid')
self.assertTrue(drvr.capabilities['supports_trusted_certs'],
'Driver capabilities for '
'\'supports_trusted_certs\' '
'is invalid')
self.assertTrue(drvr.capabilities['supports_image_type_qcow2'],
'Driver capabilities for '
'\'supports_image_type_qcow2\' '
'is invalid')
self.assertFalse(drvr.capabilities['supports_image_type_ploop'],
'Driver capabilities for '
'\'supports_image_type_ploop\' '
'is invalid')
self.assertFalse(
drvr.capabilities['supports_vtpm'],
"Driver capabilities for 'supports_vtpm' is invalid",
)
def test_driver_capabilities_qcow2_with_rbd(self):
self.flags(images_type='rbd', group='libvirt')
self.flags(force_raw_images=False)
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False)
self.assertFalse(drvr.capabilities['supports_image_type_qcow2'],
'Driver capabilities for '
'\'supports_image_type_qcow2\' '
'is invalid when \'images_type=rbd\'')
self.flags(images_type='rbd', group='libvirt')
self.flags(force_raw_images=True)
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False)
self.assertTrue(drvr.capabilities['supports_image_type_qcow2'])
def test_driver_capabilities_qcow2_with_lvm(self):
self.flags(images_type='lvm', group='libvirt')
self.flags(force_raw_images=False)
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False)
self.assertFalse(drvr.capabilities['supports_image_type_qcow2'],
'Driver capabilities for '
'\'supports_image_type_qcow2\' '
'is invalid when \'images_type=lvm\'')
self.flags(images_type='lvm', group='libvirt')
self.flags(force_raw_images=True)
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False)
self.assertTrue(drvr.capabilities['supports_image_type_qcow2'])
def test_driver_capabilities_ploop_with_virtuozzo(self):
self.flags(virt_type='kvm', group='libvirt')
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False)
self.assertFalse(drvr.capabilities['supports_image_type_ploop'],
'Driver capabilities for '
'\'supports_image_type_ploop\' '
'is invalid when virt_type=kvm')
self.flags(virt_type='parallels', group='libvirt')
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False)
self.assertTrue(drvr.capabilities['supports_image_type_ploop'])
def test_driver_capabilities_vtpm(self):
self.flags(swtpm_enabled=True, group='libvirt')
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False)
self.assertTrue(
drvr.capabilities['supports_vtpm'],
"Driver capabilities for 'supports_vtpm' is invalid when "
"'swtpm_enabled=True'"
)
@mock.patch.object(
libvirt_driver.LibvirtDriver, '_register_instance_machine_type',
new=mock.Mock())
@mock.patch.object(
host.Host, 'supports_secure_boot', new_callable=mock.PropertyMock)
def test_driver_capabilities_secure_boot(self, mock_supports):
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False)
drvr.init_host("dummyhost")
self.assertTrue(
drvr.capabilities['supports_secure_boot'],
"Driver capabilities for 'supports_secure_boot' is invalid when "
"host should support this feature"
)
mock_supports.assert_called_once_with()
def test_driver_raises_on_non_linux_platform(self):
with utils.temporary_mutation(sys, platform='darwin'):
self.assertRaises(
exception.InternalError, libvirt_driver.LibvirtDriver,
fake.FakeVirtAPI(), False)
def create_fake_libvirt_mock(self, **kwargs):
"""Defining mocks for LibvirtDriver(libvirt is not used)."""
# A fake libvirt.virConnect
class FakeLibvirtDriver(object):
def defineXML(self, xml):
return FakeVirtDomain()
# Creating mocks
fake = FakeLibvirtDriver()
# Customizing above fake if necessary
for key, val in kwargs.items():
fake.__setattr__(key, val)
self.stub_out('nova.virt.libvirt.driver.LibvirtDriver._conn', fake)
self.stub_out('nova.virt.libvirt.host.Host.get_connection',
lambda x: fake)
def fake_lookup(self, instance_name):
return FakeVirtDomain()
def fake_execute(self, *args, **kwargs):
open(args[-1], "a").close()
def _create_service(self, **kwargs):
service_ref = {'host': kwargs.get('host', 'dummy'),
'disabled': kwargs.get('disabled', False),
'binary': 'nova-compute',
'topic': 'compute',
'report_count': 0}
return objects.Service(**service_ref)
def _get_pause_flag(self, drvr, network_info, power_on=True,
vifs_already_plugged=False):
timeout = CONF.vif_plugging_timeout
events = []
if (
CONF.libvirt.virt_type in ('kvm', 'qemu') and
not vifs_already_plugged and power_on and timeout
):
events = drvr._get_neutron_events(network_info)
return bool(events)
def test_public_api_signatures(self):
baseinst = driver.ComputeDriver(None)
inst = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertPublicAPISignatures(baseinst, inst)
def test_legacy_block_device_info(self):
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertFalse(drvr.need_legacy_block_device_info)
@mock.patch.object(libvirt_driver.LibvirtDriver, '_get_cpu_traits')
@mock.patch.object(libvirt_driver.LibvirtDriver, '_get_storage_bus_traits')
@mock.patch.object(libvirt_driver.LibvirtDriver, '_get_video_model_traits')
@mock.patch.object(libvirt_driver.LibvirtDriver, '_get_vif_model_traits')
def test_static_traits(
self, mock_vif_traits, mock_video_traits, mock_storage_traits,
mock_cpu_traits,
):
"""Ensure driver capabilities are correctly retrieved and cached."""
# we don't mock out calls to os_traits intentionally, so we need to
# return valid traits here
mock_cpu_traits.return_value = {'HW_CPU_HYPERTHREADING': True}
mock_storage_traits.return_value = {'COMPUTE_STORAGE_BUS_VIRTIO': True}
mock_video_traits.return_value = {'COMPUTE_GRAPHICS_MODEL_VGA': True}
mock_vif_traits.return_value = {'COMPUTE_NET_VIF_MODEL_VIRTIO': True}
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False)
expected = {
'HW_CPU_HYPERTHREADING': True,
'COMPUTE_STORAGE_BUS_VIRTIO': True,
'COMPUTE_GRAPHICS_MODEL_VGA': True,
'COMPUTE_NET_VIF_MODEL_VIRTIO': True,
'COMPUTE_SECURITY_TPM_1_2': False,
'COMPUTE_SECURITY_TPM_2_0': False,
'COMPUTE_SOCKET_PCI_NUMA_AFFINITY': True,
}
static_traits = drvr.static_traits
# check that results are as expected and the individual helper
# functions were called once each
self.assertEqual(expected, static_traits)
for mock_traits in (
mock_vif_traits, mock_video_traits, mock_storage_traits,
mock_cpu_traits,
):
mock_traits.assert_called_once_with()
mock_traits.reset_mock()
static_traits = drvr.static_traits
# now check that the results are still as expected but the helpers
# weren't called since the value was cached
self.assertEqual(expected, static_traits)
for mock_traits in (
mock_vif_traits, mock_video_traits, mock_storage_traits,
mock_cpu_traits,
):
mock_traits.assert_not_called()
@mock.patch.object(libvirt_driver.LOG, 'debug')
@mock.patch.object(libvirt_driver.LibvirtDriver, '_get_cpu_traits')
@mock.patch.object(libvirt_driver.LibvirtDriver, '_get_storage_bus_traits')
@mock.patch.object(libvirt_driver.LibvirtDriver, '_get_video_model_traits')
@mock.patch.object(libvirt_driver.LibvirtDriver, '_get_vif_model_traits')
def test_static_traits__invalid_trait(
self, mock_vif_traits, mock_video_traits, mock_storage_traits,
mock_cpu_traits, mock_log,
):
"""Ensure driver capabilities are correctly retrieved and cached."""
mock_cpu_traits.return_value = {'foo': True}
mock_storage_traits.return_value = {'bar': True}
mock_video_traits.return_value = {'baz': True}
mock_vif_traits.return_value = {'COMPUTE_NET_VIF_MODEL_VIRTIO': True}
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False)
expected = {
'COMPUTE_NET_VIF_MODEL_VIRTIO': True,
'COMPUTE_SECURITY_TPM_1_2': False,
'COMPUTE_SECURITY_TPM_2_0': False,
'COMPUTE_SOCKET_PCI_NUMA_AFFINITY': True,
}
static_traits = drvr.static_traits
self.assertEqual(expected, static_traits)
mock_log.assert_has_calls([
mock.call("Trait '%s' is not valid; ignoring.", "foo"),
mock.call("Trait '%s' is not valid; ignoring.", "bar"),
mock.call("Trait '%s' is not valid; ignoring.", "baz"),
],
any_order=True)
@mock.patch.object(libvirt_driver.LibvirtDriver,
'_register_instance_machine_type', new=mock.Mock())
@mock.patch.object(host.Host, "has_min_version")
def test_min_version_start_ok(self, mock_version):
mock_version.return_value = True
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr.init_host("dummyhost")
@mock.patch.object(host.Host, "has_min_version")
def test_min_version_start_abort(self, mock_version):
mock_version.return_value = False
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertRaises(exception.NovaException,
drvr.init_host,
"dummyhost")
@mock.patch.object(libvirt_driver.LibvirtDriver,
'_register_instance_machine_type', new=mock.Mock())
@mock.patch.object(fakelibvirt.Connection, 'getLibVersion',
return_value=versionutils.convert_version_to_int(
libvirt_driver.NEXT_MIN_LIBVIRT_VERSION) - 1)
@mock.patch.object(libvirt_driver.LOG, 'warning')
def test_next_min_version_deprecation_warning(self, mock_warning,
mock_get_libversion):
# Skip test if there's no currently planned new min version
if (versionutils.convert_version_to_int(
libvirt_driver.NEXT_MIN_LIBVIRT_VERSION) ==
versionutils.convert_version_to_int(
libvirt_driver.MIN_LIBVIRT_VERSION)):
self.skipTest("NEXT_MIN_LIBVIRT_VERSION == MIN_LIBVIRT_VERSION")
# Test that a warning is logged if the libvirt version is less than
# the next required minimum version.
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr.init_host("dummyhost")
# assert that the next min version is in a warning message
expected_arg = {'version': versionutils.convert_version_to_str(
versionutils.convert_version_to_int(
libvirt_driver.NEXT_MIN_LIBVIRT_VERSION))}
version_arg_found = False
for call in mock_warning.call_args_list:
if call[0][1] == expected_arg:
version_arg_found = True
break
self.assertTrue(version_arg_found)
@mock.patch.object(libvirt_driver.LibvirtDriver,
'_register_instance_machine_type', new=mock.Mock())
@mock.patch.object(fakelibvirt.Connection, 'getVersion',
return_value=versionutils.convert_version_to_int(
libvirt_driver.NEXT_MIN_QEMU_VERSION) - 1)
@mock.patch.object(libvirt_driver.LOG, 'warning')
def test_next_min_qemu_version_deprecation_warning(self, mock_warning,
mock_get_libversion):
# Skip test if there's no currently planned new min version
if (versionutils.convert_version_to_int(
libvirt_driver.NEXT_MIN_QEMU_VERSION) ==
versionutils.convert_version_to_int(
libvirt_driver.MIN_QEMU_VERSION)):
self.skipTest("NEXT_MIN_QEMU_VERSION == MIN_QEMU_VERSION")
# Test that a warning is logged if the libvirt version is less than
# the next required minimum version.
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr.init_host("dummyhost")
# assert that the next min version is in a warning message
expected_arg = {'version': versionutils.convert_version_to_str(
versionutils.convert_version_to_int(
libvirt_driver.NEXT_MIN_QEMU_VERSION))}
version_arg_found = False
for call in mock_warning.call_args_list:
if call[0][1] == expected_arg:
version_arg_found = True
break
self.assertTrue(version_arg_found)
@mock.patch.object(libvirt_driver.LibvirtDriver,
'_register_instance_machine_type', new=mock.Mock())
@mock.patch.object(fakelibvirt.Connection, 'getLibVersion',
return_value=versionutils.convert_version_to_int(
libvirt_driver.NEXT_MIN_LIBVIRT_VERSION))
@mock.patch.object(libvirt_driver.LOG, 'warning')
def test_next_min_version_ok(self, mock_warning, mock_get_libversion):
# Skip test if there's no currently planned new min version
if (versionutils.convert_version_to_int(
libvirt_driver.NEXT_MIN_LIBVIRT_VERSION) ==
versionutils.convert_version_to_int(
libvirt_driver.MIN_LIBVIRT_VERSION)):
self.skipTest("NEXT_MIN_LIBVIRT_VERSION == MIN_LIBVIRT_VERSION")
# Test that a warning is not logged if the libvirt version is greater
# than or equal to NEXT_MIN_LIBVIRT_VERSION.
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr.init_host("dummyhost")
# assert that the next min version is in a warning message
expected_arg = {'version': versionutils.convert_version_to_str(
versionutils.convert_version_to_int(
libvirt_driver.NEXT_MIN_LIBVIRT_VERSION))}
version_arg_found = False
for call in mock_warning.call_args_list:
if call[0][1] == expected_arg:
version_arg_found = True
break
self.assertFalse(version_arg_found)
@mock.patch.object(libvirt_driver.LibvirtDriver,
'_register_instance_machine_type', new=mock.Mock())
@mock.patch.object(fakelibvirt.Connection, 'getVersion',
return_value=versionutils.convert_version_to_int(
libvirt_driver.NEXT_MIN_QEMU_VERSION))
@mock.patch.object(libvirt_driver.LOG, 'warning')
def test_next_min_qemu_version_ok(self, mock_warning, mock_get_libversion):
# Skip test if there's no currently planned new min version
if (versionutils.convert_version_to_int(
libvirt_driver.NEXT_MIN_QEMU_VERSION) ==
versionutils.convert_version_to_int(
libvirt_driver.MIN_QEMU_VERSION)):
self.skipTest("NEXT_MIN_QEMU_VERSION == MIN_QEMU_VERSION")
# Test that a warning is not logged if the libvirt version is greater
# than or equal to NEXT_MIN_QEMU_VERSION.
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr.init_host("dummyhost")
# assert that the next min version is in a warning message
expected_arg = {'version': versionutils.convert_version_to_str(
versionutils.convert_version_to_int(
libvirt_driver.NEXT_MIN_QEMU_VERSION))}
version_arg_found = False
for call in mock_warning.call_args_list:
if call[0][1] == expected_arg:
version_arg_found = True
break
self.assertFalse(version_arg_found)
@mock.patch.object(libvirt_driver.LibvirtDriver,
'_register_instance_machine_type', new=mock.Mock())
def test_min_version_ppc_ok(self):
self.mock_uname.return_value = fakelibvirt.os_uname(
'Linux', '', '5.4.0-0-generic', '', fields.Architecture.PPC64)
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr.init_host("dummyhost")
@mock.patch.object(libvirt_driver.LibvirtDriver,
'_register_instance_machine_type', new=mock.Mock())
def test_min_version_s390_ok(self):
self.mock_uname.return_value = fakelibvirt.os_uname(
'Linux', '', '5.4.0-0-generic', '', fields.Architecture.S390X)
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr.init_host("dummyhost")
@mock.patch.object(libvirt_driver.LibvirtDriver,
'_register_instance_machine_type', new=mock.Mock())
def test_file_backed_memory_support_called(self):
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
with mock.patch.object(drvr,
'_check_file_backed_memory_support') as mock_check_fb_support:
drvr.init_host("dummyhost")
self.assertTrue(mock_check_fb_support.called)
def test_min_version_file_backed_ok(self):
self.flags(file_backed_memory=1024, group='libvirt')
self.flags(ram_allocation_ratio=1.0)
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr._check_file_backed_memory_support()
def test_min_version_file_backed_bad_ram_allocation_ratio(self):
self.flags(file_backed_memory=1024, group="libvirt")
self.flags(ram_allocation_ratio=1.5)
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertRaises(exception.InternalError,
drvr._check_file_backed_memory_support)
def test__check_file_backed_memory_support__total_lt_reserved(self):
"""Ensure an error is raised if total memory < reserved.
Placement won't allow $resource.total < $resource.reserved, so we need
to catch this early.
"""
self.flags(file_backed_memory=1024, group='libvirt')
self.flags(ram_allocation_ratio=1.0, reserved_host_memory_mb=4096)
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertRaises(
exception.InternalError, drvr._check_file_backed_memory_support,
)
@mock.patch.object(libvirt_driver.LOG, 'warning')
def test__check_file_backed_memory_support__has_reserved(self, mock_log):
"""Ensure a warning is issued if memory is reserved.
It doesn't make sense to "reserve" memory when file-backed memory is in
use. We should report things so as to avoid confusion.
"""
self.flags(file_backed_memory=8192, group='libvirt')
self.flags(ram_allocation_ratio=1.0)
# we don't need to configure '[DEFAULT] reserved_host_memory_mb' since
# it defaults to 512 (MB)
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr._check_file_backed_memory_support()
mock_log.assert_called_once()
self.assertIn(
"Reserving memory via '[DEFAULT] reserved_host_memory_mb' is not "
"compatible",
str(mock_log.call_args[0]),
)
@mock.patch.object(libvirt_driver.LibvirtDriver,
'_register_instance_machine_type', new=mock.Mock())
def test__prepare_cpu_flag(self):
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
# The `+` means the guest "require[s]" (i.e. enable in libvirt
# parlance) the said CPU feature; and `-` will disable it
feat1 = drvr._prepare_cpu_flag('+md-clear')
feat2 = drvr._prepare_cpu_flag('pdpe1gb')
feat3 = drvr._prepare_cpu_flag('-ssbd')
self.assertIsInstance(feat1, vconfig.LibvirtConfigGuestCPUFeature)
self.assertIsInstance(feat2, vconfig.LibvirtConfigGuestCPUFeature)
self.assertIsInstance(feat3, vconfig.LibvirtConfigGuestCPUFeature)
cpu = vconfig.LibvirtConfigGuestCPU()
cpu.add_feature(feat1)
cpu.add_feature(feat2)
cpu.add_feature(feat3)
# Verify that the resulting guest XML records both enabled
# _and_ disabled CPU features
expected_xml = '''
<cpu match='exact'>
<feature policy='require' name='md-clear'/>
<feature policy='require' name='pdpe1gb'/>
<feature policy='disable' name='ssbd'/>
</cpu>'''
self.assertXmlEqual(expected_xml, cpu.to_xml())
@mock.patch.object(libvirt_driver.LibvirtDriver,
'_register_instance_machine_type', new=mock.Mock())
def test__check_cpu_compatibility_start_ok(self):
self.flags(cpu_mode="custom",
cpu_models=["Penryn"],
group="libvirt")
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr.init_host("dummyhost")
def test__check_cpu_compatibility_none_models(self):
self.flags(cpu_mode="custom",
cpu_models=[],
group="libvirt")
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertRaises(exception.Invalid, drvr.init_host, "dummyhost")
def test__check_cpu_compatibility_none_mode(self):
self.flags(cpu_mode="none",
cpu_models=["Penryn"],
group="libvirt")
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertRaises(exception.Invalid, drvr.init_host, "dummyhost")
@mock.patch('nova.virt.libvirt.host.libvirt.Connection.compareCPU')
def test__check_cpu_compatibility_advance_model(self, mocked_compare):
mocked_compare.side_effect = (2, 0)
self.flags(cpu_mode="custom",
cpu_models=["qemu64", "Broadwell-noTSX"],
group="libvirt")
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertRaises(exception.InvalidCPUInfo,
drvr.init_host, "dummyhost")
@mock.patch.object(libvirt_driver.LibvirtDriver,
'_register_instance_machine_type', new=mock.Mock())
def test__check_cpu_compatibility_with_flag(self):
self.flags(cpu_mode="custom",
cpu_models=["Penryn"],
cpu_model_extra_flags = ["aes"],
group="libvirt")
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr.init_host("dummyhost")
@mock.patch('nova.virt.libvirt.host.libvirt.Connection.compareCPU')
def test__check_cpu_compatibility_advance_flag(self, mocked_compare):
mocked_compare.side_effect = (2, 0)
self.flags(cpu_mode="custom",
cpu_models=["qemu64"],
cpu_model_extra_flags = ["avx", "avx2"],
group="libvirt")
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertRaises(exception.InvalidCPUInfo,
drvr.init_host, "dummyhost")
@mock.patch('nova.virt.libvirt.host.libvirt.Connection.compareCPU')
def test__check_cpu_compatibility_wrong_flag(self, mocked_compare):
# here, and in the surrounding similar tests, the non-zero error
# code in the compareCPU() side effect indicates error
mocked_compare.side_effect = (2, 0)
self.flags(cpu_mode="custom",
cpu_models=["Broadwell-noTSX"],
cpu_model_extra_flags = ["a v x"],
group="libvirt")
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertRaises(exception.InvalidCPUInfo,
drvr.init_host, "dummyhost")
@mock.patch('nova.virt.libvirt.host.libvirt.Connection.compareCPU')
def test__check_cpu_compatibility_enabled_and_disabled_flags(
self, mocked_compare
):
mocked_compare.side_effect = (2, 0)
self.flags(
cpu_mode="custom",
cpu_models=["Cascadelake-Server"],
cpu_model_extra_flags = ["-hle", "-rtm", "+ssbd", "mttr"],
group="libvirt")
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertRaises(exception.InvalidCPUInfo,
drvr.init_host, "dummyhost")
def test__check_cpu_compatibility_invalid_virt_type(self):
"""Test getting CPU traits when using a virt_type that doesn't support
the feature, only kvm and qemu supports reporting CPU traits.
"""
self.flags(cpu_mode='custom',
cpu_models=['IvyBridge'],
virt_type='lxc',
group='libvirt')
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertRaises(exception.Invalid, drvr.init_host, "dummyhost")
@mock.patch.object(libvirt_driver.LibvirtDriver,
'_register_instance_machine_type', new=mock.Mock())
def test__check_cpu_compatibility_aarch64_qemu_custom_start_OK(self):
"""Test getting CPU traits when using a virt_type that doesn't support
the feature, only kvm and qemu supports reporting CPU traits.
"""
self.flags(cpu_mode='custom',
cpu_models=['max'],
virt_type='qemu',
group='libvirt')
caps = vconfig.LibvirtConfigCaps()
caps.host = vconfig.LibvirtConfigCapsHost()
caps.host.cpu = vconfig.LibvirtConfigCPU()
caps.host.cpu.arch = fields.Architecture.AARCH64
with mock.patch.object(host.Host, "get_capabilities",
return_value=caps):
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr.init_host("dummyhost")
def test__check_vtpm_support_non_qemu(self):
"""Test checking for vTPM support when we're not using QEMU or KVM."""
self.flags(swtpm_enabled=True, virt_type='lxc', group='libvirt')
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
exc = self.assertRaises(exception.InvalidConfiguration,
drvr.init_host, 'dummyhost')
self.assertIn("vTPM support requires '[libvirt] virt_type' of 'qemu' "
"or 'kvm'; found 'lxc'.", str(exc))
@mock.patch.object(host.Host, 'has_min_version', return_value=True)
@mock.patch('shutil.which')
def test__check_vtpm_support_missing_exe(self, mock_which, mock_version):
"""Test checking for vTPM support when the swtpm binaries are
missing.
"""
self.flags(swtpm_enabled=True, virt_type='kvm', group='libvirt')
mock_which.return_value = False
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
exc = self.assertRaises(exception.InvalidConfiguration,
drvr.init_host, "dummyhost")
self.assertIn(
"vTPM support is configured but the 'swtpm' and 'swtpm_setup' "
"binaries could not be found on PATH.",
str(exc),
)
mock_which.assert_has_calls(
[mock.call('swtpm_setup'), mock.call('swtpm')],
)
@mock.patch.object(host.Host, 'has_min_version', return_value=True)
@mock.patch('shutil.which')
@mock.patch('pwd.getpwnam')
def test__check_vtpm_support_invalid_user(
self, mock_getpwnam, mock_which, mock_version,
):
"""Test checking for vTPM support when the configured user is
invalid.
"""
self.flags(
swtpm_user='lionel', swtpm_enabled=True, virt_type='kvm',
group='libvirt')
mock_which.return_value = True
mock_getpwnam.side_effect = KeyError
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
exc = self.assertRaises(
exception.InvalidConfiguration,
drvr.init_host, "dummyhost")
self.assertIn(
"The user configured in '[libvirt] swtpm_user' does not exist "
"on this host; expected 'lionel'.",
str(exc),
)
mock_getpwnam.assert_called_with('lionel')
@mock.patch.object(host.Host, 'has_min_version', return_value=True)
@mock.patch('shutil.which')
@mock.patch('pwd.getpwnam')
@mock.patch('grp.getgrnam')
def test__check_vtpm_support_invalid_group(
self, mock_getgrnam, mock_getpwnam, mock_which, mock_version,
):
"""Test checking for vTPM support when the configured group is
invalid.
"""
self.flags(
swtpm_group='admins', swtpm_enabled=True, virt_type='kvm',
group='libvirt')
mock_which.return_value = True
mock_getgrnam.side_effect = KeyError
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
exc = self.assertRaises(
exception.InvalidConfiguration,
drvr.init_host, "dummyhost")
self.assertIn(
"The group configured in '[libvirt] swtpm_group' does not exist "
"on this host; expected 'admins'.",
str(exc),
)
mock_getgrnam.assert_called_with('admins')
@mock.patch.object(libvirt_driver.LibvirtDriver,
'_register_instance_machine_type', new=mock.Mock())
@mock.patch('shutil.which')
@mock.patch('pwd.getpwnam')
@mock.patch('grp.getgrnam')
def test__check_vtpm_support(
self, mock_getgrnam, mock_getpwnam, mock_which
):
"""Test checking for vTPM support when everything is configured
correctly.
"""
self.flags(swtpm_enabled=True, virt_type='kvm', group='libvirt')
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr.init_host('dummyhost')
mock_which.assert_has_calls(
[mock.call('swtpm_setup'), mock.call().__bool__()],
)
@mock.patch.object(libvirt_driver.LOG, 'warning')
def test_check_cpu_set_configuration__no_configuration(self, mock_log):
"""Test that configuring no CPU option results no errors or logs.
"""
self.flags(vcpu_pin_set=None, reserved_host_cpus=None)
self.flags(cpu_shared_set=None, cpu_dedicated_set=None,
group='compute')
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr._check_cpu_set_configuration()
mock_log.assert_not_called()
def test_check_cpu_set_configuration__cpu_shared_set_cpu_dedicated_set(
self):
"""Test that configuring 'cpu_shared_set' and 'cpu_dedicated_set' such
that they overlap (are not disjoint) results in an error stating that
this is not allowed.
"""
self.flags(vcpu_pin_set=None, reserved_host_cpus=None)
self.flags(cpu_shared_set='0-3', cpu_dedicated_set='3-5',
group='compute')
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertRaises(exception.InvalidConfiguration,
drvr._check_cpu_set_configuration)
def test_check_cpu_set_configuration__reserved_host_cpus_cpu_shared_set(
self):
"""Test that configuring 'reserved_host_cpus' with one of the new
options, in this case '[compute] cpu_shared_set', results in an error
stating that this is not allowed.
"""
self.flags(vcpu_pin_set=None, reserved_host_cpus=1)
self.flags(cpu_shared_set='1-10', cpu_dedicated_set=None,
group='compute')
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
ex = self.assertRaises(exception.InvalidConfiguration,
drvr._check_cpu_set_configuration)
self.assertIn("The 'reserved_host_cpus' config option cannot be "
"defined alongside ", str(ex))
@mock.patch.object(libvirt_driver.LOG, 'warning')
def test_check_cpu_set_configuration__vcpu_pin_set(self, mock_log):
"""Test that configuring only 'vcpu_pin_set' results in a warning that
the option is being used for VCPU inventory but this is deprecated
behavior.
"""
self.flags(vcpu_pin_set='0-3', reserved_host_cpus=None)
self.flags(cpu_shared_set=None, cpu_dedicated_set=None,
group='compute')
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr._check_cpu_set_configuration()
mock_log.assert_called_once()
self.assertIn("When defined, 'vcpu_pin_set' will be used to calculate "
"'VCPU' inventory and schedule instances that have "
"'VCPU' allocations.",
str(mock_log.call_args[0]))
@mock.patch.object(libvirt_driver.LOG, 'warning')
def test_check_cpu_set_configuration__vcpu_pin_set_cpu_shared_set(
self, mock_log):
"""Test that configuring both 'vcpu_pin_set' and 'cpu_shared_set'
results in a warning that 'cpu_shared_set' is being ignored for
calculating VCPU inventory.
"""
self.flags(vcpu_pin_set='0-3', reserved_host_cpus=None)
self.flags(cpu_shared_set='4-5', cpu_dedicated_set=None,
group='compute')
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr._check_cpu_set_configuration()
mock_log.assert_called_once()
self.assertIn("The '[compute] cpu_shared_set' and 'vcpu_pin_set' "
"config options have both been defined.",
str(mock_log.call_args[0]))
def test_check_cpu_set_configuration__vcpu_pin_set_cpu_dedicated_set(
self):
"""Test that configuring both 'vcpu_pin_set' and 'cpu_dedicated_set'
results in an error stating that the two options cannot co-exist.
"""
self.flags(vcpu_pin_set='0-3', reserved_host_cpus=None)
self.flags(cpu_shared_set=None, cpu_dedicated_set='4-5',
group='compute')
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
ex = self.assertRaises(exception.InvalidConfiguration,
drvr._check_cpu_set_configuration)
self.assertIn("The 'vcpu_pin_set' config option has been deprecated "
"and cannot be defined alongside '[compute] "
"cpu_dedicated_set'.", str(ex))
def _do_test_parse_migration_flags(self, lm_expected=None,
bm_expected=None):
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr._parse_migration_flags()
if lm_expected is not None:
self.assertEqual(lm_expected, drvr._live_migration_flags)
if bm_expected is not None:
self.assertEqual(bm_expected, drvr._block_migration_flags)
def test_parse_live_migration_flags_default(self):
self._do_test_parse_migration_flags(
lm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_LIVE))
def test_parse_live_migration_flags(self):
self._do_test_parse_migration_flags(
lm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_LIVE))
def test_parse_block_migration_flags_default(self):
self._do_test_parse_migration_flags(
bm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_LIVE |
libvirt_driver.libvirt.VIR_MIGRATE_NON_SHARED_INC))
def test_parse_block_migration_flags(self):
self._do_test_parse_migration_flags(
bm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_LIVE |
libvirt_driver.libvirt.VIR_MIGRATE_NON_SHARED_INC))
def test_live_migration_tunnelled_true(self):
self.flags(live_migration_tunnelled=True, group='libvirt')
self._do_test_parse_migration_flags(
lm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_LIVE |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_TUNNELLED),
bm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_LIVE |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_NON_SHARED_INC |
libvirt_driver.libvirt.VIR_MIGRATE_TUNNELLED))
def test_live_migration_with_native_tls(self):
self.flags(live_migration_with_native_tls=True, group='libvirt')
self._do_test_parse_migration_flags(
lm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_LIVE |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_TLS),
bm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_LIVE |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_NON_SHARED_INC |
libvirt_driver.libvirt.VIR_MIGRATE_TLS))
def test_live_migration_permit_postcopy_true(self):
self.flags(live_migration_permit_post_copy=True, group='libvirt')
self._do_test_parse_migration_flags(
lm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_LIVE |
libvirt_driver.libvirt.VIR_MIGRATE_POSTCOPY),
bm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_LIVE |
libvirt_driver.libvirt.VIR_MIGRATE_NON_SHARED_INC |
libvirt_driver.libvirt.VIR_MIGRATE_POSTCOPY))
def test_live_migration_permit_auto_converge_true(self):
self.flags(live_migration_permit_auto_converge=True, group='libvirt')
self._do_test_parse_migration_flags(
lm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_LIVE |
libvirt_driver.libvirt.VIR_MIGRATE_AUTO_CONVERGE),
bm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_LIVE |
libvirt_driver.libvirt.VIR_MIGRATE_NON_SHARED_INC |
libvirt_driver.libvirt.VIR_MIGRATE_AUTO_CONVERGE))
def test_live_migration_permit_auto_converge_and_post_copy_true(self):
self.flags(live_migration_permit_auto_converge=True, group='libvirt')
self.flags(live_migration_permit_post_copy=True, group='libvirt')
self._do_test_parse_migration_flags(
lm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_LIVE |
libvirt_driver.libvirt.VIR_MIGRATE_POSTCOPY),
bm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_LIVE |
libvirt_driver.libvirt.VIR_MIGRATE_NON_SHARED_INC |
libvirt_driver.libvirt.VIR_MIGRATE_POSTCOPY))
def test_live_migration_permit_postcopy_false(self):
self._do_test_parse_migration_flags(
lm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_LIVE),
bm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_LIVE |
libvirt_driver.libvirt.VIR_MIGRATE_NON_SHARED_INC))
def test_live_migration_permit_autoconverge_false(self):
self._do_test_parse_migration_flags(
lm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_LIVE),
bm_expected=(libvirt_driver.libvirt.VIR_MIGRATE_UNDEFINE_SOURCE |
libvirt_driver.libvirt.VIR_MIGRATE_PERSIST_DEST |
libvirt_driver.libvirt.VIR_MIGRATE_PEER2PEER |
libvirt_driver.libvirt.VIR_MIGRATE_LIVE |
libvirt_driver.libvirt.VIR_MIGRATE_NON_SHARED_INC))
@mock.patch('nova.utils.get_image_from_system_metadata')
@mock.patch.object(host.Host,
'has_min_version', return_value=True)
@mock.patch('nova.virt.libvirt.host.Host.get_guest')
def test_set_admin_password(self, mock_get_guest, ver, mock_image):
self.flags(virt_type='kvm', group='libvirt')
instance = objects.Instance(**self.test_instance)
mock_image.return_value = {"properties": {
"hw_qemu_guest_agent": "yes"}}
mock_guest = mock.Mock(spec=libvirt_guest.Guest)
mock_get_guest.return_value = mock_guest
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr.set_admin_password(instance, "123")
mock_guest.set_user_password.assert_called_once_with("root", "123")
@mock.patch('nova.objects.Instance.save')
@mock.patch('oslo_serialization.base64.encode_as_text')
@mock.patch('nova.api.metadata.password.convert_password')
@mock.patch('nova.crypto.ssh_encrypt_text')
@mock.patch('nova.utils.get_image_from_system_metadata')
@mock.patch.object(host.Host,
'has_min_version', return_value=True)
@mock.patch('nova.virt.libvirt.host.Host.get_guest')
def test_set_admin_password_saves_sysmeta(self, mock_get_guest,
ver, mock_image, mock_encrypt,
mock_convert, mock_encode,
mock_save):
self.flags(virt_type='kvm', group='libvirt')
instance = objects.Instance(**self.test_instance)
# Password will only be saved in sysmeta if the key_data is present
instance.key_data = 'ssh-rsa ABCFEFG'
mock_image.return_value = {"properties": {
"hw_qemu_guest_agent": "yes"}}
mock_guest = mock.Mock(spec=libvirt_guest.Guest)
mock_get_guest.return_value = mock_guest
mock_convert.return_value = {'password_0': 'converted-password'}
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr.set_admin_password(instance, "123")
mock_guest.set_user_password.assert_called_once_with("root", "123")
mock_encrypt.assert_called_once_with(instance.key_data, '123')
mock_encode.assert_called_once_with(mock_encrypt.return_value)
mock_convert.assert_called_once_with(None, mock_encode.return_value)
self.assertEqual('converted-password',
instance.system_metadata['password_0'])
mock_save.assert_called_once_with()
@mock.patch('nova.virt.libvirt.host.Host.get_guest')
def test_set_admin_password_parallels(self, mock_get_guest):
self.flags(virt_type='parallels', group='libvirt')
instance = objects.Instance(**self.test_instance)
mock_guest = mock.Mock(spec=libvirt_guest.Guest)
mock_get_guest.return_value = mock_guest
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr.set_admin_password(instance, "123")
mock_guest.set_user_password.assert_called_once_with("root", "123")
@mock.patch('nova.utils.get_image_from_system_metadata')
@mock.patch.object(host.Host,
'has_min_version', return_value=True)
@mock.patch('nova.virt.libvirt.host.Host.get_guest')
def test_set_admin_password_windows(self, mock_get_guest, ver, mock_image):
self.flags(virt_type='kvm', group='libvirt')
instance = objects.Instance(**self.test_instance)
instance.os_type = "windows"
mock_image.return_value = {"properties": {
"hw_qemu_guest_agent": "yes"}}
mock_guest = mock.Mock(spec=libvirt_guest.Guest)
mock_get_guest.return_value = mock_guest
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr.set_admin_password(instance, "123")
mock_guest.set_user_password.assert_called_once_with(
"Administrator", "123")
@mock.patch('nova.utils.get_image_from_system_metadata')
@mock.patch.object(host.Host,
'has_min_version', return_value=True)
@mock.patch('nova.virt.libvirt.host.Host.get_guest')
def test_set_admin_password_image(self, mock_get_guest, ver, mock_image):
self.flags(virt_type='kvm', group='libvirt')
instance = objects.Instance(**self.test_instance)
mock_image.return_value = {"properties": {
"hw_qemu_guest_agent": "yes",
"os_admin_user": "foo"
}}
mock_guest = mock.Mock(spec=libvirt_guest.Guest)
mock_get_guest.return_value = mock_guest
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr.set_admin_password(instance, "123")
mock_guest.set_user_password.assert_called_once_with("foo", "123")
@mock.patch('nova.utils.get_image_from_system_metadata')
@mock.patch.object(host.Host,
'has_min_version', return_value=True)
def test_set_admin_password_bad_hyp(self, mock_svc, mock_image):
self.flags(virt_type='lxc', group='libvirt')
instance = objects.Instance(**self.test_instance)
mock_image.return_value = {"properties": {
"hw_qemu_guest_agent": "yes"}}
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertRaises(exception.SetAdminPasswdNotSupported,
drvr.set_admin_password, instance, "123")
@mock.patch.object(host.Host,
'has_min_version', return_value=True)
def test_set_admin_password_guest_agent_not_running(self, mock_svc):
self.flags(virt_type='kvm', group='libvirt')
instance = objects.Instance(**self.test_instance)
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertRaises(exception.QemuGuestAgentNotEnabled,
drvr.set_admin_password, instance, "123")
@mock.patch('nova.utils.get_image_from_system_metadata')
@mock.patch.object(host.Host,
'has_min_version', return_value=True)
@mock.patch('nova.virt.libvirt.host.Host.get_guest')
def test_set_admin_password_error(self, mock_get_guest, ver, mock_image):
self.flags(virt_type='kvm', group='libvirt')
instance = objects.Instance(**self.test_instance)
mock_image.return_value = {"properties": {
"hw_qemu_guest_agent": "yes"}}
mock_guest = mock.Mock(spec=libvirt_guest.Guest)
mock_guest.set_user_password.side_effect = (
fakelibvirt.libvirtError("error"))
mock_get_guest.return_value = mock_guest
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
with mock.patch.object(
drvr, '_save_instance_password_if_sshkey_present') as save_p:
self.assertRaises(exception.NovaException,
drvr.set_admin_password, instance, "123")
save_p.assert_not_called()
@mock.patch('nova.utils.get_image_from_system_metadata')
@mock.patch.object(host.Host,
'has_min_version', return_value=True)
@mock.patch('nova.virt.libvirt.host.Host.get_guest')
def test_set_admin_password_error_with_unicode(
self, mock_get_guest, ver, mock_image):
self.flags(virt_type='kvm', group='libvirt')
instance = objects.Instance(**self.test_instance)
mock_image.return_value = {"properties": {
"hw_qemu_guest_agent": "yes"}}
mock_guest = mock.Mock(spec=libvirt_guest.Guest)
mock_guest.set_user_password.side_effect = (
fakelibvirt.libvirtError(
b"failed: \xe9\x94\x99\xe8\xaf\xaf\xe3\x80\x82"))
mock_get_guest.return_value = mock_guest
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertRaises(exception.NovaException,
drvr.set_admin_password, instance, "123")
@mock.patch('nova.utils.get_image_from_system_metadata')
@mock.patch.object(host.Host,
'has_min_version', return_value=True)
@mock.patch('nova.virt.libvirt.host.Host.get_guest')
def test_set_admin_password_not_implemented(
self, mock_get_guest, ver, mock_image):
self.flags(virt_type='kvm', group='libvirt')
instance = objects.Instance(**self.test_instance)
mock_image.return_value = {"properties": {
"hw_qemu_guest_agent": "yes"}}
mock_guest = mock.Mock(spec=libvirt_guest.Guest)
not_implemented = fakelibvirt.make_libvirtError(
fakelibvirt.libvirtError,
"Guest agent disappeared while executing command",
error_code=fakelibvirt.VIR_ERR_AGENT_UNRESPONSIVE)
mock_guest.set_user_password.side_effect = not_implemented
mock_get_guest.return_value = mock_guest
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertRaises(NotImplementedError,
drvr.set_admin_password, instance, "123")
@mock.patch(
'nova.virt.libvirt.driver.LibvirtDriver._handle_conn_event',
new=mock.Mock())
@mock.patch.object(objects.Service, 'save')
@mock.patch.object(objects.Service, 'get_by_compute_host')
def test_set_host_enabled_with_disable(self, mock_svc, mock_save):
# Tests disabling an enabled host.
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
svc = self._create_service(host='fake-mini')
mock_svc.return_value = svc
with mock.patch.object(
drvr, '_update_compute_provider_status') as ucps:
drvr._set_host_enabled(False)
ucps.assert_called_once_with(
test.MatchType(context.RequestContext), svc)
self.assertTrue(svc.disabled)
mock_save.assert_called_once_with()
@mock.patch(
'nova.virt.libvirt.driver.LibvirtDriver._handle_conn_event',
new=mock.Mock())
@mock.patch.object(objects.Service, 'save')
@mock.patch.object(objects.Service, 'get_by_compute_host')
def test_set_host_enabled_with_enable(self, mock_svc, mock_save):
# Tests enabling a disabled host.
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
svc = self._create_service(disabled=True, host='fake-mini')
mock_svc.return_value = svc
with mock.patch.object(
drvr, '_update_compute_provider_status') as ucps:
drvr._set_host_enabled(True)
ucps.assert_not_called()
# since disabled_reason is not set and not prefixed with "AUTO:",
# service should not be enabled.
mock_save.assert_not_called()
self.assertTrue(svc.disabled)
@mock.patch(
'nova.virt.libvirt.driver.LibvirtDriver._handle_conn_event',
new=mock.Mock())
@mock.patch.object(objects.Service, 'save')
@mock.patch.object(objects.Service, 'get_by_compute_host')
def test_set_host_enabled_with_enable_state_enabled(self, mock_svc,
mock_save):
# Tests enabling an enabled host.
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
svc = self._create_service(disabled=False, host='fake-mini')
mock_svc.return_value = svc
with mock.patch.object(
drvr, '_update_compute_provider_status') as ucps:
drvr._set_host_enabled(True)
ucps.assert_not_called()
self.assertFalse(svc.disabled)
mock_save.assert_not_called()
@mock.patch(
'nova.virt.libvirt.driver.LibvirtDriver._handle_conn_event',
new=mock.Mock())
@mock.patch.object(objects.Service, 'save')
@mock.patch.object(objects.Service, 'get_by_compute_host')
def test_set_host_enabled_with_disable_state_disabled(self, mock_svc,
mock_save):
# Tests disabling a disabled host.
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
svc = self._create_service(disabled=True, host='fake-mini')
mock_svc.return_value = svc
with mock.patch.object(
drvr, '_update_compute_provider_status') as ucps:
drvr._set_host_enabled(False)
ucps.assert_not_called()
mock_save.assert_not_called()
self.assertTrue(svc.disabled)
def test_set_host_enabled_swallows_exceptions(self):
# Tests that set_host_enabled will swallow exceptions coming from the
# db_api code so they don't break anything calling it, e.g. the
# _get_new_connection method.
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
with mock.patch.object(db, 'service_get_by_compute_host') as db_mock:
# Make db.service_get_by_compute_host raise NovaException; this
# is more robust than just raising ComputeHostNotFound.
db_mock.side_effect = exception.NovaException
drvr._set_host_enabled(False)
def test_update_compute_provider_status(self):
"""Tests happy path of calling _update_compute_provider_status"""
virtapi = mock.Mock()
drvr = libvirt_driver.LibvirtDriver(virtapi, read_only=True)
ctxt = context.get_admin_context()
service = self._create_service()
service.compute_node = objects.ComputeNode(uuid=uuids.rp_uuid)
drvr._update_compute_provider_status(ctxt, service)
virtapi.update_compute_provider_status.assert_called_once_with(
ctxt, uuids.rp_uuid, enabled=not service.disabled)
def test_update_compute_provider_status_swallows_exceptions(self):
"""Tests error path handling in _update_compute_provider_status"""
# First we'll make Service.compute_node loading raise an exception
# by not setting the field and we cannot lazy-load it from an orphaned
# Service object.
virtapi = mock.Mock()
drvr = libvirt_driver.LibvirtDriver(virtapi, read_only=True)
ctxt = context.get_admin_context()
service = self._create_service(host='fake-host', disabled=True)
drvr._update_compute_provider_status(ctxt, service)
virtapi.update_compute_provider_status.assert_not_called()
self.assertIn('An error occurred while updating compute node resource '
'provider status to "disabled" for provider: fake-host',
self.stdlog.logger.output)
# Now fix Service.compute_node loading but make the VirtAPI call fail.
service.compute_node = objects.ComputeNode(uuid=uuids.rp_uuid)
service.disabled = False # make sure the log message logic works
error = exception.TraitRetrievalFailed(error='oops')
virtapi.update_compute_provider_status.side_effect = error
drvr._update_compute_provider_status(ctxt, service)
virtapi.update_compute_provider_status.assert_called_once_with(
ctxt, uuids.rp_uuid, enabled=True)
log_output = self.stdlog.logger.output
self.assertIn('An error occurred while updating compute node resource '
'provider status to "enabled" for provider: %s' %
uuids.rp_uuid, log_output)
# The error should have been logged as well.
self.assertIn(str(error), log_output)
@mock.patch('nova.context.get_admin_context')
@mock.patch('nova.compute.utils.notify_about_libvirt_connect_error')
def test_versioned_notification(self, mock_notify, mock_get):
mock_get.return_value = self.context
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
fake_error = fakelibvirt.make_libvirtError(
fakelibvirt.libvirtError, "Failed to connect to host",
error_code=fakelibvirt.VIR_ERR_INTERNAL_ERROR)
with mock.patch('nova.virt.libvirt.host.Host._get_connection',
side_effect=fake_error):
self.assertRaises(exception.HypervisorUnavailable,
drvr._host.get_connection)
mock_get.assert_called_once_with()
mock_notify.assert_called_once_with(self.context, ip=CONF.my_ip,
exception=fake_error)
@mock.patch.object(host.Host, "has_min_version", return_value=False)
def test_device_metadata(self, mock_version):
xml = """
<domain>
<name>dummy</name>
<uuid>32dfcb37-5af1-552b-357c-be8c3aa38310</uuid>
<memory>1048576</memory>
<vcpu>1</vcpu>
<os>
<type arch='x86_64' machine='pc-i440fx-2.4'>hvm</type>
</os>
<devices>
<disk type='block' device='disk'>
<driver name='qemu' type='qcow2'/>
<source dev='/dev/mapper/generic'/>
<target dev='sda' bus='scsi'/>
<address type='drive' controller='0' bus='0' target='0' unit='0'/>
</disk>
<disk type='block' device='disk'>
<driver name='qemu' type='qcow2'/>
<source dev='/dev/mapper/generic-1'/>
<target dev='hda' bus='ide'/>
<address type='drive' controller='0' bus='1' target='0' unit='0'/>
</disk>
<disk type='block' device='disk'>
<driver name='qemu' type='qcow2'/>
<source dev='/dev/mapper/generic-2'/>
<target dev='hdb' bus='ide'/>
<address type='drive' controller='0' bus='1' target='1' unit='1'/>
</disk>
<disk type='block' device='disk'>
<driver name='qemu' type='qcow2'/>
<source dev='/dev/mapper/aa1'/>
<target dev='sdb' bus='usb'/>
</disk>
<disk type='block' device='disk'>
<driver name='qemu' type='qcow2'/>
<source dev='/var/lib/libvirt/images/centos'/>
<backingStore/>
<target dev='vda' bus='virtio'/>
<boot order='1'/>
<alias name='virtio-disk0'/>
<address type='pci' domain='0x0000' bus='0x00' slot='0x09'
function='0x0'/>
</disk>
<disk type='file' device='disk'>
<driver name='qemu' type='qcow2' cache='none'/>
<source file='/var/lib/libvirt/images/generic.qcow2'/>
<target dev='vdb' bus='virtio'/>
<address type='virtio-mmio'/>
</disk>
<disk type='file' device='disk'>
<driver name='qemu' type='qcow2'/>
<source file='/var/lib/libvirt/images/test.qcow2'/>
<backingStore/>
<target dev='vdc' bus='virtio'/>
<alias name='virtio-disk1'/>
<address type='ccw' cssid='0xfe' ssid='0x0' devno='0x0000'/>
</disk>
<interface type='network'>
<mac address='52:54:00:f6:35:8f'/>
<source network='default'/>
<model type='virtio'/>
<address type='pci' domain='0x0000' bus='0x00' slot='0x03'
function='0x0'/>
</interface>
<interface type='network'>
<mac address='51:5a:2c:a4:5e:1b'/>
<source network='default'/>
<model type='virtio'/>
<address type='pci' domain='0x0000' bus='0x00' slot='0x04'
function='0x1'/>
</interface>
<interface type='network'>
<mac address='fa:16:3e:d1:28:e4'/>
<source network='default'/>
<model type='virtio'/>
<address type='virtio-mmio'/>
</interface>
<interface type='network'>
<mac address='52:54:00:14:6f:50'/>
<source network='default' bridge='virbr0'/>
<target dev='vnet0'/>
<model type='virtio'/>
<alias name='net0'/>
<address type='ccw' cssid='0xfe' ssid='0x0' devno='0x0001'/>
</interface>
<hostdev mode="subsystem" type="pci" managed="yes">
<source>
<address bus="0x06" domain="0x0000" function="0x1"
slot="0x00"/>
</source>
</hostdev>
</devices>
</domain>"""
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
dom = fakelibvirt.Domain(drvr._get_connection(), xml, False)
guest = libvirt_guest.Guest(dom)
instance_ref = objects.Instance(**self.test_instance)
bdms = block_device_obj.block_device_make_list_from_dicts(
self.context, [
fake_block_device.FakeDbBlockDeviceDict(
{'id': 1,
'source_type': 'volume', 'destination_type': 'volume',
'device_name': '/dev/sda', 'tag': "db",
'volume_id': uuids.volume_1}),
fake_block_device.FakeDbBlockDeviceDict(
{'id': 2,
'source_type': 'volume', 'destination_type': 'volume',
'device_name': '/dev/hda', 'tag': "nfvfunc1",
'volume_id': uuids.volume_2}),
fake_block_device.FakeDbBlockDeviceDict(
{'id': 3,
'source_type': 'volume', 'destination_type': 'volume',
'device_name': '/dev/sdb', 'tag': "nfvfunc2",
'volume_id': uuids.volume_3}),
fake_block_device.FakeDbBlockDeviceDict(
{'id': 4,
'source_type': 'volume', 'destination_type': 'volume',
'device_name': '/dev/hdb',
'volume_id': uuids.volume_4}),
fake_block_device.FakeDbBlockDeviceDict(
{'id': 5,
'source_type': 'volume', 'destination_type': 'volume',
'device_name': '/dev/vda', 'tag': "nfvfunc3",
'volume_id': uuids.volume_5}),
fake_block_device.FakeDbBlockDeviceDict(
{'id': 6,
'source_type': 'volume', 'destination_type': 'volume',
'device_name': '/dev/vdb', 'tag': "nfvfunc4",
'volume_id': uuids.volume_6}),
fake_block_device.FakeDbBlockDeviceDict(
{'id': 7,
'source_type': 'volume', 'destination_type': 'volume',
'device_name': '/dev/vdc', 'tag': "nfvfunc5",
'volume_id': uuids.volume_7}),
]
)
vif = obj_vif.VirtualInterface(context=self.context)
vif.address = '52:54:00:f6:35:8f'
vif.network_id = 123
vif.instance_uuid = '32dfcb37-5af1-552b-357c-be8c3aa38310'
vif.uuid = '12ec4b21-ef22-6c21-534b-ba3e3ab3a311'
vif.tag = 'mytag1'
vif1 = obj_vif.VirtualInterface(context=self.context)
vif1.address = '51:5a:2c:a4:5e:1b'
vif1.network_id = 123
vif1.instance_uuid = '32dfcb37-5af1-552b-357c-be8c3aa38310'
vif1.uuid = 'abec4b21-ef22-6c21-534b-ba3e3ab3a312'
vif2 = obj_vif.VirtualInterface(context=self.context)
vif2.address = 'fa:16:3e:d1:28:e4'
vif2.network_id = 123
vif2.instance_uuid = '32dfcb37-5af1-552b-357c-be8c3aa38310'
vif2.uuid = '645686e4-7086-4eab-8c2f-c41f017a1b16'
vif2.tag = 'mytag2'
vif3 = obj_vif.VirtualInterface(context=self.context)
vif3.address = '52:54:00:14:6f:50'
vif3.network_id = 123
vif3.instance_uuid = '32dfcb37-5af1-552b-357c-be8c3aa38310'
vif3.uuid = '99cc3604-782d-4a32-a27c-bc33ac56ce86'
vif3.tag = 'mytag3'
vif4 = obj_vif.VirtualInterface(context=self.context)
vif4.address = 'da:d1:f2:91:95:c1'
vif4.tag = 'pf_tag'
vifs = [vif, vif1, vif2, vif3, vif4]
network_info = _fake_network_info(self)
network_info[0]['vnic_type'] = network_model.VNIC_TYPE_DIRECT_PHYSICAL
network_info[0]['address'] = "51:5a:2c:a4:5e:1b"
network_info[0]['details'] = dict(vlan='2145')
network_info[0]['profile'] = dict(trusted='true')
instance_ref.info_cache = objects.InstanceInfoCache(
network_info=network_info)
with test.nested(
mock.patch('nova.objects.VirtualInterfaceList'
'.get_by_instance_uuid', return_value=vifs),
mock.patch('nova.objects.BlockDeviceMappingList'
'.get_by_instance_uuid', return_value=bdms),
mock.patch('nova.virt.libvirt.host.Host.get_guest',
return_value=guest),
mock.patch.object(nova.virt.libvirt.guest.Guest, 'get_xml_desc',
return_value=xml),
mock.patch.object(pci_utils, 'get_mac_by_pci_address',
return_value='da:d1:f2:91:95:c1')):
metadata_obj = drvr._build_device_metadata(self.context,
instance_ref)
metadata = metadata_obj.devices
self.assertEqual(11, len(metadata))
self.assertIsInstance(metadata[0],
objects.DiskMetadata)
self.assertIsInstance(metadata[0].bus,
objects.SCSIDeviceBus)
self.assertEqual(['db'], metadata[0].tags)
self.assertEqual(uuids.volume_1, metadata[0].serial)
self.assertFalse(metadata[0].bus.obj_attr_is_set('address'))
self.assertEqual(['nfvfunc1'], metadata[1].tags)
self.assertEqual(uuids.volume_2, metadata[1].serial)
self.assertIsInstance(metadata[1],
objects.DiskMetadata)
self.assertIsInstance(metadata[1].bus,
objects.IDEDeviceBus)
self.assertEqual(['nfvfunc1'], metadata[1].tags)
self.assertFalse(metadata[1].bus.obj_attr_is_set('address'))
self.assertIsInstance(metadata[2],
objects.DiskMetadata)
self.assertIsInstance(metadata[2].bus,
objects.USBDeviceBus)
self.assertEqual(['nfvfunc2'], metadata[2].tags)
self.assertEqual(uuids.volume_3, metadata[2].serial)
self.assertFalse(metadata[2].bus.obj_attr_is_set('address'))
self.assertIsInstance(metadata[3],
objects.DiskMetadata)
self.assertIsInstance(metadata[3].bus,
objects.PCIDeviceBus)
self.assertEqual(['nfvfunc3'], metadata[3].tags)
# NOTE(artom) We're not checking volume 4 because it's not tagged
# and only tagged devices appear in the metadata
self.assertEqual(uuids.volume_5, metadata[3].serial)
self.assertEqual('0000:00:09.0', metadata[3].bus.address)
self.assertIsInstance(metadata[4],
objects.DiskMetadata)
self.assertEqual(['nfvfunc4'], metadata[4].tags)
self.assertEqual(uuids.volume_6, metadata[4].serial)
self.assertIsInstance(metadata[5],
objects.DiskMetadata)
self.assertEqual(['nfvfunc5'], metadata[5].tags)
self.assertEqual(uuids.volume_7, metadata[5].serial)
self.assertIsInstance(metadata[6],
objects.NetworkInterfaceMetadata)
self.assertIsInstance(metadata[6].bus,
objects.PCIDeviceBus)
self.assertEqual(['mytag1'], metadata[6].tags)
self.assertEqual('0000:00:03.0', metadata[6].bus.address)
self.assertFalse(metadata[6].vf_trusted)
# Make sure that interface with vlan is exposed to the metadata
self.assertIsInstance(metadata[7],
objects.NetworkInterfaceMetadata)
self.assertEqual('51:5a:2c:a4:5e:1b', metadata[7].mac)
self.assertEqual(2145, metadata[7].vlan)
self.assertTrue(metadata[7].vf_trusted)
self.assertIsInstance(metadata[8],
objects.NetworkInterfaceMetadata)
self.assertEqual(['mytag2'], metadata[8].tags)
self.assertFalse(metadata[8].vf_trusted)
self.assertIsInstance(metadata[9],
objects.NetworkInterfaceMetadata)
self.assertEqual(['mytag3'], metadata[9].tags)
self.assertFalse(metadata[9].vf_trusted)
self.assertIsInstance(metadata[10],
objects.NetworkInterfaceMetadata)
self.assertEqual(['pf_tag'], metadata[10].tags)
self.assertEqual('da:d1:f2:91:95:c1', metadata[10].mac)
self.assertEqual('0000:06:00.1', metadata[10].bus.address)
@mock.patch.object(host.Host, 'get_connection')
@mock.patch.object(nova.virt.libvirt.guest.Guest, 'get_xml_desc')
def test_detach_pci_devices(self, mocked_get_xml_desc, mock_conn):
fake_domXML1_with_pci = (
"""<domain> <devices>
<disk type='file' device='disk'>
<driver name='qemu' type='qcow2' cache='none'/>
<source file='xxx'/>
<target dev='vda' bus='virtio'/>
<alias name='virtio-disk0'/>
<address type='pci' domain='0x0000' bus='0x00'
slot='0x04' function='0x0'/>
</disk>
<hostdev mode="subsystem" type="pci" managed="yes">
<source>
<address function="0x1" slot="0x10" domain="0x0001"
bus="0x04"/>
</source>
</hostdev></devices></domain>""")
fake_domXML1_without_pci = (
"""<domain> <devices>
<disk type='file' device='disk'>
<driver name='qemu' type='qcow2' cache='none'/>
<source file='xxx'/>
<target dev='vda' bus='virtio'/>
<alias name='virtio-disk0'/>
<address type='pci' domain='0x0001' bus='0x00'
slot='0x04' function='0x0'/>
</disk></devices></domain>""")
pci_device_info = {'compute_node_id': 1,
'instance_uuid': 'uuid',
'address': '0001:04:10.1'}
pci_device = objects.PciDevice(**pci_device_info)
pci_devices = [pci_device]
mocked_get_xml_desc.return_value = fake_domXML1_without_pci
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
dom = fakelibvirt.Domain(
drvr._get_connection(), fake_domXML1_with_pci, False)
guest = libvirt_guest.Guest(dom)
drvr._detach_pci_devices(guest, pci_devices)
@mock.patch.object(host.Host, 'get_connection')
@mock.patch.object(nova.virt.libvirt.guest.Guest, 'get_xml_desc')
def test_detach_pci_devices_timeout(self, mocked_get_xml_desc, mock_conn):
fake_domXML1_with_pci = (
"""<domain> <devices>
<disk type='file' device='disk'>
<driver name='qemu' type='qcow2' cache='none'/>
<source file='xxx'/>
<target dev='vda' bus='virtio'/>
<alias name='virtio-disk0'/>
<address type='pci' domain='0x0000' bus='0x00'
slot='0x04' function='0x0'/>
</disk>
<hostdev mode="subsystem" type="pci" managed="yes">
<source>
<address function="0x1" slot="0x10" domain="0x0001"
bus="0x04"/>
</source>
</hostdev></devices></domain>""")
pci_device_info = {'compute_node_id': 1,
'instance_uuid': 'uuid',
'address': '0001:04:10.1'}
pci_device = objects.PciDevice(**pci_device_info)
pci_devices = [pci_device]
mocked_get_xml_desc.return_value = fake_domXML1_with_pci
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
dom = fakelibvirt.Domain(
drvr._get_connection(), fake_domXML1_with_pci, False)
guest = libvirt_guest.Guest(dom)
self.assertRaises(exception.PciDeviceDetachFailed,
drvr._detach_pci_devices, guest, pci_devices)
@mock.patch.object(connector, 'get_connector_properties')
def test_get_connector(self, fake_get_connector):
initiator = 'fake.initiator.iqn'
ip = 'fakeip'
host = 'fakehost'
wwpns = ['100010604b019419']
wwnns = ['200010604b019419']
self.flags(my_ip=ip)
self.flags(host=host)
expected = {
'ip': ip,
'initiator': initiator,
'host': host,
'wwpns': wwpns,
'wwnns': wwnns
}
volume = {
'id': 'fake'
}
# TODO(walter-boring) add the fake in os-brick
fake_get_connector.return_value = expected
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
result = drvr.get_volume_connector(volume)
self.assertThat(expected, matchers.DictMatches(result))
@mock.patch.object(connector, 'get_connector_properties')
def test_get_connector_storage_ip(self, fake_get_connector):
ip = '100.100.100.100'
storage_ip = '101.101.101.101'
self.flags(my_block_storage_ip=storage_ip, my_ip=ip)
volume = {
'id': 'fake'
}
expected = {
'ip': storage_ip
}
# TODO(walter-boring) add the fake in os-brick
fake_get_connector.return_value = expected
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
result = drvr.get_volume_connector(volume)
self.assertEqual(storage_ip, result['ip'])
@mock.patch.object(libvirt_driver.LibvirtDriver,
'_register_instance_machine_type', new=mock.Mock())
def test_lifecycle_event_registration(self):
calls = []
def fake_registerErrorHandler(*args, **kwargs):
calls.append('fake_registerErrorHandler')
def fake_get_host_capabilities(**args):
cpu = vconfig.LibvirtConfigGuestCPU()
cpu.arch = fields.Architecture.ARMV7
caps = vconfig.LibvirtConfigCaps()
caps.host = vconfig.LibvirtConfigCapsHost()
caps.host.cpu = cpu
calls.append('fake_get_host_capabilities')
return caps
@mock.patch.object(fakelibvirt, 'registerErrorHandler',
side_effect=fake_registerErrorHandler)
@mock.patch.object(host.Host, "get_capabilities",
side_effect=fake_get_host_capabilities)
def test_init_host(get_host_capabilities, register_error_handler):
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr.init_host("test_host")
test_init_host()
# NOTE(dkliban): Will fail if get_host_capabilities is called before
# registerErrorHandler
self.assertEqual('fake_registerErrorHandler', calls[0])
self.assertEqual('fake_get_host_capabilities', calls[1])
def test_sanitize_log_to_xml(self):
# setup fake data
data = {'auth_password': 'scrubme'}
bdm = [{'connection_info': {'data': data}}]
bdi = {'block_device_mapping': bdm}
# Tests that the parameters to the _get_guest_xml method
# are sanitized for passwords when logged.
def fake_debug(*args, **kwargs):
if 'auth_password' in args[0]:
self.assertNotIn('scrubme', args[0])
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
conf = mock.Mock()
with test.nested(
mock.patch.object(libvirt_driver.LOG, 'debug',
side_effect=fake_debug),
mock.patch.object(drvr, '_get_guest_config', return_value=conf)
) as (
debug_mock, conf_mock
):
drvr._get_guest_xml(self.context, self.test_instance,
network_info={}, disk_info={},
image_meta={}, block_device_info=bdi)
# we don't care what the log message is, we just want to make sure
# our stub method is called which asserts the password is scrubbed
self.assertTrue(debug_mock.called)
def test_get_guest_config_meta_with_no_port(self):
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
meta = drvr._get_guest_config_meta(
objects.Instance(**self.test_instance),
_fake_network_info(self, num_networks=0))
self.assertEqual(len(meta.ports.ports), 0)
def test_get_guest_config_meta_with_multiple_ports(self):
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
meta = drvr._get_guest_config_meta(
objects.Instance(**self.test_instance),
_fake_network_info(self, num_networks=2))
self.assertEqual(len(meta.ports.ports), 2)
# first port
self.assertEqual(meta.ports.ports[0].uuid, getattr(uuids, 'vif1'))
self.assertEqual(len(meta.ports.ports[0].ips), 2)
self.assertEqual(meta.ports.ports[0].ips[0].address, '192.168.1.100')
self.assertEqual(meta.ports.ports[0].ips[0].ip_type, 'fixed')
self.assertEqual(meta.ports.ports[0].ips[0].ip_version, 4)
self.assertEqual(meta.ports.ports[0].ips[1].address,
'2001:db8:0:1:dcad:beff:feef:1')
self.assertEqual(meta.ports.ports[0].ips[1].ip_type, 'fixed')
self.assertEqual(meta.ports.ports[0].ips[1].ip_version, 6)
# second port
self.assertEqual(meta.ports.ports[1].uuid, getattr(uuids, 'vif2'))
self.assertEqual(len(meta.ports.ports[0].ips), 2)
self.assertEqual(meta.ports.ports[1].ips[0].address, '192.168.2.100')
self.assertEqual(meta.ports.ports[1].ips[0].ip_type, 'fixed')
self.assertEqual(meta.ports.ports[1].ips[0].ip_version, 4)
self.assertEqual(meta.ports.ports[1].ips[1].address,
'2001:db8:0:2:dcad:beff:feef:1')
self.assertEqual(meta.ports.ports[1].ips[1].ip_type, 'fixed')
self.assertEqual(meta.ports.ports[1].ips[1].ip_version, 6)
@mock.patch.object(time, "time")
def test_get_guest_config(self, time_mock):
time_mock.return_value = 1234567.89
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
test_instance = copy.deepcopy(self.test_instance)
test_instance["display_name"] = "purple tomatoes"
test_instance['system_metadata']['owner_project_name'] = 'sweetshop'
test_instance['system_metadata']['owner_user_name'] = 'cupcake'
ctxt = context.RequestContext(project_id=123,
project_name="aubergine",
user_id=456,
user_name="pie")
flavor = objects.Flavor(name='m1.small',
memory_mb=6,
vcpus=28,
root_gb=496,
ephemeral_gb=8128,
swap=33550336,
extra_specs={})
instance_ref = objects.Instance(**test_instance)
instance_ref.flavor = flavor
image_meta = objects.ImageMeta.from_dict(self.test_image_meta)
disk_info = blockinfo.get_disk_info(CONF.libvirt.virt_type,
instance_ref,
image_meta)
cfg = drvr._get_guest_config(instance_ref,
_fake_network_info(self),
image_meta, disk_info,
context=ctxt)
self.assertEqual(cfg.uuid, instance_ref["uuid"])
self.assertEqual(2, len(cfg.features))
self.assertIsInstance(cfg.features[0],
vconfig.LibvirtConfigGuestFeatureACPI)
self.assertIsInstance(cfg.features[1],
vconfig.LibvirtConfigGuestFeatureAPIC)
self.assertEqual(cfg.memory, 6 * units.Ki)
self.assertEqual(cfg.vcpus, 28)
self.assertEqual(cfg.os_type, fields.VMMode.HVM)
self.assertEqual(cfg.os_boot_dev, ["hd"])
self.assertIsNone(cfg.os_root)
self.assertEqual(len(cfg.devices), 11)
self.assertIsInstance(cfg.devices[0],
vconfig.LibvirtConfigGuestDisk)
self.assertIsInstance(cfg.devices[1],
vconfig.LibvirtConfigGuestDisk)
self.assertIsInstance(cfg.devices[2],
vconfig.LibvirtConfigGuestDisk)
self.assertIsInstance(cfg.devices[3],
vconfig.LibvirtConfigGuestInterface)
self.assertIsInstance(cfg.devices[4],
vconfig.LibvirtConfigGuestSerial)
self.assertIsInstance(cfg.devices[5],
vconfig.LibvirtConfigGuestGraphics)
self.assertIsInstance(cfg.devices[6],
vconfig.LibvirtConfigGuestVideo)
self.assertIsInstance(cfg.devices[7],
vconfig.LibvirtConfigGuestInput)
self.assertIsInstance(cfg.devices[8],
vconfig.LibvirtConfigGuestRng)
self.assertIsInstance(cfg.devices[9],
vconfig.LibvirtConfigGuestUSBHostController)
self.assertIsInstance(cfg.devices[10],
vconfig.LibvirtConfigMemoryBalloon)
self.assertEqual(len(cfg.metadata), 1)
self.assertIsInstance(cfg.metadata[0],
vconfig.LibvirtConfigGuestMetaNovaInstance)
self.assertEqual(version.version_string_with_package(),
cfg.metadata[0].package)
self.assertEqual("purple tomatoes",
cfg.metadata[0].name)
self.assertEqual(1234567.89,
cfg.metadata[0].creationTime)
self.assertEqual("image",
cfg.metadata[0].roottype)
self.assertEqual(str(instance_ref["image_ref"]),
cfg.metadata[0].rootid)
self.assertIsInstance(cfg.metadata[0].owner,
vconfig.LibvirtConfigGuestMetaNovaOwner)
self.assertEqual("838a72b0-0d54-4827-8fd6-fb1227633ceb",
cfg.metadata[0].owner.userid)
self.assertEqual("cupcake",
cfg.metadata[0].owner.username)
self.assertEqual("fake",
cfg.metadata[0].owner.projectid)
self.assertEqual("sweetshop",
cfg.metadata[0].owner.projectname)
self.assertIsInstance(cfg.metadata[0].flavor,
vconfig.LibvirtConfigGuestMetaNovaFlavor)
self.assertEqual("m1.small",
cfg.metadata[0].flavor.name)
self.assertEqual(6,
cfg.metadata[0].flavor.memory)
self.assertEqual(28,
cfg.metadata[0].flavor.vcpus)
self.assertEqual(496,
cfg.metadata[0].flavor.disk)
self.assertEqual(8128,
cfg.metadata[0].flavor.ephemeral)
self.assertEqual(33550336,
cfg.metadata[0].flavor.swap)
def test_get_guest_config_q35(self):
self.flags(virt_type="kvm",
group='libvirt')
TEST_AMOUNT_OF_PCIE_SLOTS = 8
CONF.set_override("num_pcie_ports", TEST_AMOUNT_OF_PCIE_SLOTS,
group='libvirt')
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
instance_ref = objects.Instance(**self.test_instance)
image_meta = objects.ImageMeta.from_dict({
"disk_format": "raw",
"properties": {"hw_machine_type":
"pc-q35-test"}})
disk_info = blockinfo.get_disk_info(CONF.libvirt.virt_type,
instance_ref,
image_meta)
cfg = drvr._get_guest_config(instance_ref,
_fake_network_info(self),
image_meta, disk_info)
num_ports = 0
for device in cfg.devices:
try:
if (device.root_name == 'controller' and
device.model == 'pcie-root-port'):
num_ports += 1
except AttributeError:
pass
self.assertEqual(TEST_AMOUNT_OF_PCIE_SLOTS, num_ports)
def test_get_guest_config_pcie_i440fx(self):
self.flags(virt_type="kvm",
group='libvirt')
TEST_AMOUNT_OF_PCIE_SLOTS = 8
CONF.set_override("num_pcie_ports", TEST_AMOUNT_OF_PCIE_SLOTS,
group='libvirt')
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
instance_ref = objects.Instance(**self.test_instance)
image_meta = objects.ImageMeta.from_dict({
"disk_format": "raw",
"properties": {"hw_machine_type":
"pc-i440fx-test"}})
disk_info = blockinfo.get_disk_info(CONF.libvirt.virt_type,
instance_ref,
image_meta)
cfg = drvr._get_guest_config(instance_ref,
_fake_network_info(self),
image_meta, disk_info)
num_ports = 0
for device in cfg.devices:
try:
if (device.root_name == 'controller' and
device.model == 'pcie-root-port'):
num_ports += 1
except AttributeError:
pass
# i440fx is not pcie machine so there should be no pcie ports
self.assertEqual(0, num_ports)
@mock.patch('nova.virt.libvirt.utils.get_default_machine_type',
new=mock.Mock(return_value='config-machine_type'))
def test_get_guest_config_records_machine_type_in_instance(self):
# Assert that the config derived machine type is used when it
# isn't present in the image_meta of an instance.
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
instance = objects.Instance(**self.test_instance)
image_meta = objects.ImageMeta.from_dict({})
disk_info = blockinfo.get_disk_info(
CONF.libvirt.virt_type,
instance,
image_meta
)
cfg = drvr._get_guest_config(
instance,
_fake_network_info(self),
image_meta,
disk_info
)
self.assertEqual(
'config-machine_type',
instance.system_metadata.get('image_hw_machine_type'),
)
self.assertEqual(
'config-machine_type',
cfg.os_mach_type,
)
def test_get_guest_config_missing_ownership_info(self):
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
test_instance = copy.deepcopy(self.test_instance)
ctxt = context.RequestContext(project_id=123,
project_name="aubergine",
user_id=456,
user_name="pie")
flavor = objects.Flavor(name='m1.small',
memory_mb=6,
vcpus=28,
root_gb=496,
ephemeral_gb=8128,
swap=33550336,
extra_specs={})
instance_ref = objects.Instance(**test_instance)
instance_ref.flavor = flavor
image_meta = objects.ImageMeta.from_dict(self.test_image_meta)
disk_info = blockinfo.get_disk_info(CONF.libvirt.virt_type,
instance_ref,
image_meta)
cfg = drvr._get_guest_config(instance_ref,
_fake_network_info(self),
image_meta, disk_info,
context=ctxt)
self.assertEqual("N/A",
cfg.metadata[0].owner.username)
self.assertEqual("N/A",
cfg.metadata[0].owner.projectname)
def test_get_guest_config_lxc(self):
self.flags(virt_type='lxc', group='libvirt')
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
instance_ref = objects.Instance(**self.test_instance)
image_meta = objects.ImageMeta.from_dict(self.test_image_meta)
cfg = drvr._get_guest_config(instance_ref,
_fake_network_info(self),
image_meta, {'mapping': {}})
self.assertEqual(instance_ref["uuid"], cfg.uuid)
self.assertEqual(instance_ref.flavor.memory_mb * units.Ki, cfg.memory)
self.assertEqual(instance_ref.flavor.vcpus, cfg.vcpus)
self.assertEqual(fields.VMMode.EXE, cfg.os_type)
self.assertEqual("/sbin/init", cfg.os_init_path)
self.assertEqual("console=tty0 console=ttyS0 console=hvc0",
cfg.os_cmdline)
self.assertEqual("OpenStack Nova", cfg.os_init_env['product_name'])
self.assertIsNone(cfg.os_root)
self.assertEqual(4, len(cfg.devices))
self.assertIsInstance(cfg.devices[0],
vconfig.LibvirtConfigGuestFilesys)
self.assertIsInstance(cfg.devices[1],
vconfig.LibvirtConfigGuestInterface)
self.assertIsInstance(cfg.devices[2],
vconfig.LibvirtConfigGuestConsole)
self.assertIsInstance(cfg.devices[3],
vconfig.LibvirtConfigGuestUSBHostController)
def test_get_guest_config_lxc_with_id_maps(self):
self.flags(virt_type='lxc', group='libvirt')
self.flags(uid_maps=['0:1000:100'], group='libvirt')
self.flags(gid_maps=['0:1000:100'], group='libvirt')
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
instance_ref = objects.Instance(**self.test_instance)
image_meta = objects.ImageMeta.from_dict(self.test_image_meta)
cfg = drvr._get_guest_config(instance_ref,
_fake_network_info(self),
image_meta, {'mapping': {}})
self.assertEqual(instance_ref["uuid"], cfg.uuid)
self.assertEqual(instance_ref.flavor.memory_mb * units.Ki, cfg.memory)
self.assertEqual(instance_ref.vcpus, cfg.vcpus)
self.assertEqual(fields.VMMode.EXE, cfg.os_type)
self.assertEqual("/sbin/init", cfg.os_init_path)
self.assertEqual("console=tty0 console=ttyS0 console=hvc0",
cfg.os_cmdline)
self.assertIsNone(cfg.os_root)
self.assertEqual(4, len(cfg.devices))
self.assertIsInstance(cfg.devices[0],
vconfig.LibvirtConfigGuestFilesys)
self.assertIsInstance(cfg.devices[1],
vconfig.LibvirtConfigGuestInterface)
self.assertIsInstance(cfg.devices[2],
vconfig.LibvirtConfigGuestConsole)
self.assertIsInstance(cfg.devices[3],
vconfig.LibvirtConfigGuestUSBHostController)
self.assertEqual(len(cfg.idmaps), 2)
self.assertIsInstance(cfg.idmaps[0],
vconfig.LibvirtConfigGuestUIDMap)
self.assertIsInstance(cfg.idmaps[1],
vconfig.LibvirtConfigGuestGIDMap)
def test_post_claim_migrate_data(self):
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
instance = objects.Instance(**self.test_instance)
md = objects.LibvirtLiveMigrateData()
claim = mock.Mock(autospec=True)
claimed_numa_topology = objects.InstanceNUMATopology()
claim.claimed_numa_topology = claimed_numa_topology
claim.instance_type = instance.flavor
numa_info = objects.LibvirtLiveMigrateNUMAInfo()
with test.nested(
mock.patch.object(drvr, '_get_live_migrate_numa_info',
return_value=numa_info),
mock.patch('nova.objects.Instance.image_meta',
new_callable=mock.PropertyMock,
return_value='fake-image-meta')
) as (mock_get_lm_numa_info, mock_image_meta):
claim.image_meta = instance.image_meta
post_claim_md = drvr.post_claim_migrate_data(
self.context, instance, md, claim)
self.assertEqual(post_claim_md.dst_numa_info, numa_info)
mock_get_lm_numa_info.assert_called_with(
claimed_numa_topology, instance.flavor, 'fake-image-meta')
@mock.patch.object(hardware, 'get_vcpu_pin_set', new=mock.Mock())
def test_get_live_migrate_numa_info(self):
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
vcpupin1 = vconfig.LibvirtConfigGuestCPUTuneVCPUPin()
vcpupin1.id = 0
vcpupin1.cpuset = set([0, 1])
vcpupin2 = vconfig.LibvirtConfigGuestCPUTuneVCPUPin()
vcpupin2.id = 1
vcpupin2.cpuset = set([2, 3])
emulatorpin = vconfig.LibvirtConfigGuestCPUTuneEmulatorPin()
emulatorpin.cpuset = set([4, 5])
guest_cpu_tune = vconfig.LibvirtConfigGuestCPUTune()
guest_cpu_tune.vcpupin = [vcpupin1, vcpupin2]
guest_cpu_tune.emulatorpin = emulatorpin
guest_cpu_tune.vcpusched = [
vconfig.LibvirtConfigGuestCPUTuneVCPUSched()]
guest_cpu_tune.vcpusched[0].vcpus = set([6, 7])
guest_cpu_tune.vcpusched[0].priority = 8
memnode1 = vconfig.LibvirtConfigGuestNUMATuneMemNode()
memnode1.cellid = 2
memnode1.nodeset = [6, 7]
memnode2 = vconfig.LibvirtConfigGuestNUMATuneMemNode()
memnode2.cellid = 3
memnode2.nodeset = [8, 9]
guest_numa_tune = vconfig.LibvirtConfigGuestNUMATune()
guest_numa_tune.memnodes = [memnode1, memnode2]
expected_numa_info = objects.LibvirtLiveMigrateNUMAInfo(
cpu_pins={'0': set([0, 1]), '1': set([2, 3])},
cell_pins={'2': set([6, 7]), '3': set([8, 9])},
emulator_pins=set([4, 5]),
sched_vcpus=set([7, 6]),
sched_priority=8)
# NOTE(artom) This is a
# (cpu_set, guest_cpu_tune, guest_cpu_numa, guest_numa_tune)
# tuple. See _get_guest_numa_config() docstring for full documenation.
# _get_live_migrate_numa_info() only cares about guest_cpu_tune for CPU
# pinning and emulator thread pinning, and guest_numa_tune for cell
# pinning; so only include those 2 in the tuple.
guest_numa_config = (None, guest_cpu_tune, None, guest_numa_tune)
with mock.patch.object(drvr, '_get_guest_numa_config',
return_value=guest_numa_config):
self.assertEqual(
expected_numa_info.obj_to_primitive(),
drvr._get_live_migrate_numa_info(
'fake-instance-numa-topology',
'fake-flavor', 'fake-image-meta').obj_to_primitive())
@mock.patch.object(hardware, 'get_vcpu_pin_set')
def test_get_live_migrate_numa_info_empty(self, _):
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
guest_numa_config = (None, None, None, None)
with mock.patch.object(drvr, '_get_guest_numa_config',
return_value=guest_numa_config):
self.assertEqual(
objects.LibvirtLiveMigrateNUMAInfo().obj_to_primitive(),
drvr._get_live_migrate_numa_info(
'fake-instance-numa-topology',
'fake-flavor', 'fake-image-meta').obj_to_primitive())
@mock.patch.object(
host.Host, "is_cpu_control_policy_capable", return_value=True)
def test_get_guest_config_numa_host_instance_fits(self, is_able):
self.flags(cpu_shared_set=None, cpu_dedicated_set=None,
group='compute')
instance_ref = objects.Instance(**self.test_instance)
image_meta = objects.ImageMeta.from_dict(self.test_image_meta)
flavor = objects.Flavor(memory_mb=1, vcpus=2, root_gb=496,
ephemeral_gb=8128, swap=33550336, name='fake',
extra_specs={})
instance_ref.flavor = flavor
caps = vconfig.LibvirtConfigCaps()
caps.host = vconfig.LibvirtConfigCapsHost()
caps.host.cpu = vconfig.LibvirtConfigCPU()
caps.host.cpu.arch = fields.Architecture.X86_64
caps.host.topology = fakelibvirt.NUMATopology()
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
disk_info = blockinfo.get_disk_info(CONF.libvirt.virt_type,
instance_ref,
image_meta)
with test.nested(
mock.patch.object(host.Host, 'has_min_version',
return_value=True),
mock.patch.object(host.Host, "get_capabilities",
return_value=caps),
mock.patch.object(host.Host, 'get_online_cpus',
return_value=set([0, 1])),
):
cfg = drvr._get_guest_config(instance_ref, [],
image_meta, disk_info)
self.assertIsNone(cfg.cpuset)
self.assertEqual(0, len(cfg.cputune.vcpupin))
self.assertIsNone(cfg.cpu.numa)
@mock.patch('nova.privsep.utils.supports_direct_io',
new=mock.Mock(return_value=True))
@mock.patch.object(
host.Host, "is_cpu_control_policy_capable", return_value=True)
def test_get_guest_config_numa_host_instance_no_fit(self, is_able):
instance_ref = objects.Instance(**self.test_instance)
image_meta = objects.ImageMeta.from_dict(self.test_image_meta)
flavor = objects.Flavor(memory_mb=4096, vcpus=4, root_gb=496,
ephemeral_gb=8128, swap=33550336, name='fake',
extra_specs={})
instance_ref.flavor = flavor
caps = vconfig.LibvirtConfigCaps()
caps.host = vconfig.LibvirtConfigCapsHost()
caps.host.cpu = vconfig.LibvirtConfigCPU()
caps.host.cpu.arch = fields.Architecture.X86_64
caps.host.topology = fakelibvirt.NUMATopology()
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
disk_info = blockinfo.get_disk_info(CONF.libvirt.virt_type,
instance_ref,
image_meta)
with test.nested(
mock.patch.object(host.Host, "get_capabilities",
return_value=caps),
mock.patch.object(random, 'choice'),
mock.patch.object(drvr, '_has_numa_support',
return_value=False)
) as (_, choice_mock, _):
cfg = drvr._get_guest_config(instance_ref, [],
image_meta, disk_info)
self.assertFalse(choice_mock.called)
self.assertIsNone(cfg.cpuset)
self.assertEqual(0, len(cfg.cputune.vcpupin))
self.assertIsNone(cfg.cpu.numa)
def _test_get_guest_memory_backing_config(
self, host_topology, inst_topology, numatune):
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
flavor = objects.Flavor(memory_mb=4096, vcpus=4, root_gb=496,
ephemeral_gb=8128, swap=33550336, name='fake',
extra_specs={})
image_meta = objects.ImageMeta.from_dict(self.test_image_meta)
with mock.patch.object(
drvr, "_get_host_numa_topology",
return_value=host_topology):
return drvr._get_guest_memory_backing_config(
inst_topology, numatune, flavor, image_meta)
@mock.patch.object(host.Host,
'has_min_version', return_value=True)
def test_get_guest_memory_backing_config_large_success(self, mock_version):
host_topology = objects.NUMATopology(cells=[
objects.NUMACell(
id=3,
cpuset=set([1]),
pcpuset=set(),
siblings=[set([1])],
memory=1024,
mempages=[
objects.NUMAPagesTopology(size_kb=4, total=2000, used=0),
objects.NUMAPagesTopology(size_kb=2048, total=512, used=0),
objects.NUMAPagesTopology(size_kb=1048576, total=0,
used=0),
])])
inst_topology = objects.InstanceNUMATopology(cells=[
objects.InstanceNUMACell(
id=3, cpuset=set([0, 1]), pcpuset=set(), memory=1024,
pagesize=2048),
])
numa_tune = vconfig.LibvirtConfigGuestNUMATune()
numa_tune.memnodes = [vconfig.LibvirtConfigGuestNUMATuneMemNode()]
numa_tune.memnodes[0].cellid = 0
numa_tune.memnodes[0].nodeset = [3]
result = self._test_get_guest_memory_backing_config(
host_topology, inst_topology, numa_tune)
self.assertEqual(1, len(result.hugepages))
self.assertEqual(2048, result.hugepages[0].size_kb)
self.assertEqual([0], result.hugepages[0].nodeset)
@mock.patch.object(host.Host,
'has_min_version', return_value=True)
def test_get_guest_memory_backing_config_smallest(self, mock_version):
host_topology = objects.NUMATopology(cells=[
objects.NUMACell(
id=3,
cpuset=set([1]),
pcpuset=set(),
siblings=[set([1])],
memory=1024,
mempages=[
objects.NUMAPagesTopology(size_kb=4, total=2000, used=0),
objects.NUMAPagesTopology(size_kb=2048, total=512, used=0),
objects.NUMAPagesTopology(size_kb=1048576, total=0,
used=0),
])])
inst_topology = objects.InstanceNUMATopology(cells=[
objects.InstanceNUMACell(
id=3, cpuset=set([0, 1]), pcpuset=set(), memory=1024,
pagesize=4),
])
numa_tune = vconfig.LibvirtConfigGuestNUMATune()
numa_tune.memnodes = [vconfig.LibvirtConfigGuestNUMATuneMemNode()]
numa_tune.memnodes[0].cellid = 0
numa_tune.memnodes[0].nodeset = [3]
result = self._test_get_guest_memory_backing_config(
host_topology, inst_topology, numa_tune)
self.assertIsNone(result)
def test_get_guest_memory_backing_config_realtime(self):
extra_specs = {
"hw:cpu_realtime": "yes",
"hw:cpu_policy": "dedicated"
}
flavor = objects.Flavor(name='m1.small',
memory_mb=6,
vcpus=28,
root_gb=496,
ephemeral_gb=8128,
swap=33550336,
extra_specs=extra_specs)
image_meta = objects.ImageMeta.from_dict(self.test_image_meta)
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
membacking = drvr._get_guest_memory_backing_config(
None, None, flavor, image_meta)
self.assertTrue(membacking.locked)
self.assertFalse(membacking.sharedpages)
def test_get_guest_memory_backing_config_realtime_invalid_share(self):
"""Test behavior when there is no pool of shared CPUS on which to place
the emulator threads, isolating them from the instance CPU processes.
"""
extra_specs = {
"hw:cpu_realtime": "yes",
"hw:cpu_policy": "dedicated",
"hw:emulator_threads_policy": "share",
}
flavor = objects.Flavor(
name='m1.small',
memory_mb=6,
vcpus=28,
root_gb=496,
ephemeral_gb=8128,
swap=33550336,
extra_specs=extra_specs)
image_meta = objects.ImageMeta.from_dict(self.test_image_meta)
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
# this should fail because there is nowhere to place the emulator
# threads
self.assertRaises(
exception.RealtimeMaskNotFoundOrInvalid,
drvr._get_guest_memory_backing_config,
None, None, flavor, image_meta,
)
def _test_sev_enabled(self, expected=None, host_sev_enabled=False,
enc_extra_spec=None, enc_image_prop=None,
hw_machine_type=None, hw_firmware_type=None):
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr._host._supports_amd_sev = host_sev_enabled
extra_specs = {}
if enc_extra_spec is not None:
extra_specs['hw:mem_encryption'] = enc_extra_spec
flavor = objects.Flavor(name='m1.fake')
flavor.extra_specs = extra_specs
image_props = {}
image_props['hw_architecture'] = fields.Architecture.X86_64
if hw_machine_type is not None:
image_props['hw_machine_type'] = hw_machine_type
if hw_firmware_type is not None:
image_props['hw_firmware_type'] = hw_firmware_type
if enc_image_prop is not None:
image_props['hw_mem_encryption'] = enc_image_prop
image_meta = fake_image.fake_image_obj(
{'id': '150d530b-1c57-4367-b754-1f1b5237923d'},
{}, image_props)
enabled = drvr._sev_enabled(flavor, image_meta)
if expected is None:
self.fail("_test_sev_enabled called without an expected "
"return value. Maybe you expected an exception?")
self.assertEqual(expected, enabled)
def test_sev_enabled_no_host_support(self):
self._test_sev_enabled(False)
def test_sev_enabled_host_support_no_flavor_image(self):
self._test_sev_enabled(False, host_sev_enabled=True)
def test_sev_enabled_no_host_support_flavor_requested(self):
self._test_sev_enabled(False, enc_extra_spec=True)
def test_sev_enabled_no_host_support_image_requested(self):
self._test_sev_enabled(False, enc_image_prop=True)
def test_sev_enabled_host_support_flavor_requested(self):
self._test_sev_enabled(True, host_sev_enabled=True,
enc_extra_spec=True, hw_firmware_type='uefi',
hw_machine_type='q35')
def test_sev_enabled_host_support_image_requested(self):
self._test_sev_enabled(True, host_sev_enabled=True,
enc_image_prop=True, hw_firmware_type='uefi',
hw_machine_type='q35')
# The cases where the flavor and image requests contradict each other
# are already covered by test_hardware.MemEncryptionConflictTestCase
# so we don't need to test them in great detail here.
def test_sev_enabled_host_extra_spec_image_conflict(self):
exc = self.assertRaises(exception.FlavorImageConflict,
self._test_sev_enabled,
host_sev_enabled=True, enc_extra_spec=False,
enc_image_prop=True)
self.assertEqual(
"Flavor m1.fake has hw:mem_encryption extra spec explicitly set "
"to False, conflicting with image fake_image which has "
"hw_mem_encryption property explicitly set to True", str(exc))
def test_sev_enabled_host_extra_spec_no_uefi(self):
exc = self.assertRaises(exception.FlavorImageConflict,
self._test_sev_enabled,
host_sev_enabled=True, enc_extra_spec=True)
self.assertEqual(
"Memory encryption requested by hw:mem_encryption extra spec in "
"m1.fake flavor but image fake_image doesn't have "
"'hw_firmware_type' property set to 'uefi'", str(exc))
def test_sev_enabled_host_extra_spec_no_machine_type(self):
exc = self.assertRaises(exception.InvalidMachineType,
self._test_sev_enabled,
host_sev_enabled=True, enc_extra_spec=True,
hw_firmware_type='uefi')
self.assertEqual(
"Machine type 'pc' is not compatible with image fake_image "
"(150d530b-1c57-4367-b754-1f1b5237923d): q35 type is required "
"for SEV to work", str(exc))
def test_sev_enabled_host_extra_spec_pc(self):
exc = self.assertRaises(exception.InvalidMachineType,
self._test_sev_enabled,
host_sev_enabled=True, enc_extra_spec=True,
hw_firmware_type='uefi', hw_machine_type='pc')
self.assertEqual(
"Machine type 'pc' is not compatible with image fake_image "
"(150d530b-1c57-4367-b754-1f1b5237923d): q35 type is required "
"for SEV to work", str(exc))
def _setup_fake_domain_caps(self, fake_domain_caps):
sev_feature = vconfig.LibvirtConfigDomainCapsFeatureSev()
sev_feature.cbitpos = 47
sev_feature.reduced_phys_bits = 1
domain_caps = vconfig.LibvirtConfigDomainCaps()
domain_caps._features = vconfig.LibvirtConfigDomainCapsFeatures()
domain_caps._features.features = [sev_feature]
domain_caps._os = vconfig.LibvirtConfigDomainCapsOS()
domain_caps._os.loader_paths = ['foo']
fake_domain_caps.return_value = collections.defaultdict(
dict, {'x86_64': {'q35': domain_caps}})
@mock.patch.object(host.Host, 'get_domain_capabilities')
def test_find_sev_feature_missing_arch(self, fake_domain_caps):
self._setup_fake_domain_caps(fake_domain_caps)
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertIsNone(drvr._find_sev_feature('arm1', 'q35'))
@mock.patch.object(host.Host, 'get_domain_capabilities')
def test_find_sev_feature_missing_mach_type(self, fake_domain_caps):
self._setup_fake_domain_caps(fake_domain_caps)
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
self.assertIsNone(drvr._find_sev_feature('x86_64', 'g3beige'))
@mock.patch.object(host.Host, 'get_domain_capabilities')
def test_find_sev_feature(self, fake_domain_caps):
self._setup_fake_domain_caps(fake_domain_caps)
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
feature = drvr._find_sev_feature('x86_64', 'q35')
self.assertIsInstance(feature,
vconfig.LibvirtConfigDomainCapsFeatureSev)
self.assertEqual(47, feature.cbitpos)
self.assertEqual(1, feature.reduced_phys_bits)
def _setup_sev_guest(self, extra_image_properties=None):
drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
drvr._host._supports_uefi = True
drvr._host._supports_amd_sev = True
ctxt = context.RequestContext(project_id=123,
project_name="aubergine",
user_id=456,
user_name="pie")
extra_specs = {
"hw:mem_encryption": True,
}
flavor = objects.Flavor(name='m1.small',
memory_mb=6,
vcpus=28,
root_gb=496,
ephemeral_gb=8128,
swap=33550336,
extra_specs=extra_specs)
instance_ref = objects.Instance(**self.test_instance)
instance_ref.flavor = flavor
image_meta_properties = {
'hw_firmware_type': 'uefi',
'hw_machine_type': 'q35'}
if extra_image_properties:
image_meta_properties.update(extra_image_properties)
image_meta = objects.ImageMeta.from_dict({
'id': 'd9c6aeee-8258-4bdb-bca4-39940461b182',
'name': 'fakeimage',
'disk_format': 'raw',
'properties': image_meta_properties})
disk_info = blockinfo.get_disk_info(CONF.libvirt.virt_type,
instance_ref,
image_meta)
return drvr._get_guest_config(instance_ref,
_fake_network_info(self),
image_meta, disk_info,
context=ctxt)
def test_get_guest_config_sev_no_feature(self):
self.assertRaises(exception.MissingDomainCapabilityFeatureException,
self._setup_sev_guest)
@mock.patch.object(host.Host, 'get_domain_capabilities')
@mock.patch.object(designer, 'set_driver_iommu_for_sev')
def test_get_guest_config_sev(self, mock_designer, fake_domain_caps):
self._setup_fake_domain_caps(fake_domain_caps)
cfg = self._setup_sev_guest()
# SEV-related tag should be set
self.assertIsInstance(cfg.launch_security,
vconfig.LibvirtConfigGuestSEVLaunchSecurity)
self.assertIsInstance(cfg.membacking,
vconfig.LibvirtConfigGuestMemoryBacking)
self.assertTrue(cfg.membacking.locked)
mock_designer.assert_called_once_with(cfg)
def test_get_guest_memory_backing_config_file_backed(self):
self.flags(file_backed_memory=1024, group="libvirt")
result = self._test_get_guest_memory_backing_config(
None, None, None
)
self.assertTrue(result.sharedaccess)
self.assertTrue(result.filesource)
self.assertTrue(result.allocateimmediate)
def test_get_guest_memory_backing_config_file_backed_hugepages(self):
self.flags(file_backed_memory=1024, group="libvirt")
host_topology = objects.NUMATopology(cells=[
objects.NUMACell(
id=3,
cpuset=set([1]),
pcpuset=set(),
siblings=[set([1])],
memory=1024,
mempages=[
objects.NUMAPagesTopology(size_kb=4, total=2000, used=0),
objects.NUMAPagesTopology(size_kb=2048, total=512, used=0),
objects.NUMAPagesTopology(size_kb=1048576, total=0,
used=0),
])])
inst_topology = objects.InstanceNUMATopology(cells=[
objects.InstanceNUMACell(
id=3, cpuset=set([0, 1]), pcpuset=set(), memory=1024,
pagesize=2048),
])
numa_tune = vconfig.LibvirtConfigGuestNUMATune()
numa_tune.memnodes = [vconfig.LibvirtConfigGuestNUMATuneMemNode()]
numa_tune.memnodes[0].cellid = 0
numa_tune.memnodes[0].nodeset = [3]
self.assertRaises(exception.MemoryPagesUnsupported,
self._test_get_guest_memory_backing_config,
host_topology, inst_topology, numa_tune)
@mock.patch.object(
host.Host, "is_cpu_control_policy_capable", return_value=True)
def test_get_guest_config_numa_host_instance_pci_no_numa_info(
self, is_able):
self.flags(cpu_shared_set='3', cpu_dedicated_set=None,
group='compute')
instance_ref = objects.Instance(**self.test_instance)
image_meta = objects.ImageMeta.from_dict(self.test_image_meta)
flavor = objects.Flavor(memory_mb=1, vcpus=2, root_gb=496,
ephemeral_gb=8128, swap=33550336, name='fake',
extra_specs={})
instance_ref.flavor = flavor
caps = vconfig.LibvirtConfigCaps()
caps.host = vconfig.LibvirtConfigCapsHost()
caps.host.cpu = vconfig.LibvirtConfigCPU()
caps.host.cpu.arch = fields.Architecture.X86_64
caps.host.topology = fakelibvirt.NUMATopology()
conn = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), True)
disk_info = blockinfo.get_disk_info(CONF.libvirt.virt_type,
instance_ref,
image_meta)
pci_device_info = dict(test_pci_device.fake_db_dev)
pci_device_info.update(compute_node_id=1,
label='fake',
status=fields.PciDeviceStatus.AVAILABLE,
address='0000:00:00.1',
instance_uuid=None,
request_id=None,
extra_info={},
numa_node=None)
pci_device = objects.PciDevice(**pci_device_info)
with test.nested(
mock.patch.object(host.Host, 'has_min_version',
return_value=True),
mock.patch.object(host.Host, "get_capabilities",
return_value=caps),
mock.patch.object(host.Host, 'get_online_cpus',