Merge "xenapi: Moving tests under tests/virt/xenapi/"
This commit is contained in:
Jenkins
Gerrit Code Review
commit
6dd1709bce
|
|
@ -15,16 +15,45 @@
|
|||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
|
||||
import contextlib
|
||||
|
||||
import fixtures
|
||||
import mox
|
||||
|
||||
from nova.compute import flavors
|
||||
from nova import context
|
||||
from nova import db
|
||||
from nova import exception
|
||||
from nova import test
|
||||
from nova.tests.virt.xenapi import stubs
|
||||
from nova import utils
|
||||
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'
|
||||
|
||||
|
||||
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]
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
|
|
@ -414,3 +443,347 @@ class CheckVDISizeTestCase(test.TestCase):
|
|||
|
||||
vm_utils._check_vdi_size(self.context, self.session, self.instance,
|
||||
self.vdi_uuid)
|
||||
|
||||
|
||||
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()
|
||||
|
||||
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
|
||||
|
||||
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)
|
||||
|
|
|
|||
|
|
@ -17,7 +17,7 @@
|
|||
import collections
|
||||
|
||||
from nova import test
|
||||
from nova.tests.xenapi import stubs
|
||||
from nova.tests.virt.xenapi import stubs
|
||||
from nova.virt.xenapi import volumeops
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -42,7 +42,7 @@ from nova.tests import fake_network
|
|||
from nova.tests import fake_processutils
|
||||
import nova.tests.image.fake as fake_image
|
||||
from nova.tests import matchers
|
||||
from nova.tests.xenapi import stubs
|
||||
from nova.tests.virt.xenapi import stubs
|
||||
from nova.virt import fake
|
||||
from nova.virt.xenapi import agent
|
||||
from nova.virt.xenapi import driver as xenapi_conn
|
||||
|
|
@ -396,7 +396,7 @@ class XenAPIVMTestCase(stubs.XenAPITestBase):
|
|||
def test_get_diagnostics(self):
|
||||
def fake_get_rrd(host, vm_uuid):
|
||||
path = os.path.dirname(os.path.realpath(__file__))
|
||||
with open(os.path.join(path, 'xenapi/vm_rrd.xml')) as f:
|
||||
with open(os.path.join(path, 'vm_rrd.xml')) as f:
|
||||
return re.sub(r'\s', '', f.read())
|
||||
self.stubs.Set(vm_utils, '_get_rrd', fake_get_rrd)
|
||||
|
||||
|
|
@ -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,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)
|
||||
Loading…
Reference in New Issue