Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Merge "VMware: Consolidate fake_session in test_(vm|ds)_util"

  • Loading branch information...
commit 0b3f4588e8bce48fbed862de0b7d0c4eb279584f 2 parents 27fbfac + 13b8714
Jenkins authored openstack-gerrit committed
View
34 nova/tests/virt/vmwareapi/fake.py
@@ -994,6 +994,40 @@ def create(self, obj_name):
return DataObject(obj_name)
+class FakeSession(object):
+ """Fake Session Class."""
+
+ def __init__(self):
+ self.vim = FakeVim()
+
+ def _get_vim(self):
+ return self.vim
+
+ def _call_method(self, module, method, *args, **kwargs):
+ raise NotImplementedError()
+
+ def _wait_for_task(self, task_ref):
+ raise NotImplementedError()
+
+
+class FakeObjectRetrievalSession(FakeSession):
+ """A session for faking object retrieval tasks.
+
+ _call_method() returns a given set of objects
+ sequentially, regardless of the method called.
+ """
+
+ def __init__(self, *ret):
+ super(FakeObjectRetrievalSession, self).__init__()
+ self.ret = ret
+ self.ind = 0
+
+ def _call_method(self, module, method, *args, **kwargs):
+ # return fake objects in a circular manner
+ self.ind = (self.ind + 1) % len(self.ret)
+ return self.ret[self.ind - 1]
+
+
class FakeVim(object):
"""Fake VIM Class."""
View
81 nova/tests/virt/vmwareapi/test_ds_util.py
@@ -23,32 +23,10 @@
from nova.virt.vmwareapi import error_util
-class fake_session(object):
- def __init__(self, ret=None):
- self.ret = ret
-
- def _get_vim(self):
- return fake.FakeVim()
-
- def _call_method(self, module, method, *args, **kwargs):
- return self.ret
-
- def _wait_for_task(self, task_ref):
- task_info = self._call_method('module', "get_dynamic_property",
- task_ref, "Task", "info")
- if task_info.state == 'success':
- return task_info
- else:
- error_info = 'fake error'
- error = task_info.error
- name = error.fault.__class__.__name__
- raise error_util.get_fault_class(name)(error_info)
-
-
class DsUtilTestCase(test.NoDBTestCase):
def setUp(self):
super(DsUtilTestCase, self).setUp()
- self.session = fake_session()
+ self.session = fake.FakeSession()
self.flags(api_retry_count=1, group='vmware')
fake.reset()
@@ -126,22 +104,31 @@ def fake_call_method(module, method, *args, **kwargs):
datastorePath = kwargs.get('datastorePath')
self.assertEqual('fake-path', datastorePath)
return 'fake_exists_task'
- elif method == 'get_dynamic_property':
- info = fake.DataObject()
- info.name = 'search_task'
- info.state = 'success'
+
+ # Should never get here
+ self.fail()
+
+ def fake_wait_for_task(task_ref):
+ if task_ref == 'fake_exists_task':
+ result_file = fake.DataObject()
+ result_file.path = 'fake-file'
+
result = fake.DataObject()
- result.path = 'fake-path'
- matched = fake.DataObject()
- matched.path = 'fake-file'
- result.file = [matched]
- info.result = result
- return info
+ result.file = [result_file]
+
+ task_info = fake.DataObject()
+ task_info.result = result
+
+ return task_info
+
# Should never get here
self.fail()
- with mock.patch.object(self.session, '_call_method',
- fake_call_method):
+ with contextlib.nested(
+ mock.patch.object(self.session, '_call_method',
+ fake_call_method),
+ mock.patch.object(self.session, '_wait_for_task',
+ fake_wait_for_task)):
file_exists = ds_util.file_exists(self.session,
'fake-browser', 'fake-path', 'fake-file')
self.assertTrue(file_exists)
@@ -150,20 +137,22 @@ def test_file_exists_fails(self):
def fake_call_method(module, method, *args, **kwargs):
if method == 'SearchDatastore_Task':
return 'fake_exists_task'
- elif method == 'get_dynamic_property':
- info = fake.DataObject()
- info.name = 'search_task'
- info.state = 'error'
- error = fake.DataObject()
- error.localizedMessage = "Error message"
- error.fault = fake.FileNotFound()
- info.error = error
- return info
+
# Should never get here
self.fail()
- with mock.patch.object(self.session, '_call_method',
- fake_call_method):
+ def fake_wait_for_task(task_ref):
+ if task_ref == 'fake_exists_task':
+ raise error_util.FileNotFoundException()
+
+ # Should never get here
+ self.fail()
+
+ with contextlib.nested(
+ mock.patch.object(self.session, '_call_method',
+ fake_call_method),
+ mock.patch.object(self.session, '_wait_for_task',
+ fake_wait_for_task)):
file_exists = ds_util.file_exists(self.session,
'fake-browser', 'fake-path', 'fake-file')
self.assertFalse(file_exists)
View
3  nova/tests/virt/vmwareapi/test_vif.py
@@ -24,7 +24,6 @@
from nova.tests import matchers
from nova.tests import utils
from nova.tests.virt.vmwareapi import fake
-from nova.tests.virt.vmwareapi import test_vm_util
from nova.virt.vmwareapi import error_util
from nova.virt.vmwareapi import network_util
from nova.virt.vmwareapi import vif
@@ -54,7 +53,7 @@ def setUp(self):
ovs_interfaceid=None,
rxtx_cap=3)
])[0]
- self.session = test_vm_util.fake_session()
+ self.session = fake.FakeSession()
self.cluster = None
def tearDown(self):
View
89 nova/tests/virt/vmwareapi/test_vm_util.py
@@ -31,24 +31,6 @@
from nova.virt.vmwareapi import vm_util
-class fake_session(object):
- def __init__(self, *ret):
- self.ret = ret
- self.ind = 0
- self.vim = fake.FakeVim()
-
- def _call_method(self, *args):
- # return fake objects in circular manner
- self.ind = (self.ind + 1) % len(self.ret)
- return self.ret[self.ind - 1]
-
- def _wait_for_task(self, task_ref):
- pass
-
- def _get_vim(self):
- return self.vim
-
-
class partialObject(object):
def __init__(self, path='fake-path'):
self.path = path
@@ -69,7 +51,7 @@ def test_get_datastore(self):
fake_objects = fake.FakeRetrieveResult()
fake_objects.add_object(fake.Datastore())
result = vm_util.get_datastore(
- fake_session(fake_objects))
+ fake.FakeObjectRetrievalSession(fake_objects))
self.assertEqual("fake-ds", result.name)
self.assertEqual(units.Ti, result.capacity)
@@ -83,7 +65,8 @@ def test_get_datastore_with_regex(self):
fake_objects.add_object(fake.Datastore("fake-ds0"))
fake_objects.add_object(fake.Datastore("fake-ds1"))
result = vm_util.get_datastore(
- fake_session(fake_objects), None, None, datastore_valid_regex)
+ fake.FakeObjectRetrievalSession(fake_objects),
+ None, None, datastore_valid_regex)
self.assertEqual("openstack-ds0", result.name)
def _test_get_stats_from_cluster(self, connection_state="connected",
@@ -126,7 +109,6 @@ def _test_get_stats_from_cluster(self, connection_state="connected",
respool_resource_usage = fake.DataObject()
respool_resource_usage.maxUsage = 5368709120
respool_resource_usage.overallUsage = 2147483648
- session = fake_session()
def fake_call_method(*args):
if "get_dynamic_properties" in args:
@@ -135,8 +117,9 @@ def fake_call_method(*args):
return fake_objects
else:
return respool_resource_usage
- with mock.patch.object(fake_session, '_call_method',
- fake_call_method):
+
+ session = fake.FakeSession()
+ with mock.patch.object(session, '_call_method', fake_call_method):
result = vm_util.get_stats_from_cluster(session, "cluster1")
cpu_info = {}
mem_info = {}
@@ -175,7 +158,7 @@ def test_get_datastore_with_token(self):
fake1.add_object(fake.Datastore("ds2", 10 * units.Gi, 8 * units.Gi))
fake1.add_object(fake.Datastore("ds3", 10 * units.Gi, 1 * units.Gi))
result = vm_util.get_datastore(
- fake_session(fake0, fake1), None, None, regex)
+ fake.FakeObjectRetrievalSession(fake0, fake1), None, None, regex)
self.assertEqual("ds2", result.name)
def test_get_datastore_with_list(self):
@@ -186,7 +169,8 @@ def test_get_datastore_with_list(self):
fake_objects.add_object(fake.Datastore("openstack-ds1"))
fake_objects.add_object(fake.Datastore("openstack-ds2"))
result = vm_util.get_datastore(
- fake_session(fake_objects), None, None, datastore_valid_regex)
+ fake.FakeObjectRetrievalSession(fake_objects),
+ None, None, datastore_valid_regex)
self.assertNotEqual("openstack-ds1", result.name)
def test_get_datastore_with_regex_error(self):
@@ -202,7 +186,7 @@ def test_get_datastore_with_regex_error(self):
# try/catch block, but it's available only from Py 2.7.
try:
vm_util.get_datastore(
- fake_session(fake_objects), None, None,
+ fake.FakeObjectRetrievalSession(fake_objects), None, None,
datastore_invalid_regex)
except exception.DatastoreNotFound as e:
self.assertEqual(exp_message, e.args[0])
@@ -214,11 +198,11 @@ def test_get_datastore_without_datastore(self):
self.assertRaises(exception.DatastoreNotFound,
vm_util.get_datastore,
- fake_session(None), host="fake-host")
+ fake.FakeObjectRetrievalSession(None), host="fake-host")
self.assertRaises(exception.DatastoreNotFound,
vm_util.get_datastore,
- fake_session(None), cluster="fake-cluster")
+ fake.FakeObjectRetrievalSession(None), cluster="fake-cluster")
def test_get_host_ref_from_id(self):
fake_host_name = "ha-host"
@@ -227,7 +211,8 @@ def test_get_host_ref_from_id(self):
fake_objects = fake.FakeRetrieveResult()
fake_objects.add_object(fake_host_sys)
ref = vm_util.get_host_ref_from_id(
- fake_session(fake_objects), fake_host_id, ['name'])
+ fake.FakeObjectRetrievalSession(fake_objects),
+ fake_host_id, ['name'])
self.assertIsInstance(ref, fake.HostSystem)
self.assertEqual(fake_host_id, ref.obj.value)
@@ -239,12 +224,12 @@ def test_get_host_ref_from_id(self):
def test_get_host_ref_no_hosts_in_cluster(self):
self.assertRaises(exception.NoValidHost,
vm_util.get_host_ref,
- fake_session(""), 'fake_cluster')
+ fake.FakeObjectRetrievalSession(""), 'fake_cluster')
def test_get_datastore_no_host_in_cluster(self):
self.assertRaises(exception.DatastoreNotFound,
vm_util.get_datastore,
- fake_session(""), 'fake_cluster')
+ fake.FakeObjectRetrievalSession(""), 'fake_cluster')
@mock.patch.object(vm_util, '_get_vm_ref_from_vm_uuid',
return_value=None)
@@ -257,12 +242,12 @@ def test_get_host_name_for_vm(self, _get_ref_from_uuid):
fake_objects.add_object(fake_vm)
vm_ref = vm_util.get_vm_ref_from_name(
- fake_session(fake_objects), 'vm-123')
+ fake.FakeObjectRetrievalSession(fake_objects), 'vm-123')
self.assertIsNotNone(vm_ref)
host_id = vm_util.get_host_id_from_vm_ref(
- fake_session(fake_objects), vm_ref)
+ fake.FakeObjectRetrievalSession(fake_objects), vm_ref)
self.assertEqual(fake_host_id, host_id)
@@ -329,7 +314,7 @@ def test_get_datastore_inaccessible_ds(self):
self.assertRaises(exception.DatastoreNotFound,
vm_util.get_datastore,
- fake_session(fake_objects))
+ fake.FakeObjectRetrievalSession(fake_objects))
def test_get_resize_spec(self):
fake_instance = {'id': 7, 'name': 'fake!',
@@ -548,7 +533,8 @@ def test_get_vnc_port(self):
fake_vms = self._create_fake_vms()
self.flags(vnc_port=5900, group='vmware')
self.flags(vnc_port_total=10000, group='vmware')
- actual = vm_util.get_vnc_port(fake_session(fake_vms))
+ actual = vm_util.get_vnc_port(
+ fake.FakeObjectRetrievalSession(fake_vms))
self.assertEqual(actual, 5910)
def test_get_vnc_port_exhausted(self):
@@ -557,26 +543,26 @@ def test_get_vnc_port_exhausted(self):
self.flags(vnc_port_total=10, group='vmware')
self.assertRaises(exception.ConsolePortRangeExhausted,
vm_util.get_vnc_port,
- fake_session(fake_vms))
+ fake.FakeObjectRetrievalSession(fake_vms))
def test_get_all_cluster_refs_by_name_none(self):
fake_objects = fake.FakeRetrieveResult()
- refs = vm_util.get_all_cluster_refs_by_name(fake_session(fake_objects),
- ['fake_cluster'])
+ refs = vm_util.get_all_cluster_refs_by_name(
+ fake.FakeObjectRetrievalSession(fake_objects), ['fake_cluster'])
self.assertTrue(not refs)
def test_get_all_cluster_refs_by_name_exists(self):
fake_objects = fake.FakeRetrieveResult()
fake_objects.add_object(fake.ClusterComputeResource(name='cluster'))
- refs = vm_util.get_all_cluster_refs_by_name(fake_session(fake_objects),
- ['cluster'])
+ refs = vm_util.get_all_cluster_refs_by_name(
+ fake.FakeObjectRetrievalSession(fake_objects), ['cluster'])
self.assertEqual(1, len(refs))
def test_get_all_cluster_refs_by_name_missing(self):
fake_objects = fake.FakeRetrieveResult()
fake_objects.add_object(partialObject(path='cluster'))
- refs = vm_util.get_all_cluster_refs_by_name(fake_session(fake_objects),
- ['cluster'])
+ refs = vm_util.get_all_cluster_refs_by_name(
+ fake.FakeObjectRetrievalSession(fake_objects), ['cluster'])
self.assertTrue(not refs)
def test_propset_dict_simple(self):
@@ -674,7 +660,7 @@ def fake_wait_for_task(self, *args):
task_info = mock.Mock(state="success", result="fake_vm_ref")
return task_info
- session = fake_session()
+ session = fake.FakeSession()
fake_instance = mock.MagicMock()
fake_call_mock = mock.Mock(side_effect=fake_call_method)
fake_wait_mock = mock.Mock(side_effect=fake_wait_for_task)
@@ -714,7 +700,7 @@ def test_convert_vif_model(self):
"InvalidVifModel")
def test_power_on_instance_with_vm_ref(self):
- session = fake_session()
+ session = fake.FakeSession()
fake_instance = mock.MagicMock()
with contextlib.nested(
mock.patch.object(session, "_call_method",
@@ -729,7 +715,7 @@ def test_power_on_instance_with_vm_ref(self):
fake_wait_for_task.assert_called_once_with('fake-task')
def test_power_on_instance_without_vm_ref(self):
- session = fake_session()
+ session = fake.FakeSession()
fake_instance = mock.MagicMock()
with contextlib.nested(
mock.patch.object(vm_util, "get_vm_ref",
@@ -746,7 +732,7 @@ def test_power_on_instance_without_vm_ref(self):
fake_wait_for_task.assert_called_once_with('fake-task')
def test_power_on_instance_with_exception(self):
- session = fake_session()
+ session = fake.FakeSession()
fake_instance = mock.MagicMock()
with contextlib.nested(
mock.patch.object(session, "_call_method",
@@ -764,7 +750,7 @@ def test_power_on_instance_with_exception(self):
fake_wait_for_task.assert_called_once_with('fake-task')
def test_power_on_instance_with_power_state_exception(self):
- session = fake_session()
+ session = fake.FakeSession()
fake_instance = mock.MagicMock()
with contextlib.nested(
mock.patch.object(session, "_call_method",
@@ -781,7 +767,7 @@ def test_power_on_instance_with_power_state_exception(self):
fake_wait_for_task.assert_called_once_with('fake-task')
def test_create_virtual_disk(self):
- session = fake_session()
+ session = fake.FakeSession()
dm = session._get_vim().get_service_content().virtualDiskManager
with contextlib.nested(
mock.patch.object(vm_util, "get_vmdk_create_spec",
@@ -807,7 +793,7 @@ def test_create_virtual_disk(self):
fake_wait_for_task.assert_called_once_with('fake-task')
def test_copy_virtual_disk(self):
- session = fake_session()
+ session = fake.FakeSession()
dm = session._get_vim().get_service_content().virtualDiskManager
with contextlib.nested(
mock.patch.object(session, "_call_method",
@@ -837,14 +823,15 @@ def test_get_values(self):
'summary.guest.toolsStatus',
'summary.guest.toolsRunningStatus']
query = vm_util.get_values_from_object_properties(
- fake_session(objects), objects, lst_properties)
+ fake.FakeObjectRetrievalSession(objects),
+ objects, lst_properties)
self.assertEqual('poweredOn', query['runtime.powerState'])
self.assertEqual('guestToolsRunning',
query['summary.guest.toolsRunningStatus'])
self.assertEqual('toolsOk', query['summary.guest.toolsStatus'])
def test_reconfigure_vm(self):
- session = fake_session()
+ session = fake.FakeSession()
with contextlib.nested(
mock.patch.object(session, '_call_method',
return_value='fake_reconfigure_task'),
Please sign in to comment.
Something went wrong with that request. Please try again.