x/python-ganttclient
Code Issues Proposed changes

first cut of unittest framework for xenapi

Browse Source
This commit is contained in:
Armando Migliaccio
2010-12-14 14:27:56 +00:00
parent 478fb2844e 9ec2c03b22
commit 8f2a292319
3 changed files with 189 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
nova/tests/virt_unittest.py
View File

@@ -30,9 +30,6 @@ from nova.virt.xenapi import volume_utils
FLAGS = flags.FLAGS
flags.DECLARE('instances_path', 'nova.compute.manager')
# Those are XenAPI related
flags.DECLARE('target_host', 'nova.virt.xenapi_conn')
FLAGS.target_host = '127.0.0.1'
class LibvirtConnTestCase(test.TrialTestCase):
@@ -262,74 +259,3 @@ class NWFilterTestCase(test.TrialTestCase):
d.addCallback(lambda _: self.teardown_security_group())
return d
class XenAPIVolumeTestCase(test.TrialTestCase):
def setUp(self):
super(XenAPIVolumeTestCase, self).setUp()
self.flags(xenapi_use_fake_session=True)
self.session = fake.FakeXenAPISession()
self.helper = volume_utils.VolumeHelper
self.helper.late_import()
def _create_volume(self, size='0'):
"""Create a volume object."""
vol = {}
vol['size'] = size
vol['user_id'] = 'fake'
vol['project_id'] = 'fake'
vol['host'] = 'localhost'
vol['availability_zone'] = FLAGS.storage_availability_zone
vol['status'] = "creating"
vol['attach_status'] = "detached"
return db.volume_create(context.get_admin_context(), vol)
def test_create_iscsi_storage_raise_no_exception(self):
vol = self._create_volume()
info = yield self.helper.parse_volume_info(vol['ec2_id'], '/dev/sdc')
label = None # For testing new SRs
description = 'Test-SR'
self.session.fail_next_call = False
sr_ref = self.helper.create_iscsi_storage_blocking(self.session,
info,
label,
description)
self.assertEqual(sr_ref, self.session.SR.FAKE_REF)
db.volume_destroy(context.get_admin_context(), vol['id'])
def test_create_iscsi_storage_raise_unable_to_create_sr_exception(self):
vol = self._create_volume()
info = yield self.helper.parse_volume_info(vol['ec2_id'], '/dev/sdc')
label = None # For testing new SRs
description = None
self.session.fail_next_call = True
self.assertRaises(volume_utils.StorageError,
self.helper.create_iscsi_storage_blocking,
self.session,
info,
label,
description)
def test_find_sr_from_vbd_raise_no_exception(self):
sr_ref = yield self.helper.find_sr_from_vbd(self.session,
self.session.VBD.FAKE_REF)
self.assertEqual(sr_ref, self.session.SR.FAKE_REF)
def test_destroy_iscsi_storage(self):
sr_ref = self.session.SR.FAKE_REF
self.helper.destroy_iscsi_storage_blocking(self.session, sr_ref)
def test_introduce_vdi_raise_no_exception(self):
sr_ref = self.session.SR.FAKE_REF
self.helper.introduce_vdi_blocking(self.session, sr_ref)
def test_introduce_vdi_raise_unable_get_vdi_record_exception(self):
sr_ref = self.session.SR.FAKE_REF
self.session.fail_next_call = True
self.assertRaises(volume_utils.StorageError,
self.helper.introduce_vdi_blocking,
self.session, sr_ref)
def tearDown(self):
super(XenAPIVolumeTestCase, self).tearDown()

188
nova/tests/xenapi_unittest.py Normal file
View File

@@ -0,0 +1,188 @@
# 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.
# 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.
import uuid
from twisted.internet import defer
from twisted.internet import threads
from nova import db
from nova import context
from nova import flags
from nova import test
from nova import utils
from nova.auth import manager
from nova.compute import instance_types
from nova.compute import power_state
from nova.virt import xenapi_conn
from nova.virt.xenapi import fake
from nova.virt.xenapi import volume_utils
FLAGS = flags.FLAGS
class XenAPIVolumeTestCase(test.TrialTestCase):
def setUp(self):
super(XenAPIVolumeTestCase, self).setUp()
FLAGS.xenapi_use_fake_session = True
FLAGS.target_host = '127.0.0.1'
FLAGS.xenapi_connection_url = 'test_url'
FLAGS.xenapi_connection_password = 'test_pass'
fake.reset()
def _create_volume(self, size='0'):
"""Create a volume object."""
vol = {}
vol['size'] = size
vol['user_id'] = 'fake'
vol['project_id'] = 'fake'
vol['host'] = 'localhost'
vol['availability_zone'] = FLAGS.storage_availability_zone
vol['status'] = "creating"
vol['attach_status'] = "detached"
return db.volume_create(context.get_admin_context(), vol)
def test_create_iscsi_storage_raise_no_exception(self):
session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass')
helper = volume_utils.VolumeHelper
helper.late_import(FLAGS)
vol = self._create_volume()
info = yield helper.parse_volume_info(vol['ec2_id'], '/dev/sdc')
label = 'SR-%s' % vol['ec2_id']
description = 'Test-SR'
sr_ref = helper.create_iscsi_storage_blocking(session,
info,
label,
description)
db.volume_destroy(context.get_admin_context(), vol['id'])
def test_attach_volume(self):
conn = xenapi_conn.get_connection(False)
volume = self._create_volume()
instance = FakeInstance(1, 'fake', 'fake', 1, 2, 3,
'm1.large', 'aa:bb:cc:dd:ee:ff')
fake.create_vm(instance.name, 'Running')
result = conn.attach_volume(instance.name, volume['ec2_id'],
'/dev/sdc')
def check(_):
# check that
# 1. the SR has been created
# 2. the instance has a VBD attached to it
pass
result.addCallback(check)
return result
def tearDown(self):
super(XenAPIVolumeTestCase, self).tearDown()
class XenAPIVMTestCase(test.TrialTestCase):
def setUp(self):
super(XenAPIVMTestCase, self).setUp()
self.manager = manager.AuthManager()
self.user = self.manager.create_user('fake', 'fake', 'fake',
admin=True)
self.project = self.manager.create_project('fake', 'fake', 'fake')
self.network = utils.import_object(FLAGS.network_manager)
FLAGS.xenapi_use_fake_session = True
FLAGS.xenapi_connection_url = 'test_url'
FLAGS.xenapi_connection_password = 'test_pass'
fake.reset()
fake.create_network('fake', FLAGS.flat_network_bridge)
def test_list_instances_0(self):
conn = xenapi_conn.get_connection(False)
instances = conn.list_instances()
self.assertEquals(instances, [])
test_list_instances_0.skip = "E"
def test_spawn(self):
conn = xenapi_conn.get_connection(False)
instance = FakeInstance(1, self.project.id, self.user.id, 1, 2, 3,
'm1.large', 'aa:bb:cc:dd:ee:ff')
result = conn.spawn(instance)
def check(_):
instances = conn.list_instances()
self.assertEquals(instances, [1])
# Get Nova record for VM
vm_info = conn.get_info(1)
# Get XenAPI record for VM
vms = fake.get_all('VM')
vm = fake.get_record('VM', vms[0])
# Check that m1.large above turned into the right thing.
instance_type = instance_types.INSTANCE_TYPES['m1.large']
mem_kib = long(instance_type['memory_mb']) << 10
mem_bytes = str(mem_kib << 10)
vcpus = instance_type['vcpus']
self.assertEquals(vm_info['max_mem'], mem_kib)
self.assertEquals(vm_info['mem'], mem_kib)
self.assertEquals(vm['memory_static_max'], mem_bytes)
self.assertEquals(vm['memory_dynamic_max'], mem_bytes)
self.assertEquals(vm['memory_dynamic_min'], mem_bytes)
self.assertEquals(vm['VCPUs_max'], str(vcpus))
self.assertEquals(vm['VCPUs_at_startup'], str(vcpus))
# Check that the VM is running according to Nova
self.assertEquals(vm_info['state'], power_state.RUNNING)
# Check that the VM is running according to XenAPI.
self.assertEquals(vm['power_state'], 'Running')
result.addCallback(check)
return result
def tearDown(self):
super(XenAPIVMTestCase, self).tearDown()
self.manager.delete_project(self.project)
self.manager.delete_user(self.user)
class FakeInstance():
def __init__(self, name, project_id, user_id, image_id, kernel_id,
ramdisk_id, instance_type, mac_address):
self.name = name
self.project_id = project_id
self.user_id = user_id
self.image_id = image_id
self.kernel_id = kernel_id
self.ramdisk_id = ramdisk_id
self.instance_type = instance_type
self.mac_address = mac_address

2
run_tests.py
View File

@@ -65,8 +65,8 @@ from nova.tests.service_unittest import *
from nova.tests.twistd_unittest import *
from nova.tests.validator_unittest import *
from nova.tests.virt_unittest import *
from nova.tests.virt_unittest import *
from nova.tests.volume_unittest import *
from nova.tests.xenapi_unittest import *
FLAGS = flags.FLAGS