x/python-ganttclient
Code Issues Proposed changes

Added VLAN networking support for XenAPI

Browse Source
This commit is contained in:
Armando Migliaccio
2011-03-29 15:23:15 +00:00
committed by Tarmac
parent 8e59b6a532 3632bcc2ae
commit e2879b1a35
1 changed files with 54 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
nova/tests/test_xenapi.py
View File

@@ -14,9 +14,7 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
""" """Test suite for XenAPI."""
Test suite for XenAPI
"""
import functools import functools
import os import os
@@ -65,9 +63,7 @@ def stub_vm_utils_with_vdi_attached_here(function, should_return=True):
class XenAPIVolumeTestCase(test.TestCase): class XenAPIVolumeTestCase(test.TestCase):
""" """Unit tests for Volume operations."""
Unit tests for Volume operations
"""
def setUp(self): def setUp(self):
super(XenAPIVolumeTestCase, self).setUp() super(XenAPIVolumeTestCase, self).setUp()
self.stubs = stubout.StubOutForTesting() self.stubs = stubout.StubOutForTesting()
@@ -101,7 +97,7 @@ class XenAPIVolumeTestCase(test.TestCase):
return db.volume_create(self.context, vol) return db.volume_create(self.context, vol)
def test_create_iscsi_storage(self): def test_create_iscsi_storage(self):
""" This shows how to test helper classes' methods """ """This shows how to test helper classes' methods."""
stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests) stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests)
session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass') session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass')
helper = volume_utils.VolumeHelper helper = volume_utils.VolumeHelper
@@ -116,7 +112,7 @@ class XenAPIVolumeTestCase(test.TestCase):
db.volume_destroy(context.get_admin_context(), vol['id']) db.volume_destroy(context.get_admin_context(), vol['id'])
def test_parse_volume_info_raise_exception(self): def test_parse_volume_info_raise_exception(self):
""" This shows how to test helper classes' methods """ """This shows how to test helper classes' methods."""
stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests) stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests)
session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass') session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass')
helper = volume_utils.VolumeHelper helper = volume_utils.VolumeHelper
@@ -130,7 +126,7 @@ class XenAPIVolumeTestCase(test.TestCase):
db.volume_destroy(context.get_admin_context(), vol['id']) db.volume_destroy(context.get_admin_context(), vol['id'])
def test_attach_volume(self): def test_attach_volume(self):
""" This shows how to test Ops classes' methods """ """This shows how to test Ops classes' methods."""
stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests) stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests)
conn = xenapi_conn.get_connection(False) conn = xenapi_conn.get_connection(False)
volume = self._create_volume() volume = self._create_volume()
@@ -149,7 +145,7 @@ class XenAPIVolumeTestCase(test.TestCase):
check() check()
def test_attach_volume_raise_exception(self): def test_attach_volume_raise_exception(self):
""" This shows how to test when exceptions are raised """ """This shows how to test when exceptions are raised."""
stubs.stubout_session(self.stubs, stubs.stubout_session(self.stubs,
stubs.FakeSessionForVolumeFailedTests) stubs.FakeSessionForVolumeFailedTests)
conn = xenapi_conn.get_connection(False) conn = xenapi_conn.get_connection(False)
@@ -172,9 +168,7 @@ def reset_network(*args):
class XenAPIVMTestCase(test.TestCase): class XenAPIVMTestCase(test.TestCase):
""" """Unit tests for VM operations."""
Unit tests for VM operations
"""
def setUp(self): def setUp(self):
super(XenAPIVMTestCase, self).setUp() super(XenAPIVMTestCase, self).setUp()
self.manager = manager.AuthManager() self.manager = manager.AuthManager()
@@ -188,6 +182,7 @@ class XenAPIVMTestCase(test.TestCase):
instance_name_template='%d') instance_name_template='%d')
xenapi_fake.reset() xenapi_fake.reset()
xenapi_fake.create_local_srs() xenapi_fake.create_local_srs()
xenapi_fake.create_local_pifs()
db_fakes.stub_out_db_instance_api(self.stubs) db_fakes.stub_out_db_instance_api(self.stubs)
xenapi_fake.create_network('fake', FLAGS.flat_network_bridge) xenapi_fake.create_network('fake', FLAGS.flat_network_bridge)
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests) stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
@@ -247,12 +242,12 @@ class XenAPIVMTestCase(test.TestCase):
check() check()
def create_vm_record(self, conn, os_type): def create_vm_record(self, conn, os_type, instance_id=1):
instances = conn.list_instances() instances = conn.list_instances()
self.assertEquals(instances, ['1']) self.assertEquals(instances, [str(instance_id)])
# Get Nova record for VM # Get Nova record for VM
vm_info = conn.get_info(1) vm_info = conn.get_info(instance_id)
# Get XenAPI record for VM # Get XenAPI record for VM
vms = [rec for ref, rec vms = [rec for ref, rec
in xenapi_fake.get_all_records('VM').iteritems() in xenapi_fake.get_all_records('VM').iteritems()
@@ -286,19 +281,19 @@ class XenAPIVMTestCase(test.TestCase):
key = 'vm-data/networking/aabbccddeeff' key = 'vm-data/networking/aabbccddeeff'
xenstore_value = xenstore_data[key] xenstore_value = xenstore_data[key]
tcpip_data = ast.literal_eval(xenstore_value) tcpip_data = ast.literal_eval(xenstore_value)
self.assertEquals(tcpip_data, { self.assertEquals(tcpip_data,
'label': 'test_network', {'label': 'fake_flat_network',
'broadcast': '10.0.0.255', 'broadcast': '10.0.0.255',
'ips': [{'ip': '10.0.0.3', 'ips': [{'ip': '10.0.0.3',
'netmask':'255.255.255.0', 'netmask':'255.255.255.0',
'enabled':'1'}], 'enabled':'1'}],
'ip6s': [{'ip': 'fe80::a8bb:ccff:fedd:eeff', 'ip6s': [{'ip': 'fe80::a8bb:ccff:fedd:eeff',
'netmask': '120', 'netmask': '120',
'enabled': '1', 'enabled': '1',
'gateway': 'fe80::a00:1'}], 'gateway': 'fe80::a00:1'}],
'mac': 'aa:bb:cc:dd:ee:ff', 'mac': 'aa:bb:cc:dd:ee:ff',
'dns': ['10.0.0.2'], 'dns': ['10.0.0.2'],
'gateway': '10.0.0.1'}) 'gateway': '10.0.0.1'})
def check_vm_params_for_windows(self): def check_vm_params_for_windows(self):
self.assertEquals(self.vm['platform']['nx'], 'true') self.assertEquals(self.vm['platform']['nx'], 'true')
@@ -334,9 +329,9 @@ class XenAPIVMTestCase(test.TestCase):
def _test_spawn(self, image_id, kernel_id, ramdisk_id, def _test_spawn(self, image_id, kernel_id, ramdisk_id,
instance_type="m1.large", os_type="linux", instance_type="m1.large", os_type="linux",
check_injection=False): instance_id=1, check_injection=False):
stubs.stubout_loopingcall_start(self.stubs) stubs.stubout_loopingcall_start(self.stubs)
values = {'id': 1, values = {'id': instance_id,
'project_id': self.project.id, 'project_id': self.project.id,
'user_id': self.user.id, 'user_id': self.user.id,
'image_id': image_id, 'image_id': image_id,
@@ -347,7 +342,7 @@ class XenAPIVMTestCase(test.TestCase):
'os_type': os_type} 'os_type': os_type}
instance = db.instance_create(self.context, values) instance = db.instance_create(self.context, values)
self.conn.spawn(instance) self.conn.spawn(instance)
self.create_vm_record(self.conn, os_type) self.create_vm_record(self.conn, os_type, instance_id)
self.check_vm_record(self.conn, check_injection) self.check_vm_record(self.conn, check_injection)
def test_spawn_not_enough_memory(self): def test_spawn_not_enough_memory(self):
@@ -468,6 +463,28 @@ class XenAPIVMTestCase(test.TestCase):
# guest agent is detected # guest agent is detected
self.assertFalse(self._tee_executed) self.assertFalse(self._tee_executed)
def test_spawn_vlanmanager(self):
self.flags(xenapi_image_service='glance',
network_manager='nova.network.manager.VlanManager',
network_driver='nova.network.xenapi_net',
vlan_interface='fake0')
# Reset network table
xenapi_fake.reset_table('network')
# Instance id = 2 will use vlan network (see db/fakes.py)
fake_instance_id = 2
network_bk = self.network
# Ensure we use xenapi_net driver
self.network = utils.import_object(FLAGS.network_manager)
self.network.setup_compute_network(None, fake_instance_id)
self._test_spawn(glance_stubs.FakeGlance.IMAGE_MACHINE,
glance_stubs.FakeGlance.IMAGE_KERNEL,
glance_stubs.FakeGlance.IMAGE_RAMDISK,
instance_id=fake_instance_id)
# TODO(salvatore-orlando): a complete test here would require
# a check for making sure the bridge for the VM's VIF is
# consistent with bridge specified in nova db
self.network = network_bk
def test_spawn_with_network_qos(self): def test_spawn_with_network_qos(self):
self._create_instance() self._create_instance()
for vif_ref in xenapi_fake.get_all('VIF'): for vif_ref in xenapi_fake.get_all('VIF'):
@@ -497,7 +514,7 @@ class XenAPIVMTestCase(test.TestCase):
self.stubs.UnsetAll() self.stubs.UnsetAll()
def _create_instance(self): def _create_instance(self):
"""Creates and spawns a test instance""" """Creates and spawns a test instance."""
stubs.stubout_loopingcall_start(self.stubs) stubs.stubout_loopingcall_start(self.stubs)
values = { values = {
'id': 1, 'id': 1,
@@ -515,9 +532,7 @@ class XenAPIVMTestCase(test.TestCase):
class XenAPIDiffieHellmanTestCase(test.TestCase): class XenAPIDiffieHellmanTestCase(test.TestCase):
""" """Unit tests for Diffie-Hellman code."""
Unit tests for Diffie-Hellman code
"""
def setUp(self): def setUp(self):
super(XenAPIDiffieHellmanTestCase, self).setUp() super(XenAPIDiffieHellmanTestCase, self).setUp()
self.alice = SimpleDH() self.alice = SimpleDH()
@@ -541,9 +556,7 @@ class XenAPIDiffieHellmanTestCase(test.TestCase):
class XenAPIMigrateInstance(test.TestCase): class XenAPIMigrateInstance(test.TestCase):
""" """Unit test for verifying migration-related actions."""
Unit test for verifying migration-related actions
"""
def setUp(self): def setUp(self):
super(XenAPIMigrateInstance, self).setUp() super(XenAPIMigrateInstance, self).setUp()
@@ -598,9 +611,7 @@ class XenAPIMigrateInstance(test.TestCase):
class XenAPIDetermineDiskImageTestCase(test.TestCase): class XenAPIDetermineDiskImageTestCase(test.TestCase):
""" """Unit tests for code that detects the ImageType."""
Unit tests for code that detects the ImageType
"""
def setUp(self): def setUp(self):
super(XenAPIDetermineDiskImageTestCase, self).setUp() super(XenAPIDetermineDiskImageTestCase, self).setUp()
glance_stubs.stubout_glance_client(self.stubs, glance_stubs.stubout_glance_client(self.stubs,
@@ -619,9 +630,7 @@ class XenAPIDetermineDiskImageTestCase(test.TestCase):
self.assertEqual(disk_type, dt) self.assertEqual(disk_type, dt)
def test_instance_disk(self): def test_instance_disk(self):
""" """If a kernel is specified, the image type is DISK (aka machine)."""
If a kernel is specified then the image type is DISK (aka machine)
"""
FLAGS.xenapi_image_service = 'objectstore' FLAGS.xenapi_image_service = 'objectstore'
self.fake_instance.image_id = glance_stubs.FakeGlance.IMAGE_MACHINE self.fake_instance.image_id = glance_stubs.FakeGlance.IMAGE_MACHINE
self.fake_instance.kernel_id = glance_stubs.FakeGlance.IMAGE_KERNEL self.fake_instance.kernel_id = glance_stubs.FakeGlance.IMAGE_KERNEL