Merge "xenapi: Moving tests under tests/virt/xenapi/"
This commit is contained in:
Jenkins
File diff suppressed because it is too large
Load Diff
@@ -1,20 +0,0 @@
|
||||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright (c) 2010 Citrix Systems, 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.
|
||||
|
||||
"""
|
||||
:mod:`xenapi` -- Stubs for XenAPI
|
||||
=================================
|
||||
"""
|
||||
@@ -1,353 +0,0 @@
|
||||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright (c) 2010 Citrix Systems, 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.
|
||||
|
||||
"""Stubouts, mocks and fixtures for the test suite."""
|
||||
|
||||
import pickle
|
||||
import random
|
||||
|
||||
from nova.openstack.common import jsonutils
|
||||
from nova import test
|
||||
import nova.tests.image.fake
|
||||
from nova.virt.xenapi import driver as xenapi_conn
|
||||
from nova.virt.xenapi import fake
|
||||
from nova.virt.xenapi import vm_utils
|
||||
from nova.virt.xenapi import vmops
|
||||
|
||||
|
||||
def stubout_firewall_driver(stubs, conn):
|
||||
|
||||
def fake_none(self, *args):
|
||||
return
|
||||
|
||||
_vmops = conn._vmops
|
||||
stubs.Set(_vmops.firewall_driver, 'prepare_instance_filter', fake_none)
|
||||
stubs.Set(_vmops.firewall_driver, 'instance_filter_exists', fake_none)
|
||||
|
||||
|
||||
def stubout_instance_snapshot(stubs):
|
||||
def fake_fetch_image(context, session, instance, name_label, image, type):
|
||||
return {'root': dict(uuid=_make_fake_vdi(), file=None),
|
||||
'kernel': dict(uuid=_make_fake_vdi(), file=None),
|
||||
'ramdisk': dict(uuid=_make_fake_vdi(), file=None)}
|
||||
|
||||
stubs.Set(vm_utils, '_fetch_image', fake_fetch_image)
|
||||
|
||||
def fake_wait_for_vhd_coalesce(*args):
|
||||
#TODO(sirp): Should we actually fake out the data here
|
||||
return "fakeparent", "fakebase"
|
||||
|
||||
stubs.Set(vm_utils, '_wait_for_vhd_coalesce', fake_wait_for_vhd_coalesce)
|
||||
|
||||
|
||||
def stubout_session(stubs, cls, product_version=(5, 6, 2),
|
||||
product_brand='XenServer', **opt_args):
|
||||
"""Stubs out methods from XenAPISession."""
|
||||
stubs.Set(xenapi_conn.XenAPISession, '_create_session',
|
||||
lambda s, url: cls(url, **opt_args))
|
||||
stubs.Set(xenapi_conn.XenAPISession, '_get_product_version_and_brand',
|
||||
lambda s: (product_version, product_brand))
|
||||
|
||||
|
||||
def stubout_get_this_vm_uuid(stubs):
|
||||
def f():
|
||||
vms = [rec['uuid'] for ref, rec
|
||||
in fake.get_all_records('VM').iteritems()
|
||||
if rec['is_control_domain']]
|
||||
return vms[0]
|
||||
stubs.Set(vm_utils, 'get_this_vm_uuid', f)
|
||||
|
||||
|
||||
def stubout_image_service_download(stubs):
|
||||
def fake_download(*args, **kwargs):
|
||||
pass
|
||||
stubs.Set(nova.tests.image.fake._FakeImageService,
|
||||
'download', fake_download)
|
||||
|
||||
|
||||
def stubout_stream_disk(stubs):
|
||||
def fake_stream_disk(*args, **kwargs):
|
||||
pass
|
||||
stubs.Set(vm_utils, '_stream_disk', fake_stream_disk)
|
||||
|
||||
|
||||
def stubout_is_vdi_pv(stubs):
|
||||
def f(_1):
|
||||
return False
|
||||
stubs.Set(vm_utils, '_is_vdi_pv', f)
|
||||
|
||||
|
||||
def stubout_determine_is_pv_objectstore(stubs):
|
||||
"""Assumes VMs stu have PV kernels."""
|
||||
|
||||
def f(*args):
|
||||
return False
|
||||
stubs.Set(vm_utils, '_determine_is_pv_objectstore', f)
|
||||
|
||||
|
||||
def stubout_is_snapshot(stubs):
|
||||
"""Always returns true
|
||||
xenapi fake driver does not create vmrefs for snapshots """
|
||||
|
||||
def f(*args):
|
||||
return True
|
||||
stubs.Set(vm_utils, 'is_snapshot', f)
|
||||
|
||||
|
||||
def stubout_lookup_image(stubs):
|
||||
"""Simulates a failure in lookup image."""
|
||||
def f(_1, _2, _3, _4):
|
||||
raise Exception("Test Exception raised by fake lookup_image")
|
||||
stubs.Set(vm_utils, 'lookup_image', f)
|
||||
|
||||
|
||||
def stubout_fetch_disk_image(stubs, raise_failure=False):
|
||||
"""Simulates a failure in fetch image_glance_disk."""
|
||||
|
||||
def _fake_fetch_disk_image(context, session, instance, name_label, image,
|
||||
image_type):
|
||||
if raise_failure:
|
||||
raise fake.Failure("Test Exception raised by "
|
||||
"fake fetch_image_glance_disk")
|
||||
elif image_type == vm_utils.ImageType.KERNEL:
|
||||
filename = "kernel"
|
||||
elif image_type == vm_utils.ImageType.RAMDISK:
|
||||
filename = "ramdisk"
|
||||
else:
|
||||
filename = "unknown"
|
||||
|
||||
vdi_type = vm_utils.ImageType.to_string(image_type)
|
||||
return {vdi_type: dict(uuid=None, file=filename)}
|
||||
|
||||
stubs.Set(vm_utils, '_fetch_disk_image', _fake_fetch_disk_image)
|
||||
|
||||
|
||||
def stubout_create_vm(stubs):
|
||||
"""Simulates a failure in create_vm."""
|
||||
|
||||
def f(*args):
|
||||
raise fake.Failure("Test Exception raised by fake create_vm")
|
||||
stubs.Set(vm_utils, 'create_vm', f)
|
||||
|
||||
|
||||
def stubout_attach_disks(stubs):
|
||||
"""Simulates a failure in _attach_disks."""
|
||||
|
||||
def f(*args):
|
||||
raise fake.Failure("Test Exception raised by fake _attach_disks")
|
||||
stubs.Set(vmops.VMOps, '_attach_disks', f)
|
||||
|
||||
|
||||
def _make_fake_vdi():
|
||||
sr_ref = fake.get_all('SR')[0]
|
||||
vdi_ref = fake.create_vdi('', sr_ref)
|
||||
vdi_rec = fake.get_record('VDI', vdi_ref)
|
||||
return vdi_rec['uuid']
|
||||
|
||||
|
||||
class FakeSessionForVMTests(fake.SessionBase):
|
||||
"""Stubs out a XenAPISession for VM tests."""
|
||||
|
||||
_fake_iptables_save_output = ("# Generated by iptables-save v1.4.10 on "
|
||||
"Sun Nov 6 22:49:02 2011\n"
|
||||
"*filter\n"
|
||||
":INPUT ACCEPT [0:0]\n"
|
||||
":FORWARD ACCEPT [0:0]\n"
|
||||
":OUTPUT ACCEPT [0:0]\n"
|
||||
"COMMIT\n"
|
||||
"# Completed on Sun Nov 6 22:49:02 2011\n")
|
||||
|
||||
def host_call_plugin(self, _1, _2, plugin, method, _5):
|
||||
if (plugin, method) == ('glance', 'download_vhd'):
|
||||
root_uuid = _make_fake_vdi()
|
||||
return pickle.dumps(dict(root=dict(uuid=root_uuid)))
|
||||
elif (plugin, method) == ("xenhost", "iptables_config"):
|
||||
return fake.as_json(out=self._fake_iptables_save_output,
|
||||
err='')
|
||||
else:
|
||||
return (super(FakeSessionForVMTests, self).
|
||||
host_call_plugin(_1, _2, plugin, method, _5))
|
||||
|
||||
def VM_start(self, _1, ref, _2, _3):
|
||||
vm = fake.get_record('VM', ref)
|
||||
if vm['power_state'] != 'Halted':
|
||||
raise fake.Failure(['VM_BAD_POWER_STATE', ref, 'Halted',
|
||||
vm['power_state']])
|
||||
vm['power_state'] = 'Running'
|
||||
vm['is_a_template'] = False
|
||||
vm['is_control_domain'] = False
|
||||
vm['domid'] = random.randrange(1, 1 << 16)
|
||||
return vm
|
||||
|
||||
def VM_start_on(self, _1, vm_ref, host_ref, _2, _3):
|
||||
vm_rec = self.VM_start(_1, vm_ref, _2, _3)
|
||||
vm_rec['resident_on'] = host_ref
|
||||
|
||||
def VDI_snapshot(self, session_ref, vm_ref, _1):
|
||||
sr_ref = "fakesr"
|
||||
return fake.create_vdi('fakelabel', sr_ref, read_only=True)
|
||||
|
||||
def SR_scan(self, session_ref, sr_ref):
|
||||
pass
|
||||
|
||||
|
||||
class FakeSessionForFirewallTests(FakeSessionForVMTests):
|
||||
"""Stubs out a XenApi Session for doing IPTable Firewall tests."""
|
||||
|
||||
def __init__(self, uri, test_case=None):
|
||||
super(FakeSessionForFirewallTests, self).__init__(uri)
|
||||
if hasattr(test_case, '_in_rules'):
|
||||
self._in_rules = test_case._in_rules
|
||||
if hasattr(test_case, '_in6_filter_rules'):
|
||||
self._in6_filter_rules = test_case._in6_filter_rules
|
||||
self._test_case = test_case
|
||||
|
||||
def host_call_plugin(self, _1, _2, plugin, method, args):
|
||||
"""Mock method four host_call_plugin to be used in unit tests
|
||||
for the dom0 iptables Firewall drivers for XenAPI
|
||||
|
||||
"""
|
||||
if plugin == "xenhost" and method == "iptables_config":
|
||||
# The command to execute is a json-encoded list
|
||||
cmd_args = args.get('cmd_args', None)
|
||||
cmd = jsonutils.loads(cmd_args)
|
||||
if not cmd:
|
||||
ret_str = ''
|
||||
else:
|
||||
output = ''
|
||||
process_input = args.get('process_input', None)
|
||||
if cmd == ['ip6tables-save', '-c']:
|
||||
output = '\n'.join(self._in6_filter_rules)
|
||||
if cmd == ['iptables-save', '-c']:
|
||||
output = '\n'.join(self._in_rules)
|
||||
if cmd == ['iptables-restore', '-c', ]:
|
||||
lines = process_input.split('\n')
|
||||
if '*filter' in lines:
|
||||
if self._test_case is not None:
|
||||
self._test_case._out_rules = lines
|
||||
output = '\n'.join(lines)
|
||||
if cmd == ['ip6tables-restore', '-c', ]:
|
||||
lines = process_input.split('\n')
|
||||
if '*filter' in lines:
|
||||
output = '\n'.join(lines)
|
||||
ret_str = fake.as_json(out=output, err='')
|
||||
return ret_str
|
||||
|
||||
|
||||
def stub_out_vm_methods(stubs):
|
||||
def fake_acquire_bootlock(self, vm):
|
||||
pass
|
||||
|
||||
def fake_release_bootlock(self, vm):
|
||||
pass
|
||||
|
||||
def fake_generate_ephemeral(*args):
|
||||
pass
|
||||
|
||||
def fake_wait_for_device(dev):
|
||||
pass
|
||||
|
||||
stubs.Set(vmops.VMOps, "_acquire_bootlock", fake_acquire_bootlock)
|
||||
stubs.Set(vmops.VMOps, "_release_bootlock", fake_release_bootlock)
|
||||
stubs.Set(vm_utils, 'generate_ephemeral', fake_generate_ephemeral)
|
||||
stubs.Set(vm_utils, '_wait_for_device', fake_wait_for_device)
|
||||
|
||||
|
||||
class FakeSessionForVolumeTests(fake.SessionBase):
|
||||
"""Stubs out a XenAPISession for Volume tests."""
|
||||
def VDI_introduce(self, _1, uuid, _2, _3, _4, _5,
|
||||
_6, _7, _8, _9, _10, _11):
|
||||
valid_vdi = False
|
||||
refs = fake.get_all('VDI')
|
||||
for ref in refs:
|
||||
rec = fake.get_record('VDI', ref)
|
||||
if rec['uuid'] == uuid:
|
||||
valid_vdi = True
|
||||
if not valid_vdi:
|
||||
raise fake.Failure([['INVALID_VDI', 'session', self._session]])
|
||||
|
||||
|
||||
class FakeSessionForVolumeFailedTests(FakeSessionForVolumeTests):
|
||||
"""Stubs out a XenAPISession for Volume tests: it injects failures."""
|
||||
def VDI_introduce(self, _1, uuid, _2, _3, _4, _5,
|
||||
_6, _7, _8, _9, _10, _11):
|
||||
# This is for testing failure
|
||||
raise fake.Failure([['INVALID_VDI', 'session', self._session]])
|
||||
|
||||
def PBD_unplug(self, _1, ref):
|
||||
rec = fake.get_record('PBD', ref)
|
||||
rec['currently-attached'] = False
|
||||
|
||||
def SR_forget(self, _1, ref):
|
||||
pass
|
||||
|
||||
|
||||
def stub_out_migration_methods(stubs):
|
||||
fakesr = fake.create_sr()
|
||||
|
||||
def fake_move_disks(self, instance, disk_info):
|
||||
vdi_ref = fake.create_vdi(instance['name'], fakesr)
|
||||
vdi_rec = fake.get_record('VDI', vdi_ref)
|
||||
vdi_rec['other_config']['nova_disk_type'] = 'root'
|
||||
return {'uuid': vdi_rec['uuid'], 'ref': vdi_ref}
|
||||
|
||||
def fake_get_vdi(session, vm_ref):
|
||||
vdi_ref_parent = fake.create_vdi('derp-parent', fakesr)
|
||||
vdi_rec_parent = fake.get_record('VDI', vdi_ref_parent)
|
||||
vdi_ref = fake.create_vdi('derp', fakesr,
|
||||
sm_config={'vhd-parent': vdi_rec_parent['uuid']})
|
||||
vdi_rec = session.call_xenapi("VDI.get_record", vdi_ref)
|
||||
return vdi_ref, vdi_rec
|
||||
|
||||
def fake_sr(session, *args):
|
||||
return fakesr
|
||||
|
||||
def fake_get_sr_path(*args):
|
||||
return "fake"
|
||||
|
||||
def fake_destroy(*args, **kwargs):
|
||||
pass
|
||||
|
||||
def fake_generate_ephemeral(*args):
|
||||
pass
|
||||
|
||||
stubs.Set(vmops.VMOps, '_destroy', fake_destroy)
|
||||
stubs.Set(vm_utils, 'move_disks', fake_move_disks)
|
||||
stubs.Set(vm_utils, 'scan_default_sr', fake_sr)
|
||||
stubs.Set(vm_utils, 'get_vdi_for_vm_safely', fake_get_vdi)
|
||||
stubs.Set(vm_utils, 'get_sr_path', fake_get_sr_path)
|
||||
stubs.Set(vm_utils, 'generate_ephemeral', fake_generate_ephemeral)
|
||||
|
||||
|
||||
class FakeSessionForFailedMigrateTests(FakeSessionForVMTests):
|
||||
def VM_assert_can_migrate(self, session, vmref, migrate_data,
|
||||
live, vdi_map, vif_map, options):
|
||||
raise fake.Failure("XenAPI VM.assert_can_migrate failed")
|
||||
|
||||
def host_migrate_receive(self, session, hostref, networkref, options):
|
||||
raise fake.Failure("XenAPI host.migrate_receive failed")
|
||||
|
||||
def VM_migrate_send(self, session, vmref, migrate_data, islive, vdi_map,
|
||||
vif_map, options):
|
||||
raise fake.Failure("XenAPI VM.migrate_send failed")
|
||||
|
||||
|
||||
class XenAPITestBase(test.TestCase):
|
||||
def setUp(self):
|
||||
super(XenAPITestBase, self).setUp()
|
||||
|
||||
self.useFixture(test.ReplaceModule('XenAPI', fake))
|
||||
|
||||
fake.reset()
|
||||
@@ -1,377 +0,0 @@
|
||||
import mox
|
||||
from nova import context
|
||||
from nova import db
|
||||
from nova import exception
|
||||
from nova import test
|
||||
from nova.tests.xenapi import stubs
|
||||
from nova.virt.xenapi import driver as xenapi_conn
|
||||
from nova.virt.xenapi import fake
|
||||
from nova.virt.xenapi import vm_utils
|
||||
from nova.virt.xenapi import volume_utils
|
||||
|
||||
|
||||
XENSM_TYPE = 'xensm'
|
||||
ISCSI_TYPE = 'iscsi'
|
||||
|
||||
|
||||
class FakeSession():
|
||||
def call_xenapi(self, operation, *args, **kwargs):
|
||||
# VDI.add_to_other_config -> VDI_add_to_other_config
|
||||
method = getattr(self, operation.replace('.', '_'), None)
|
||||
if method:
|
||||
return method(*args, **kwargs)
|
||||
|
||||
self.operation = operation
|
||||
self.args = args
|
||||
self.kwargs = kwargs
|
||||
|
||||
|
||||
def get_fake_connection_data(sr_type):
|
||||
fakes = {XENSM_TYPE: {'sr_uuid': 'falseSR',
|
||||
'name_label': 'fake_storage',
|
||||
'name_description': 'test purposes',
|
||||
'server': 'myserver',
|
||||
'serverpath': '/local/scratch/myname',
|
||||
'sr_type': 'nfs',
|
||||
'introduce_sr_keys': ['server',
|
||||
'serverpath',
|
||||
'sr_type'],
|
||||
'vdi_uuid': 'falseVDI'},
|
||||
ISCSI_TYPE: {'volume_id': 'fake_volume_id',
|
||||
'target_lun': 1,
|
||||
'target_iqn': 'fake_iqn:volume-fake_volume_id',
|
||||
'target_portal': u'localhost:3260',
|
||||
'target_discovered': False}, }
|
||||
return fakes[sr_type]
|
||||
|
||||
|
||||
class GetInstanceForVdisForSrTestCase(stubs.XenAPITestBase):
|
||||
def setUp(self):
|
||||
super(GetInstanceForVdisForSrTestCase, self).setUp()
|
||||
self.flags(disable_process_locking=True,
|
||||
instance_name_template='%d',
|
||||
firewall_driver='nova.virt.xenapi.firewall.'
|
||||
'Dom0IptablesFirewallDriver',
|
||||
xenapi_connection_url='test_url',
|
||||
xenapi_connection_password='test_pass',)
|
||||
|
||||
def test_get_instance_vdis_for_sr(self):
|
||||
vm_ref = fake.create_vm("foo", "Running")
|
||||
sr_ref = fake.create_sr()
|
||||
|
||||
vdi_1 = fake.create_vdi('vdiname1', sr_ref)
|
||||
vdi_2 = fake.create_vdi('vdiname2', sr_ref)
|
||||
|
||||
for vdi_ref in [vdi_1, vdi_2]:
|
||||
fake.create_vbd(vm_ref, vdi_ref)
|
||||
|
||||
stubs.stubout_session(self.stubs, fake.SessionBase)
|
||||
driver = xenapi_conn.XenAPIDriver(False)
|
||||
|
||||
result = list(vm_utils.get_instance_vdis_for_sr(
|
||||
driver._session, vm_ref, sr_ref))
|
||||
|
||||
self.assertEquals([vdi_1, vdi_2], result)
|
||||
|
||||
def test_get_instance_vdis_for_sr_no_vbd(self):
|
||||
vm_ref = fake.create_vm("foo", "Running")
|
||||
sr_ref = fake.create_sr()
|
||||
|
||||
stubs.stubout_session(self.stubs, fake.SessionBase)
|
||||
driver = xenapi_conn.XenAPIDriver(False)
|
||||
|
||||
result = list(vm_utils.get_instance_vdis_for_sr(
|
||||
driver._session, vm_ref, sr_ref))
|
||||
|
||||
self.assertEquals([], result)
|
||||
|
||||
def test_get_vdi_uuid_for_volume_with_sr_uuid(self):
|
||||
connection_data = get_fake_connection_data(XENSM_TYPE)
|
||||
stubs.stubout_session(self.stubs, fake.SessionBase)
|
||||
driver = xenapi_conn.XenAPIDriver(False)
|
||||
|
||||
vdi_uuid = vm_utils.get_vdi_uuid_for_volume(
|
||||
driver._session, connection_data)
|
||||
self.assertEquals(vdi_uuid, 'falseVDI')
|
||||
|
||||
def test_get_vdi_uuid_for_volume_failure(self):
|
||||
stubs.stubout_session(self.stubs, fake.SessionBase)
|
||||
driver = xenapi_conn.XenAPIDriver(False)
|
||||
|
||||
def bad_introduce_sr(session, sr_uuid, label, sr_params):
|
||||
return None
|
||||
|
||||
self.stubs.Set(volume_utils, 'introduce_sr', bad_introduce_sr)
|
||||
connection_data = get_fake_connection_data(XENSM_TYPE)
|
||||
self.assertRaises(exception.NovaException,
|
||||
vm_utils.get_vdi_uuid_for_volume,
|
||||
driver._session, connection_data)
|
||||
|
||||
def test_get_vdi_uuid_for_volume_from_iscsi_vol_missing_sr_uuid(self):
|
||||
connection_data = get_fake_connection_data(ISCSI_TYPE)
|
||||
stubs.stubout_session(self.stubs, fake.SessionBase)
|
||||
driver = xenapi_conn.XenAPIDriver(False)
|
||||
|
||||
vdi_uuid = vm_utils.get_vdi_uuid_for_volume(
|
||||
driver._session, connection_data)
|
||||
self.assertNotEquals(vdi_uuid, None)
|
||||
|
||||
|
||||
class VMRefOrRaiseVMFoundTestCase(test.TestCase):
|
||||
|
||||
def test_lookup_call(self):
|
||||
mock = mox.Mox()
|
||||
mock.StubOutWithMock(vm_utils, 'lookup')
|
||||
|
||||
vm_utils.lookup('session', 'somename').AndReturn('ignored')
|
||||
|
||||
mock.ReplayAll()
|
||||
vm_utils.vm_ref_or_raise('session', 'somename')
|
||||
mock.VerifyAll()
|
||||
|
||||
def test_return_value(self):
|
||||
mock = mox.Mox()
|
||||
mock.StubOutWithMock(vm_utils, 'lookup')
|
||||
|
||||
vm_utils.lookup(mox.IgnoreArg(), mox.IgnoreArg()).AndReturn('vmref')
|
||||
|
||||
mock.ReplayAll()
|
||||
self.assertEquals(
|
||||
'vmref', vm_utils.vm_ref_or_raise('session', 'somename'))
|
||||
mock.VerifyAll()
|
||||
|
||||
|
||||
class VMRefOrRaiseVMNotFoundTestCase(test.TestCase):
|
||||
|
||||
def test_exception_raised(self):
|
||||
mock = mox.Mox()
|
||||
mock.StubOutWithMock(vm_utils, 'lookup')
|
||||
|
||||
vm_utils.lookup('session', 'somename').AndReturn(None)
|
||||
|
||||
mock.ReplayAll()
|
||||
self.assertRaises(
|
||||
exception.InstanceNotFound,
|
||||
lambda: vm_utils.vm_ref_or_raise('session', 'somename')
|
||||
)
|
||||
mock.VerifyAll()
|
||||
|
||||
def test_exception_msg_contains_vm_name(self):
|
||||
mock = mox.Mox()
|
||||
mock.StubOutWithMock(vm_utils, 'lookup')
|
||||
|
||||
vm_utils.lookup('session', 'somename').AndReturn(None)
|
||||
|
||||
mock.ReplayAll()
|
||||
try:
|
||||
vm_utils.vm_ref_or_raise('session', 'somename')
|
||||
except exception.InstanceNotFound as e:
|
||||
self.assertTrue(
|
||||
'somename' in str(e))
|
||||
mock.VerifyAll()
|
||||
|
||||
|
||||
class BittorrentTestCase(stubs.XenAPITestBase):
|
||||
def setUp(self):
|
||||
super(BittorrentTestCase, self).setUp()
|
||||
self.context = context.get_admin_context()
|
||||
|
||||
def test_image_uses_bittorrent(self):
|
||||
sys_meta = {'image_bittorrent': True}
|
||||
instance = db.instance_create(self.context,
|
||||
{'system_metadata': sys_meta})
|
||||
instance = db.instance_get_by_uuid(self.context, instance['uuid'])
|
||||
self.flags(xenapi_torrent_images='some')
|
||||
self.assertTrue(vm_utils._image_uses_bittorrent(self.context,
|
||||
instance))
|
||||
|
||||
def _test_create_image(self, cache_type):
|
||||
sys_meta = {'image_cache_in_nova': True}
|
||||
instance = db.instance_create(self.context,
|
||||
{'system_metadata': sys_meta})
|
||||
instance = db.instance_get_by_uuid(self.context, instance['uuid'])
|
||||
self.flags(cache_images=cache_type)
|
||||
|
||||
was = {'called': None}
|
||||
|
||||
def fake_create_cached_image(*args):
|
||||
was['called'] = 'some'
|
||||
return {}
|
||||
self.stubs.Set(vm_utils, '_create_cached_image',
|
||||
fake_create_cached_image)
|
||||
|
||||
def fake_fetch_image(*args):
|
||||
was['called'] = 'none'
|
||||
return {}
|
||||
self.stubs.Set(vm_utils, '_fetch_image',
|
||||
fake_fetch_image)
|
||||
|
||||
vm_utils._create_image(self.context, None, instance,
|
||||
'foo', 'bar', 'baz')
|
||||
|
||||
self.assertEqual(was['called'], cache_type)
|
||||
|
||||
def test_create_image_cached(self):
|
||||
self._test_create_image('some')
|
||||
|
||||
def test_create_image_uncached(self):
|
||||
self._test_create_image('none')
|
||||
|
||||
|
||||
class CreateVBDTestCase(test.TestCase):
|
||||
def setUp(self):
|
||||
super(CreateVBDTestCase, self).setUp()
|
||||
self.session = FakeSession()
|
||||
self.mock = mox.Mox()
|
||||
self.mock.StubOutWithMock(self.session, 'call_xenapi')
|
||||
self.vbd_rec = self._generate_vbd_rec()
|
||||
|
||||
def _generate_vbd_rec(self):
|
||||
vbd_rec = {}
|
||||
vbd_rec['VM'] = 'vm_ref'
|
||||
vbd_rec['VDI'] = 'vdi_ref'
|
||||
vbd_rec['userdevice'] = '0'
|
||||
vbd_rec['bootable'] = False
|
||||
vbd_rec['mode'] = 'RW'
|
||||
vbd_rec['type'] = 'disk'
|
||||
vbd_rec['unpluggable'] = True
|
||||
vbd_rec['empty'] = False
|
||||
vbd_rec['other_config'] = {}
|
||||
vbd_rec['qos_algorithm_type'] = ''
|
||||
vbd_rec['qos_algorithm_params'] = {}
|
||||
vbd_rec['qos_supported_algorithms'] = []
|
||||
return vbd_rec
|
||||
|
||||
def test_create_vbd_default_args(self):
|
||||
self.session.call_xenapi('VBD.create',
|
||||
self.vbd_rec).AndReturn("vbd_ref")
|
||||
self.mock.ReplayAll()
|
||||
|
||||
result = vm_utils.create_vbd(self.session, "vm_ref", "vdi_ref", 0)
|
||||
self.assertEquals(result, "vbd_ref")
|
||||
self.mock.VerifyAll()
|
||||
|
||||
def test_create_vbd_osvol(self):
|
||||
self.session.call_xenapi('VBD.create',
|
||||
self.vbd_rec).AndReturn("vbd_ref")
|
||||
self.session.call_xenapi('VBD.add_to_other_config', "vbd_ref",
|
||||
"osvol", "True")
|
||||
self.mock.ReplayAll()
|
||||
result = vm_utils.create_vbd(self.session, "vm_ref", "vdi_ref", 0,
|
||||
osvol=True)
|
||||
self.assertEquals(result, "vbd_ref")
|
||||
self.mock.VerifyAll()
|
||||
|
||||
def test_create_vbd_extra_args(self):
|
||||
self.vbd_rec['VDI'] = 'OpaqueRef:NULL'
|
||||
self.vbd_rec['type'] = 'a'
|
||||
self.vbd_rec['mode'] = 'RO'
|
||||
self.vbd_rec['bootable'] = True
|
||||
self.vbd_rec['empty'] = True
|
||||
self.vbd_rec['unpluggable'] = False
|
||||
self.session.call_xenapi('VBD.create',
|
||||
self.vbd_rec).AndReturn("vbd_ref")
|
||||
self.mock.ReplayAll()
|
||||
|
||||
result = vm_utils.create_vbd(self.session, "vm_ref", None, 0,
|
||||
vbd_type="a", read_only=True, bootable=True,
|
||||
empty=True, unpluggable=False)
|
||||
self.assertEquals(result, "vbd_ref")
|
||||
self.mock.VerifyAll()
|
||||
|
||||
def test_attach_cd(self):
|
||||
self.mock.StubOutWithMock(vm_utils, 'create_vbd')
|
||||
|
||||
vm_utils.create_vbd(self.session, "vm_ref", None, 1,
|
||||
vbd_type='cd', read_only=True, bootable=True,
|
||||
empty=True, unpluggable=False).AndReturn("vbd_ref")
|
||||
self.session.call_xenapi('VBD.insert', "vbd_ref", "vdi_ref")
|
||||
self.mock.ReplayAll()
|
||||
|
||||
result = vm_utils.attach_cd(self.session, "vm_ref", "vdi_ref", 1)
|
||||
self.assertEquals(result, "vbd_ref")
|
||||
self.mock.VerifyAll()
|
||||
|
||||
|
||||
class VDIOtherConfigTestCase(stubs.XenAPITestBase):
|
||||
"""Tests to ensure that the code is populating VDI's `other_config`
|
||||
attribute with the correct metadta.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
super(VDIOtherConfigTestCase, self).setUp()
|
||||
self.session = FakeSession()
|
||||
self.context = context.get_admin_context()
|
||||
self.fake_instance = {'uuid': 'aaaa-bbbb-cccc-dddd',
|
||||
'name': 'myinstance'}
|
||||
|
||||
def test_create_vdi(self):
|
||||
# Some images are registered with XenServer explicitly by calling
|
||||
# `create_vdi`
|
||||
vm_utils.create_vdi(self.session, 'sr_ref', self.fake_instance,
|
||||
'myvdi', 'root', 1024, read_only=True)
|
||||
|
||||
expected = {'nova_disk_type': 'root',
|
||||
'nova_instance_uuid': 'aaaa-bbbb-cccc-dddd'}
|
||||
|
||||
self.assertEqual(expected, self.session.args[0]['other_config'])
|
||||
|
||||
def test_create_image(self):
|
||||
# Other images are registered implicitly when they are dropped into
|
||||
# the SR by a dom0 plugin or some other process
|
||||
self.flags(cache_images='none')
|
||||
|
||||
def fake_fetch_image(*args):
|
||||
return {'root': {'uuid': 'fake-uuid'}}
|
||||
|
||||
self.stubs.Set(vm_utils, '_fetch_image', fake_fetch_image)
|
||||
|
||||
other_config = {}
|
||||
|
||||
def VDI_add_to_other_config(ref, key, value):
|
||||
other_config[key] = value
|
||||
|
||||
def VDI_get_record(ref):
|
||||
return {'other_config': {}}
|
||||
|
||||
# Stubbing on the session object and not class so we don't pollute
|
||||
# other tests
|
||||
self.session.VDI_add_to_other_config = VDI_add_to_other_config
|
||||
self.session.VDI_get_record = VDI_get_record
|
||||
|
||||
vm_utils._create_image(self.context, self.session, self.fake_instance,
|
||||
'myvdi', 'image1', vm_utils.ImageType.DISK_VHD)
|
||||
|
||||
expected = {'nova_disk_type': 'root',
|
||||
'nova_instance_uuid': 'aaaa-bbbb-cccc-dddd'}
|
||||
|
||||
self.assertEqual(expected, other_config)
|
||||
|
||||
def test_move_disks(self):
|
||||
# Migrated images should preserve the `other_config`
|
||||
other_config = {}
|
||||
|
||||
def VDI_add_to_other_config(ref, key, value):
|
||||
other_config[key] = value
|
||||
|
||||
def VDI_get_record(ref):
|
||||
return {'other_config': {}}
|
||||
|
||||
def call_plugin_serialized(*args, **kwargs):
|
||||
return {'root': {'uuid': 'aaaa-bbbb-cccc-dddd'}}
|
||||
|
||||
# Stubbing on the session object and not class so we don't pollute
|
||||
# other tests
|
||||
self.session.VDI_add_to_other_config = VDI_add_to_other_config
|
||||
self.session.VDI_get_record = VDI_get_record
|
||||
self.session.call_plugin_serialized = call_plugin_serialized
|
||||
|
||||
self.stubs.Set(vm_utils, 'get_sr_path', lambda *a, **k: None)
|
||||
self.stubs.Set(vm_utils, 'scan_default_sr', lambda *a, **k: None)
|
||||
|
||||
vm_utils.move_disks(self.session, self.fake_instance, {})
|
||||
|
||||
expected = {'nova_disk_type': 'root',
|
||||
'nova_instance_uuid': 'aaaa-bbbb-cccc-dddd'}
|
||||
|
||||
self.assertEqual(expected, other_config)
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user