# -*- coding: utf-8 -*- """ :codeauthor: :email:`Alexandru Bleotu ` Tests functions in salt.utils.vsan """ # Import python libraries from __future__ import absolute_import, print_function, unicode_literals import logging # Import Salt libraries from salt.exceptions import ( VMwareApiError, VMwareObjectRetrievalError, VMwareRuntimeError, ) from salt.utils import vsan # Import Salt testing libraries from tests.support.mixins import LoaderModuleMockMixin from tests.support.mock import MagicMock, PropertyMock, patch from tests.support.unit import TestCase, skipIf try: from pyVmomi import vim, vmodl # pylint: disable=no-name-in-module HAS_PYVMOMI = True except ImportError: HAS_PYVMOMI = False HAS_PYVSAN = vsan.HAS_PYVSAN # Get Logging Started log = logging.getLogger(__name__) @skipIf(not HAS_PYVMOMI, "The 'pyvmomi' library is missing") @skipIf(not HAS_PYVSAN, "The 'vsan' ext library is missing") class VsanSupportedTestCase(TestCase): """Tests for salt.utils.vsan.vsan_supported""" def test_supported_api_version(self): mock_si = MagicMock(content=MagicMock(about=MagicMock())) type(mock_si.content.about).apiVersion = PropertyMock(return_value="6.0") self.assertTrue(vsan.vsan_supported(mock_si)) def test_unsupported_api_version(self): mock_si = MagicMock(content=MagicMock(about=MagicMock())) type(mock_si.content.about).apiVersion = PropertyMock(return_value="5.0") self.assertFalse(vsan.vsan_supported(mock_si)) def test_api_version_raises_no_permission(self): exc = vim.fault.NoPermission() exc.privilegeId = "Fake privilege" mock_si = MagicMock(content=MagicMock(about=MagicMock())) type(mock_si.content.about).apiVersion = PropertyMock(side_effect=exc) with self.assertRaises(VMwareApiError) as excinfo: vsan.vsan_supported(mock_si) self.assertEqual( excinfo.exception.strerror, "Not enough permissions. Required privilege: " "Fake privilege", ) def test_api_version_raises_vim_fault(self): exc = vim.fault.VimFault() exc.msg = "VimFault msg" mock_si = MagicMock(content=MagicMock(about=MagicMock())) type(mock_si.content.about).apiVersion = PropertyMock(side_effect=exc) with self.assertRaises(VMwareApiError) as excinfo: vsan.vsan_supported(mock_si) self.assertEqual(excinfo.exception.strerror, "VimFault msg") def test_api_version_raises_runtime_fault(self): exc = vmodl.RuntimeFault() exc.msg = "RuntimeFault msg" mock_si = MagicMock(content=MagicMock(about=MagicMock())) type(mock_si.content.about).apiVersion = PropertyMock(side_effect=exc) with self.assertRaises(VMwareRuntimeError) as excinfo: vsan.vsan_supported(mock_si) self.assertEqual(excinfo.exception.strerror, "RuntimeFault msg") @skipIf(not HAS_PYVMOMI, "The 'pyvmomi' library is missing") @skipIf(not HAS_PYVSAN, "The 'vsan' ext library is missing") class GetVsanClusterConfigSystemTestCase(TestCase, LoaderModuleMockMixin): """Tests for salt.utils.vsan.get_vsan_cluster_config_system""" def setup_loader_modules(self): return { vsan: { "__virtual__": MagicMock(return_value="vsan"), "sys": MagicMock(), "ssl": MagicMock(), } } def setUp(self): self.mock_si = MagicMock() self.mock_ret = MagicMock() patches = ( ( "salt.utils.vsan.vsanapiutils.GetVsanVcMos", MagicMock(return_value={"vsan-cluster-config-system": self.mock_ret}), ), ) for mod, mock in patches: patcher = patch(mod, mock) patcher.start() self.addCleanup(patcher.stop) type(vsan.sys).version_info = PropertyMock(return_value=(2, 7, 9)) self.mock_context = MagicMock() self.mock_create_default_context = MagicMock(return_value=self.mock_context) vsan.ssl.create_default_context = self.mock_create_default_context def tearDown(self): for attr in ( "mock_si", "mock_ret", "mock_context", "mock_create_default_context", ): delattr(self, attr) def test_ssl_default_context_loaded(self): vsan.get_vsan_cluster_config_system(self.mock_si) self.mock_create_default_context.assert_called_once_with() self.assertFalse(self.mock_context.check_hostname) self.assertEqual(self.mock_context.verify_mode, vsan.ssl.CERT_NONE) def test_ssl_default_context_not_loaded(self): type(vsan.sys).version_info = PropertyMock(return_value=(2, 7, 8)) vsan.get_vsan_cluster_config_system(self.mock_si) self.assertEqual(self.mock_create_default_context.call_count, 0) def test_GetVsanVcMos_call(self): mock_get_vsan_vc_mos = MagicMock() with patch("salt.utils.vsan.vsanapiutils.GetVsanVcMos", mock_get_vsan_vc_mos): vsan.get_vsan_cluster_config_system(self.mock_si) mock_get_vsan_vc_mos.assert_called_once_with( self.mock_si._stub, context=self.mock_context ) def test_return(self): ret = vsan.get_vsan_cluster_config_system(self.mock_si) self.assertEqual(ret, self.mock_ret) @skipIf(not HAS_PYVMOMI, "The 'pyvmomi' library is missing") @skipIf(not HAS_PYVSAN, "The 'pyvsan' bindings are missing") class GetVsanDiskManagementSystemTestCase(TestCase, LoaderModuleMockMixin): """Tests for salt.utils.vsan.get_vsan_disk_management_system""" def setup_loader_modules(self): return { vsan: { "__virtual__": MagicMock(return_value="vsan"), "sys": MagicMock(), "ssl": MagicMock(), } } def setUp(self): self.mock_si = MagicMock() self.mock_ret = MagicMock() patches = ( ( "salt.utils.vsan.vsanapiutils.GetVsanVcMos", MagicMock(return_value={"vsan-disk-management-system": self.mock_ret}), ), ) for mod, mock in patches: patcher = patch(mod, mock) patcher.start() self.addCleanup(patcher.stop) type(vsan.sys).version_info = PropertyMock(return_value=(2, 7, 9)) self.mock_context = MagicMock() self.mock_create_default_context = MagicMock(return_value=self.mock_context) vsan.ssl.create_default_context = self.mock_create_default_context def tearDown(self): for attr in ( "mock_si", "mock_ret", "mock_context", "mock_create_default_context", ): delattr(self, attr) def test_ssl_default_context_loaded(self): vsan.get_vsan_disk_management_system(self.mock_si) self.mock_create_default_context.assert_called_once_with() self.assertFalse(self.mock_context.check_hostname) self.assertEqual(self.mock_context.verify_mode, vsan.ssl.CERT_NONE) def test_ssl_default_context_not_loaded(self): type(vsan.sys).version_info = PropertyMock(return_value=(2, 7, 8)) vsan.get_vsan_disk_management_system(self.mock_si) self.assertEqual(self.mock_create_default_context.call_count, 0) def test_GetVsanVcMos_call(self): mock_get_vsan_vc_mos = MagicMock() with patch("salt.utils.vsan.vsanapiutils.GetVsanVcMos", mock_get_vsan_vc_mos): vsan.get_vsan_disk_management_system(self.mock_si) mock_get_vsan_vc_mos.assert_called_once_with( self.mock_si._stub, context=self.mock_context ) def test_return(self): ret = vsan.get_vsan_disk_management_system(self.mock_si) self.assertEqual(ret, self.mock_ret) @skipIf(not HAS_PYVMOMI, "The 'pyvmomi' library is missing") @skipIf(not HAS_PYVSAN, "The 'vsan' ext library is missing") class GetHostVsanSystemTestCase(TestCase): """Tests for salt.utils.vsan.get_host_vsan_system""" def setUp(self): self.mock_host_ref = MagicMock() self.mock_si = MagicMock() self.mock_traversal_spec = MagicMock() self.mock_vsan_system = MagicMock() patches = ( ( "salt.utils.vmware.get_managed_object_name", MagicMock(return_value="fake_hostname"), ), ( "salt.utils.vsan.vmodl.query.PropertyCollector.TraversalSpec", MagicMock(return_value=self.mock_traversal_spec), ), ( "salt.utils.vmware.get_mors_with_properties", MagicMock(return_value=self.mock_traversal_spec), ), ( "salt.utils.vmware.get_mors_with_properties", MagicMock(return_value=[{"object": self.mock_vsan_system}]), ), ) for mod, mock in patches: patcher = patch(mod, mock) patcher.start() self.addCleanup(patcher.stop) def test_get_hostname(self): mock_get_managed_object_name = MagicMock(return_value="fake_hostname") with patch( "salt.utils.vmware.get_managed_object_name", mock_get_managed_object_name ): vsan.get_host_vsan_system(self.mock_si, self.mock_host_ref) mock_get_managed_object_name.assert_called_once_with(self.mock_host_ref) def test_hostname_argument(self): mock_get_managed_object_name = MagicMock(return_value="fake_hostname") with patch( "salt.utils.vmware.get_managed_object_name", MagicMock(return_value="fake_hostname"), ): vsan.get_host_vsan_system( self.mock_si, self.mock_host_ref, hostname="passedin_hostname" ) self.assertEqual(mock_get_managed_object_name.call_count, 0) def test_traversal_spec(self): mock_traversal_spec = MagicMock(return_value=self.mock_traversal_spec) with patch( "salt.utils.vmware.vmodl.query.PropertyCollector.TraversalSpec", mock_traversal_spec, ): vsan.get_host_vsan_system(self.mock_si, self.mock_host_ref) mock_traversal_spec.assert_called_once_with( path="configManager.vsanSystem", type=vim.HostSystem, skip=False ) def test_get_mors_with_properties(self): mock_get_mors = MagicMock(return_value=[{"object": self.mock_vsan_system}]) with patch("salt.utils.vmware.get_mors_with_properties", mock_get_mors): vsan.get_host_vsan_system(self.mock_si, self.mock_host_ref) mock_get_mors.assert_called_once_with( self.mock_si, vim.HostVsanSystem, property_list=["config.enabled"], container_ref=self.mock_host_ref, traversal_spec=self.mock_traversal_spec, ) def test_empty_mors_result(self): mock_get_mors = MagicMock(return_value=None) with patch("salt.utils.vmware.get_mors_with_properties", mock_get_mors): with self.assertRaises(VMwareObjectRetrievalError) as excinfo: vsan.get_host_vsan_system(self.mock_si, self.mock_host_ref) self.assertEqual( excinfo.exception.strerror, "Host's 'fake_hostname' VSAN system was " "not retrieved", ) def test_valid_mors_result(self): res = vsan.get_host_vsan_system(self.mock_si, self.mock_host_ref) self.assertEqual(res, self.mock_vsan_system) @skipIf(not HAS_PYVMOMI, "The 'pyvmomi' library is missing") @skipIf(not HAS_PYVSAN, "The 'vsan' ext library is missing") class CreateDiskgroupTestCase(TestCase): """Tests for salt.utils.vsan.create_diskgroup""" def setUp(self): self.mock_si = MagicMock() self.mock_task = MagicMock() self.mock_initialise_disk_mapping = MagicMock(return_value=self.mock_task) self.mock_vsan_disk_mgmt_system = MagicMock( InitializeDiskMappings=self.mock_initialise_disk_mapping ) self.mock_host_ref = MagicMock() self.mock_cache_disk = MagicMock() self.mock_cap_disk1 = MagicMock() self.mock_cap_disk2 = MagicMock() self.mock_spec = MagicMock() patches = ( ( "salt.utils.vmware.get_managed_object_name", MagicMock(return_value="fake_hostname"), ), ( "salt.utils.vsan.vim.VimVsanHostDiskMappingCreationSpec", MagicMock(return_value=self.mock_spec), ), ("salt.utils.vsan._wait_for_tasks", MagicMock()), ) for mod, mock in patches: patcher = patch(mod, mock) patcher.start() self.addCleanup(patcher.stop) def test_get_hostname(self): mock_get_managed_object_name = MagicMock(return_value="fake_hostname") with patch( "salt.utils.vmware.get_managed_object_name", mock_get_managed_object_name ): vsan.create_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_cache_disk, [self.mock_cap_disk1, self.mock_cap_disk2], ) mock_get_managed_object_name.assert_called_once_with(self.mock_host_ref) def test_vsan_spec_all_flash(self): self.mock_cap_disk1.ssd = True vsan.create_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_cache_disk, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertEqual( self.mock_spec.capacityDisks, [self.mock_cap_disk1, self.mock_cap_disk2] ) self.assertEqual(self.mock_spec.cacheDisks, [self.mock_cache_disk]) self.assertEqual(self.mock_spec.creationType, "allFlash") self.assertEqual(self.mock_spec.host, self.mock_host_ref) def test_vsan_spec_hybrid(self): self.mock_cap_disk1.ssd = False vsan.create_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_cache_disk, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.mock_cap_disk1.ssd = False self.assertEqual(self.mock_spec.creationType, "hybrid") def test_initialize_disk_mapping(self): vsan.create_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_cache_disk, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.mock_initialise_disk_mapping.assert_called_once_with(self.mock_spec) def test_initialize_disk_mapping_raise_no_permission(self): err = vim.fault.NoPermission() err.privilegeId = "Fake privilege" self.mock_vsan_disk_mgmt_system.InitializeDiskMappings = MagicMock( side_effect=err ) with self.assertRaises(VMwareApiError) as excinfo: vsan.create_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_cache_disk, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertEqual( excinfo.exception.strerror, "Not enough permissions. Required privilege: " "Fake privilege", ) def test_initialize_disk_mapping_raise_vim_fault(self): err = vim.fault.VimFault() err.msg = "vim_fault" self.mock_vsan_disk_mgmt_system.InitializeDiskMappings = MagicMock( side_effect=err ) with self.assertRaises(VMwareApiError) as excinfo: vsan.create_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_cache_disk, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertEqual(excinfo.exception.strerror, "vim_fault") def test_initialize_disk_mapping_raise_method_not_found(self): err = vmodl.fault.MethodNotFound() err.method = "fake_method" self.mock_vsan_disk_mgmt_system.InitializeDiskMappings = MagicMock( side_effect=err ) with self.assertRaises(VMwareRuntimeError) as excinfo: vsan.create_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_cache_disk, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertEqual(excinfo.exception.strerror, "Method 'fake_method' not found") def test_initialize_disk_mapping_raise_runtime_fault(self): err = vmodl.RuntimeFault() err.msg = "runtime_fault" self.mock_vsan_disk_mgmt_system.InitializeDiskMappings = MagicMock( side_effect=err ) with self.assertRaises(VMwareRuntimeError) as excinfo: vsan.create_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_cache_disk, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertEqual(excinfo.exception.strerror, "runtime_fault") def test__wait_for_tasks(self): mock___wait_for_tasks = MagicMock() with patch("salt.utils.vsan._wait_for_tasks", mock___wait_for_tasks): vsan.create_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_cache_disk, [self.mock_cap_disk1, self.mock_cap_disk2], ) mock___wait_for_tasks.assert_called_once_with([self.mock_task], self.mock_si) def test_result(self): res = vsan.create_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_cache_disk, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertTrue(res) @skipIf(not HAS_PYVMOMI, "The 'pyvmomi' library is missing") @skipIf(not HAS_PYVSAN, "The 'vsan' ext library is missing") class AddCapacityToDiskGroupTestCase(TestCase): """Tests for salt.utils.vsan.add_capacity_to_diskgroup""" def setUp(self): self.mock_si = MagicMock() self.mock_task = MagicMock() self.mock_initialise_disk_mapping = MagicMock(return_value=self.mock_task) self.mock_vsan_disk_mgmt_system = MagicMock( InitializeDiskMappings=self.mock_initialise_disk_mapping ) self.mock_host_ref = MagicMock() self.mock_cache_disk = MagicMock() self.mock_diskgroup = MagicMock(ssd=self.mock_cache_disk) self.mock_cap_disk1 = MagicMock() self.mock_cap_disk2 = MagicMock() self.mock_spec = MagicMock() patches = ( ( "salt.utils.vmware.get_managed_object_name", MagicMock(return_value="fake_hostname"), ), ( "salt.utils.vsan.vim.VimVsanHostDiskMappingCreationSpec", MagicMock(return_value=self.mock_spec), ), ("salt.utils.vsan._wait_for_tasks", MagicMock()), ) for mod, mock in patches: patcher = patch(mod, mock) patcher.start() self.addCleanup(patcher.stop) def test_get_hostname(self): mock_get_managed_object_name = MagicMock(return_value="fake_hostname") with patch( "salt.utils.vmware.get_managed_object_name", mock_get_managed_object_name ): vsan.add_capacity_to_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) mock_get_managed_object_name.assert_called_once_with(self.mock_host_ref) def test_vsan_spec_all_flash(self): self.mock_cap_disk1.ssd = True vsan.add_capacity_to_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertEqual( self.mock_spec.capacityDisks, [self.mock_cap_disk1, self.mock_cap_disk2] ) self.assertEqual(self.mock_spec.cacheDisks, [self.mock_cache_disk]) self.assertEqual(self.mock_spec.creationType, "allFlash") self.assertEqual(self.mock_spec.host, self.mock_host_ref) def test_vsan_spec_hybrid(self): self.mock_cap_disk1.ssd = False vsan.add_capacity_to_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.mock_cap_disk1.ssd = False self.assertEqual(self.mock_spec.creationType, "hybrid") def test_initialize_disk_mapping(self): vsan.add_capacity_to_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.mock_initialise_disk_mapping.assert_called_once_with(self.mock_spec) def test_initialize_disk_mapping_raise_no_permission(self): err = vim.fault.NoPermission() err.privilegeId = "Fake privilege" self.mock_vsan_disk_mgmt_system.InitializeDiskMappings = MagicMock( side_effect=err ) with self.assertRaises(VMwareApiError) as excinfo: vsan.add_capacity_to_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertEqual( excinfo.exception.strerror, "Not enough permissions. Required privilege: " "Fake privilege", ) def test_initialize_disk_mapping_raise_vim_fault(self): err = vim.fault.VimFault() err.msg = "vim_fault" self.mock_vsan_disk_mgmt_system.InitializeDiskMappings = MagicMock( side_effect=err ) with self.assertRaises(VMwareApiError) as excinfo: vsan.add_capacity_to_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertEqual(excinfo.exception.strerror, "vim_fault") def test_initialize_disk_mapping_raise_method_not_found(self): err = vmodl.fault.MethodNotFound() err.method = "fake_method" self.mock_vsan_disk_mgmt_system.InitializeDiskMappings = MagicMock( side_effect=err ) with self.assertRaises(VMwareRuntimeError) as excinfo: vsan.add_capacity_to_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertEqual(excinfo.exception.strerror, "Method 'fake_method' not found") def test_initialize_disk_mapping_raise_runtime_fault(self): err = vmodl.RuntimeFault() err.msg = "runtime_fault" self.mock_vsan_disk_mgmt_system.InitializeDiskMappings = MagicMock( side_effect=err ) with self.assertRaises(VMwareRuntimeError) as excinfo: vsan.add_capacity_to_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertEqual(excinfo.exception.strerror, "runtime_fault") def test__wait_for_tasks(self): mock___wait_for_tasks = MagicMock() with patch("salt.utils.vsan._wait_for_tasks", mock___wait_for_tasks): vsan.add_capacity_to_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) mock___wait_for_tasks.assert_called_once_with([self.mock_task], self.mock_si) def test_result(self): res = vsan.add_capacity_to_diskgroup( self.mock_si, self.mock_vsan_disk_mgmt_system, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertTrue(res) @skipIf(not HAS_PYVMOMI, "The 'pyvmomi' library is missing") @skipIf(not HAS_PYVSAN, "The 'vsan' ext library is missing") class RemoveCapacityFromDiskGroup(TestCase): """Tests for salt.utils.vsan.remove_capacity_from_diskgroup""" def setUp(self): self.mock_si = MagicMock() self.mock_task = MagicMock() self.mock_remove_disk = MagicMock(return_value=self.mock_task) self.mock_host_vsan_system = MagicMock(RemoveDisk_Task=self.mock_remove_disk) self.mock_host_ref = MagicMock() self.mock_cache_disk = MagicMock() self.mock_diskgroup = MagicMock(ssd=self.mock_cache_disk) self.mock_cap_disk1 = MagicMock() self.mock_cap_disk2 = MagicMock() self.mock_spec = MagicMock() patches = ( ( "salt.utils.vmware.get_managed_object_name", MagicMock(return_value="fake_hostname"), ), ( "salt.utils.vsan.get_host_vsan_system", MagicMock(return_value=self.mock_host_vsan_system), ), ( "salt.utils.vsan.vim.HostMaintenanceSpec", MagicMock(return_value=self.mock_spec), ), ("salt.utils.vsan.vim.VsanHostDecommissionMode", MagicMock()), ("salt.utils.vmware.wait_for_task", MagicMock()), ) for mod, mock in patches: patcher = patch(mod, mock) patcher.start() self.addCleanup(patcher.stop) def test_get_hostname(self): mock_get_managed_object_name = MagicMock(return_value="fake_hostname") with patch( "salt.utils.vmware.get_managed_object_name", mock_get_managed_object_name ): vsan.remove_capacity_from_diskgroup( self.mock_si, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) mock_get_managed_object_name.assert_called_once_with(self.mock_host_ref) def test_maintenance_mode_evacuate_all_data(self): vsan.remove_capacity_from_diskgroup( self.mock_si, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertEqual( self.mock_spec.vsanMode.objectAction, vim.VsanHostDecommissionModeObjectAction.evacuateAllData, ) def test_maintenance_mode_no_action(self): vsan.remove_capacity_from_diskgroup( self.mock_si, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], data_evacuation=False, ) self.assertEqual( self.mock_spec.vsanMode.objectAction, vim.VsanHostDecommissionModeObjectAction.noAction, ) def test_remove_disk(self): vsan.remove_capacity_from_diskgroup( self.mock_si, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.mock_remove_disk.assert_called_once_with( disk=[self.mock_cap_disk1, self.mock_cap_disk2], maintenanceSpec=self.mock_spec, ) def test_remove_disk_raise_no_permission(self): err = vim.fault.NoPermission() err.privilegeId = "Fake privilege" self.mock_host_vsan_system.RemoveDisk_Task = MagicMock(side_effect=err) with self.assertRaises(VMwareApiError) as excinfo: vsan.remove_capacity_from_diskgroup( self.mock_si, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertEqual( excinfo.exception.strerror, "Not enough permissions. Required privilege: " "Fake privilege", ) def test_remove_disk_raise_vim_fault(self): err = vim.fault.VimFault() err.msg = "vim_fault" self.mock_host_vsan_system.RemoveDisk_Task = MagicMock(side_effect=err) with self.assertRaises(VMwareApiError) as excinfo: vsan.remove_capacity_from_diskgroup( self.mock_si, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertEqual(excinfo.exception.strerror, "vim_fault") def test_remove_disk_raise_runtime_fault(self): err = vmodl.RuntimeFault() err.msg = "runtime_fault" self.mock_host_vsan_system.RemoveDisk_Task = MagicMock(side_effect=err) with self.assertRaises(VMwareRuntimeError) as excinfo: vsan.remove_capacity_from_diskgroup( self.mock_si, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertEqual(excinfo.exception.strerror, "runtime_fault") def test_wait_for_tasks(self): mock_wait_for_task = MagicMock() with patch("salt.utils.vmware.wait_for_task", mock_wait_for_task): vsan.remove_capacity_from_diskgroup( self.mock_si, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) mock_wait_for_task.assert_called_once_with( self.mock_task, "fake_hostname", "remove_capacity" ) def test_result(self): res = vsan.remove_capacity_from_diskgroup( self.mock_si, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertTrue(res) @skipIf(not HAS_PYVMOMI, "The 'pyvmomi' library is missing") @skipIf(not HAS_PYVSAN, "The 'vsan' ext library is missing") class RemoveDiskgroup(TestCase): """Tests for salt.utils.vsan.remove_diskgroup""" def setUp(self): self.mock_si = MagicMock() self.mock_task = MagicMock() self.mock_remove_disk_mapping = MagicMock(return_value=self.mock_task) self.mock_host_vsan_system = MagicMock( RemoveDiskMapping_Task=self.mock_remove_disk_mapping ) self.mock_host_ref = MagicMock() self.mock_cache_disk = MagicMock() self.mock_diskgroup = MagicMock(ssd=self.mock_cache_disk) self.mock_cap_disk1 = MagicMock() self.mock_cap_disk2 = MagicMock() self.mock_spec = MagicMock() patches = ( ( "salt.utils.vmware.get_managed_object_name", MagicMock(return_value="fake_hostname"), ), ( "salt.utils.vsan.get_host_vsan_system", MagicMock(return_value=self.mock_host_vsan_system), ), ( "salt.utils.vsan.vim.HostMaintenanceSpec", MagicMock(return_value=self.mock_spec), ), ("salt.utils.vsan.vim.VsanHostDecommissionMode", MagicMock()), ("salt.utils.vmware.wait_for_task", MagicMock()), ) for mod, mock in patches: patcher = patch(mod, mock) patcher.start() self.addCleanup(patcher.stop) def test_get_hostname(self): mock_get_managed_object_name = MagicMock(return_value="fake_hostname") with patch( "salt.utils.vmware.get_managed_object_name", mock_get_managed_object_name ): vsan.remove_diskgroup(self.mock_si, self.mock_host_ref, self.mock_diskgroup) mock_get_managed_object_name.assert_called_once_with(self.mock_host_ref) def test_maintenance_mode_evacuate_all_data(self): vsan.remove_diskgroup(self.mock_si, self.mock_host_ref, self.mock_diskgroup) vsan.remove_capacity_from_diskgroup( self.mock_si, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.assertEqual( self.mock_spec.vsanMode.objectAction, vim.VsanHostDecommissionModeObjectAction.evacuateAllData, ) def test_maintenance_mode_no_action(self): vsan.remove_diskgroup(self.mock_si, self.mock_host_ref, self.mock_diskgroup) vsan.remove_capacity_from_diskgroup( self.mock_si, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], data_evacuation=False, ) self.assertEqual( self.mock_spec.vsanMode.objectAction, vim.VsanHostDecommissionModeObjectAction.noAction, ) def test_remove_disk_mapping(self): vsan.remove_diskgroup(self.mock_si, self.mock_host_ref, self.mock_diskgroup) vsan.remove_capacity_from_diskgroup( self.mock_si, self.mock_host_ref, self.mock_diskgroup, [self.mock_cap_disk1, self.mock_cap_disk2], ) self.mock_remove_disk_mapping.assert_called_once_with( mapping=[self.mock_diskgroup], maintenanceSpec=self.mock_spec ) def test_remove_disk_mapping_raise_no_permission(self): vsan.remove_diskgroup(self.mock_si, self.mock_host_ref, self.mock_diskgroup) err = vim.fault.NoPermission() err.privilegeId = "Fake privilege" self.mock_host_vsan_system.RemoveDiskMapping_Task = MagicMock(side_effect=err) with self.assertRaises(VMwareApiError) as excinfo: vsan.remove_diskgroup(self.mock_si, self.mock_host_ref, self.mock_diskgroup) self.assertEqual( excinfo.exception.strerror, "Not enough permissions. Required privilege: " "Fake privilege", ) def test_remove_disk_mapping_raise_vim_fault(self): err = vim.fault.VimFault() err.msg = "vim_fault" self.mock_host_vsan_system.RemoveDiskMapping_Task = MagicMock(side_effect=err) with self.assertRaises(VMwareApiError) as excinfo: vsan.remove_diskgroup(self.mock_si, self.mock_host_ref, self.mock_diskgroup) self.assertEqual(excinfo.exception.strerror, "vim_fault") def test_remove_disk_mapping_raise_runtime_fault(self): err = vmodl.RuntimeFault() err.msg = "runtime_fault" self.mock_host_vsan_system.RemoveDiskMapping_Task = MagicMock(side_effect=err) with self.assertRaises(VMwareRuntimeError) as excinfo: vsan.remove_diskgroup(self.mock_si, self.mock_host_ref, self.mock_diskgroup) self.assertEqual(excinfo.exception.strerror, "runtime_fault") def test_wait_for_tasks(self): mock_wait_for_task = MagicMock() with patch("salt.utils.vmware.wait_for_task", mock_wait_for_task): vsan.remove_diskgroup(self.mock_si, self.mock_host_ref, self.mock_diskgroup) mock_wait_for_task.assert_called_once_with( self.mock_task, "fake_hostname", "remove_diskgroup" ) def test_result(self): res = vsan.remove_diskgroup( self.mock_si, self.mock_host_ref, self.mock_diskgroup ) self.assertTrue(res) @skipIf(not HAS_PYVMOMI, "The 'pyvmomi' library is missing") @skipIf(not HAS_PYVSAN, "The 'vsan' ext library is missing") class GetClusterVsanInfoTestCase(TestCase, LoaderModuleMockMixin): """Tests for salt.utils.vsan.get_cluster_vsan_info""" def setup_loader_modules(self): return {vsan: {"__virtual__": MagicMock(return_value="vsan")}} def setUp(self): self.mock_cl_ref = MagicMock() self.mock_si = MagicMock() patches = ( ("salt.utils.vmware.get_managed_object_name", MagicMock()), ( "salt.utils.vmware.get_service_instance_from_managed_object", MagicMock(return_value=self.mock_si), ), ("salt.utils.vsan.get_vsan_cluster_config_system", MagicMock()), ) for mod, mock in patches: patcher = patch(mod, mock) patcher.start() self.addCleanup(patcher.stop) def tearDown(self): for attr in ("mock_si", "mock_cl_ref"): delattr(self, attr) 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 ): vsan.get_cluster_vsan_info(self.mock_cl_ref) mock_get_managed_object_name.assert_called_once_with(self.mock_cl_ref) def test_get_vsan_cluster_config_system_call(self): mock_get_vsan_cl_syst = MagicMock() with patch( "salt.utils.vsan.get_vsan_cluster_config_system", mock_get_vsan_cl_syst ): vsan.get_cluster_vsan_info(self.mock_cl_ref) mock_get_vsan_cl_syst.assert_called_once_with(self.mock_si) def test_VsanClusterGetConfig_call(self): mock_vsan_sys = MagicMock() with patch( "salt.utils.vsan.get_vsan_cluster_config_system", MagicMock(return_value=mock_vsan_sys), ): vsan.get_cluster_vsan_info(self.mock_cl_ref) mock_vsan_sys.VsanClusterGetConfig.assert_called_once_with(self.mock_cl_ref) def test_VsanClusterGetConfig_raises_no_permission(self): exc = vim.fault.NoPermission() exc.privilegeId = "Fake privilege" with patch( "salt.utils.vsan.get_vsan_cluster_config_system", MagicMock( return_value=MagicMock(VsanClusterGetConfig=MagicMock(side_effect=exc)) ), ): with self.assertRaises(VMwareApiError) as excinfo: vsan.get_cluster_vsan_info(self.mock_cl_ref) self.assertEqual( excinfo.exception.strerror, "Not enough permissions. Required privilege: " "Fake privilege", ) def test_VsanClusterGetConfig_raises_vim_fault(self): exc = vim.fault.VimFault() exc.msg = "VimFault msg" with patch( "salt.utils.vsan.get_vsan_cluster_config_system", MagicMock( return_value=MagicMock(VsanClusterGetConfig=MagicMock(side_effect=exc)) ), ): with self.assertRaises(VMwareApiError) as excinfo: vsan.get_cluster_vsan_info(self.mock_cl_ref) self.assertEqual(excinfo.exception.strerror, "VimFault msg") def test_VsanClusterGetConfig_raises_runtime_fault(self): exc = vmodl.RuntimeFault() exc.msg = "RuntimeFault msg" with patch( "salt.utils.vsan.get_vsan_cluster_config_system", MagicMock( return_value=MagicMock(VsanClusterGetConfig=MagicMock(side_effect=exc)) ), ): with self.assertRaises(VMwareRuntimeError) as excinfo: vsan.get_cluster_vsan_info(self.mock_cl_ref) self.assertEqual(excinfo.exception.strerror, "RuntimeFault msg") @skipIf(not HAS_PYVMOMI, "The 'pyvmomi' library is missing") @skipIf(not HAS_PYVSAN, "The 'vsan' ext library is missing") class ReconfigureClusterVsanTestCase(TestCase): """Tests for salt.utils.vsan.reconfigure_cluster_vsan""" def setUp(self): self.mock_si = MagicMock() self.mock_task = MagicMock() self.mock_cl_reconf = MagicMock(return_value=self.mock_task) self.mock_get_vsan_conf_sys = MagicMock( return_value=MagicMock(VsanClusterReconfig=self.mock_cl_reconf) ) self.mock_cl_ref = MagicMock() self.mock_cl_vsan_spec = MagicMock() patches = ( ("salt.utils.vmware.get_managed_object_name", MagicMock()), ( "salt.utils.vmware.get_service_instance_from_managed_object", MagicMock(return_value=self.mock_si), ), ( "salt.utils.vsan.get_vsan_cluster_config_system", self.mock_get_vsan_conf_sys, ), ("salt.utils.vsan._wait_for_tasks", MagicMock()), ) for mod, mock in patches: patcher = patch(mod, mock) patcher.start() self.addCleanup(patcher.stop) def tearDown(self): for attr in ( "mock_si", "mock_cl_reconf", "mock_get_vsan_conf_sys", "mock_cl_ref", "mock_cl_vsan_spec", "mock_task", ): delattr(self, attr) def test_get_cluster_name_call(self): get_managed_object_name_mock = MagicMock() with patch( "salt.utils.vmware.get_managed_object_name", get_managed_object_name_mock ): vsan.reconfigure_cluster_vsan(self.mock_cl_ref, self.mock_cl_vsan_spec) get_managed_object_name_mock.assert_called_once_with(self.mock_cl_ref) def test_get_service_instance_call(self): get_service_instance_from_managed_object_mock = MagicMock() with patch( "salt.utils.vmware.get_service_instance_from_managed_object", get_service_instance_from_managed_object_mock, ): vsan.reconfigure_cluster_vsan(self.mock_cl_ref, self.mock_cl_vsan_spec) get_service_instance_from_managed_object_mock.assert_called_once_with( self.mock_cl_ref ) def test_get_vsan_cluster_config_system_call(self): vsan.reconfigure_cluster_vsan(self.mock_cl_ref, self.mock_cl_vsan_spec) self.mock_get_vsan_conf_sys.assert_called_once_with(self.mock_si) def test_cluster_reconfig_call(self): vsan.reconfigure_cluster_vsan(self.mock_cl_ref, self.mock_cl_vsan_spec) self.mock_cl_reconf.assert_called_once_with( self.mock_cl_ref, self.mock_cl_vsan_spec ) def test_cluster_reconfig_raises_no_permission(self): exc = vim.fault.NoPermission() exc.privilegeId = "Fake privilege" with patch( "salt.utils.vsan.get_vsan_cluster_config_system", MagicMock( return_value=MagicMock(VsanClusterReconfig=MagicMock(side_effect=exc)) ), ): with self.assertRaises(VMwareApiError) as excinfo: vsan.reconfigure_cluster_vsan(self.mock_cl_ref, self.mock_cl_vsan_spec) self.assertEqual( excinfo.exception.strerror, "Not enough permissions. Required privilege: " "Fake privilege", ) def test_cluster_reconfig_raises_vim_fault(self): exc = vim.fault.VimFault() exc.msg = "VimFault msg" with patch( "salt.utils.vsan.get_vsan_cluster_config_system", MagicMock( return_value=MagicMock(VsanClusterReconfig=MagicMock(side_effect=exc)) ), ): with self.assertRaises(VMwareApiError) as excinfo: vsan.reconfigure_cluster_vsan(self.mock_cl_ref, self.mock_cl_vsan_spec) self.assertEqual(excinfo.exception.strerror, "VimFault msg") def test_cluster_reconfig_raises_vmodl_runtime_error(self): exc = vmodl.RuntimeFault() exc.msg = "VimRuntime msg" with patch( "salt.utils.vsan.get_vsan_cluster_config_system", MagicMock( return_value=MagicMock(VsanClusterReconfig=MagicMock(side_effect=exc)) ), ): with self.assertRaises(VMwareRuntimeError) as excinfo: vsan.reconfigure_cluster_vsan(self.mock_cl_ref, self.mock_cl_vsan_spec) self.assertEqual(excinfo.exception.strerror, "VimRuntime msg") def test__wait_for_tasks_call(self): mock_wait_for_tasks = MagicMock() with patch("salt.utils.vsan._wait_for_tasks", mock_wait_for_tasks): vsan.reconfigure_cluster_vsan(self.mock_cl_ref, self.mock_cl_vsan_spec) mock_wait_for_tasks.assert_called_once_with([self.mock_task], self.mock_si) @skipIf(not HAS_PYVMOMI, "The 'pyvmomi' library is missing") @skipIf(not HAS_PYVSAN, "The 'vsan' ext library is missing") class _WaitForTasks(TestCase, LoaderModuleMockMixin): """Tests for salt.utils.vsan._wait_for_tasks""" def setup_loader_modules(self): return {vsan: {"__virtual__": MagicMock(return_value="vsan")}} def setUp(self): self.mock_si = MagicMock() self.mock_tasks = MagicMock() patches = (("salt.utils.vsan.vsanapiutils.WaitForTasks", MagicMock()),) for mod, mock in patches: patcher = patch(mod, mock) patcher.start() self.addCleanup(patcher.stop) def tearDown(self): for attr in ("mock_si", "mock_tasks"): delattr(self, attr) def test_wait_for_tasks_call(self): mock_wait_for_tasks = MagicMock() with patch("salt.utils.vsan.vsanapiutils.WaitForTasks", mock_wait_for_tasks): vsan._wait_for_tasks(self.mock_tasks, self.mock_si) mock_wait_for_tasks.assert_called_once_with(self.mock_tasks, self.mock_si) def test_wait_for_tasks_raises_no_permission(self): exc = vim.fault.NoPermission() exc.privilegeId = "Fake privilege" with patch( "salt.utils.vsan.vsanapiutils.WaitForTasks", MagicMock(side_effect=exc) ): with self.assertRaises(VMwareApiError) as excinfo: vsan._wait_for_tasks(self.mock_tasks, self.mock_si) self.assertEqual( excinfo.exception.strerror, "Not enough permissions. Required privilege: " "Fake privilege", ) def test_wait_for_tasks_raises_vim_fault(self): exc = vim.fault.VimFault() exc.msg = "VimFault msg" with patch( "salt.utils.vsan.vsanapiutils.WaitForTasks", MagicMock(side_effect=exc) ): with self.assertRaises(VMwareApiError) as excinfo: vsan._wait_for_tasks(self.mock_tasks, self.mock_si) self.assertEqual(excinfo.exception.strerror, "VimFault msg") def test_wait_for_tasks_raises_vmodl_runtime_error(self): exc = vmodl.RuntimeFault() exc.msg = "VimRuntime msg" with patch( "salt.utils.vsan.vsanapiutils.WaitForTasks", MagicMock(side_effect=exc) ): with self.assertRaises(VMwareRuntimeError) as excinfo: vsan._wait_for_tasks(self.mock_tasks, self.mock_si) self.assertEqual(excinfo.exception.strerror, "VimRuntime msg")