saltstack/salt
1
0
Fork 0
mirror of https://github.com/saltstack/salt.git synced 2025-04-17 10:10:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Adding unit tests for vmware_test

Browse source
This commit is contained in:
Nitin Madhok 2017-03-30 00:54:57 -04:00
parent 36edf0af64
commit dd62310941
5 changed files with 2226 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

249
tests/unit/utils/vmware_test/test_cluster.py Normal file
View file

@ -0,0 +1,249 @@
# -*- coding: utf-8 -*-
'''
:codeauthor: :email:`Alexandru Bleotu <alexandru.bleotu@morganstanley.com>`
Tests for cluster related functions in salt.utils.vmware
'''
# Import python libraries
from __future__ import absolute_import
import logging
# Import Salt testing libraries
from salttesting import TestCase, skipIf
from salttesting.mock import NO_MOCK, NO_MOCK_REASON, patch, MagicMock, call
# Import Salt libraries
from salt.exceptions import VMwareApiError, VMwareRuntimeError, \
VMwareObjectRetrievalError
import salt.utils.vmware as vmware
# Import Third Party Libs
try:
from pyVmomi import vim, vmodl
HAS_PYVMOMI = True
except ImportError:
HAS_PYVMOMI = False
# Get Logging Started
log = logging.getLogger(__name__)
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
@patch('salt.utils.vmware.get_managed_object_name',
MagicMock())
@patch('salt.utils.vmware.get_service_instance_from_managed_object',
MagicMock())
@patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=[{'name': 'fake_cluster',
'object': MagicMock()}]))
class GetClusterTestCase(TestCase):
'''Tests for salt.utils.vmware.get_cluster'''
def setUp(self):
self.mock_si = MagicMock()
self.mock_dc = MagicMock()
self.mock_cluster1 = MagicMock()
self.mock_cluster2 = MagicMock()
self.mock_entries = [{'name': 'fake_cluster1',
'object': self.mock_cluster1},
{'name': 'fake_cluster2',
'object': self.mock_cluster2}]
def test_get_managed_object_name_call(self):
mock_get_managed_object_name = MagicMock()
with patch('salt.utils.vmware.get_managed_object_name',
mock_get_managed_object_name):
vmware.get_cluster(self.mock_dc, 'fake_cluster')
mock_get_managed_object_name.assert_called_once_with(self.mock_dc)
def test_get_service_instance_from_managed_object(self):
mock_dc_name = MagicMock()
mock_get_service_instance_from_managed_object = MagicMock()
with patch('salt.utils.vmware.get_managed_object_name',
MagicMock(return_value=mock_dc_name)):
with patch(
'salt.utils.vmware.get_service_instance_from_managed_object',
mock_get_service_instance_from_managed_object):
vmware.get_cluster(self.mock_dc, 'fake_cluster')
mock_get_service_instance_from_managed_object.assert_called_once_with(
self.mock_dc, name=mock_dc_name)
def test_traversal_spec_init(self):
mock_dc_name = MagicMock()
mock_traversal_spec = MagicMock()
mock_traversal_spec_ini = MagicMock(return_value=mock_traversal_spec)
mock_get_service_instance_from_managed_object = MagicMock()
patch_traversal_spec_str = \
'salt.utils.vmware.vmodl.query.PropertyCollector.TraversalSpec'
with patch(patch_traversal_spec_str, mock_traversal_spec_ini):
vmware.get_cluster(self.mock_dc, 'fake_cluster')
mock_traversal_spec_ini.assert_has_calls(
[call(path='childEntity',
skip=False,
type=vim.Folder),
call(path='hostFolder',
skip=True,
type=vim.Datacenter,
selectSet=[mock_traversal_spec])])
def test_get_mors_with_properties_call(self):
mock_get_mors_with_properties = MagicMock(
return_value=[{'name': 'fake_cluster', 'object': MagicMock()}])
mock_traversal_spec = MagicMock()
patch_traversal_spec_str = \
'salt.utils.vmware.vmodl.query.PropertyCollector.TraversalSpec'
with patch(
'salt.utils.vmware.get_service_instance_from_managed_object',
MagicMock(return_value=self.mock_si)):
with patch('salt.utils.vmware.get_mors_with_properties',
mock_get_mors_with_properties):
with patch(patch_traversal_spec_str,
MagicMock(return_value=mock_traversal_spec)):
vmware.get_cluster(self.mock_dc, 'fake_cluster')
mock_get_mors_with_properties.assert_called_once_with(
self.mock_si, vim.ClusterComputeResource,
container_ref=self.mock_dc,
property_list=['name'],
traversal_spec=mock_traversal_spec)
def test_get_mors_with_properties_returns_empty_array(self):
with patch('salt.utils.vmware.get_managed_object_name',
MagicMock(return_value='fake_dc')):
with patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=[])):
with self.assertRaises(VMwareObjectRetrievalError) as excinfo:
vmware.get_cluster(self.mock_dc, 'fake_cluster')
self.assertEqual(excinfo.exception.strerror,
'Cluster \'fake_cluster\' was not found in '
'datacenter \'fake_dc\'')
def test_cluster_not_found(self):
with patch('salt.utils.vmware.get_managed_object_name',
MagicMock(return_value='fake_dc')):
with patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=self.mock_entries)):
with self.assertRaises(VMwareObjectRetrievalError) as excinfo:
vmware.get_cluster(self.mock_dc, 'fake_cluster')
self.assertEqual(excinfo.exception.strerror,
'Cluster \'fake_cluster\' was not found in '
'datacenter \'fake_dc\'')
def test_cluster_found(self):
with patch('salt.utils.vmware.get_managed_object_name',
MagicMock(return_value='fake_dc')):
with patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=self.mock_entries)):
res = vmware.get_cluster(self.mock_dc, 'fake_cluster2')
self.assertEqual(res, self.mock_cluster2)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
@patch('salt.utils.vmware.get_managed_object_name', MagicMock())
class CreateClusterTestCase(TestCase):
'''Tests for salt.utils.vmware.create_cluster'''
def setUp(self):
self.mock_create_cluster_ex = MagicMock()
self.mock_dc = MagicMock(
hostFolder=MagicMock(CreateClusterEx=self.mock_create_cluster_ex))
self.mock_cluster_spec = MagicMock()
def test_get_managed_object_name(self):
mock_get_managed_object_name = MagicMock()
with patch('salt.utils.vmware.get_managed_object_name',
mock_get_managed_object_name):
vmware.create_cluster(self.mock_dc, 'fake_cluster',
self.mock_cluster_spec)
mock_get_managed_object_name.assert_called_once_with(self.mock_dc)
def test_create_cluster_call(self):
vmware.create_cluster(self.mock_dc, 'fake_cluster',
self.mock_cluster_spec)
self.mock_create_cluster_ex.assert_called_once_with(
'fake_cluster', self.mock_cluster_spec)
def test_create_cluster_raise_vim_fault(self):
exc = vim.fault.VimFault()
exc.msg = 'VimFault msg'
self.mock_dc.hostFolder.CreateClusterEx = MagicMock(
side_effect=exc)
with self.assertRaises(VMwareApiError) as excinfo:
vmware.create_cluster(self.mock_dc, 'fake_cluster',
self.mock_cluster_spec)
self.assertEqual(excinfo.exception.strerror, 'VimFault msg')
def test_create_cluster_raise_runtime_fault(self):
exc = vmodl.RuntimeFault()
exc.msg = 'RuntimeFault msg'
self.mock_dc.hostFolder.CreateClusterEx = MagicMock(
side_effect=exc)
with self.assertRaises(VMwareRuntimeError) as excinfo:
vmware.create_cluster(self.mock_dc, 'fake_cluster',
self.mock_cluster_spec)
self.assertEqual(excinfo.exception.strerror, 'RuntimeFault msg')
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
@patch('salt.utils.vmware.get_managed_object_name', MagicMock())
@patch('salt.utils.vmware.wait_for_task', MagicMock())
class UpdateClusterTestCase(TestCase):
'''Tests for salt.utils.vmware.update_cluster'''
def setUp(self):
self.mock_task = MagicMock()
self.mock_reconfigure_compute_resource_task = \
MagicMock(return_value=self.mock_task)
self.mock_cluster = MagicMock(ReconfigureComputeResource_Task=
self.mock_reconfigure_compute_resource_task)
self.mock_cluster_spec = MagicMock()
def test_get_managed_object_name(self):
mock_get_managed_object_name = MagicMock()
with patch('salt.utils.vmware.get_managed_object_name',
mock_get_managed_object_name):
vmware.update_cluster(self.mock_cluster, self.mock_cluster_spec)
mock_get_managed_object_name.assert_called_once_with(self.mock_cluster)
def test_reconfigure_compute_resource_task_call(self):
vmware.update_cluster(self.mock_cluster, self.mock_cluster_spec)
self.mock_reconfigure_compute_resource_task.assert_called_once_with(
self.mock_cluster_spec, modify=True)
def test_reconfigure_compute_resource_raise_vim_fault(self):
exc = vim.fault.VimFault()
exc.msg = 'VimFault msg'
self.mock_cluster.ReconfigureComputeResource_Task = \
MagicMock(side_effect=exc)
with self.assertRaises(VMwareApiError) as excinfo:
vmware.update_cluster(self.mock_cluster, self.mock_cluster_spec)
self.assertEqual(excinfo.exception.strerror, 'VimFault msg')
def test_reconfigure_compute_resource_raise_runtime_fault(self):
exc = vmodl.RuntimeFault()
exc.msg = 'RuntimeFault msg'
self.mock_cluster.ReconfigureComputeResource_Task = \
MagicMock(side_effect=exc)
with self.assertRaises(VMwareRuntimeError) as excinfo:
vmware.update_cluster(self.mock_cluster, self.mock_cluster_spec)
self.assertEqual(excinfo.exception.strerror, 'RuntimeFault msg')
def test_wait_for_task_call(self):
mock_wait_for_task = MagicMock()
with patch('salt.utils.vmware.get_managed_object_name',
MagicMock(return_value='fake_cluster')):
with patch('salt.utils.vmware.wait_for_task', mock_wait_for_task):
vmware.update_cluster(self.mock_cluster,
self.mock_cluster_spec)
mock_wait_for_task.assert_called_once_with(
self.mock_task, 'fake_cluster', 'ClusterUpdateTask')
if __name__ == '__main__':
from integration import run_tests
run_tests(GetClusterTestCase, needs_daemon=False)
run_tests(CreateClusterTestCase, needs_daemon=False)
run_tests(UpdateClusterTestCase, needs_daemon=False)

800
tests/unit/utils/vmware_test/test_common.py Normal file
View file

@ -0,0 +1,800 @@
# -*- coding: utf-8 -*-
'''
:codeauthor: :email:`Alexandru Bleotu <alexandru.bleotu@morganstanley.com>`
Tests for common functions in salt.utils.vmware
'''
# Import python libraries
from __future__ import absolute_import
import logging
# Import Salt testing libraries
from salttesting import TestCase, skipIf
from salttesting.mock import NO_MOCK, NO_MOCK_REASON, patch, MagicMock, call, \
PropertyMock
# Import Salt libraries
import salt.exceptions as excs
import salt.utils.vmware
# Import Third Party Libs
try:
from pyVmomi import vim, vmodl
HAS_PYVMOMI = True
except ImportError:
HAS_PYVMOMI = False
# Get Logging Started
log = logging.getLogger(__name__)
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
@patch('salt.utils.vmware.time.time', MagicMock(return_value=1))
@patch('salt.utils.vmware.time.sleep', MagicMock(return_value=None))
class WaitForTaskTestCase(TestCase):
'''Tests for salt.utils.vmware.wait_for_task'''
def test_first_task_info_raise_vim_fault(self):
exc = vim.fault.VimFault()
exc.msg = 'VimFault msg'
mock_task = MagicMock()
type(mock_task).info = PropertyMock(side_effect=exc)
with self.assertRaises(excs.VMwareApiError) as excinfo:
salt.utils.vmware.wait_for_task(mock_task,
'fake_instance_name',
'task_type')
self.assertEqual(excinfo.exception.strerror, 'VimFault msg')
def test_first_task_info_raise_runtime_fault(self):
exc = vmodl.RuntimeFault()
exc.msg = 'RuntimeFault msg'
mock_task = MagicMock()
type(mock_task).info = PropertyMock(side_effect=exc)
with self.assertRaises(excs.VMwareRuntimeError) as excinfo:
salt.utils.vmware.wait_for_task(mock_task,
'fake_instance_name',
'task_type')
self.assertEqual(excinfo.exception.strerror, 'RuntimeFault msg')
def test_inner_loop_task_info_raise_vim_fault(self):
exc = vim.fault.VimFault()
exc.msg = 'VimFault msg'
mock_task = MagicMock()
mock_info1 = MagicMock()
type(mock_task).info = PropertyMock(
side_effect=[mock_info1, exc])
type(mock_info1).state = PropertyMock(side_effect=['running', 'bad'])
with self.assertRaises(excs.VMwareApiError) as excinfo:
salt.utils.vmware.wait_for_task(mock_task,
'fake_instance_name',
'task_type')
self.assertEqual(excinfo.exception.strerror, 'VimFault msg')
def test_inner_loop_task_info_raise_runtime_fault(self):
exc = vmodl.RuntimeFault()
exc.msg = 'RuntimeFault msg'
mock_task = MagicMock()
mock_info1 = MagicMock()
type(mock_task).info = PropertyMock(
side_effect=[mock_info1, exc])
type(mock_info1).state = PropertyMock(side_effect=['running', 'bad'])
with self.assertRaises(excs.VMwareRuntimeError) as excinfo:
salt.utils.vmware.wait_for_task(mock_task,
'fake_instance_name',
'task_type')
self.assertEqual(excinfo.exception.strerror, 'RuntimeFault msg')
def test_info_state_running(self):
# The 'bad' values are invalid in the while loop
mock_task = MagicMock()
prop_mock_state = PropertyMock(side_effect=['running', 'bad', 'bad',
'success'])
prop_mock_result = PropertyMock()
type(mock_task.info).state = prop_mock_state
type(mock_task.info).result = prop_mock_result
salt.utils.vmware.wait_for_task(mock_task,
'fake_instance_name',
'task_type')
self.assertEqual(prop_mock_state.call_count, 4)
self.assertEqual(prop_mock_result.call_count, 1)
def test_info_state_running_continues_loop(self):
mock_task = MagicMock()
# The 'fake' values are required to match all the lookups and end the
# loop
prop_mock_state = PropertyMock(side_effect=['running', 'fake', 'fake',
'success'])
prop_mock_result = PropertyMock()
type(mock_task.info).state = prop_mock_state
type(mock_task.info).result = prop_mock_result
salt.utils.vmware.wait_for_task(mock_task,
'fake_instance_name',
'task_type')
self.assertEqual(prop_mock_state.call_count, 4)
self.assertEqual(prop_mock_result.call_count, 1)
def test_info_state_queued_continues_loop(self):
mock_task = MagicMock()
# The 'fake' values are required to match all the lookups and end the
# loop
prop_mock_state = PropertyMock(side_effect=['fake', 'queued', 'fake',
'fake', 'success'])
prop_mock_result = PropertyMock()
type(mock_task.info).state = prop_mock_state
type(mock_task.info).result = prop_mock_result
salt.utils.vmware.wait_for_task(mock_task,
'fake_instance_name',
'task_type')
self.assertEqual(prop_mock_state.call_count, 5)
self.assertEqual(prop_mock_result.call_count, 1)
def test_info_state_success(self):
mock_task = MagicMock()
prop_mock_state = PropertyMock(return_value='success')
prop_mock_result = PropertyMock()
type(mock_task.info).state = prop_mock_state
type(mock_task.info).result = prop_mock_result
salt.utils.vmware.wait_for_task(mock_task,
'fake_instance_name',
'task_type')
self.assertEqual(prop_mock_state.call_count, 3)
self.assertEqual(prop_mock_result.call_count, 1)
def test_info_error_exception(self):
mock_task = MagicMock()
prop_mock_state = PropertyMock(return_value='error')
prop_mock_error = PropertyMock(side_effect=Exception('error exc'))
type(mock_task.info).state = prop_mock_state
type(mock_task.info).error = prop_mock_error
with self.assertRaises(Exception) as excinfo:
salt.utils.vmware.wait_for_task(mock_task,
'fake_instance_name',
'task_type')
self.assertEqual(excinfo.exception.message, 'error exc')
def test_info_error_vim_fault(self):
exc = vim.fault.VimFault()
exc.msg = 'VimFault msg'
mock_task = MagicMock()
prop_mock_state = PropertyMock(return_value='error')
prop_mock_error = PropertyMock(side_effect=exc)
type(mock_task.info).state = prop_mock_state
type(mock_task.info).error = prop_mock_error
with self.assertRaises(excs.VMwareApiError) as excinfo:
salt.utils.vmware.wait_for_task(mock_task,
'fake_instance_name',
'task_type')
self.assertEqual(excinfo.exception.strerror, 'VimFault msg')
def test_info_error_system_fault(self):
exc = vmodl.fault.SystemError()
exc.msg = 'SystemError msg'
mock_task = MagicMock()
prop_mock_state = PropertyMock(return_value='error')
prop_mock_error = PropertyMock(side_effect=exc)
type(mock_task.info).state = prop_mock_state
type(mock_task.info).error = prop_mock_error
with self.assertRaises(excs.VMwareSystemError) as excinfo:
salt.utils.vmware.wait_for_task(mock_task,
'fake_instance_name',
'task_type')
self.assertEqual(excinfo.exception.strerror, 'SystemError msg')
def test_info_error_invalid_argument_no_fault_message(self):
exc = vmodl.fault.InvalidArgument()
exc.faultMessage = None
exc.msg = 'InvalidArgumentFault msg'
mock_task = MagicMock()
prop_mock_state = PropertyMock(return_value='error')
prop_mock_error = PropertyMock(side_effect=exc)
type(mock_task.info).state = prop_mock_state
type(mock_task.info).error = prop_mock_error
with self.assertRaises(excs.VMwareApiError) as excinfo:
salt.utils.vmware.wait_for_task(mock_task,
'fake_instance_name',
'task_type')
self.assertEqual(excinfo.exception.strerror,
'InvalidArgumentFault msg')
def test_info_error_invalid_argument_with_fault_message(self):
exc = vmodl.fault.InvalidArgument()
fault_message = vim.LocalizableMessage()
fault_message.message = 'LocalFault msg'
exc.faultMessage = [fault_message]
exc.msg = 'InvalidArgumentFault msg'
mock_task = MagicMock()
prop_mock_state = PropertyMock(return_value='error')
prop_mock_error = PropertyMock(side_effect=exc)
type(mock_task.info).state = prop_mock_state
type(mock_task.info).error = prop_mock_error
with self.assertRaises(excs.VMwareApiError) as excinfo:
salt.utils.vmware.wait_for_task(mock_task,
'fake_instance_name',
'task_type')
self.assertEqual(excinfo.exception.strerror,
'InvalidArgumentFault msg (LocalFault msg)')
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
class GetMorsWithPropertiesTestCase(TestCase):
'''Tests for salt.utils.get_mors_with_properties'''
si = None
obj_type = None
prop_list = None
container_ref = None
traversal_spec = None
def setUp(self):
self.si = MagicMock()
self.obj_type = MagicMock()
self.prop_list = MagicMock()
self.container_ref = MagicMock()
self.traversal_spec = MagicMock()
def test_empty_content(self):
get_content = MagicMock(return_value=[])
with patch('salt.utils.vmware.get_content', get_content):
ret = salt.utils.vmware.get_mors_with_properties(
self.si, self.obj_type, self.prop_list,
self.container_ref, self.traversal_spec)
get_content.assert_called_once_with(
self.si, self.obj_type,
property_list=self.prop_list,
container_ref=self.container_ref,
traversal_spec=self.traversal_spec,
local_properties=False)
self.assertEqual(ret, [])
def test_local_properties_set(self):
obj_mock = MagicMock()
# obj.propSet
propSet_prop = PropertyMock(return_value=[])
type(obj_mock).propSet = propSet_prop
# obj.obj
inner_obj_mock = MagicMock()
obj_prop = PropertyMock(return_value=inner_obj_mock)
type(obj_mock).obj = obj_prop
get_content = MagicMock(return_value=[obj_mock])
with patch('salt.utils.vmware.get_content', get_content):
ret = salt.utils.vmware.get_mors_with_properties(
self.si, self.obj_type, self.prop_list,
self.container_ref, self.traversal_spec,
local_properties=True)
get_content.assert_called_once_with(
self.si, self.obj_type,
property_list=self.prop_list,
container_ref=self.container_ref,
traversal_spec=self.traversal_spec,
local_properties=True)
def test_one_element_content(self):
obj_mock = MagicMock()
# obj.propSet
propSet_prop = PropertyMock(return_value=[])
type(obj_mock).propSet = propSet_prop
# obj.obj
inner_obj_mock = MagicMock()
obj_prop = PropertyMock(return_value=inner_obj_mock)
type(obj_mock).obj = obj_prop
get_content = MagicMock(return_value=[obj_mock])
with patch('salt.utils.vmware.get_content', get_content):
ret = salt.utils.vmware.get_mors_with_properties(
self.si, self.obj_type, self.prop_list,
self.container_ref, self.traversal_spec)
get_content.assert_called_once_with(
self.si, self.obj_type,
property_list=self.prop_list,
container_ref=self.container_ref,
traversal_spec=self.traversal_spec,
local_properties=False)
self.assertEqual(propSet_prop.call_count, 1)
self.assertEqual(obj_prop.call_count, 1)
self.assertEqual(len(ret), 1)
self.assertDictEqual(ret[0], {'object': inner_obj_mock})
def test_multiple_element_content(self):
# obj1
obj1_mock = MagicMock()
# obj1.propSet
obj1_propSet_prop = PropertyMock(return_value=[])
type(obj1_mock).propSet = obj1_propSet_prop
# obj1.obj
obj1_inner_obj_mock = MagicMock()
obj1_obj_prop = PropertyMock(return_value=obj1_inner_obj_mock)
type(obj1_mock).obj = obj1_obj_prop
# obj2
obj2_mock = MagicMock()
# obj2.propSet
obj2_propSet_prop = PropertyMock(return_value=[])
type(obj2_mock).propSet = obj2_propSet_prop
# obj2.obj
obj2_inner_obj_mock = MagicMock()
obj2_obj_prop = PropertyMock(return_value=obj2_inner_obj_mock)
type(obj2_mock).obj = obj2_obj_prop
get_content = MagicMock(return_value=[obj1_mock, obj2_mock])
with patch('salt.utils.vmware.get_content', get_content):
ret = salt.utils.vmware.get_mors_with_properties(
self.si, self.obj_type, self.prop_list,
self.container_ref, self.traversal_spec)
get_content.assert_called_once_with(
self.si, self.obj_type,
property_list=self.prop_list,
container_ref=self.container_ref,
traversal_spec=self.traversal_spec,
local_properties=False)
self.assertEqual(obj1_propSet_prop.call_count, 1)
self.assertEqual(obj2_propSet_prop.call_count, 1)
self.assertEqual(obj1_obj_prop.call_count, 1)
self.assertEqual(obj2_obj_prop.call_count, 1)
self.assertEqual(len(ret), 2)
self.assertDictEqual(ret[0], {'object': obj1_inner_obj_mock})
self.assertDictEqual(ret[1], {'object': obj2_inner_obj_mock})
def test_one_elem_one_property(self):
obj_mock = MagicMock()
# property mock
prop_set_obj_mock = MagicMock()
prop_set_obj_name_prop = PropertyMock(return_value='prop_name')
prop_set_obj_val_prop = PropertyMock(return_value='prop_value')
type(prop_set_obj_mock).name = prop_set_obj_name_prop
type(prop_set_obj_mock).val = prop_set_obj_val_prop
# obj.propSet
propSet_prop = PropertyMock(return_value=[prop_set_obj_mock])
type(obj_mock).propSet = propSet_prop
# obj.obj
inner_obj_mock = MagicMock()
obj_prop = PropertyMock(return_value=inner_obj_mock)
type(obj_mock).obj = obj_prop
get_content = MagicMock(return_value=[obj_mock])
with patch('salt.utils.vmware.get_content', get_content):
ret = salt.utils.vmware.get_mors_with_properties(
self.si, self.obj_type, self.prop_list,
self.container_ref, self.traversal_spec,
local_properties=False)
get_content.assert_called_once_with(
self.si, self.obj_type,
property_list=self.prop_list,
container_ref=self.container_ref,
traversal_spec=self.traversal_spec,
local_properties=False)
self.assertEqual(propSet_prop.call_count, 1)
self.assertEqual(prop_set_obj_name_prop.call_count, 1)
self.assertEqual(prop_set_obj_val_prop.call_count, 1)
self.assertEqual(obj_prop.call_count, 1)
self.assertEqual(len(ret), 1)
self.assertDictEqual(ret[0], {'prop_name': 'prop_value',
'object': inner_obj_mock})
def test_one_elem_multiple_properties(self):
obj_mock = MagicMock()
# property1 mock
prop_set_obj1_mock = MagicMock()
prop_set_obj1_name_prop = PropertyMock(return_value='prop_name1')
prop_set_obj1_val_prop = PropertyMock(return_value='prop_value1')
type(prop_set_obj1_mock).name = prop_set_obj1_name_prop
type(prop_set_obj1_mock).val = prop_set_obj1_val_prop
# property2 mock
prop_set_obj2_mock = MagicMock()
prop_set_obj2_name_prop = PropertyMock(return_value='prop_name2')
prop_set_obj2_val_prop = PropertyMock(return_value='prop_value2')
type(prop_set_obj2_mock).name = prop_set_obj2_name_prop
type(prop_set_obj2_mock).val = prop_set_obj2_val_prop
# obj.propSet
propSet_prop = PropertyMock(return_value=[prop_set_obj1_mock,
prop_set_obj2_mock])
type(obj_mock).propSet = propSet_prop
# obj.obj
inner_obj_mock = MagicMock()
obj_prop = PropertyMock(return_value=inner_obj_mock)
type(obj_mock).obj = obj_prop
get_content = MagicMock(return_value=[obj_mock])
with patch('salt.utils.vmware.get_content', get_content):
ret = salt.utils.vmware.get_mors_with_properties(
self.si, self.obj_type, self.prop_list,
self.container_ref, self.traversal_spec)
get_content.assert_called_once_with(
self.si, self.obj_type,
property_list=self.prop_list,
container_ref=self.container_ref,
traversal_spec=self.traversal_spec,
local_properties=False)
self.assertEqual(propSet_prop.call_count, 1)
self.assertEqual(prop_set_obj1_name_prop.call_count, 1)
self.assertEqual(prop_set_obj1_val_prop.call_count, 1)
self.assertEqual(prop_set_obj2_name_prop.call_count, 1)
self.assertEqual(prop_set_obj2_val_prop.call_count, 1)
self.assertEqual(obj_prop.call_count, 1)
self.assertEqual(len(ret), 1)
self.assertDictEqual(ret[0], {'prop_name1': 'prop_value1',
'prop_name2': 'prop_value2',
'object': inner_obj_mock})
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
@patch('salt.utils.vmware.get_service_instance_from_managed_object',
MagicMock())
@patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=[MagicMock()]))
class GetPropertiesOfManagedObjectTestCase(TestCase):
'''Tests for salt.utils.get_properties_of_managed_object'''
def setUp(self):
self.mock_si = MagicMock()
self.fake_mo_ref = vim.ManagedEntity('Fake')
self.mock_props = MagicMock()
self.mock_item_name = {'name': 'fake_name'}
self.mock_item = MagicMock()
def test_get_service_instance_from_managed_object_call(self):
mock_get_instance_from_managed_object = MagicMock()
with patch(
'salt.utils.vmware.get_service_instance_from_managed_object',
mock_get_instance_from_managed_object):
salt.utils.vmware.get_properties_of_managed_object(
self.fake_mo_ref, self.mock_props)
mock_get_instance_from_managed_object.assert_called_once_with(
self.fake_mo_ref)
def test_get_mors_with_properties_calls(self):
mock_get_mors_with_properties = MagicMock(return_value=[MagicMock()])
with patch(
'salt.utils.vmware.get_service_instance_from_managed_object',
MagicMock(return_value=self.mock_si)):
with patch('salt.utils.vmware.get_mors_with_properties',
mock_get_mors_with_properties):
salt.utils.vmware.get_properties_of_managed_object(
self.fake_mo_ref, self.mock_props)
mock_get_mors_with_properties.assert_has_calls(
[call(self.mock_si, vim.ManagedEntity,
container_ref=self.fake_mo_ref,
property_list=['name'],
local_properties=True),
call(self.mock_si, vim.ManagedEntity,
container_ref=self.fake_mo_ref,
property_list=self.mock_props,
local_properties=True)])
def test_managed_object_no_name_property(self):
with patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(side_effect=[vmodl.query.InvalidProperty(), []])):
with self.assertRaises(excs.VMwareApiError) as excinfo:
salt.utils.vmware.get_properties_of_managed_object(
self.fake_mo_ref, self.mock_props)
self.assertEqual('Properties of managed object \'<unnamed>\' weren\'t '
'retrieved', excinfo.exception.strerror)
def test_no_items_named_object(self):
with patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(side_effect=[[self.mock_item_name], []])):
with self.assertRaises(excs.VMwareApiError) as excinfo:
salt.utils.vmware.get_properties_of_managed_object(
self.fake_mo_ref, self.mock_props)
self.assertEqual('Properties of managed object \'fake_name\' weren\'t '
'retrieved', excinfo.exception.strerror)
@patch('salt.utils.vmware.get_properties_of_managed_object',
MagicMock(return_value={'key': 'value'}))
class GetManagedObjectName(TestCase):
'''Tests for salt.utils.get_managed_object_name'''
def setUp(self):
self.mock_mo_ref = MagicMock()
def test_get_properties_of_managed_object_call(self):
mock_get_properties_of_managed_object = MagicMock()
with patch('salt.utils.vmware.get_properties_of_managed_object',
mock_get_properties_of_managed_object):
salt.utils.vmware.get_managed_object_name(self.mock_mo_ref)
mock_get_properties_of_managed_object.assert_called_once_with(
self.mock_mo_ref, ['name'])
def test_no_name_in_property_dict(self):
ret = salt.utils.vmware.get_managed_object_name(self.mock_mo_ref)
self.assertIsNone(ret)
def test_return_managed_object_name(self):
mock_get_properties_of_managed_object = MagicMock()
with patch('salt.utils.vmware.get_properties_of_managed_object',
MagicMock(return_value={'name': 'fake_name'})):
ret = salt.utils.vmware.get_managed_object_name(self.mock_mo_ref)
self.assertEqual(ret, 'fake_name')
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
@patch('salt.utils.vmware.get_root_folder', MagicMock())
@patch('salt.utils.vmware.vmodl.query.PropertyCollector.TraversalSpec',
MagicMock(return_value=MagicMock()))
@patch('salt.utils.vmware.vmodl.query.PropertyCollector.PropertySpec',
MagicMock(return_value=MagicMock()))
@patch('salt.utils.vmware.vmodl.query.PropertyCollector.ObjectSpec',
MagicMock(return_value=MagicMock()))
@patch('salt.utils.vmware.vmodl.query.PropertyCollector.FilterSpec',
MagicMock(return_value=MagicMock()))
class GetContentTestCase(TestCase):
'''Tests for salt.utils.get_content'''
# Method names to be patched
traversal_spec_method_name = \
'salt.utils.vmware.vmodl.query.PropertyCollector.TraversalSpec'
property_spec_method_name = \
'salt.utils.vmware.vmodl.query.PropertyCollector.PropertySpec'
obj_spec_method_name = \
'salt.utils.vmware.vmodl.query.PropertyCollector.ObjectSpec'
filter_spec_method_name = \
'salt.utils.vmware.vmodl.query.PropertyCollector.FilterSpec'
# Class variables
si_mock = None
root_folder_mock = None
root_folder_prop = None
container_view_mock = None
create_container_view_mock = None
result_mock = None
retrieve_contents_mock = None
destroy_mock = None
obj_type_mock = None
traversal_spec_ret_mock = None
traversal_spec_mock = None
property_spec_ret_mock = None
property_spec_mock = None
obj_spec_ret_mock = None
obj_spec_mock = None
filter_spec_ret_mock = None
filter_spec_mock = None
def setUp(self):
# setup the service instance
self.si_mock = MagicMock()
# RootFolder
self.root_folder_mock = MagicMock()
self.get_root_folder_mock = \
MagicMock(return_value=self.root_folder_mock)
# CreateContainerView()
self.container_view_mock = MagicMock()
self.create_container_view_mock = \
MagicMock(return_value=self.container_view_mock)
self.si_mock.content.viewManager.CreateContainerView = \
self.create_container_view_mock
# RetrieveContents()
self.result_mock = MagicMock()
self.retrieve_contents_mock = MagicMock(return_value=self.result_mock)
self.si_mock.content.propertyCollector.RetrieveContents = \
self.retrieve_contents_mock
# Destroy()
self.destroy_mock = MagicMock()
self.container_view_mock.Destroy = self.destroy_mock
# override mocks
self.obj_type_mock = MagicMock()
self.traversal_spec_ret_mock = MagicMock()
self.traversal_spec_mock = \
MagicMock(return_value=self.traversal_spec_ret_mock)
self.property_spec_ret_mock = MagicMock()
self.property_spec_mock = \
MagicMock(return_value=self.property_spec_ret_mock)
self.obj_spec_ret_mock = MagicMock()
self.obj_spec_mock = \
MagicMock(return_value=self.obj_spec_ret_mock)
self.filter_spec_ret_mock = MagicMock()
self.filter_spec_mock = \
MagicMock(return_value=self.filter_spec_ret_mock)
def test_empty_container_ref(self):
with patch('salt.utils.vmware.get_root_folder',
self.get_root_folder_mock):
salt.utils.vmware.get_content(self.si_mock, self.obj_type_mock)
self.get_root_folder_mock.assert_called_once_with(self.si_mock)
self.create_container_view_mock.assert_called_once_with(
self.root_folder_mock, [self.obj_type_mock], True)
def test_defined_container_ref(self):
container_ref_mock = MagicMock()
with patch('salt.utils.vmware.get_root_folder',
self.get_root_folder_mock):
with patch(self.obj_spec_method_name, self.obj_type_mock):
salt.utils.vmware.get_content(
self.si_mock, self.obj_type_mock,
container_ref=container_ref_mock)
self.assertEqual(self.get_root_folder_mock.call_count, 0)
self.create_container_view_mock.assert_called_once_with(
container_ref_mock, [self.obj_type_mock], True)
# Also checks destroy is called
def test_local_traversal_spec(self):
with patch('salt.utils.vmware.get_root_folder',
self.get_root_folder_mock):
with patch(self.traversal_spec_method_name,
self.traversal_spec_mock):
with patch(self.obj_spec_method_name, self.obj_spec_mock):
ret = salt.utils.vmware.get_content(self.si_mock,
self.obj_type_mock)
self.create_container_view_mock.assert_called_once_with(
self.root_folder_mock, [self.obj_type_mock], True)
self.traversal_spec_mock.assert_called_once_with(
name='traverseEntities', path='view', skip=False,
type=vim.view.ContainerView)
self.obj_spec_mock.assert_called_once_with(
obj=self.container_view_mock,
skip=True,
selectSet=[self.traversal_spec_ret_mock])
# check destroy is called
self.assertEqual(self.destroy_mock.call_count, 1)
def test_create_container_view_raise_vim_fault(self):
exc = vim.fault.VimFault()
exc.msg = 'VimFault msg'
self.si_mock.content.viewManager.CreateContainerView = \
MagicMock(side_effect=exc)
with patch('salt.utils.vmware.get_root_folder',
self.get_root_folder_mock):
with self.assertRaises(excs.VMwareApiError) as excinfo:
salt.utils.vmware.get_content(self.si_mock, self.obj_type_mock)
self.assertEqual(excinfo.exception.strerror, 'VimFault msg')
def test_create_container_view_raise_runtime_fault(self):
exc = vmodl.RuntimeFault()
exc.msg = 'RuntimeFault msg'
self.si_mock.content.viewManager.CreateContainerView = \
MagicMock(side_effect=exc)
with patch('salt.utils.vmware.get_root_folder',
self.get_root_folder_mock):
with self.assertRaises(excs.VMwareRuntimeError) as excinfo:
salt.utils.vmware.get_content(self.si_mock, self.obj_type_mock)
self.assertEqual(excinfo.exception.strerror, 'RuntimeFault msg')
def test_destroy_raise_vim_fault(self):
exc = vim.fault.VimFault()
exc.msg = 'VimFault msg'
self.si_mock.content.viewManager.CreateContainerView = MagicMock(
return_value=MagicMock(Destroy=MagicMock(side_effect=exc)))
with patch('salt.utils.vmware.get_root_folder',
self.get_root_folder_mock):
with self.assertRaises(excs.VMwareApiError) as excinfo:
salt.utils.vmware.get_content(self.si_mock, self.obj_type_mock)
self.assertEqual(excinfo.exception.strerror, 'VimFault msg')
def test_destroy_raise_runtime_fault(self):
exc = vmodl.RuntimeFault()
exc.msg = 'RuntimeFault msg'
self.si_mock.content.viewManager.CreateContainerView = MagicMock(
return_value=MagicMock(Destroy=MagicMock(side_effect=exc)))
with patch('salt.utils.vmware.get_root_folder',
self.get_root_folder_mock):
with self.assertRaises(excs.VMwareRuntimeError) as excinfo:
salt.utils.vmware.get_content(self.si_mock, self.obj_type_mock)
self.assertEqual(excinfo.exception.strerror, 'RuntimeFault msg')
# Also checks destroy is not called
def test_external_traversal_spec(self):
traversal_spec_obj_mock = MagicMock()
with patch('salt.utils.vmware.get_root_folder',
self.get_root_folder_mock):
with patch(self.traversal_spec_method_name,
self.traversal_spec_mock):
with patch(self.obj_spec_method_name, self.obj_spec_mock):
salt.utils.vmware.get_content(
self.si_mock,
self.obj_type_mock,
traversal_spec=traversal_spec_obj_mock)
self.obj_spec_mock.assert_called_once_with(
obj=self.root_folder_mock,
skip=True,
selectSet=[traversal_spec_obj_mock])
# Check local traversal methods are not called
self.assertEqual(self.create_container_view_mock.call_count, 0)
self.assertEqual(self.traversal_spec_mock.call_count, 0)
# check destroy is not called
self.assertEqual(self.destroy_mock.call_count, 0)
def test_property_obj_filter_specs_and_contents(self):
with patch(self.traversal_spec_method_name, self.traversal_spec_mock):
with patch(self.property_spec_method_name, self.property_spec_mock):
with patch(self.obj_spec_method_name, self.obj_spec_mock):
with patch(self.filter_spec_method_name,
self.filter_spec_mock):
ret = salt.utils.vmware.get_content(
self.si_mock,
self.obj_type_mock)
self.traversal_spec_mock.assert_called_once_with(
name='traverseEntities', path='view', skip=False,
type=vim.view.ContainerView)
self.property_spec_mock.assert_called_once_with(
type=self.obj_type_mock, all=True, pathSet=None)
self.obj_spec_mock.assert_called_once_with(
obj=self.container_view_mock, skip=True,
selectSet=[self.traversal_spec_ret_mock])
self.retrieve_contents_mock.assert_called_once_with(
[self.filter_spec_ret_mock])
self.assertEqual(ret, self.result_mock)
def test_retrieve_contents_raise_vim_fault(self):
exc = vim.fault.VimFault()
exc.msg = 'VimFault msg'
self.si_mock.content.propertyCollector.RetrieveContents = \
MagicMock(side_effect=exc)
with self.assertRaises(excs.VMwareApiError) as excinfo:
salt.utils.vmware.get_content(self.si_mock, self.obj_type_mock)
self.assertEqual(excinfo.exception.strerror, 'VimFault msg')
def test_retrieve_contents_raise_runtime_fault(self):
exc = vmodl.RuntimeFault()
exc.msg = 'RuntimeFault msg'
self.si_mock.content.propertyCollector.RetrieveContents = \
MagicMock(side_effect=exc)
with self.assertRaises(excs.VMwareRuntimeError) as excinfo:
salt.utils.vmware.get_content(self.si_mock, self.obj_type_mock)
self.assertEqual(excinfo.exception.strerror, 'RuntimeFault msg')
def test_local_properties_set(self):
container_ref_mock = MagicMock()
with patch(self.traversal_spec_method_name, self.traversal_spec_mock):
with patch(self.property_spec_method_name, self.property_spec_mock):
with patch(self.obj_spec_method_name, self.obj_spec_mock):
salt.utils.vmware.get_content(
self.si_mock,
self.obj_type_mock,
container_ref=container_ref_mock,
local_properties=True)
self.assertEqual(self.traversal_spec_mock.call_count, 0)
self.obj_spec_mock.assert_called_once_with(
obj=container_ref_mock, skip=False, selectSet=None)
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
class GetRootFolderTestCase(TestCase):
'''Tests for salt.utils.get_root_folder'''
def setUp(self):
self.mock_root_folder = MagicMock()
self.mock_content = MagicMock(rootFolder=self.mock_root_folder)
self.mock_si = MagicMock(
RetrieveContent=MagicMock(return_value=self.mock_content))
def test_raise_vim_fault(self):
exc = vim.fault.VimFault()
exc.msg = 'VimFault msg'
type(self.mock_content).rootFolder = PropertyMock(side_effect=exc)
with self.assertRaises(excs.VMwareApiError) as excinfo:
salt.utils.vmware.get_root_folder(self.mock_si)
self.assertEqual(excinfo.exception.strerror, 'VimFault msg')
def test_raise_runtime_fault(self):
exc = vmodl.RuntimeFault()
exc.msg = 'RuntimeFault msg'
type(self.mock_content).rootFolder = PropertyMock(side_effect=exc)
with self.assertRaises(excs.VMwareRuntimeError) as excinfo:
salt.utils.vmware.get_root_folder(self.mock_si)
self.assertEqual(excinfo.exception.strerror, 'RuntimeFault msg')
def test_return(self):
ret = salt.utils.vmware.get_root_folder(self.mock_si)
self.assertEqual(ret, self.mock_root_folder)
if __name__ == '__main__':
from integration import run_tests
run_tests(WaitForTaskTestCase, needs_daemon=False)
run_tests(GetMorsWithPropertiesTestCase, needs_daemon=False)
run_tests(GetPropertiesOfManagedObjectTestCase, needs_daemon=False)
run_tests(GetManagedObjectName, needs_daemon=False)
run_tests(GetContentTestCase, needs_daemon=False)
run_tests(GetRootFolderTestCase, needs_daemon=False)

837
tests/unit/utils/vmware_test/test_connection.py Normal file
View file

@ -0,0 +1,837 @@
# -*- coding: utf-8 -*-
'''
:codeauthor: :email:`Alexandru Bleotu <alexandru.bleotu@morganstanley.com>`
Tests for connections related functions in salt.utils.vmware
'''
# Import python libraries
from __future__ import absolute_import
import logging
import base64
import ssl
import sys
# Import Salt testing libraries
from salttesting import TestCase, skipIf
from salttesting.mock import NO_MOCK, NO_MOCK_REASON, patch, MagicMock, call, \
PropertyMock
import salt.exceptions as excs
# Import Salt libraries
import salt.utils.vmware
# Import Third Party Libs
try:
from pyVmomi import vim, vmodl
HAS_PYVMOMI = True
except ImportError:
HAS_PYVMOMI = False
try:
import gssapi
HAS_GSSAPI = True
except ImportError:
HAS_GSSAPI = False
if sys.version_info[:3] > (2, 7, 8):
SSL_VALIDATION = True
else:
SSL_VALIDATION = False
# Get Logging Started
log = logging.getLogger(__name__)
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
@skipIf(not HAS_GSSAPI, 'The \'gssapi\' library is missing')
@patch('gssapi.Name', MagicMock(return_value='service'))
@patch('gssapi.InitContext', MagicMock())
class GssapiTokenTest(TestCase):
'''
Test cases for salt.utils.vmware.get_gssapi_token
'''
@patch('salt.utils.vmware.HAS_GSSAPI', False)
def test_no_gssapi(self):
with self.assertRaises(ImportError) as excinfo:
salt.utils.vmware.get_gssapi_token('principal', 'host', 'domain')
self.assertIn('The gssapi library is not imported.',
excinfo.exception.message)
@skipIf(not HAS_GSSAPI, 'The \'gssapi\' library is missing')
def test_service_name(self):
mock_name = MagicMock()
with patch.object(salt.utils.vmware.gssapi, 'Name', mock_name):
with self.assertRaises(excs.CommandExecutionError):
salt.utils.vmware.get_gssapi_token('principal', 'host',
'domain')
mock_name.assert_called_once_with('principal/host@domain',
gssapi.C_NT_USER_NAME)
@skipIf(not HAS_GSSAPI, 'The \'gssapi\' library is missing')
def test_out_token_defined(self):
mock_context = MagicMock(return_value=MagicMock())
mock_context.return_value.established = False
mock_context.return_value.step = MagicMock(return_value='out_token')
with patch.object(salt.utils.vmware.gssapi, 'InitContext',
mock_context):
ret = salt.utils.vmware.get_gssapi_token('principal', 'host',
'domain')
self.assertEqual(mock_context.return_value.step.called, 1)
self.assertEqual(ret, base64.b64encode('out_token'))
@skipIf(not HAS_GSSAPI, 'The \'gssapi\' library is missing')
def test_out_token_undefined(self):
mock_context = MagicMock(return_value=MagicMock())
mock_context.return_value.established = False
mock_context.return_value.step = MagicMock(return_value=None)
with patch.object(salt.utils.vmware.gssapi, 'InitContext',
mock_context):
with self.assertRaises(excs.CommandExecutionError) as excinfo:
salt.utils.vmware.get_gssapi_token('principal', 'host',
'domain')
self.assertEqual(mock_context.return_value.step.called, 1)
self.assertIn('Can\'t receive token',
excinfo.exception.strerror)
@skipIf(not HAS_GSSAPI, 'The \'gssapi\' library is missing')
def test_context_extablished(self):
mock_context = MagicMock(return_value=MagicMock())
mock_context.return_value.established = True
mock_context.return_value.step = MagicMock(return_value='out_token')
with patch.object(salt.utils.vmware.gssapi, 'InitContext',
mock_context):
mock_context.established = True
mock_context.step = MagicMock(return_value=None)
with self.assertRaises(excs.CommandExecutionError) as excinfo:
salt.utils.vmware.get_gssapi_token('principal', 'host',
'domain')
self.assertEqual(mock_context.step.called, 0)
self.assertIn('Context established, but didn\'t receive token',
excinfo.exception.strerror)
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
@patch('salt.utils.vmware.SmartConnect', MagicMock())
@patch('salt.utils.vmware.Disconnect', MagicMock())
@patch('salt.utils.vmware.get_gssapi_token',
MagicMock(return_value='fake_token'))
class PrivateGetServiceInstanceTestCase(TestCase):
'''Tests for salt.utils.vmware._get_service_instance'''
def test_invalid_mechianism(self):
with self.assertRaises(excs.CommandExecutionError) as excinfo:
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='invalid_mechanism',
principal='fake principal',
domain='fake_domain')
self.assertIn('Unsupported mechanism', excinfo.exception.strerror)
def test_userpass_mechanism_empty_username(self):
with self.assertRaises(excs.CommandExecutionError) as excinfo:
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username=None,
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='userpass',
principal='fake principal',
domain='fake_domain')
self.assertIn('mandatory parameter \'username\'',
excinfo.exception.strerror)
def test_userpass_mechanism_empty_password(self):
with self.assertRaises(excs.CommandExecutionError) as excinfo:
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='fake_username',
password=None,
protocol='fake_protocol',
port=1,
mechanism='userpass',
principal='fake principal',
domain='fake_domain')
self.assertIn('mandatory parameter \'password\'',
excinfo.exception.strerror)
def test_userpass_mechanism_no_domain(self):
mock_sc = MagicMock()
with patch('salt.utils.vmware.SmartConnect', mock_sc):
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='userpass',
principal='fake principal',
domain=None)
mock_sc.assert_called_once_with(
host='fake_host.fqdn',
user='fake_username',
pwd='fake_password',
protocol='fake_protocol',
port=1,
b64token=None,
mechanism='userpass')
def test_userpass_mech_domain_unused(self):
mock_sc = MagicMock()
with patch('salt.utils.vmware.SmartConnect', mock_sc):
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='fake_username@domain',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='userpass',
principal='fake principal',
domain='fake_domain')
mock_sc.assert_called_once_with(
host='fake_host.fqdn',
user='fake_username@domain',
pwd='fake_password',
protocol='fake_protocol',
port=1,
b64token=None,
mechanism='userpass')
mock_sc.reset_mock()
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='domain\\fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='userpass',
principal='fake principal',
domain='fake_domain')
mock_sc.assert_called_once_with(
host='fake_host.fqdn',
user='domain\\fake_username',
pwd='fake_password',
protocol='fake_protocol',
port=1,
b64token=None,
mechanism='userpass')
def test_sspi_empty_principal(self):
with self.assertRaises(excs.CommandExecutionError) as excinfo:
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='sspi',
principal=None,
domain='fake_domain')
self.assertIn('mandatory parameters are missing',
excinfo.exception.strerror)
def test_sspi_empty_domain(self):
with self.assertRaises(excs.CommandExecutionError) as excinfo:
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='sspi',
principal='fake_principal',
domain=None)
self.assertIn('mandatory parameters are missing',
excinfo.exception.strerror)
def test_sspi_get_token_error(self):
mock_token = MagicMock(side_effect=Exception('Exception'))
with patch('salt.utils.vmware.get_gssapi_token', mock_token):
with self.assertRaises(excs.VMwareConnectionError) as excinfo:
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='sspi',
principal='fake_principal',
domain='fake_domain')
mock_token.assert_called_once_with('fake_principal',
'fake_host.fqdn',
'fake_domain')
self.assertEqual('Exception', excinfo.exception.strerror)
def test_sspi_get_token_success_(self):
mock_token = MagicMock(return_value='fake_token')
mock_sc = MagicMock()
with patch('salt.utils.vmware.get_gssapi_token', mock_token):
with patch('salt.utils.vmware.SmartConnect', mock_sc):
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='sspi',
principal='fake_principal',
domain='fake_domain')
mock_token.assert_called_once_with('fake_principal',
'fake_host.fqdn',
'fake_domain')
mock_sc.assert_called_once_with(
host='fake_host.fqdn',
user='fake_username',
pwd='fake_password',
protocol='fake_protocol',
port=1,
b64token='fake_token',
mechanism='sspi')
def test_first_attempt_successful_connection(self):
mock_sc = MagicMock()
with patch('salt.utils.vmware.SmartConnect', mock_sc):
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='sspi',
principal='fake_principal',
domain='fake_domain')
mock_sc.assert_called_once_with(
host='fake_host.fqdn',
user='fake_username',
pwd='fake_password',
protocol='fake_protocol',
port=1,
b64token='fake_token',
mechanism='sspi')
@skipIf(not SSL_VALIDATION, 'SSL validation is not enabled')
@patch('ssl.SSLContext', MagicMock())
@patch('ssl._create_unverified_context', MagicMock())
def test_second_attempt_successful_connection(self):
exc = vim.fault.HostConnectFault()
exc.msg = '[SSL: CERTIFICATE_VERIFY_FAILED]'
mock_sc = MagicMock(side_effect=[exc, None])
mock_ssl = MagicMock()
with patch('salt.utils.vmware.SmartConnect', mock_sc):
with patch('ssl._create_unverified_context',
mock_ssl):
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='sspi',
principal='fake_principal',
domain='fake_domain')
mock_ssl.assert_called_once_with()
calls = [call(host='fake_host.fqdn',
user='fake_username',
pwd='fake_password',
protocol='fake_protocol',
port=1,
b64token='fake_token',
mechanism='sspi'),
call(host='fake_host.fqdn',
user='fake_username',
pwd='fake_password',
protocol='fake_protocol',
port=1,
sslContext=mock_ssl.return_value,
b64token='fake_token',
mechanism='sspi')]
mock_sc.assert_has_calls(calls)
@skipIf(not SSL_VALIDATION, 'SSL validation is not enabled')
@patch('ssl.SSLContext', MagicMock())
@patch('ssl._create_unverified_context', MagicMock())
def test_third_attempt_successful_connection(self):
exc = vim.fault.HostConnectFault()
exc.msg = '[SSL: CERTIFICATE_VERIFY_FAILED]'
exc2 = Exception('certificate verify failed')
mock_sc = MagicMock(side_effect=[exc, exc2, None])
mock_ssl_unverif = MagicMock()
mock_ssl_context = MagicMock()
with patch('salt.utils.vmware.SmartConnect', mock_sc):
with patch('ssl._create_unverified_context',
mock_ssl_unverif):
with patch('ssl.SSLContext', mock_ssl_context):
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='sspi',
principal='fake_principal',
domain='fake_domain')
mock_ssl_context.assert_called_once_with(ssl.PROTOCOL_TLSv1)
mock_ssl_unverif.assert_called_once_with()
calls = [call(host='fake_host.fqdn',
user='fake_username',
pwd='fake_password',
protocol='fake_protocol',
port=1,
b64token='fake_token',
mechanism='sspi'),
call(host='fake_host.fqdn',
user='fake_username',
pwd='fake_password',
protocol='fake_protocol',
port=1,
sslContext=mock_ssl_unverif.return_value,
b64token='fake_token',
mechanism='sspi'),
call(host='fake_host.fqdn',
user='fake_username',
pwd='fake_password',
protocol='fake_protocol',
port=1,
sslContext=mock_ssl_context.return_value,
b64token='fake_token',
mechanism='sspi'),
]
mock_sc.assert_has_calls(calls)
def test_first_attempt_unsuccessful_connection_default_error(self):
exc = Exception('Exception')
mock_sc = MagicMock(side_effect=exc)
with patch('salt.utils.vmware.SmartConnect', mock_sc):
with self.assertRaises(excs.VMwareConnectionError) as excinfo:
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='sspi',
principal='fake_principal',
domain='fake_domain')
self.assertEqual(mock_sc.call_count, 1)
self.assertIn('Could not connect to host \'fake_host.fqdn\'',
excinfo.Exception.message)
def test_first_attempt_unsuccessful_connection_vim_fault(self):
exc = vim.fault.VimFault()
exc.msg = 'VimFault'
mock_sc = MagicMock(side_effect=exc)
with patch('salt.utils.vmware.SmartConnect', mock_sc):
with self.assertRaises(excs.VMwareConnectionError) as excinfo:
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='sspi',
principal='fake_principal',
domain='fake_domain')
self.assertEqual(mock_sc.call_count, 1)
self.assertEqual('VimFault', excinfo.Exception.message)
@skipIf(not SSL_VALIDATION, 'SSL validation is not enabled')
@patch('ssl.SSLContext', MagicMock())
@patch('ssl._create_unverified_context', MagicMock())
def test_second_attempt_unsuccsessful_connection_default_error(self):
exc = vim.fault.HostConnectFault()
exc.msg = '[SSL: CERTIFICATE_VERIFY_FAILED]'
exc2 = Exception('Exception')
mock_sc = MagicMock(side_effect=[exc, exc2])
with patch('salt.utils.vmware.SmartConnect', mock_sc):
with self.assertRaises(excs.VMwareConnectionError) as excinfo:
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='sspi',
principal='fake_principal',
domain='fake_domain')
self.assertEqual(mock_sc.call_count, 2)
self.assertIn('Could not connect to host \'fake_host.fqdn\'',
excinfo.Exception.message)
@skipIf(not SSL_VALIDATION, 'SSL validation is not enabled')
@patch('ssl.SSLContext', MagicMock())
@patch('ssl._create_unverified_context', MagicMock())
def test_second_attempt_unsuccsessful_connection_vim_fault(self):
exc = vim.fault.HostConnectFault()
exc.msg = '[SSL: CERTIFICATE_VERIFY_FAILED]'
exc2 = vim.fault.VimFault()
exc2.msg = 'VimFault'
mock_sc = MagicMock(side_effect=[exc, exc2])
with patch('salt.utils.vmware.SmartConnect', mock_sc):
with self.assertRaises(excs.VMwareConnectionError) as excinfo:
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='sspi',
principal='fake_principal',
domain='fake_domain')
self.assertEqual(mock_sc.call_count, 2)
self.assertIn('VimFault', excinfo.Exception.message)
@skipIf(not SSL_VALIDATION, 'SSL validation is not enabled')
@patch('ssl.SSLContext', MagicMock())
@patch('ssl._create_unverified_context', MagicMock())
def test_third_attempt_unsuccessful_connection_detault_error(self):
exc = vim.fault.HostConnectFault()
exc.msg = '[SSL: CERTIFICATE_VERIFY_FAILED]'
exc2 = Exception('certificate verify failed')
exc3 = Exception('Exception')
mock_sc = MagicMock(side_effect=[exc, exc2, exc3])
with patch('salt.utils.vmware.SmartConnect', mock_sc):
with self.assertRaises(excs.VMwareConnectionError) as excinfo:
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='sspi',
principal='fake_principal',
domain='fake_domain')
self.assertEqual(mock_sc.call_count, 3)
self.assertIn('Exception', excinfo.Exception.message)
@skipIf(not SSL_VALIDATION, 'SSL validation is not enabled')
@patch('ssl.SSLContext', MagicMock())
@patch('ssl._create_unverified_context', MagicMock())
def test_third_attempt_unsuccessful_connection_vim_fault(self):
exc = vim.fault.HostConnectFault()
exc.msg = '[SSL: CERTIFICATE_VERIFY_FAILED]'
exc2 = Exception('certificate verify failed')
exc3 = vim.fault.VimFault()
exc3.msg = 'VimFault'
mock_sc = MagicMock(side_effect=[exc, exc2, exc3])
with patch('salt.utils.vmware.SmartConnect', mock_sc):
with self.assertRaises(excs.VMwareConnectionError) as excinfo:
salt.utils.vmware._get_service_instance(
host='fake_host.fqdn',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='sspi',
principal='fake_principal',
domain='fake_domain')
self.assertEqual(mock_sc.call_count, 3)
self.assertIn('VimFault', excinfo.Exception.message)
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
@patch('salt.utils.vmware.GetSi', MagicMock(return_value=None))
@patch('salt.utils.vmware._get_service_instance',
MagicMock(return_value=MagicMock()))
class GetServiceInstanceTestCase(TestCase):
'''Tests for salt.utils.vmware.get_service_instance'''
def test_default_params(self):
mock_get_si = MagicMock()
with patch('salt.utils.vmware._get_service_instance', mock_get_si):
salt.utils.vmware.get_service_instance(
host='fake_host'
)
mock_get_si.assert_called_once_with('fake_host', None, None,
'https', 443, 'userpass', None,
None)
@patch('salt.utils.is_proxy', MagicMock(return_value=True))
def test_no_cached_service_instance_same_host_on_proxy(self):
# Service instance is uncached when using class default mock objs
mock_get_si = MagicMock()
with patch('salt.utils.vmware._get_service_instance', mock_get_si):
salt.utils.vmware.get_service_instance(
host='fake_host',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='fake_mechanism',
principal='fake_principal',
domain='fake_domain'
)
mock_get_si.assert_called_once_with('fake_host',
'fake_username',
'fake_password',
'fake_protocol',
1,
'fake_mechanism',
'fake_principal',
'fake_domain')
def test_cached_service_instance_different_host(self):
mock_si = MagicMock()
mock_si_stub = MagicMock()
mock_disconnect = MagicMock()
mock_get_si = MagicMock(return_value=mock_si)
mock_getstub = MagicMock()
with patch('salt.utils.vmware.GetSi', mock_get_si):
with patch('salt.utils.vmware.GetStub', mock_getstub):
with patch('salt.utils.vmware.Disconnect', mock_disconnect):
salt.utils.vmware.get_service_instance(
host='fake_host',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='fake_mechanism',
principal='fake_principal',
domain='fake_domain'
)
self.assertEqual(mock_get_si.call_count, 1)
self.assertEqual(mock_getstub.call_count, 1)
self.assertEqual(mock_disconnect.call_count, 1)
def test_uncached_service_instance(self):
# Service instance is uncached when using class default mock objs
mock_get_si = MagicMock()
with patch('salt.utils.vmware._get_service_instance', mock_get_si):
salt.utils.vmware.get_service_instance(
host='fake_host',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='fake_mechanism',
principal='fake_principal',
domain='fake_domain'
)
mock_get_si.assert_called_once_with('fake_host',
'fake_username',
'fake_password',
'fake_protocol',
1,
'fake_mechanism',
'fake_principal',
'fake_domain')
def test_unauthenticated_service_instance(self):
mock_si_current_time = MagicMock(side_effect=vim.fault.NotAuthenticated)
mock_si = MagicMock()
mock_get_si = MagicMock(return_value=mock_si)
mock_si.CurrentTime = mock_si_current_time
mock_disconnect = MagicMock()
with patch('salt.utils.vmware._get_service_instance', mock_get_si):
with patch('salt.utils.vmware.Disconnect', mock_disconnect):
salt.utils.vmware.get_service_instance(
host='fake_host',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='fake_mechanism',
principal='fake_principal',
domain='fake_domain'
)
self.assertEqual(mock_si_current_time.call_count, 1)
self.assertEqual(mock_disconnect.call_count, 1)
self.assertEqual(mock_get_si.call_count, 2)
def test_current_time_raise_vim_fault(self):
exc = vim.fault.VimFault()
exc.msg = 'VimFault msg'
with patch('salt.utils.vmware._get_service_instance',
MagicMock(return_value=MagicMock(
CurrentTime=MagicMock(side_effect=exc)))):
with self.assertRaises(excs.VMwareApiError) as excinfo:
salt.utils.vmware.get_service_instance(
host='fake_host',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='fake_mechanism',
principal='fake_principal',
domain='fake_domain')
self.assertEqual(excinfo.exception.strerror, 'VimFault msg')
def test_current_time_raise_runtime_fault(self):
exc = vmodl.RuntimeFault()
exc.msg = 'RuntimeFault msg'
with patch('salt.utils.vmware._get_service_instance',
MagicMock(return_value=MagicMock(
CurrentTime=MagicMock(side_effect=exc)))):
with self.assertRaises(excs.VMwareRuntimeError) as excinfo:
salt.utils.vmware.get_service_instance(
host='fake_host',
username='fake_username',
password='fake_password',
protocol='fake_protocol',
port=1,
mechanism='fake_mechanism',
principal='fake_principal',
domain='fake_domain')
self.assertEqual(excinfo.exception.strerror, 'RuntimeFault msg')
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
class DisconnectTestCase(TestCase):
'''Tests for salt.utils.vmware.disconnect'''
def setUp(self):
self.mock_si = MagicMock()
def test_disconnect(self):
mock_disconnect = MagicMock()
with patch('salt.utils.vmware.Disconnect', mock_disconnect):
salt.utils.vmware.disconnect(
service_instance=self.mock_si)
mock_disconnect.assert_called_once_with(self.mock_si)
def test_disconnect_raise_vim_fault(self):
exc = vim.fault.VimFault()
exc.msg = 'VimFault msg'
with patch('salt.utils.vmware.Disconnect', MagicMock(side_effect=exc)):
with self.assertRaises(excs.VMwareApiError) as excinfo:
salt.utils.vmware.disconnect(
service_instance=self.mock_si)
self.assertEqual(excinfo.exception.strerror, 'VimFault msg')
def test_disconnect_raise_runtime_fault(self):
exc = vmodl.RuntimeFault()
exc.msg = 'RuntimeFault msg'
with patch('salt.utils.vmware.Disconnect', MagicMock(side_effect=exc)):
with self.assertRaises(excs.VMwareRuntimeError) as excinfo:
salt.utils.vmware.disconnect(
service_instance=self.mock_si)
self.assertEqual(excinfo.exception.strerror, 'RuntimeFault msg')
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
class IsConnectionToAVCenterTestCase(TestCase):
'''Tests for salt.utils.vmware.is_connection_to_a_vcenter'''
def test_api_type_raise_vim_fault(self):
exc = vim.fault.VimFault()
exc.msg = 'VimFault msg'
mock_si = MagicMock()
type(mock_si.content.about).apiType = PropertyMock(side_effect=exc)
with self.assertRaises(excs.VMwareApiError) as excinfo:
salt.utils.vmware.is_connection_to_a_vcenter(mock_si)
self.assertEqual(excinfo.exception.strerror, 'VimFault msg')
def test_api_type_raise_runtime_fault(self):
exc = vmodl.RuntimeFault()
exc.msg = 'RuntimeFault msg'
mock_si = MagicMock()
type(mock_si.content.about).apiType = PropertyMock(side_effect=exc)
with self.assertRaises(excs.VMwareRuntimeError) as excinfo:
salt.utils.vmware.is_connection_to_a_vcenter(mock_si)
self.assertEqual(excinfo.exception.strerror, 'RuntimeFault msg')
def test_connected_to_a_vcenter(self):
mock_si = MagicMock()
mock_si.content.about.apiType = 'VirtualCenter'
ret = salt.utils.vmware.is_connection_to_a_vcenter(mock_si)
self.assertTrue(ret)
def test_connected_to_a_host(self):
mock_si = MagicMock()
mock_si.content.about.apiType = 'HostAgent'
ret = salt.utils.vmware.is_connection_to_a_vcenter(mock_si)
self.assertFalse(ret)
def test_connected_to_invalid_entity(self):
mock_si = MagicMock()
mock_si.content.about.apiType = 'UnsupportedType'
with self.assertRaises(excs.VMwareApiError) as excinfo:
salt.utils.vmware.is_connection_to_a_vcenter(mock_si)
self.assertIn('Unexpected api type \'UnsupportedType\'',
excinfo.exception.strerror)
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
@patch('salt.utils.vmware.vim.ServiceInstance', MagicMock())
class GetServiceInstanceFromManagedObjectTestCase(TestCase):
'''Tests for salt.utils.vmware.get_managed_instance_from_managed_object'''
def setUp(self):
self.mock_si = MagicMock()
self.mock_stub = PropertyMock()
self.mock_mo_ref = MagicMock(_stub=self.mock_stub)
def test_default_name_parameter(self):
mock_trace = MagicMock()
type(salt.utils.vmware.log).trace = mock_trace
salt.utils.vmware.get_service_instance_from_managed_object(
self.mock_mo_ref)
mock_trace.assert_called_once_with('[<unnamed>] Retrieving service '
'instance from managed object')
def test_name_parameter_passed_in(self):
mock_trace = MagicMock()
type(salt.utils.vmware.log).trace = mock_trace
salt.utils.vmware.get_service_instance_from_managed_object(
self.mock_mo_ref, 'fake_mo_name')
mock_trace.assert_called_once_with('[fake_mo_name] Retrieving service '
'instance from managed object')
def test_service_instance_instantiation(self):
mock_service_instance_ini = MagicMock()
with patch('salt.utils.vmware.vim.ServiceInstance',
mock_service_instance_ini):
salt.utils.vmware.get_service_instance_from_managed_object(
self.mock_mo_ref)
mock_service_instance_ini.assert_called_once_with('ServiceInstance')
def test_si_return_and_stub_assignment(self):
with patch('salt.utils.vmware.vim.ServiceInstance',
MagicMock(return_value=self.mock_si)):
ret = salt.utils.vmware.get_service_instance_from_managed_object(
self.mock_mo_ref)
self.assertEqual(ret, self.mock_si)
self.assertEqual(ret._stub, self.mock_stub)
if __name__ == '__main__':
from integration import run_tests
run_tests(GssapiTokenTest, needs_daemon=False)
run_tests(PrivateGetServiceInstanceTestCase, needs_daemon=False)
run_tests(GetServiceInstanceTestCase, needs_daemon=False)
run_tests(DisconnectTestCase, needs_daemon=False)
run_tests(IsConnectionToAVCenterTestCase, needs_daemon=False)
run_tests(GetServiceInstanceFromManagedObjectTestCase, needs_daemon=False)

183
tests/unit/utils/vmware_test/test_datacenter.py Normal file
View file

@ -0,0 +1,183 @@
# -*- coding: utf-8 -*-
'''
:codeauthor: :email:`Alexandru Bleotu <alexandru.bleotu@morganstanley.com>`
Tests for datacenter related functions in salt.utils.vmware
'''
# Import python libraries
from __future__ import absolute_import
import logging
# Import Salt testing libraries
from salttesting import TestCase, skipIf
from salttesting.mock import NO_MOCK, NO_MOCK_REASON, patch, MagicMock
from salt.exceptions import VMwareObjectRetrievalError, VMwareApiError, \
VMwareRuntimeError
# Import Salt libraries
import salt.utils.vmware as vmware
# Import Third Party Libs
try:
from pyVmomi import vim, vmodl
HAS_PYVMOMI = True
except ImportError:
HAS_PYVMOMI = False
# Get Logging Started
log = logging.getLogger(__name__)
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
@patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=[{'name': 'fake_dc', 'object': MagicMock()}]))
class GetDatacentersTestCase(TestCase):
'''Tests for salt.utils.vmware.get_datacenters'''
def setUp(self):
self.mock_si = MagicMock()
self.mock_dc1 = MagicMock()
self.mock_dc2 = MagicMock()
self.mock_entries = [{'name': 'fake_dc1',
'object': self.mock_dc1},
{'name': 'fake_dc2',
'object': self.mock_dc2}]
def test_get_mors_with_properties_call(self):
mock_get_mors_with_properties = MagicMock(
return_value=[{'name': 'fake_dc', 'object': MagicMock()}])
with patch('salt.utils.vmware.get_mors_with_properties',
mock_get_mors_with_properties):
vmware.get_datacenters(self.mock_si, datacenter_names=['fake_dc1'])
mock_get_mors_with_properties.assert_called_once_with(
self.mock_si, vim.Datacenter, property_list=['name'])
def test_get_mors_with_properties_returns_empty_array(self):
with patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=[])):
res = vmware.get_datacenters(self.mock_si,
datacenter_names=['fake_dc1'])
self.assertEqual(res, [])
def test_no_parameters(self):
with patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=self.mock_entries)):
res = vmware.get_datacenters(self.mock_si)
self.assertEqual(res, [])
def test_datastore_not_found(self):
with patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=self.mock_entries)):
res = vmware.get_datacenters(self.mock_si,
datacenter_names=['fake_dc'])
self.assertEqual(res, [])
def test_datastore_found(self):
with patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=self.mock_entries)):
res = vmware.get_datacenters(
self.mock_si, datacenter_names=['fake_dc2'])
self.assertEqual(res, [self.mock_dc2])
def test_get_all_datastores(self):
with patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=self.mock_entries)):
res = vmware.get_datacenters(
self.mock_si, get_all_datacenters=True)
self.assertEqual(res, [self.mock_dc1, self.mock_dc2])
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
@patch('salt.utils.vmware.get_datacenters',
MagicMock(return_value=[MagicMock()]))
class GetDatacenterTestCase(TestCase):
'''Tests for salt.utils.vmware.get_datacenter'''
def setUp(self):
self.mock_si = MagicMock()
self.mock_dc = MagicMock()
def test_get_datacenters_call(self):
mock_get_datacenters = MagicMock(return_value=[MagicMock()])
with patch('salt.utils.vmware.get_datacenters',
mock_get_datacenters):
vmware.get_datacenter(self.mock_si, 'fake_dc1')
mock_get_datacenters.assert_called_once_with(
self.mock_si, datacenter_names=['fake_dc1'])
def test_no_datacenters_returned(self):
with patch('salt.utils.vmware.get_datacenters',
MagicMock(return_value=[])):
with self.assertRaises(VMwareObjectRetrievalError) as excinfo:
vmware.get_datacenter(self.mock_si, 'fake_dc1')
self.assertEqual('Datacenter \'fake_dc1\' was not found',
excinfo.exception.strerror)
def test_get_datacenter_return(self):
with patch('salt.utils.vmware.get_datacenters',
MagicMock(return_value=[self.mock_dc])):
res = vmware.get_datacenter(self.mock_si, 'fake_dc1')
self.assertEqual(res, self.mock_dc)
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
@patch('salt.utils.vmware.get_root_folder', MagicMock())
class CreateDatacenterTestCase(TestCase):
'''Tests for salt.utils.vmware.create_datacenter'''
def setUp(self):
self.mock_si = MagicMock()
self.mock_dc = MagicMock()
self.mock_create_datacenter = MagicMock(return_value=self.mock_dc)
self.mock_root_folder = MagicMock(
CreateDatacenter=self.mock_create_datacenter)
def test_get_root_folder(self):
mock_get_root_folder = MagicMock()
with patch('salt.utils.vmware.get_root_folder', mock_get_root_folder):
vmware.create_datacenter(self.mock_si, 'fake_dc')
mock_get_root_folder.assert_called_once_with(self.mock_si)
def test_create_datacenter_call(self):
with patch('salt.utils.vmware.get_root_folder',
MagicMock(return_value=self.mock_root_folder)):
vmware.create_datacenter(self.mock_si, 'fake_dc')
self.mock_create_datacenter.assert_called_once_with('fake_dc')
def test_create_datacenter_raise_vim_fault(self):
exc = vim.VimFault()
exc.msg = 'VimFault msg'
self.mock_root_folder = MagicMock(
CreateDatacenter=MagicMock(side_effect=exc))
with patch('salt.utils.vmware.get_root_folder',
MagicMock(return_value=self.mock_root_folder)):
with self.assertRaises(VMwareApiError) as excinfo:
vmware.create_datacenter(self.mock_si, 'fake_dc')
self.assertEqual(excinfo.exception.strerror, 'VimFault msg')
def test_create_datacenter_runtime_fault(self):
exc = vmodl.RuntimeFault()
exc.msg = 'RuntimeFault msg'
self.mock_root_folder = MagicMock(
CreateDatacenter=MagicMock(side_effect=exc))
with patch('salt.utils.vmware.get_root_folder',
MagicMock(return_value=self.mock_root_folder)):
with self.assertRaises(VMwareRuntimeError) as excinfo:
vmware.create_datacenter(self.mock_si, 'fake_dc')
self.assertEqual(excinfo.exception.strerror, 'RuntimeFault msg')
def test_datastore_successfully_created(self):
with patch('salt.utils.vmware.get_root_folder',
MagicMock(return_value=self.mock_root_folder)):
res = vmware.create_datacenter(self.mock_si, 'fake_dc')
self.assertEqual(res, self.mock_dc)
if __name__ == '__main__':
from integration import run_tests
run_tests(GetDatacentersTestCase, needs_daemon=False)
run_tests(GetDatacenterTestCase, needs_daemon=False)
run_tests(CreateDatacenterTestCase, needs_daemon=False)

157
tests/unit/utils/vmware_test/test_host.py Normal file
View file

@ -0,0 +1,157 @@
# -*- coding: utf-8 -*-
'''
:codeauthor: :email:`Alexandru Bleotu <alexandru.bleotu@morganstanley.com>`
Tests for host functions in salt.utils.vmware
'''
# Import python libraries
from __future__ import absolute_import
import logging
# Import Salt testing libraries
from salttesting import TestCase, skipIf
from salttesting.mock import NO_MOCK, NO_MOCK_REASON, patch, MagicMock
# Import Salt libraries
import salt.utils.vmware
# Import Third Party Libs
try:
from pyVmomi import vim
HAS_PYVMOMI = True
except ImportError:
HAS_PYVMOMI = False
# Get Logging Started
log = logging.getLogger(__name__)
@skipIf(NO_MOCK, NO_MOCK_REASON)
@skipIf(not HAS_PYVMOMI, 'The \'pyvmomi\' library is missing')
@patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=[]))
@patch('salt.utils.vmware.get_datacenter',
MagicMock(return_value=None))
@patch('salt.utils.vmware.get_cluster',
MagicMock(return_value=None))
class GetHostsTestCase(TestCase):
'''Tests for salt.utils.vmware.get_hosts'''
def setUp(self):
self.mock_root_folder = MagicMock()
self.mock_si = MagicMock()
self.mock_host1, self.mock_host2, self.mock_host3 = MagicMock(), \
MagicMock(), MagicMock()
self.mock_prop_host1 = {'name': 'fake_hostname1',
'object': self.mock_host1}
self.mock_prop_host2 = {'name': 'fake_hostname2',
'object': self.mock_host2}
self.mock_prop_host3 = {'name': 'fake_hostname3',
'object': self.mock_host3}
self.mock_prop_hosts = [self.mock_prop_host1, self.mock_prop_host2,
self.mock_prop_host3]
def test_get_si_no_datacenter_no_cluster(self):
mock_get_mors = MagicMock()
mock_get_root_folder = MagicMock(return_value=self.mock_root_folder)
with patch('salt.utils.vmware.get_root_folder', mock_get_root_folder):
with patch('salt.utils.vmware.get_mors_with_properties',
mock_get_mors):
salt.utils.vmware.get_hosts(self.mock_si)
mock_get_root_folder.assert_called_once_with(self.mock_si)
mock_get_mors.assert_called_once_with(
self.mock_si, vim.HostSystem, container_ref=self.mock_root_folder,
property_list=['name'])
def test_get_si_datacenter_name_no_cluster_name(self):
mock_dc = MagicMock()
mock_get_dc = MagicMock(return_value=mock_dc)
mock_get_mors = MagicMock()
with patch('salt.utils.vmware.get_datacenter', mock_get_dc):
with patch('salt.utils.vmware.get_mors_with_properties',
mock_get_mors):
salt.utils.vmware.get_hosts(self.mock_si,
datacenter_name='fake_datacenter')
mock_get_dc.assert_called_once_with(self.mock_si, 'fake_datacenter')
mock_get_mors.assert_called_once_with(self.mock_si,
vim.HostSystem,
container_ref=mock_dc,
property_list=['name'])
def test_get_si_datacenter_name_and_cluster_name(self):
mock_dc = MagicMock()
mock_get_dc = MagicMock(return_value=mock_dc)
mock_get_cl = MagicMock()
mock_get_mors = MagicMock()
with patch('salt.utils.vmware.get_datacenter', mock_get_dc):
with patch('salt.utils.vmware.get_cluster', mock_get_cl):
with patch('salt.utils.vmware.get_mors_with_properties',
mock_get_mors):
salt.utils.vmware.get_hosts(
self.mock_si, datacenter_name='fake_datacenter',
cluster_name='fake_cluster')
mock_get_dc.assert_called_once_with(self.mock_si, 'fake_datacenter')
mock_get_cl.assert_called_once_with(mock_dc, 'fake_cluster')
mock_get_mors.assert_called_once_with(self.mock_si,
vim.HostSystem,
container_ref=mock_dc,
property_list=['name', 'parent'])
def test_host_get_all_hosts(self):
with patch('salt.utils.vmware.get_root_folder',
MagicMock(return_value=self.mock_root_folder)):
with patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=self.mock_prop_hosts)):
res = salt.utils.vmware.get_hosts(self.mock_si, get_all_hosts=True)
self.assertEqual(res, [self.mock_host1, self.mock_host2,
self.mock_host3])
def test_filter_hostname(self):
with patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=self.mock_prop_hosts)):
res = salt.utils.vmware.get_hosts(self.mock_si,
host_names=['fake_hostname1',
'fake_hostname2'])
self.assertEqual(res, [self.mock_host1, self.mock_host2])
def test_get_all_host_flag_not_set_and_no_host_names(self):
with patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=self.mock_prop_hosts)):
res = salt.utils.vmware.get_hosts(self.mock_si)
self.assertEqual(res, [])
def test_filter_cluster(self):
cluster1 = vim.ClusterComputeResource('fake_good_cluster')
cluster2 = vim.ClusterComputeResource('fake_bad_cluster')
# Mock cluster1.name and cluster2.name
cluster1._stub = MagicMock(InvokeAccessor=MagicMock(
return_value='fake_good_cluster'))
cluster2._stub = MagicMock(InvokeAccessor=MagicMock(
return_value='fake_bad_cluster'))
self.mock_prop_host1['parent'] = cluster2
self.mock_prop_host2['parent'] = cluster1
self.mock_prop_host3['parent'] = cluster1
with patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=self.mock_prop_hosts)):
res = salt.utils.vmware.get_hosts(self.mock_si,
datacenter_name='fake_datacenter',
cluster_name='fake_good_cluster',
get_all_hosts=True)
self.assertEqual(res, [self.mock_host2, self.mock_host3])
def test_no_hosts(self):
with patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=[])):
res = salt.utils.vmware.get_hosts(self.mock_si, get_all_hosts=True)
self.assertEqual(res, [])
def test_one_host_returned(self):
with patch('salt.utils.vmware.get_mors_with_properties',
MagicMock(return_value=[self.mock_prop_host1])):
res = salt.utils.vmware.get_hosts(self.mock_si, get_all_hosts=True)
self.assertEqual(res, [self.mock_host1])
if __name__ == '__main__':
from integration import run_tests
run_tests(GetHostsTestCase, needs_daemon=False)