x/python-ganttclient
Code Issues Proposed changes

Remove vm_utils.VMHelper class

Browse Source 
The XenAPI driver has the concept of helper classes. All methods
have been classmethods and the classes themselves are never
instantiated. As a result, they only add an extra namespace
lookup and provide no value.

Change-Id: If8a20501c9a9353359234441580552e41471722b
This commit is contained in:
Johannes Erdfelt
2012-05-31 18:13:03 +00:00
parent 14dc493355
commit f1b206e733
2 changed files with 37 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
nova/tests/test_xenapi.py
View File

@@ -703,7 +703,7 @@ class XenAPIVMTestCase(test.TestCase):
def test_rescue(self): def test_rescue(self):
instance = self._create_instance() instance = self._create_instance()
session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass') session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass')
vm_ref = vm_utils.VMHelper.lookup(session, instance.name) vm_ref = vm_utils.lookup(session, instance.name)
xenapi_fake.create_vbd(vm_ref, "swap", userdevice=1) xenapi_fake.create_vbd(vm_ref, "swap", userdevice=1)
xenapi_fake.create_vbd(vm_ref, "rootfs", userdevice=0) xenapi_fake.create_vbd(vm_ref, "rootfs", userdevice=0)
@@ -715,7 +715,7 @@ class XenAPIVMTestCase(test.TestCase):
vm = xenapi_fake.get_record('VM', vm_ref) vm = xenapi_fake.get_record('VM', vm_ref)
rescue_name = "%s-rescue" % vm["name_label"] rescue_name = "%s-rescue" % vm["name_label"]
rescue_ref = vm_utils.VMHelper.lookup(session, rescue_name) rescue_ref = vm_utils.lookup(session, rescue_name)
rescue_vm = xenapi_fake.get_record('VM', rescue_ref) rescue_vm = xenapi_fake.get_record('VM', rescue_ref)
vdi_uuids = [] vdi_uuids = []
@@ -1035,7 +1035,7 @@ class XenAPIDetermineDiskImageTestCase(test.TestCase):
self.fake_instance.architecture = 'x86-64' self.fake_instance.architecture = 'x86-64'
def assert_disk_type(self, image_meta, expected_disk_type): def assert_disk_type(self, image_meta, expected_disk_type):
actual = vm_utils.VMHelper.determine_disk_image_type(image_meta) actual = vm_utils.determine_disk_image_type(image_meta)
self.assertEqual(expected_disk_type, actual) self.assertEqual(expected_disk_type, actual)
def test_machine(self): def test_machine(self):
@@ -1154,14 +1154,11 @@ class XenAPIAutoDiskConfigTestCase(test.TestCase):
self.context = context.RequestContext(self.user_id, self.project_id) self.context = context.RequestContext(self.user_id, self.project_id)
@classmethod def fake_create_vbd(session, vm_ref, vdi_ref, userdevice,
def fake_create_vbd(cls, session, vm_ref, vdi_ref, userdevice,
vbd_type='disk', read_only=False, bootable=True): vbd_type='disk', read_only=False, bootable=True):
pass pass
self.stubs.Set(vm_utils.VMHelper, self.stubs.Set(vm_utils, 'create_vbd', fake_create_vbd)
"create_vbd",
fake_create_vbd)
def assertIsPartitionCalled(self, called): def assertIsPartitionCalled(self, called):
marker = {"partition_called": False} marker = {"partition_called": False}
@@ -1251,14 +1248,11 @@ class XenAPIGenerateLocal(test.TestCase):
self.context = context.RequestContext(self.user_id, self.project_id) self.context = context.RequestContext(self.user_id, self.project_id)
@classmethod def fake_create_vbd(session, vm_ref, vdi_ref, userdevice,
def fake_create_vbd(cls, session, vm_ref, vdi_ref, userdevice,
vbd_type='disk', read_only=False, bootable=True): vbd_type='disk', read_only=False, bootable=True):
pass pass
self.stubs.Set(vm_utils.VMHelper, self.stubs.Set(vm_utils, 'create_vbd', fake_create_vbd)
"create_vbd",
fake_create_vbd)
def assertCalled(self, instance): def assertCalled(self, instance):
ctx = context.RequestContext(self.user_id, self.project_id) ctx = context.RequestContext(self.user_id, self.project_id)
@@ -1281,11 +1275,9 @@ class XenAPIGenerateLocal(test.TestCase):
instance = db.instance_update(self.context, instance['id'], instance = db.instance_update(self.context, instance['id'],
{'instance_type_id': 5}) {'instance_type_id': 5})
@classmethod def fake_generate_swap(*args, **kwargs):
def fake_generate_swap(cls, *args, **kwargs):
self.called = True self.called = True
self.stubs.Set(vm_utils.VMHelper, 'generate_swap', self.stubs.Set(vm_utils, 'generate_swap', fake_generate_swap)
fake_generate_swap)
self.assertCalled(instance) self.assertCalled(instance)
@@ -1295,11 +1287,9 @@ class XenAPIGenerateLocal(test.TestCase):
instance = db.instance_update(self.context, instance['id'], instance = db.instance_update(self.context, instance['id'],
{'instance_type_id': 4}) {'instance_type_id': 4})
@classmethod def fake_generate_ephemeral(*args):
def fake_generate_ephemeral(cls, *args):
self.called = True self.called = True
self.stubs.Set(vm_utils.VMHelper, 'generate_ephemeral', self.stubs.Set(vm_utils, 'generate_ephemeral', fake_generate_ephemeral)
fake_generate_ephemeral)
self.assertCalled(instance) self.assertCalled(instance)
@@ -1307,7 +1297,7 @@ class XenAPIGenerateLocal(test.TestCase):
class XenAPIBWUsageTestCase(test.TestCase): class XenAPIBWUsageTestCase(test.TestCase):
def setUp(self): def setUp(self):
super(XenAPIBWUsageTestCase, self).setUp() super(XenAPIBWUsageTestCase, self).setUp()
self.stubs.Set(vm_utils.VMHelper, "compile_metrics", self.stubs.Set(vm_utils, 'compile_metrics',
XenAPIBWUsageTestCase._fake_compile_metrics) XenAPIBWUsageTestCase._fake_compile_metrics)
self.flags(target_host='127.0.0.1', self.flags(target_host='127.0.0.1',
xenapi_connection_url='test_url', xenapi_connection_url='test_url',
@@ -1651,7 +1641,7 @@ class XenAPISRSelectionTestCase(test.TestCase):
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests) stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass') session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass')
self.assertRaises(exception.StorageRepositoryNotFound, self.assertRaises(exception.StorageRepositoryNotFound,
vm_utils.VMHelper.safe_find_sr, session) vm_utils.safe_find_sr, session)
def test_safe_find_sr_local_storage(self): def test_safe_find_sr_local_storage(self):
"""Ensure the default local-storage is found.""" """Ensure the default local-storage is found."""
@@ -1666,7 +1656,7 @@ class XenAPISRSelectionTestCase(test.TestCase):
'Local storage', 'Local storage',
'i18n-key': 'local-storage'}, 'i18n-key': 'local-storage'},
host_ref=host_ref) host_ref=host_ref)
expected = vm_utils.VMHelper.safe_find_sr(session) expected = vm_utils.safe_find_sr(session)
self.assertEqual(local_sr, expected) self.assertEqual(local_sr, expected)
def test_safe_find_sr_by_other_criteria(self): def test_safe_find_sr_by_other_criteria(self):
@@ -1679,7 +1669,7 @@ class XenAPISRSelectionTestCase(test.TestCase):
type='lvm', type='lvm',
other_config={'my_fake_sr': 'true'}, other_config={'my_fake_sr': 'true'},
host_ref=host_ref) host_ref=host_ref)
expected = vm_utils.VMHelper.safe_find_sr(session) expected = vm_utils.safe_find_sr(session)
self.assertEqual(local_sr, expected) self.assertEqual(local_sr, expected)
def test_safe_find_sr_default(self): def test_safe_find_sr_default(self):
@@ -1688,7 +1678,7 @@ class XenAPISRSelectionTestCase(test.TestCase):
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests) stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass') session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass')
pool_ref = xenapi_fake.create_pool('') pool_ref = xenapi_fake.create_pool('')
expected = vm_utils.VMHelper.safe_find_sr(session) expected = vm_utils.safe_find_sr(session)
self.assertEqual(session.call_xenapi('pool.get_default_SR', pool_ref), self.assertEqual(session.call_xenapi('pool.get_default_SR', pool_ref),
expected) expected)

52
nova/tests/xenapi/stubs.py
View File

@@ -37,13 +37,12 @@ def stubout_firewall_driver(stubs, conn):
def stubout_instance_snapshot(stubs): def stubout_instance_snapshot(stubs):
@classmethod def fake_fetch_image(context, session, instance, image, type):
def fake_fetch_image(cls, context, session, instance, image, type):
return {'root': dict(uuid=_make_fake_vdi(), file=None), return {'root': dict(uuid=_make_fake_vdi(), file=None),
'kernel': dict(uuid=_make_fake_vdi(), file=None), 'kernel': dict(uuid=_make_fake_vdi(), file=None),
'ramdisk': dict(uuid=_make_fake_vdi(), file=None)} 'ramdisk': dict(uuid=_make_fake_vdi(), file=None)}
stubs.Set(vm_utils.VMHelper, 'fetch_image', fake_fetch_image) stubs.Set(vm_utils, 'fetch_image', fake_fetch_image)
def fake_wait_for_vhd_coalesce(*args): def fake_wait_for_vhd_coalesce(*args):
#TODO(sirp): Should we actually fake out the data here #TODO(sirp): Should we actually fake out the data here
@@ -97,20 +96,18 @@ def stubout_is_vdi_pv(stubs):
def stubout_determine_is_pv_objectstore(stubs): def stubout_determine_is_pv_objectstore(stubs):
"""Assumes VMs stu have PV kernels""" """Assumes VMs stu have PV kernels"""
@classmethod def f(*args):
def f(cls, *args):
return False return False
stubs.Set(vm_utils.VMHelper, '_determine_is_pv_objectstore', f) stubs.Set(vm_utils, '_determine_is_pv_objectstore', f)
def stubout_is_snapshot(stubs): def stubout_is_snapshot(stubs):
""" Always returns true """ Always returns true
xenapi fake driver does not create vmrefs for snapshots """ xenapi fake driver does not create vmrefs for snapshots """
@classmethod def f(*args):
def f(cls, *args):
return True return True
stubs.Set(vm_utils.VMHelper, 'is_snapshot', f) stubs.Set(vm_utils, 'is_snapshot', f)
def stubout_lookup_image(stubs): def stubout_lookup_image(stubs):
@@ -123,8 +120,7 @@ def stubout_lookup_image(stubs):
def stubout_fetch_image_glance_disk(stubs, raise_failure=False): def stubout_fetch_image_glance_disk(stubs, raise_failure=False):
"""Simulates a failure in fetch image_glance_disk.""" """Simulates a failure in fetch image_glance_disk."""
@classmethod def _fake_fetch_image_glance_disk(context, session, instance, image,
def _fake_fetch_image_glance_disk(cls, context, session, instance, image,
image_type): image_type):
if raise_failure: if raise_failure:
raise fake.Failure("Test Exception raised by " raise fake.Failure("Test Exception raised by "
@@ -139,18 +135,17 @@ def stubout_fetch_image_glance_disk(stubs, raise_failure=False):
vdi_type = vm_utils.ImageType.to_string(image_type) vdi_type = vm_utils.ImageType.to_string(image_type)
return {vdi_type: dict(uuid=None, file=filename)} return {vdi_type: dict(uuid=None, file=filename)}
stubs.Set(vm_utils.VMHelper, '_fetch_image_glance_disk', stubs.Set(vm_utils, '_fetch_image_glance_disk',
_fake_fetch_image_glance_disk) _fake_fetch_image_glance_disk)
def stubout_create_vm(stubs): def stubout_create_vm(stubs):
"""Simulates a failure in create_vm.""" """Simulates a failure in create_vm."""
@classmethod def f(*args):
def f(cls, *args):
raise fake.Failure("Test Exception raised by " + raise fake.Failure("Test Exception raised by " +
"fake create_vm") "fake create_vm")
stubs.Set(vm_utils.VMHelper, 'create_vm', f) stubs.Set(vm_utils, 'create_vm', f)
def _make_fake_vdi(): def _make_fake_vdi():
@@ -286,8 +281,7 @@ def stub_out_vm_methods(stubs):
def fake_release_bootlock(self, vm): def fake_release_bootlock(self, vm):
pass pass
@classmethod def fake_generate_ephemeral(*args):
def fake_generate_ephemeral(cls, *args):
pass pass
def fake_wait_for_device(dev): def fake_wait_for_device(dev):
@@ -295,7 +289,7 @@ def stub_out_vm_methods(stubs):
stubs.Set(vmops.VMOps, "_acquire_bootlock", fake_acquire_bootlock) stubs.Set(vmops.VMOps, "_acquire_bootlock", fake_acquire_bootlock)
stubs.Set(vmops.VMOps, "_release_bootlock", fake_release_bootlock) stubs.Set(vmops.VMOps, "_release_bootlock", fake_release_bootlock)
stubs.Set(vm_utils.VMHelper, 'generate_ephemeral', fake_generate_ephemeral) stubs.Set(vm_utils, 'generate_ephemeral', fake_generate_ephemeral)
stubs.Set(vm_utils, '_wait_for_device', fake_wait_for_device) stubs.Set(vm_utils, '_wait_for_device', fake_wait_for_device)
@@ -344,32 +338,28 @@ def stub_out_migration_methods(stubs):
vdi_rec['other_config']['nova_disk_type'] = 'root' vdi_rec['other_config']['nova_disk_type'] = 'root'
return {'uuid': vdi_rec['uuid'], 'ref': vdi_ref} return {'uuid': vdi_rec['uuid'], 'ref': vdi_ref}
@classmethod def fake_get_vdi(session, vm_ref):
def fake_get_vdi(cls, session, vm_ref):
vdi_ref = fake.create_vdi('derp', 'herp') vdi_ref = fake.create_vdi('derp', 'herp')
vdi_rec = session.call_xenapi("VDI.get_record", vdi_ref) vdi_rec = session.call_xenapi("VDI.get_record", vdi_ref)
return vdi_ref, {'uuid': vdi_rec['uuid'], } return vdi_ref, {'uuid': vdi_rec['uuid'], }
@classmethod def fake_sr(session, *args):
def fake_sr(cls, session, *args):
pass pass
@classmethod def fake_get_sr_path(*args):
def fake_get_sr_path(cls, *args):
return "fake" return "fake"
def fake_destroy(*args, **kwargs): def fake_destroy(*args, **kwargs):
pass pass
@classmethod def fake_generate_ephemeral(*args):
def fake_generate_ephemeral(cls, *args):
pass pass
stubs.Set(vmops.VMOps, '_destroy', fake_destroy) stubs.Set(vmops.VMOps, '_destroy', fake_destroy)
stubs.Set(vmops.VMOps, '_move_disks', fake_move_disks) stubs.Set(vmops.VMOps, '_move_disks', fake_move_disks)
stubs.Set(vm_utils.VMHelper, 'scan_default_sr', fake_sr) stubs.Set(vm_utils, 'scan_default_sr', fake_sr)
stubs.Set(vm_utils.VMHelper, 'scan_sr', fake_sr) stubs.Set(vm_utils, 'scan_sr', fake_sr)
stubs.Set(vmops.VMOps, '_create_snapshot', fake_create_snapshot) stubs.Set(vmops.VMOps, '_create_snapshot', fake_create_snapshot)
stubs.Set(vm_utils.VMHelper, 'get_vdi_for_vm_safely', fake_get_vdi) stubs.Set(vm_utils, 'get_vdi_for_vm_safely', fake_get_vdi)
stubs.Set(vm_utils.VMHelper, 'get_sr_path', fake_get_sr_path) stubs.Set(vm_utils, 'get_sr_path', fake_get_sr_path)
stubs.Set(vm_utils.VMHelper, 'generate_ephemeral', fake_generate_ephemeral) stubs.Set(vm_utils, 'generate_ephemeral', fake_generate_ephemeral)