diff --git a/tests/unit/utils/vmware_test/common_test.py b/tests/unit/utils/vmware_test/common_test.py deleted file mode 100644 index 75394cbf43..0000000000 --- a/tests/unit/utils/vmware_test/common_test.py +++ /dev/null @@ -1,474 +0,0 @@ -# -*- coding: utf-8 -*- -''' - :codeauthor: :email:`Alexandru Bleotu ` - - 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, \ - PropertyMock - -# 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.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_info_state_running(self): - mock_task = MagicMock() - # The 'bad' values are invalid in the while loop - 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_queued(self): - mock_task = MagicMock() - # The 'bad' values are invalid in the while loop - prop_mock_state = PropertyMock(side_effect=['bad', 'queued', '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, 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_state_different_no_error_attr(self): - mock_task = MagicMock() - # The 'bad' values are invalid in the while loop - 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(prop_mock_state.call_count, 3) - self.assertEqual(prop_mock_error.call_count, 1) - self.assertEqual('error exc', excinfo.exception.message) - - -@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.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.root_folder_prop = PropertyMock(return_value=self.root_folder_mock) - type(self.si_mock.content).rootFolder = self.root_folder_prop - # 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): - ret = salt.utils.vmware.get_content(self.si_mock, self.obj_type_mock) - self.assertEqual(self.root_folder_prop.call_count, 1) - 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(self.obj_spec_method_name, self.obj_type_mock): - ret = salt.utils.vmware.get_content( - self.si_mock, self.obj_type_mock, - container_ref=container_ref_mock) - self.assertEqual(self.root_folder_prop.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(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) - - # Also checks destroy is not called - def test_external_traversal_spec(self): - traversal_spec_obj_mock = MagicMock() - 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, - 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_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) - - -if __name__ == '__main__': - from integration import run_tests - run_tests(WaitForTaskTestCase, needs_daemon=False) - run_tests(GetMorsWithPropertiesTestCase, needs_daemon=False) - run_tests(GetContentTestCase, needs_daemon=False) diff --git a/tests/unit/utils/vmware_test/connection_test.py b/tests/unit/utils/vmware_test/connection_test.py deleted file mode 100644 index 6ef8ca6604..0000000000 --- a/tests/unit/utils/vmware_test/connection_test.py +++ /dev/null @@ -1,702 +0,0 @@ -# -*- coding: utf-8 -*- -''' - :codeauthor: :email:`Alexandru Bleotu ` - - 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 -import salt.exceptions as excs - -# Import Salt libraries -import salt.utils.vmware -# Import Third Party Libs -try: - from pyVmomi import vim - 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) - - -@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_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) - -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(IsConnectionToAVCenterTestCase, needs_daemon=False)