diff options
| -rw-r--r-- | nova/tests/virt_unittest.py | 76 | ||||
| -rw-r--r-- | nova/virt/xenapi/fake.py | 128 |
2 files changed, 0 insertions, 204 deletions
diff --git a/nova/tests/virt_unittest.py b/nova/tests/virt_unittest.py index 1095662c3..d49383fb7 100644 --- a/nova/tests/virt_unittest.py +++ b/nova/tests/virt_unittest.py @@ -25,14 +25,9 @@ from nova import utils from nova.api.ec2 import cloud from nova.auth import manager from nova.virt import libvirt_conn -from nova.virt.xenapi import fake -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 +257,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() diff --git a/nova/virt/xenapi/fake.py b/nova/virt/xenapi/fake.py deleted file mode 100644 index 2fed28609..000000000 --- a/nova/virt/xenapi/fake.py +++ /dev/null @@ -1,128 +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. - -""" -A fake XenAPI SDK. - -Allows for xenapi helper classes testing. -""" - - -class Failure(Exception): - def __init__(self, message=None): - super(Failure, self).__init__(message) - self.details = [] - - def __str__(self): - return 'Fake XenAPI Exception' - - -class FakeXenAPISession(object): - """ The session to invoke XenAPI SDK calls """ - def __init__(self): - self.fail_next_call = False - - def get_xenapi(self): - """ Return the xenapi object """ - return self - - def get_xenapi_host(self): - """ Return the xenapi host """ - return 'FAKE_XENAPI_HOST' - - def call_xenapi(self, method, *args): - """Call the specified XenAPI method on a background thread. Returns - a Deferred for the result.""" - raise NotImplementedError() - - def async_call_plugin(self, plugin, fn, args): - """Call Async.host.call_plugin on a background thread. Returns a - Deferred with the task reference.""" - raise NotImplementedError() - - def wait_for_task(self, task): - """Return a Deferred that will give the result of the given task. - The task is polled until it completes.""" - raise NotImplementedError() - - def __getattr__(self, name): - return FakeXenAPIObject(name, self) - - -class FakeXenAPIObject(object): - def __init__(self, name, session): - self.name = name - self.session = session - self.FAKE_REF = 'FAKE_REFERENCE_%s' % name - - def get_by_name_label(self, label): - if label is None: - return '' # 'No object found' - else: - return 'FAKE_OBJECT_%s_%s' % (self.name, label) - - def getter(self, *args): - self._check_fail() - return self.FAKE_REF - - def ref_list(self, *args): - self._check_fail() - return [FakeXenAPIRecord()] - - def __getattr__(self, name): - if name == 'create': - return self._create - elif name == 'get_record': - return self._record - elif name == 'introduce' or\ - name == 'forget' or\ - name == 'unplug': - return self._fake_action - elif name.startswith('get_'): - getter = 'get_%s' % self.name - if name == getter: - return self.getter - else: - child = name[name.find('_') + 1:] - if child.endswith('s'): - return FakeXenAPIObject(child[:-1], self.session).ref_list - else: - return FakeXenAPIObject(child, self.session).getter - - def _create(self, *args): - self._check_fail() - return self.FAKE_REF - - def _record(self, *args): - self._check_fail() - return FakeXenAPIRecord() - - def _fake_action(self, *args): - self._check_fail() - pass - - def _check_fail(self): - if self.session.fail_next_call: - self.session.fail_next_call = False # Reset! - raise Failure('Unable to create %s' % self.name) - - -class FakeXenAPIRecord(dict): - def __init__(self): - pass - - def __getitem__(self, attr): - return '' |