738774b942
We use the oslo.utils save_and_reraise_exception context manager in our detach device code and catch specific exceptions that mean 'not found' and raise DeviceNotFound instead. When we do that, the save_and_reraise_exception context manager logs an ERROR traceback of the original exception, for informational purposes. This is misleading when trying to debug other issues, as it makes it look like the caught exception caused a problem. This passes the reraise=False keyword arg to the context manager and sets the 'reraise' attribute to True only if we are not going to raise a different exception. Related-Bug: #1836212 Change-Id: Icce1e31fe3ebcbf9e4897bbfa57b7f3d1fba67a3
1073 lines
44 KiB
Python
1073 lines
44 KiB
Python
# Copyright 2010 OpenStack Foundation
|
|
# Copyright 2012 University Of Minho
|
|
# Copyright 2014-2015 Red Hat, Inc
|
|
#
|
|
# 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 sys
|
|
|
|
import mock
|
|
from oslo_service import fixture as service_fixture
|
|
from oslo_utils import encodeutils
|
|
import six
|
|
import testtools
|
|
|
|
from nova import context
|
|
from nova import exception
|
|
from nova import test
|
|
from nova.tests.unit.virt.libvirt import fakelibvirt
|
|
from nova.virt.libvirt import config as vconfig
|
|
from nova.virt.libvirt import guest as libvirt_guest
|
|
from nova.virt.libvirt import host
|
|
|
|
|
|
if sys.version_info > (3,):
|
|
long = int
|
|
|
|
|
|
class GuestTestCase(test.NoDBTestCase):
|
|
|
|
def setUp(self):
|
|
super(GuestTestCase, self).setUp()
|
|
|
|
self.useFixture(fakelibvirt.FakeLibvirtFixture())
|
|
self.host = host.Host("qemu:///system")
|
|
self.context = context.get_admin_context()
|
|
|
|
self.domain = mock.Mock(spec=fakelibvirt.virDomain)
|
|
self.guest = libvirt_guest.Guest(self.domain)
|
|
|
|
# Make RetryDecorator not actually sleep on retries
|
|
self.useFixture(service_fixture.SleepFixture())
|
|
|
|
def test_repr(self):
|
|
self.domain.ID.return_value = 99
|
|
self.domain.UUIDString.return_value = "UUID"
|
|
self.domain.name.return_value = "foo"
|
|
self.assertEqual("<Guest 99 foo UUID>", repr(self.guest))
|
|
|
|
@mock.patch.object(fakelibvirt.Connection, 'defineXML')
|
|
def test_create(self, mock_define):
|
|
libvirt_guest.Guest.create("xml", self.host)
|
|
mock_define.assert_called_once_with("xml")
|
|
|
|
@mock.patch.object(fakelibvirt.Connection, 'defineXML')
|
|
def test_create_exception(self, mock_define):
|
|
mock_define.side_effect = test.TestingException
|
|
self.assertRaises(test.TestingException,
|
|
libvirt_guest.Guest.create,
|
|
"foo", self.host)
|
|
|
|
def test_launch(self):
|
|
self.guest.launch()
|
|
self.domain.createWithFlags.assert_called_once_with(0)
|
|
|
|
def test_launch_and_pause(self):
|
|
self.guest.launch(pause=True)
|
|
self.domain.createWithFlags.assert_called_once_with(
|
|
fakelibvirt.VIR_DOMAIN_START_PAUSED)
|
|
|
|
def test_shutdown(self):
|
|
self.domain.shutdown = mock.MagicMock()
|
|
self.guest.shutdown()
|
|
self.domain.shutdown.assert_called_once_with()
|
|
|
|
@mock.patch.object(encodeutils, 'safe_decode')
|
|
def test_launch_exception(self, mock_safe_decode):
|
|
self.domain.createWithFlags.side_effect = test.TestingException
|
|
mock_safe_decode.return_value = "</xml>"
|
|
self.assertRaises(test.TestingException, self.guest.launch)
|
|
self.assertEqual(1, mock_safe_decode.called)
|
|
|
|
@mock.patch('nova.privsep.libvirt.enable_hairpin')
|
|
@mock.patch.object(libvirt_guest.Guest, 'get_interfaces')
|
|
def test_enable_hairpin(self, mock_get_interfaces, mock_writefile):
|
|
mock_get_interfaces.return_value = ["vnet0", "vnet1"]
|
|
self.guest.enable_hairpin()
|
|
mock_writefile.assert_has_calls([
|
|
mock.call('vnet0'),
|
|
mock.call('vnet1')]
|
|
)
|
|
|
|
@mock.patch.object(encodeutils, 'safe_decode')
|
|
@mock.patch('nova.privsep.libvirt.enable_hairpin')
|
|
@mock.patch.object(libvirt_guest.Guest, 'get_interfaces')
|
|
def test_enable_hairpin_exception(self, mock_get_interfaces,
|
|
mock_writefile, mock_safe_decode):
|
|
mock_get_interfaces.return_value = ["foo"]
|
|
mock_writefile.side_effect = test.TestingException
|
|
|
|
self.assertRaises(test.TestingException, self.guest.enable_hairpin)
|
|
self.assertEqual(1, mock_safe_decode.called)
|
|
|
|
def test_get_interfaces(self):
|
|
self.domain.XMLDesc.return_value = """<domain>
|
|
<devices>
|
|
<interface type="network">
|
|
<target dev="vnet0"/>
|
|
</interface>
|
|
<interface type="network">
|
|
<target dev="vnet1"/>
|
|
</interface>
|
|
</devices>
|
|
</domain>"""
|
|
self.assertEqual(["vnet0", "vnet1"], self.guest.get_interfaces())
|
|
|
|
def test_get_interfaces_exception(self):
|
|
self.domain.XMLDesc.return_value = "<bad xml>"
|
|
self.assertEqual([], self.guest.get_interfaces())
|
|
|
|
def test_poweroff(self):
|
|
self.guest.poweroff()
|
|
self.domain.destroy.assert_called_once_with()
|
|
|
|
def test_resume(self):
|
|
self.guest.resume()
|
|
self.domain.resume.assert_called_once_with()
|
|
|
|
@mock.patch('time.time', return_value=1234567890.125)
|
|
def test_time_sync_no_errors(self, time_mock):
|
|
self.domain.setTime.side_effect = fakelibvirt.libvirtError('error')
|
|
self.guest.sync_guest_time()
|
|
self.domain.setTime.assert_called_once_with(time={
|
|
'nseconds': 125000000,
|
|
'seconds': 1234567890})
|
|
|
|
def test_get_vcpus_info(self):
|
|
self.domain.vcpus.return_value = ([(0, 1, int(10290000000), 2)],
|
|
[(True, True)])
|
|
vcpus = list(self.guest.get_vcpus_info())
|
|
self.assertEqual(0, vcpus[0].id)
|
|
self.assertEqual(2, vcpus[0].cpu)
|
|
self.assertEqual(1, vcpus[0].state)
|
|
self.assertEqual(int(10290000000), vcpus[0].time)
|
|
|
|
def test_delete_configuration(self):
|
|
self.guest.delete_configuration()
|
|
self.domain.undefineFlags.assert_called_once_with(
|
|
fakelibvirt.VIR_DOMAIN_UNDEFINE_MANAGED_SAVE)
|
|
|
|
def test_delete_configuration_with_nvram(self):
|
|
self.guest.delete_configuration(support_uefi=True)
|
|
self.domain.undefineFlags.assert_called_once_with(
|
|
fakelibvirt.VIR_DOMAIN_UNDEFINE_MANAGED_SAVE |
|
|
fakelibvirt.VIR_DOMAIN_UNDEFINE_NVRAM)
|
|
|
|
def test_delete_configuration_exception(self):
|
|
self.domain.undefineFlags.side_effect = fakelibvirt.libvirtError(
|
|
'oops')
|
|
self.domain.ID.return_value = 1
|
|
self.guest.delete_configuration()
|
|
self.domain.undefine.assert_called_once_with()
|
|
|
|
def test_attach_device(self):
|
|
conf = mock.Mock(spec=vconfig.LibvirtConfigGuestDevice)
|
|
conf.to_xml.return_value = "</xml>"
|
|
self.guest.attach_device(conf)
|
|
self.domain.attachDeviceFlags.assert_called_once_with(
|
|
"</xml>", flags=0)
|
|
|
|
def test_attach_device_persistent(self):
|
|
conf = mock.Mock(spec=vconfig.LibvirtConfigGuestDevice)
|
|
conf.to_xml.return_value = "</xml>"
|
|
self.guest.attach_device(conf, persistent=True)
|
|
self.domain.attachDeviceFlags.assert_called_once_with(
|
|
"</xml>", flags=fakelibvirt.VIR_DOMAIN_AFFECT_CONFIG)
|
|
|
|
def test_attach_device_live(self):
|
|
conf = mock.Mock(spec=vconfig.LibvirtConfigGuestDevice)
|
|
conf.to_xml.return_value = "</xml>"
|
|
self.guest.attach_device(conf, live=True)
|
|
self.domain.attachDeviceFlags.assert_called_once_with(
|
|
"</xml>", flags=fakelibvirt.VIR_DOMAIN_AFFECT_LIVE)
|
|
|
|
def test_attach_device_persistent_live(self):
|
|
conf = mock.Mock(spec=vconfig.LibvirtConfigGuestDevice)
|
|
conf.to_xml.return_value = "</xml>"
|
|
self.guest.attach_device(conf, persistent=True, live=True)
|
|
self.domain.attachDeviceFlags.assert_called_once_with(
|
|
"</xml>", flags=(fakelibvirt.VIR_DOMAIN_AFFECT_CONFIG |
|
|
fakelibvirt.VIR_DOMAIN_AFFECT_LIVE))
|
|
|
|
def test_detach_device(self):
|
|
conf = mock.Mock(spec=vconfig.LibvirtConfigGuestDevice)
|
|
conf.to_xml.return_value = "</xml>"
|
|
self.guest.detach_device(conf)
|
|
self.domain.detachDeviceFlags.assert_called_once_with(
|
|
"</xml>", flags=0)
|
|
|
|
def test_detach_device_persistent(self):
|
|
conf = mock.Mock(spec=vconfig.LibvirtConfigGuestDevice)
|
|
conf.to_xml.return_value = "</xml>"
|
|
self.guest.detach_device(conf, persistent=True)
|
|
self.domain.detachDeviceFlags.assert_called_once_with(
|
|
"</xml>", flags=fakelibvirt.VIR_DOMAIN_AFFECT_CONFIG)
|
|
|
|
def test_detach_device_live(self):
|
|
conf = mock.Mock(spec=vconfig.LibvirtConfigGuestDevice)
|
|
conf.to_xml.return_value = "</xml>"
|
|
self.guest.detach_device(conf, live=True)
|
|
self.domain.detachDeviceFlags.assert_called_once_with(
|
|
"</xml>", flags=fakelibvirt.VIR_DOMAIN_AFFECT_LIVE)
|
|
|
|
def test_detach_device_persistent_live(self):
|
|
conf = mock.Mock(spec=vconfig.LibvirtConfigGuestDevice)
|
|
conf.to_xml.return_value = "</xml>"
|
|
self.guest.detach_device(conf, persistent=True, live=True)
|
|
self.domain.detachDeviceFlags.assert_called_once_with(
|
|
"</xml>", flags=(fakelibvirt.VIR_DOMAIN_AFFECT_CONFIG |
|
|
fakelibvirt.VIR_DOMAIN_AFFECT_LIVE))
|
|
|
|
def test_detach_device_with_retry_from_transient_domain(self):
|
|
conf = mock.Mock(spec=vconfig.LibvirtConfigGuestDevice)
|
|
conf.to_xml.return_value = "</xml>"
|
|
get_config = mock.Mock()
|
|
get_config.side_effect = [conf, conf, conf, None, None]
|
|
dev_path = "/dev/vdb"
|
|
self.domain.isPersistent.return_value = False
|
|
retry_detach = self.guest.detach_device_with_retry(
|
|
get_config, dev_path, live=True, inc_sleep_time=.01)
|
|
self.domain.detachDeviceFlags.assert_called_once_with(
|
|
"</xml>", flags=fakelibvirt.VIR_DOMAIN_AFFECT_LIVE)
|
|
self.domain.detachDeviceFlags.reset_mock()
|
|
retry_detach()
|
|
self.assertEqual(1, self.domain.detachDeviceFlags.call_count)
|
|
|
|
def test_detach_device_with_retry_detach_success(self):
|
|
conf = mock.Mock(spec=vconfig.LibvirtConfigGuestDevice)
|
|
conf.to_xml.return_value = "</xml>"
|
|
get_config = mock.Mock()
|
|
# Force multiple retries of detach
|
|
get_config.side_effect = [conf, conf, conf, conf, conf, None, None]
|
|
dev_path = "/dev/vdb"
|
|
self.domain.isPersistent.return_value = True
|
|
|
|
retry_detach = self.guest.detach_device_with_retry(
|
|
get_config, dev_path, live=True, inc_sleep_time=.01)
|
|
# Ensure we've only done the initial detach call
|
|
self.domain.detachDeviceFlags.assert_called_once_with(
|
|
"</xml>", flags=(fakelibvirt.VIR_DOMAIN_AFFECT_CONFIG |
|
|
fakelibvirt.VIR_DOMAIN_AFFECT_LIVE))
|
|
|
|
get_config.assert_called_with(dev_path)
|
|
|
|
# Some time later, we can do the wait/retry to ensure detach succeeds
|
|
self.domain.detachDeviceFlags.reset_mock()
|
|
retry_detach()
|
|
# Should have two retries before we pretend device is detached
|
|
self.assertEqual(2, self.domain.detachDeviceFlags.call_count)
|
|
|
|
def test_detach_device_with_retry_detach_failure(self):
|
|
conf = mock.Mock(spec=vconfig.LibvirtConfigGuestDevice)
|
|
conf.to_xml.return_value = "</xml>"
|
|
# Continue to return some value for the disk config
|
|
get_config = mock.Mock(return_value=conf)
|
|
self.domain.isPersistent.return_value = True
|
|
|
|
retry_detach = self.guest.detach_device_with_retry(
|
|
get_config, "/dev/vdb", live=True, inc_sleep_time=.01,
|
|
max_retry_count=3)
|
|
# Ensure we've only done the initial detach call
|
|
self.domain.detachDeviceFlags.assert_called_once_with(
|
|
"</xml>", flags=(fakelibvirt.VIR_DOMAIN_AFFECT_CONFIG |
|
|
fakelibvirt.VIR_DOMAIN_AFFECT_LIVE))
|
|
|
|
# Some time later, we can do the wait/retry to ensure detach
|
|
self.domain.detachDeviceFlags.reset_mock()
|
|
# Should hit max # of retries
|
|
self.assertRaises(exception.DeviceDetachFailed, retry_detach)
|
|
self.assertEqual(4, self.domain.detachDeviceFlags.call_count)
|
|
|
|
def test_detach_device_with_retry_device_not_found(self):
|
|
get_config = mock.Mock(return_value=None)
|
|
self.domain.isPersistent.return_value = True
|
|
ex = self.assertRaises(
|
|
exception.DeviceNotFound, self.guest.detach_device_with_retry,
|
|
get_config, "/dev/vdb", live=True)
|
|
self.assertIn("/dev/vdb", six.text_type(ex))
|
|
|
|
def test_detach_device_with_retry_device_not_found_alt_name(self):
|
|
"""Tests to make sure we use the alternative name in errors."""
|
|
get_config = mock.Mock(return_value=None)
|
|
self.domain.isPersistent.return_value = True
|
|
ex = self.assertRaises(
|
|
exception.DeviceNotFound, self.guest.detach_device_with_retry,
|
|
get_config, mock.sentinel.device, live=True,
|
|
alternative_device_name='foo')
|
|
self.assertIn('foo', six.text_type(ex))
|
|
|
|
@mock.patch.object(libvirt_guest.Guest, "detach_device")
|
|
def test_detach_device_with_retry_operation_failed(self, mock_detach):
|
|
# This simulates a retry of the transient/live domain detach
|
|
# failing because the device is not found
|
|
conf = mock.Mock(spec=vconfig.LibvirtConfigGuestDevice)
|
|
conf.to_xml.return_value = "</xml>"
|
|
self.domain.isPersistent.return_value = True
|
|
|
|
get_config = mock.Mock(return_value=conf)
|
|
fake_device = "vdb"
|
|
fake_exc = fakelibvirt.make_libvirtError(
|
|
fakelibvirt.libvirtError, "",
|
|
error_message="operation failed: disk vdb not found",
|
|
error_code=fakelibvirt.VIR_ERR_OPERATION_FAILED,
|
|
error_domain=fakelibvirt.VIR_FROM_DOMAIN)
|
|
mock_detach.side_effect = [None, fake_exc]
|
|
retry_detach = self.guest.detach_device_with_retry(
|
|
get_config, fake_device, live=True,
|
|
inc_sleep_time=.01, max_retry_count=3)
|
|
# Some time later, we can do the wait/retry to ensure detach
|
|
self.assertRaises(exception.DeviceNotFound, retry_detach)
|
|
# Check that the save_and_reraise_exception context manager didn't log
|
|
# a traceback when the libvirtError was caught and DeviceNotFound was
|
|
# raised.
|
|
self.assertNotIn('Original exception being dropped',
|
|
self.stdlog.logger.output)
|
|
|
|
@mock.patch.object(libvirt_guest.Guest, "detach_device")
|
|
def test_detach_device_with_retry_operation_internal(self, mock_detach):
|
|
# This simulates a retry of the transient/live domain detach
|
|
# failing because the device is not found
|
|
conf = mock.Mock(spec=vconfig.LibvirtConfigGuestDevice)
|
|
conf.to_xml.return_value = "</xml>"
|
|
self.domain.isPersistent.return_value = True
|
|
|
|
get_config = mock.Mock(return_value=conf)
|
|
fake_device = "vdb"
|
|
fake_exc = fakelibvirt.make_libvirtError(
|
|
fakelibvirt.libvirtError, "",
|
|
error_message="operation failed: disk vdb not found",
|
|
error_code=fakelibvirt.VIR_ERR_INTERNAL_ERROR,
|
|
error_domain=fakelibvirt.VIR_FROM_DOMAIN)
|
|
mock_detach.side_effect = [None, fake_exc]
|
|
retry_detach = self.guest.detach_device_with_retry(
|
|
get_config, fake_device, live=True,
|
|
inc_sleep_time=.01, max_retry_count=3)
|
|
# Some time later, we can do the wait/retry to ensure detach
|
|
self.assertRaises(exception.DeviceNotFound, retry_detach)
|
|
|
|
def test_detach_device_with_retry_invalid_argument(self):
|
|
# This simulates a persistent domain detach failing because
|
|
# the device is not found
|
|
conf = mock.Mock(spec=vconfig.LibvirtConfigGuestDevice)
|
|
conf.to_xml.return_value = "</xml>"
|
|
self.domain.isPersistent.return_value = True
|
|
|
|
get_config = mock.Mock()
|
|
# Simulate the persistent domain attach attempt followed by the live
|
|
# domain attach attempt and success
|
|
get_config.side_effect = [conf, conf, conf, None, None]
|
|
fake_device = "vdb"
|
|
fake_exc = fakelibvirt.make_libvirtError(
|
|
fakelibvirt.libvirtError, "",
|
|
error_message="invalid argument: no target device vdb",
|
|
error_code=fakelibvirt.VIR_ERR_INVALID_ARG,
|
|
error_domain=fakelibvirt.VIR_FROM_DOMAIN)
|
|
# Detach from persistent raises not found, detach from live succeeds
|
|
self.domain.detachDeviceFlags.side_effect = [fake_exc, None]
|
|
retry_detach = self.guest.detach_device_with_retry(get_config,
|
|
fake_device, live=True, inc_sleep_time=.01, max_retry_count=3)
|
|
# We should have tried to detach from the persistent domain
|
|
self.domain.detachDeviceFlags.assert_called_once_with(
|
|
"</xml>", flags=(fakelibvirt.VIR_DOMAIN_AFFECT_CONFIG |
|
|
fakelibvirt.VIR_DOMAIN_AFFECT_LIVE))
|
|
# During the retry detach, should detach from the live domain
|
|
self.domain.detachDeviceFlags.reset_mock()
|
|
retry_detach()
|
|
# We should have tried to detach from the live domain
|
|
self.domain.detachDeviceFlags.assert_called_once_with(
|
|
"</xml>", flags=fakelibvirt.VIR_DOMAIN_AFFECT_LIVE)
|
|
|
|
def test_detach_device_with_retry_invalid_argument_no_live(self):
|
|
# This simulates a persistent domain detach failing because
|
|
# the device is not found
|
|
conf = mock.Mock(spec=vconfig.LibvirtConfigGuestDevice)
|
|
conf.to_xml.return_value = "</xml>"
|
|
self.domain.isPersistent.return_value = True
|
|
|
|
get_config = mock.Mock()
|
|
# Simulate the persistent domain attach attempt
|
|
get_config.return_value = conf
|
|
fake_device = "vdb"
|
|
fake_exc = fakelibvirt.make_libvirtError(
|
|
fakelibvirt.libvirtError, "",
|
|
error_message="invalid argument: no target device vdb",
|
|
error_code=fakelibvirt.VIR_ERR_INVALID_ARG,
|
|
error_domain=fakelibvirt.VIR_FROM_DOMAIN)
|
|
# Detach from persistent raises not found
|
|
self.domain.detachDeviceFlags.side_effect = fake_exc
|
|
self.assertRaises(exception.DeviceNotFound,
|
|
self.guest.detach_device_with_retry, get_config,
|
|
fake_device, live=False, inc_sleep_time=.01, max_retry_count=3)
|
|
# We should have tried to detach from the persistent domain
|
|
self.domain.detachDeviceFlags.assert_called_once_with(
|
|
"</xml>", flags=fakelibvirt.VIR_DOMAIN_AFFECT_CONFIG)
|
|
|
|
def test_get_xml_desc(self):
|
|
self.guest.get_xml_desc()
|
|
self.domain.XMLDesc.assert_called_once_with(flags=0)
|
|
|
|
def test_get_xml_desc_dump_inactive(self):
|
|
self.guest.get_xml_desc(dump_inactive=True)
|
|
self.domain.XMLDesc.assert_called_once_with(
|
|
flags=fakelibvirt.VIR_DOMAIN_XML_INACTIVE)
|
|
|
|
def test_get_xml_desc_dump_sensitive(self):
|
|
self.guest.get_xml_desc(dump_sensitive=True)
|
|
self.domain.XMLDesc.assert_called_once_with(
|
|
flags=fakelibvirt.VIR_DOMAIN_XML_SECURE)
|
|
|
|
def test_get_xml_desc_dump_inactive_dump_sensitive(self):
|
|
self.guest.get_xml_desc(dump_inactive=True, dump_sensitive=True)
|
|
self.domain.XMLDesc.assert_called_once_with(
|
|
flags=(fakelibvirt.VIR_DOMAIN_XML_INACTIVE |
|
|
fakelibvirt.VIR_DOMAIN_XML_SECURE))
|
|
|
|
def test_get_xml_desc_dump_migratable(self):
|
|
self.guest.get_xml_desc(dump_migratable=True)
|
|
self.domain.XMLDesc.assert_called_once_with(
|
|
flags=fakelibvirt.VIR_DOMAIN_XML_MIGRATABLE)
|
|
|
|
def test_has_persistent_configuration(self):
|
|
self.assertTrue(
|
|
self.guest.has_persistent_configuration())
|
|
self.domain.isPersistent.assert_called_once_with()
|
|
|
|
def test_save_memory_state(self):
|
|
self.guest.save_memory_state()
|
|
self.domain.managedSave.assert_called_once_with(0)
|
|
|
|
def test_get_block_device(self):
|
|
disk = 'vda'
|
|
gblock = self.guest.get_block_device(disk)
|
|
self.assertEqual(disk, gblock._disk)
|
|
self.assertEqual(self.guest, gblock._guest)
|
|
|
|
def test_set_user_password(self):
|
|
self.guest.set_user_password("foo", "123")
|
|
self.domain.setUserPassword.assert_called_once_with("foo", "123", 0)
|
|
|
|
def test_get_config(self):
|
|
xml = "<domain type='kvm'><name>fake</name></domain>"
|
|
self.domain.XMLDesc.return_value = xml
|
|
result = self.guest.get_config()
|
|
self.assertIsInstance(result, vconfig.LibvirtConfigGuest)
|
|
self.assertEqual('kvm', result.virt_type)
|
|
self.assertEqual('fake', result.name)
|
|
|
|
def test_get_devices(self):
|
|
xml = """
|
|
<domain type='qemu'>
|
|
<name>QEMUGuest1</name>
|
|
<uuid>c7a5fdbd-edaf-9455-926a-d65c16db1809</uuid>
|
|
<memory unit='KiB'>219136</memory>
|
|
<currentMemory unit='KiB'>219136</currentMemory>
|
|
<vcpu placement='static'>1</vcpu>
|
|
<os>
|
|
<type arch='i686' machine='pc'>hvm</type>
|
|
<boot dev='hd'/>
|
|
</os>
|
|
<clock offset='utc'/>
|
|
<on_poweroff>destroy</on_poweroff>
|
|
<on_reboot>restart</on_reboot>
|
|
<on_crash>destroy</on_crash>
|
|
<devices>
|
|
<emulator>/usr/bin/qemu</emulator>
|
|
<disk type='block' device='disk'>
|
|
<driver name='qemu' type='raw'/>
|
|
<source dev='/dev/HostVG/QEMUGuest2'/>
|
|
<target dev='hda' bus='ide'/>
|
|
<address type='drive' controller='0' bus='0' target='0' unit='0'/>
|
|
</disk>
|
|
<disk type='network' device='disk'>
|
|
<driver name='qemu' type='raw'/>
|
|
<auth username='myname'>
|
|
<secret type='iscsi' usage='mycluster_myname'/>
|
|
</auth>
|
|
<source protocol='iscsi' name='iqn.1992-01.com.example'>
|
|
<host name='example.org' port='6000'/>
|
|
</source>
|
|
<target dev='vda' bus='virtio'/>
|
|
</disk>
|
|
<disk type='network' device='disk'>
|
|
<driver name='qemu' type='raw'/>
|
|
<source protocol='iscsi' name='iqn.1992-01.com.example/1'>
|
|
<host name='example.org' port='6000'/>
|
|
</source>
|
|
<target dev='vdb' bus='virtio'/>
|
|
</disk>
|
|
<hostdev mode='subsystem' type='pci' managed='yes'>
|
|
<source>
|
|
<address domain='0x0000' bus='0x06' slot='0x12' function='0x5'/>
|
|
</source>
|
|
</hostdev>
|
|
<hostdev mode='subsystem' type='pci' managed='yes'>
|
|
<source>
|
|
<address domain='0x0000' bus='0x06' slot='0x12' function='0x6'/>
|
|
</source>
|
|
</hostdev>
|
|
<interface type="bridge">
|
|
<mac address="fa:16:3e:f9:af:ae"/>
|
|
<model type="virtio"/>
|
|
<driver name="qemu"/>
|
|
<source bridge="qbr84008d03-11"/>
|
|
<target dev="tap84008d03-11"/>
|
|
</interface>
|
|
<controller type='usb' index='0'/>
|
|
<controller type='pci' index='0' model='pci-root'/>
|
|
<memballoon model='none'/>
|
|
</devices>
|
|
</domain>
|
|
"""
|
|
|
|
self.domain.XMLDesc.return_value = xml
|
|
|
|
devs = self.guest.get_all_devices()
|
|
# Only currently parse <disk>, <hostdev> and <interface> elements
|
|
# hence we're not counting the controller/memballoon
|
|
self.assertEqual(6, len(devs))
|
|
self.assertIsInstance(devs[0], vconfig.LibvirtConfigGuestDisk)
|
|
self.assertIsInstance(devs[1], vconfig.LibvirtConfigGuestDisk)
|
|
self.assertIsInstance(devs[2], vconfig.LibvirtConfigGuestDisk)
|
|
self.assertIsInstance(devs[3], vconfig.LibvirtConfigGuestHostdev)
|
|
self.assertIsInstance(devs[4], vconfig.LibvirtConfigGuestHostdev)
|
|
self.assertIsInstance(devs[5], vconfig.LibvirtConfigGuestInterface)
|
|
|
|
devs = self.guest.get_all_devices(vconfig.LibvirtConfigGuestDisk)
|
|
self.assertEqual(3, len(devs))
|
|
self.assertIsInstance(devs[0], vconfig.LibvirtConfigGuestDisk)
|
|
self.assertIsInstance(devs[1], vconfig.LibvirtConfigGuestDisk)
|
|
self.assertIsInstance(devs[2], vconfig.LibvirtConfigGuestDisk)
|
|
|
|
devs = self.guest.get_all_disks()
|
|
self.assertEqual(3, len(devs))
|
|
self.assertIsInstance(devs[0], vconfig.LibvirtConfigGuestDisk)
|
|
self.assertIsInstance(devs[1], vconfig.LibvirtConfigGuestDisk)
|
|
self.assertIsInstance(devs[2], vconfig.LibvirtConfigGuestDisk)
|
|
|
|
devs = self.guest.get_all_devices(vconfig.LibvirtConfigGuestHostdev)
|
|
self.assertEqual(2, len(devs))
|
|
self.assertIsInstance(devs[0], vconfig.LibvirtConfigGuestHostdev)
|
|
self.assertIsInstance(devs[1], vconfig.LibvirtConfigGuestHostdev)
|
|
|
|
devs = self.guest.get_all_devices(vconfig.LibvirtConfigGuestInterface)
|
|
self.assertEqual(1, len(devs))
|
|
self.assertIsInstance(devs[0], vconfig.LibvirtConfigGuestInterface)
|
|
|
|
cfg = vconfig.LibvirtConfigGuestInterface()
|
|
cfg.parse_str("""
|
|
<interface type="bridge">
|
|
<mac address="fa:16:3e:f9:af:ae"/>
|
|
<model type="virtio"/>
|
|
<driver name="qemu"/>
|
|
<source bridge="qbr84008d03-11"/>
|
|
<target dev="tap84008d03-11"/>
|
|
</interface>""")
|
|
self.assertIsNotNone(
|
|
self.guest.get_interface_by_cfg(cfg))
|
|
self.assertIsNone(self.guest.get_interface_by_cfg(None))
|
|
|
|
def test_get_interface_by_cfg_vhostuser(self):
|
|
self.domain.XMLDesc.return_value = """<domain>
|
|
<devices>
|
|
<interface type="vhostuser">
|
|
<mac address='fa:16:3e:55:3e:e4'/>
|
|
<source type='unix' path='/var/run/openvswitch/vhued80c655-4e'
|
|
mode='server'/>
|
|
<target dev='vhued80c655-4e'/>
|
|
<model type='virtio'/>
|
|
<alias name='net0'/>
|
|
<address type='pci' domain='0x0000' bus='0x00' slot='0x03'
|
|
function='0x0'/>
|
|
</interface>
|
|
</devices>
|
|
</domain>"""
|
|
cfg = vconfig.LibvirtConfigGuestInterface()
|
|
cfg.parse_str("""<interface type="vhostuser">
|
|
<mac address='fa:16:3e:55:3e:e4'/>
|
|
<model type="virtio"/>
|
|
<source type='unix' path='/var/run/openvswitch/vhued80c655-4e'
|
|
mode='server'/>
|
|
<alias name='net0'/>
|
|
<address type='pci' domain='0x0000' bus='0x00' slot='0x03' function='0x0'/>
|
|
</interface>""")
|
|
self.assertIsNotNone(
|
|
self.guest.get_interface_by_cfg(cfg))
|
|
self.assertIsNone(self.guest.get_interface_by_cfg(None))
|
|
|
|
def test_get_info(self):
|
|
self.domain.info.return_value = (1, 2, 3, 4, 5)
|
|
self.domain.ID.return_value = 6
|
|
info = self.guest.get_info(self.host)
|
|
self.domain.info.assert_called_once_with()
|
|
self.assertEqual(1, info.state)
|
|
self.assertEqual(6, info.internal_id)
|
|
|
|
def test_get_power_state(self):
|
|
self.domain.info.return_value = (1, 2, 3, 4, 5)
|
|
power = self.guest.get_power_state(self.host)
|
|
self.assertEqual(1, power)
|
|
|
|
def test_is_active_when_domain_is_active(self):
|
|
with mock.patch.object(self.domain, "isActive", return_value=True):
|
|
self.assertTrue(self.guest.is_active())
|
|
|
|
def test_is_active_when_domain_not_active(self):
|
|
with mock.patch.object(self.domain, "isActive", return_value=False):
|
|
self.assertFalse(self.guest.is_active())
|
|
|
|
def test_freeze_filesystems(self):
|
|
self.guest.freeze_filesystems()
|
|
self.domain.fsFreeze.assert_called_once_with()
|
|
|
|
def test_thaw_filesystems(self):
|
|
self.guest.thaw_filesystems()
|
|
self.domain.fsThaw.assert_called_once_with()
|
|
|
|
def _conf_snapshot(self):
|
|
conf = mock.Mock(spec=vconfig.LibvirtConfigGuestSnapshotDisk)
|
|
conf.to_xml.return_value = '<disk/>'
|
|
return conf
|
|
|
|
def test_snapshot(self):
|
|
conf = self._conf_snapshot()
|
|
self.guest.snapshot(conf)
|
|
self.domain.snapshotCreateXML('<disk/>', flags=0)
|
|
conf.to_xml.assert_called_once_with()
|
|
|
|
def test_snapshot_no_metadata(self):
|
|
conf = self._conf_snapshot()
|
|
self.guest.snapshot(conf, no_metadata=True)
|
|
self.domain.snapshotCreateXML(
|
|
'<disk/>',
|
|
flags=fakelibvirt.VIR_DOMAIN_SNAPSHOT_CREATE_NO_METADATA)
|
|
conf.to_xml.assert_called_once_with()
|
|
|
|
def test_snapshot_disk_only(self):
|
|
conf = self._conf_snapshot()
|
|
self.guest.snapshot(conf, disk_only=True)
|
|
self.domain.snapshotCreateXML(
|
|
'<disk/>', flags=fakelibvirt.VIR_DOMAIN_SNAPSHOT_CREATE_DISK_ONLY)
|
|
conf.to_xml.assert_called_once_with()
|
|
|
|
def test_snapshot_reuse_ext(self):
|
|
conf = self._conf_snapshot()
|
|
self.guest.snapshot(conf, reuse_ext=True)
|
|
self.domain.snapshotCreateXML(
|
|
'<disk/>', flags=fakelibvirt.VIR_DOMAIN_SNAPSHOT_CREATE_REUSE_EXT)
|
|
conf.to_xml.assert_called_once_with()
|
|
|
|
def test_snapshot_quiesce(self):
|
|
conf = self._conf_snapshot()
|
|
self.guest.snapshot(conf, quiesce=True)
|
|
self.domain.snapshotCreateXML(
|
|
'<disk/>', flags=fakelibvirt.VIR_DOMAIN_SNAPSHOT_CREATE_QUIESCE)
|
|
conf.to_xml.assert_called_once_with()
|
|
|
|
def test_snapshot_all(self):
|
|
conf = self._conf_snapshot()
|
|
self.guest.snapshot(conf, no_metadata=True,
|
|
disk_only=True, reuse_ext=True,
|
|
quiesce=True)
|
|
self.domain.snapshotCreateXML(
|
|
'<disk/>', flags=(
|
|
fakelibvirt.VIR_DOMAIN_SNAPSHOT_CREATE_REUSE_EXT |
|
|
fakelibvirt.VIR_DOMAIN_SNAPSHOT_CREATE_DISK_ONLY |
|
|
fakelibvirt.VIR_DOMAIN_SNAPSHOT_CREATE_NO_METADATA |
|
|
fakelibvirt.VIR_DOMAIN_SNAPSHOT_CREATE_QUIESCE))
|
|
conf.to_xml.assert_called_once_with()
|
|
|
|
def test_pause(self):
|
|
self.guest.pause()
|
|
self.domain.suspend.assert_called_once_with()
|
|
|
|
def test_migrate_v3(self):
|
|
self.guest.migrate('an-uri', flags=1, migrate_uri='dest-uri',
|
|
migrate_disks='disk1',
|
|
destination_xml='</xml>',
|
|
bandwidth=2)
|
|
self.domain.migrateToURI3.assert_called_once_with(
|
|
'an-uri', flags=1, params={'migrate_uri': 'dest-uri',
|
|
'migrate_disks': 'disk1',
|
|
'destination_xml': '</xml>',
|
|
'bandwidth': 2})
|
|
|
|
@testtools.skipIf(not six.PY2, 'libvirt python3 bindings accept unicode')
|
|
def test_migrate_v3_unicode(self):
|
|
dest_xml_template = "<domain type='qemu'><name>%s</name></domain>"
|
|
name = u'\u00CD\u00F1st\u00E1\u00F1c\u00E9'
|
|
dest_xml = dest_xml_template % name
|
|
expect_dest_xml = dest_xml_template % encodeutils.to_utf8(name)
|
|
self.guest.migrate('an-uri', flags=1, migrate_uri='dest-uri',
|
|
migrate_disks=[u"disk1", u"disk2"],
|
|
destination_xml=dest_xml,
|
|
bandwidth=2)
|
|
self.domain.migrateToURI3.assert_called_once_with(
|
|
'an-uri', flags=1, params={'migrate_uri': 'dest-uri',
|
|
'migrate_disks': ['disk1',
|
|
'disk2'],
|
|
'destination_xml': expect_dest_xml,
|
|
'bandwidth': 2})
|
|
|
|
def test_abort_job(self):
|
|
self.guest.abort_job()
|
|
self.domain.abortJob.assert_called_once_with()
|
|
|
|
def test_migrate_configure_max_downtime(self):
|
|
self.guest.migrate_configure_max_downtime(1000)
|
|
self.domain.migrateSetMaxDowntime.assert_called_once_with(1000)
|
|
|
|
|
|
class GuestBlockTestCase(test.NoDBTestCase):
|
|
|
|
def setUp(self):
|
|
super(GuestBlockTestCase, self).setUp()
|
|
|
|
self.useFixture(fakelibvirt.FakeLibvirtFixture())
|
|
self.host = host.Host("qemu:///system")
|
|
self.context = context.get_admin_context()
|
|
|
|
self.domain = mock.Mock(spec=fakelibvirt.virDomain)
|
|
self.guest = libvirt_guest.Guest(self.domain)
|
|
self.gblock = self.guest.get_block_device('vda')
|
|
|
|
def test_abort_job(self):
|
|
self.gblock.abort_job()
|
|
self.domain.blockJobAbort.assert_called_once_with('vda', flags=0)
|
|
|
|
def test_abort_job_async(self):
|
|
self.gblock.abort_job(async_=True)
|
|
self.domain.blockJobAbort.assert_called_once_with(
|
|
'vda', flags=fakelibvirt.VIR_DOMAIN_BLOCK_JOB_ABORT_ASYNC)
|
|
|
|
def test_abort_job_pivot(self):
|
|
self.gblock.abort_job(pivot=True)
|
|
self.domain.blockJobAbort.assert_called_once_with(
|
|
'vda', flags=fakelibvirt.VIR_DOMAIN_BLOCK_JOB_ABORT_PIVOT)
|
|
|
|
def test_get_job_info(self):
|
|
self.domain.blockJobInfo.return_value = {
|
|
"type": 1,
|
|
"bandwidth": 18,
|
|
"cur": 66,
|
|
"end": 100}
|
|
|
|
info = self.gblock.get_job_info()
|
|
self.assertEqual(1, info.job)
|
|
self.assertEqual(18, info.bandwidth)
|
|
self.assertEqual(66, info.cur)
|
|
self.assertEqual(100, info.end)
|
|
self.domain.blockJobInfo.assert_called_once_with('vda', flags=0)
|
|
|
|
def test_resize(self):
|
|
self.gblock.resize(10)
|
|
self.domain.blockResize.assert_called_once_with('vda', 10)
|
|
|
|
def test_rebase(self):
|
|
self.gblock.rebase("foo")
|
|
self.domain.blockRebase.assert_called_once_with(
|
|
'vda', "foo", 0, flags=0)
|
|
|
|
def test_rebase_shallow(self):
|
|
self.gblock.rebase("foo", shallow=True)
|
|
self.domain.blockRebase.assert_called_once_with(
|
|
'vda', "foo", 0, flags=fakelibvirt.VIR_DOMAIN_BLOCK_REBASE_SHALLOW)
|
|
|
|
def test_rebase_reuse_ext(self):
|
|
self.gblock.rebase("foo", reuse_ext=True)
|
|
self.domain.blockRebase.assert_called_once_with(
|
|
'vda', "foo", 0,
|
|
flags=fakelibvirt.VIR_DOMAIN_BLOCK_REBASE_REUSE_EXT)
|
|
|
|
def test_rebase_copy(self):
|
|
self.gblock.rebase("foo", copy=True)
|
|
self.domain.blockRebase.assert_called_once_with(
|
|
'vda', "foo", 0,
|
|
flags=fakelibvirt.VIR_DOMAIN_BLOCK_REBASE_COPY)
|
|
|
|
def test_rebase_relative(self):
|
|
self.gblock.rebase("foo", relative=True)
|
|
self.domain.blockRebase.assert_called_once_with(
|
|
'vda', "foo", 0,
|
|
flags=fakelibvirt.VIR_DOMAIN_BLOCK_REBASE_RELATIVE)
|
|
|
|
def test_rebase_copy_dev(self):
|
|
self.gblock.rebase("foo", copy_dev=True)
|
|
self.domain.blockRebase.assert_called_once_with(
|
|
'vda', "foo", 0,
|
|
flags=fakelibvirt.VIR_DOMAIN_BLOCK_REBASE_COPY_DEV)
|
|
|
|
def test_commit(self):
|
|
self.gblock.commit("foo", "top")
|
|
self.domain.blockCommit.assert_called_once_with(
|
|
'vda', "foo", "top", 0, flags=0)
|
|
|
|
def test_commit_relative(self):
|
|
self.gblock.commit("foo", "top", relative=True)
|
|
self.domain.blockCommit.assert_called_once_with(
|
|
'vda', "foo", "top", 0,
|
|
flags=fakelibvirt.VIR_DOMAIN_BLOCK_COMMIT_RELATIVE)
|
|
|
|
def test_is_job_complete_cur_end_zeros(self):
|
|
self.domain.blockJobInfo.return_value = {
|
|
"type": 4,
|
|
"bandwidth": 18,
|
|
"cur": 0,
|
|
"end": 0}
|
|
is_complete = self.gblock.is_job_complete()
|
|
self.assertFalse(is_complete)
|
|
|
|
def test_is_job_complete_current_lower_than_end(self):
|
|
self.domain.blockJobInfo.return_value = {
|
|
"type": 4,
|
|
"bandwidth": 18,
|
|
"cur": 95,
|
|
"end": 100}
|
|
is_complete = self.gblock.is_job_complete()
|
|
self.assertFalse(is_complete)
|
|
|
|
def test_is_job_complete_not_ready(self):
|
|
gblock = self.guest.get_block_device('vda')
|
|
disk = vconfig.LibvirtConfigGuestDisk()
|
|
disk.mirror = vconfig.LibvirtConfigGuestDiskMirror()
|
|
with mock.patch.object(self.guest, 'get_disk', return_value=disk):
|
|
self.domain.blockJobInfo.return_value = {
|
|
"type": 4,
|
|
"bandwidth": 18,
|
|
"cur": 100,
|
|
"end": 100}
|
|
is_complete = gblock.is_job_complete()
|
|
self.assertFalse(is_complete)
|
|
|
|
def test_is_job_complete_ready(self):
|
|
gblock = self.guest.get_block_device('vda')
|
|
disk = vconfig.LibvirtConfigGuestDisk()
|
|
disk.mirror = vconfig.LibvirtConfigGuestDiskMirror()
|
|
disk.mirror.ready = 'yes'
|
|
with mock.patch.object(self.guest, 'get_disk', return_value=disk):
|
|
self.domain.blockJobInfo.return_value = {
|
|
"type": 4,
|
|
"bandwidth": 18,
|
|
"cur": 100,
|
|
"end": 100}
|
|
is_complete = gblock.is_job_complete()
|
|
self.assertTrue(is_complete)
|
|
|
|
def test_is_job_complete_no_job(self):
|
|
self.domain.blockJobInfo.return_value = {}
|
|
is_complete = self.gblock.is_job_complete()
|
|
self.assertTrue(is_complete)
|
|
|
|
def test_is_job_complete_exception(self):
|
|
self.domain.blockJobInfo.side_effect = fakelibvirt.libvirtError('fake')
|
|
self.assertRaises(fakelibvirt.libvirtError,
|
|
self.gblock.is_job_complete)
|
|
|
|
def test_blockStats(self):
|
|
self.gblock.blockStats()
|
|
self.domain.blockStats.assert_called_once_with('vda')
|
|
|
|
|
|
class JobInfoTestCase(test.NoDBTestCase):
|
|
|
|
def setUp(self):
|
|
super(JobInfoTestCase, self).setUp()
|
|
|
|
self.useFixture(fakelibvirt.FakeLibvirtFixture())
|
|
|
|
self.conn = fakelibvirt.openAuth("qemu:///system",
|
|
[[], lambda: True])
|
|
xml = ("<domain type='kvm'>"
|
|
" <name>instance-0000000a</name>"
|
|
"</domain>")
|
|
self.dom = self.conn.createXML(xml, 0)
|
|
self.guest = libvirt_guest.Guest(self.dom)
|
|
libvirt_guest.JobInfo._have_job_stats = True
|
|
|
|
@mock.patch.object(fakelibvirt.virDomain, "jobInfo")
|
|
@mock.patch.object(fakelibvirt.virDomain, "jobStats")
|
|
def test_job_stats(self, mock_stats, mock_info):
|
|
mock_stats.return_value = {
|
|
"type": fakelibvirt.VIR_DOMAIN_JOB_UNBOUNDED,
|
|
"memory_total": 75,
|
|
"memory_processed": 50,
|
|
"memory_remaining": 33,
|
|
"some_new_libvirt_stat_we_dont_know_about": 83
|
|
}
|
|
|
|
info = self.guest.get_job_info()
|
|
|
|
self.assertIsInstance(info, libvirt_guest.JobInfo)
|
|
self.assertEqual(fakelibvirt.VIR_DOMAIN_JOB_UNBOUNDED, info.type)
|
|
self.assertEqual(75, info.memory_total)
|
|
self.assertEqual(50, info.memory_processed)
|
|
self.assertEqual(33, info.memory_remaining)
|
|
self.assertEqual(0, info.disk_total)
|
|
self.assertEqual(0, info.disk_processed)
|
|
self.assertEqual(0, info.disk_remaining)
|
|
|
|
mock_stats.assert_called_once_with()
|
|
self.assertFalse(mock_info.called)
|
|
|
|
@mock.patch.object(fakelibvirt.virDomain, "jobInfo")
|
|
@mock.patch.object(fakelibvirt.virDomain, "jobStats")
|
|
def test_job_info_no_support(self, mock_stats, mock_info):
|
|
mock_stats.side_effect = fakelibvirt.make_libvirtError(
|
|
fakelibvirt.libvirtError,
|
|
"virDomainGetJobStats not implemented",
|
|
fakelibvirt.VIR_ERR_NO_SUPPORT)
|
|
|
|
mock_info.return_value = [
|
|
fakelibvirt.VIR_DOMAIN_JOB_UNBOUNDED,
|
|
100, 99, 10, 11, 12, 75, 50, 33, 1, 2, 3]
|
|
|
|
info = self.guest.get_job_info()
|
|
|
|
self.assertIsInstance(info, libvirt_guest.JobInfo)
|
|
self.assertEqual(fakelibvirt.VIR_DOMAIN_JOB_UNBOUNDED, info.type)
|
|
self.assertEqual(100, info.time_elapsed)
|
|
self.assertEqual(99, info.time_remaining)
|
|
self.assertEqual(10, info.data_total)
|
|
self.assertEqual(11, info.data_processed)
|
|
self.assertEqual(12, info.data_remaining)
|
|
self.assertEqual(75, info.memory_total)
|
|
self.assertEqual(50, info.memory_processed)
|
|
self.assertEqual(33, info.memory_remaining)
|
|
self.assertEqual(1, info.disk_total)
|
|
self.assertEqual(2, info.disk_processed)
|
|
self.assertEqual(3, info.disk_remaining)
|
|
|
|
mock_stats.assert_called_once_with()
|
|
mock_info.assert_called_once_with()
|
|
|
|
@mock.patch.object(fakelibvirt.virDomain, "jobInfo")
|
|
@mock.patch.object(fakelibvirt.virDomain, "jobStats")
|
|
def test_job_info_attr_error(self, mock_stats, mock_info):
|
|
mock_stats.side_effect = AttributeError("No such API")
|
|
|
|
mock_info.return_value = [
|
|
fakelibvirt.VIR_DOMAIN_JOB_UNBOUNDED,
|
|
100, 99, 10, 11, 12, 75, 50, 33, 1, 2, 3]
|
|
|
|
info = self.guest.get_job_info()
|
|
|
|
self.assertIsInstance(info, libvirt_guest.JobInfo)
|
|
self.assertEqual(fakelibvirt.VIR_DOMAIN_JOB_UNBOUNDED, info.type)
|
|
self.assertEqual(100, info.time_elapsed)
|
|
self.assertEqual(99, info.time_remaining)
|
|
self.assertEqual(10, info.data_total)
|
|
self.assertEqual(11, info.data_processed)
|
|
self.assertEqual(12, info.data_remaining)
|
|
self.assertEqual(75, info.memory_total)
|
|
self.assertEqual(50, info.memory_processed)
|
|
self.assertEqual(33, info.memory_remaining)
|
|
self.assertEqual(1, info.disk_total)
|
|
self.assertEqual(2, info.disk_processed)
|
|
self.assertEqual(3, info.disk_remaining)
|
|
|
|
mock_stats.assert_called_once_with()
|
|
mock_info.assert_called_once_with()
|
|
|
|
@mock.patch.object(fakelibvirt.virDomain, "jobInfo")
|
|
@mock.patch.object(fakelibvirt.virDomain, "jobStats")
|
|
def test_job_stats_no_domain(self, mock_stats, mock_info):
|
|
mock_stats.side_effect = fakelibvirt.make_libvirtError(
|
|
fakelibvirt.libvirtError,
|
|
"No such domain with UUID blah",
|
|
fakelibvirt.VIR_ERR_NO_DOMAIN)
|
|
|
|
info = self.guest.get_job_info()
|
|
|
|
self.assertIsInstance(info, libvirt_guest.JobInfo)
|
|
self.assertEqual(fakelibvirt.VIR_DOMAIN_JOB_COMPLETED, info.type)
|
|
self.assertEqual(0, info.time_elapsed)
|
|
self.assertEqual(0, info.time_remaining)
|
|
self.assertEqual(0, info.memory_total)
|
|
self.assertEqual(0, info.memory_processed)
|
|
self.assertEqual(0, info.memory_remaining)
|
|
|
|
mock_stats.assert_called_once_with()
|
|
self.assertFalse(mock_info.called)
|
|
|
|
@mock.patch.object(fakelibvirt.virDomain, "jobInfo")
|
|
@mock.patch.object(fakelibvirt.virDomain, "jobStats")
|
|
def test_job_info_no_domain(self, mock_stats, mock_info):
|
|
mock_stats.side_effect = fakelibvirt.make_libvirtError(
|
|
fakelibvirt.libvirtError,
|
|
"virDomainGetJobStats not implemented",
|
|
fakelibvirt.VIR_ERR_NO_SUPPORT)
|
|
|
|
mock_info.side_effect = fakelibvirt.make_libvirtError(
|
|
fakelibvirt.libvirtError,
|
|
"No such domain with UUID blah",
|
|
fakelibvirt.VIR_ERR_NO_DOMAIN)
|
|
|
|
info = self.guest.get_job_info()
|
|
|
|
self.assertIsInstance(info, libvirt_guest.JobInfo)
|
|
self.assertEqual(fakelibvirt.VIR_DOMAIN_JOB_COMPLETED, info.type)
|
|
self.assertEqual(0, info.time_elapsed)
|
|
self.assertEqual(0, info.time_remaining)
|
|
self.assertEqual(0, info.memory_total)
|
|
self.assertEqual(0, info.memory_processed)
|
|
self.assertEqual(0, info.memory_remaining)
|
|
|
|
mock_stats.assert_called_once_with()
|
|
mock_info.assert_called_once_with()
|
|
|
|
@mock.patch.object(fakelibvirt.virDomain, "jobInfo")
|
|
@mock.patch.object(fakelibvirt.virDomain, "jobStats")
|
|
def test_job_stats_operation_invalid(self, mock_stats, mock_info):
|
|
mock_stats.side_effect = fakelibvirt.make_libvirtError(
|
|
fakelibvirt.libvirtError,
|
|
"Domain is not running",
|
|
fakelibvirt.VIR_ERR_OPERATION_INVALID)
|
|
|
|
info = self.guest.get_job_info()
|
|
|
|
self.assertIsInstance(info, libvirt_guest.JobInfo)
|
|
self.assertEqual(fakelibvirt.VIR_DOMAIN_JOB_COMPLETED, info.type)
|
|
self.assertEqual(0, info.time_elapsed)
|
|
self.assertEqual(0, info.time_remaining)
|
|
self.assertEqual(0, info.memory_total)
|
|
self.assertEqual(0, info.memory_processed)
|
|
self.assertEqual(0, info.memory_remaining)
|
|
|
|
mock_stats.assert_called_once_with()
|
|
self.assertFalse(mock_info.called)
|
|
|
|
@mock.patch.object(fakelibvirt.virDomain, "jobInfo")
|
|
@mock.patch.object(fakelibvirt.virDomain, "jobStats")
|
|
def test_job_info_operation_invalid(self, mock_stats, mock_info):
|
|
mock_stats.side_effect = fakelibvirt.make_libvirtError(
|
|
fakelibvirt.libvirtError,
|
|
"virDomainGetJobStats not implemented",
|
|
fakelibvirt.VIR_ERR_NO_SUPPORT)
|
|
|
|
mock_info.side_effect = fakelibvirt.make_libvirtError(
|
|
fakelibvirt.libvirtError,
|
|
"Domain is not running",
|
|
fakelibvirt.VIR_ERR_OPERATION_INVALID)
|
|
|
|
info = self.guest.get_job_info()
|
|
|
|
self.assertIsInstance(info, libvirt_guest.JobInfo)
|
|
self.assertEqual(fakelibvirt.VIR_DOMAIN_JOB_COMPLETED, info.type)
|
|
self.assertEqual(0, info.time_elapsed)
|
|
self.assertEqual(0, info.time_remaining)
|
|
self.assertEqual(0, info.memory_total)
|
|
self.assertEqual(0, info.memory_processed)
|
|
self.assertEqual(0, info.memory_remaining)
|
|
|
|
mock_stats.assert_called_once_with()
|
|
mock_info.assert_called_once_with()
|