mirror of
https://github.com/valitydev/salt.git
synced 2024-11-07 08:58:59 +00:00
Remove old vmware unit tests
This commit is contained in:
Nitin Madhok
parent
7c144888da
commit
034ef30f7c
@ -1,474 +0,0 @@
|
||||
# -*- 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, \
|
||||
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)
|
||||
@ -1,702 +0,0 @@
|
||||
# -*- 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
|
||||
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)
|
||||
Loading…
Reference in New Issue
Block a user