# 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. """ Test suite for XenAPI """ import stubout 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 as xenapi_fake from nova.virt.xenapi import volume_utils from nova.tests.db import fakes as db_fakes from nova.tests.xenapi import stubs FLAGS = flags.FLAGS class XenAPIVolumeTestCase(test.TestCase): """ Unit tests for Volume operations """ def setUp(self): super(XenAPIVolumeTestCase, self).setUp() self.stubs = stubout.StubOutForTesting() FLAGS.target_host = '127.0.0.1' FLAGS.xenapi_connection_url = 'test_url' FLAGS.xenapi_connection_password = 'test_pass' db_fakes.stub_out_db_instance_api(self.stubs) stubs.stub_out_get_target(self.stubs) xenapi_fake.reset() self.values = {'name': 1, 'id': 1, 'project_id': 'fake', 'user_id': 'fake', 'image_id': 1, 'kernel_id': 2, 'ramdisk_id': 3, 'instance_type': 'm1.large', 'mac_address': 'aa:bb:cc:dd:ee:ff', } 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(self): """ This shows how to test helper classes' methods """ stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests) session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass') helper = volume_utils.VolumeHelper helper.XenAPI = session.get_imported_xenapi() vol = self._create_volume() info = helper.parse_volume_info(vol['id'], '/dev/sdc') label = 'SR-%s' % vol['id'] description = 'Test-SR' sr_ref = helper.create_iscsi_storage(session, info, label, description) srs = xenapi_fake.get_all('SR') self.assertEqual(sr_ref, srs[0]) db.volume_destroy(context.get_admin_context(), vol['id']) def test_parse_volume_info_raise_exception(self): """ This shows how to test helper classes' methods """ stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests) session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass') helper = volume_utils.VolumeHelper helper.XenAPI = session.get_imported_xenapi() vol = self._create_volume() # oops, wrong mount point! self.assertRaises(volume_utils.StorageError, helper.parse_volume_info, vol['id'], '/dev/sd') db.volume_destroy(context.get_admin_context(), vol['id']) def test_attach_volume(self): """ This shows how to test Ops classes' methods """ stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests) conn = xenapi_conn.get_connection(False) volume = self._create_volume() instance = db.instance_create(self.values) xenapi_fake.create_vm(instance.name, 'Running') result = conn.attach_volume(instance.name, volume['id'], '/dev/sdc') def check(): # check that the VM has a VBD attached to it # Get XenAPI reference for the VM vms = xenapi_fake.get_all('VM') # Get XenAPI record for VBD vbds = xenapi_fake.get_all('VBD') vbd = xenapi_fake.get_record('VBD', vbds[0]) vm_ref = vbd['VM'] self.assertEqual(vm_ref, vms[0]) check() def test_attach_volume_raise_exception(self): """ This shows how to test when exceptions are raised """ stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeFailedTests) conn = xenapi_conn.get_connection(False) volume = self._create_volume() instance = db.instance_create(self.values) xenapi_fake.create_vm(instance.name, 'Running') self.assertRaises(Exception, conn.attach_volume, instance.name, volume['id'], '/dev/sdc') def tearDown(self): super(XenAPIVolumeTestCase, self).tearDown() self.stubs.UnsetAll() class XenAPIVMTestCase(test.TestCase): """ Unit tests for VM operations """ 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) self.stubs = stubout.StubOutForTesting() FLAGS.xenapi_connection_url = 'test_url' FLAGS.xenapi_connection_password = 'test_pass' xenapi_fake.reset() db_fakes.stub_out_db_instance_api(self.stubs) xenapi_fake.create_network('fake', FLAGS.flat_network_bridge) stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests) self.conn = xenapi_conn.get_connection(False) def test_list_instances_0(self): instances = self.conn.list_instances() self.assertEquals(instances, []) def test_get_diagnostics(self): instance = self._create_instance() self.conn.get_diagnostics(instance) def test_instance_snapshot(self): stubs.stubout_instance_snapshot(self.stubs) instance = self._create_instance() name = "MySnapshot" template_vm_ref = self.conn.snapshot(instance, name) def ensure_vm_was_torn_down(): vm_labels = [] for vm_ref in xenapi_fake.get_all('VM'): vm_rec = xenapi_fake.get_record('VM', vm_ref) if not vm_rec["is_control_domain"]: vm_labels.append(vm_rec["name_label"]) self.assertEquals(vm_labels, [1]) def ensure_vbd_was_torn_down(): vbd_labels = [] for vbd_ref in xenapi_fake.get_all('VBD'): vbd_rec = xenapi_fake.get_record('VBD', vbd_ref) vbd_labels.append(vbd_rec["vm_name_label"]) self.assertEquals(vbd_labels, [1]) def ensure_vdi_was_torn_down(): for vdi_ref in xenapi_fake.get_all('VDI'): vdi_rec = xenapi_fake.get_record('VDI', vdi_ref) name_label = vdi_rec["name_label"] self.assert_(not name_label.endswith('snapshot')) def check(): ensure_vm_was_torn_down() ensure_vbd_was_torn_down() ensure_vdi_was_torn_down() check() def test_spawn(self): instance = self._create_instance() def check(): instances = self.conn.list_instances() self.assertEquals(instances, [1]) # Get Nova record for VM vm_info = self.conn.get_info(1) # Get XenAPI record for VM vms = xenapi_fake.get_all('VM') vm = xenapi_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') check() def tearDown(self): super(XenAPIVMTestCase, self).tearDown() self.manager.delete_project(self.project) self.manager.delete_user(self.user) self.stubs.UnsetAll() def _create_instance(self): """Creates and spawns a test instance""" values = { 'name': 1, 'id': 1, 'project_id': self.project.id, 'user_id': self.user.id, 'image_id': 1, 'kernel_id': 2, 'ramdisk_id': 3, 'instance_type': 'm1.large', 'mac_address': 'aa:bb:cc:dd:ee:ff'} instance = db.instance_create(values) self.conn.spawn(instance) return instance