# Copyright 2016 Cloudbase Solutions Srl # All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import ctypes from unittest import mock import ddt from os_win import constants from os_win import exceptions from os_win.tests.unit import test_base from os_win.utils.compute import _clusapi_utils from os_win.utils.winapi import constants as w_const from os_win.utils.winapi.libs import clusapi as clusapi_def from os_win.utils.winapi import wintypes @ddt.ddt class ClusApiUtilsTestCase(test_base.OsWinBaseTestCase): _LIVE_MIGRATION_TYPE = 4 def setUp(self): super(ClusApiUtilsTestCase, self).setUp() self._clusapi = mock.patch.object( _clusapi_utils, 'clusapi', create=True).start() self._clusapi_utils = _clusapi_utils.ClusApiUtils() self._run_patcher = mock.patch.object(self._clusapi_utils, '_run_and_check_output') self._mock_run = self._run_patcher.start() def _mock_ctypes(self): self._ctypes = mock.Mock() # This is used in order to easily make assertions on the variables # passed by reference. self._ctypes.byref = lambda x: (x, "byref") self._ctypes.c_wchar_p = lambda x: (x, 'c_wchar_p') self._ctypes.sizeof = lambda x: (x, 'sizeof') self._ctypes.c_ulong = lambda x: (x, 'c_ulong') mock.patch.object(_clusapi_utils, 'ctypes', self._ctypes).start() def test_run_and_check_output(self): self._clusapi_utils._win32utils = mock.Mock() self._clusapi_utils._run_and_check_output = ( self._run_patcher.temp_original) mock_win32utils_run_and_check_output = ( self._clusapi_utils._win32utils.run_and_check_output) ret_val = self._clusapi_utils._run_and_check_output( mock.sentinel.func, mock.sentinel.arg, fake_kwarg=mock.sentinel.kwarg) mock_win32utils_run_and_check_output.assert_called_once_with( mock.sentinel.func, mock.sentinel.arg, fake_kwarg=mock.sentinel.kwarg, failure_exc=exceptions.ClusterWin32Exception) self.assertEqual(mock_win32utils_run_and_check_output.return_value, ret_val) def test_dword_align(self): self.assertEqual(8, self._clusapi_utils._dword_align(5)) self.assertEqual(4, self._clusapi_utils._dword_align(4)) def test_get_clusprop_value_struct(self): val_type = ctypes.c_ubyte * 3 expected_padding_sz = 1 clusprop_val_struct = self._clusapi_utils._get_clusprop_value_struct( val_type) expected_fields = [('syntax', wintypes.DWORD), ('length', wintypes.DWORD), ('value', val_type), ('_padding', ctypes.c_ubyte * expected_padding_sz)] self.assertEqual(expected_fields, clusprop_val_struct._fields_) def test_get_property_list_entry(self): fake_prop_name = 'fake prop name' fake_prop_syntax = 1 fake_prop_val = (ctypes.c_wchar * 10)() fake_prop_val.value = 'fake prop' entry = self._clusapi_utils.get_property_list_entry( name=fake_prop_name, syntax=fake_prop_syntax, value=fake_prop_val) self.assertEqual(w_const.CLUSPROP_SYNTAX_NAME, entry.name.syntax) self.assertEqual(fake_prop_name, entry.name.value) self.assertEqual( ctypes.sizeof(ctypes.c_wchar) * (len(fake_prop_name) + 1), entry.name.length) self.assertEqual(fake_prop_syntax, entry.value.syntax) self.assertEqual(bytearray(fake_prop_val), bytearray(entry.value.value)) self.assertEqual( ctypes.sizeof(fake_prop_val), entry.value.length) self.assertEqual(w_const.CLUSPROP_SYNTAX_ENDMARK, entry._endmark) def test_get_property_list(self): entry_0 = self._clusapi_utils.get_property_list_entry( name='fake prop name', syntax=1, value=ctypes.c_uint(2)) entry_1 = self._clusapi_utils.get_property_list_entry( name='fake prop name', syntax=2, value=ctypes.c_ubyte(5)) prop_list = self._clusapi_utils.get_property_list( [entry_0, entry_1]) self.assertEqual(2, prop_list.count) self.assertEqual(bytearray(entry_0) + bytearray(entry_1), prop_list.entries_buff) @ddt.data('fake cluster name', None) def test_open_cluster(self, cluster_name): self._mock_ctypes() handle = self._clusapi_utils.open_cluster(cluster_name) expected_handle_arg = ( self._ctypes.c_wchar_p(cluster_name) if cluster_name else None) self._mock_run.assert_called_once_with( self._clusapi.OpenCluster, expected_handle_arg, **self._clusapi_utils._open_handle_check_flags) self.assertEqual(self._mock_run.return_value, handle) def test_open_cluster_enum(self): handle = self._clusapi_utils.open_cluster_enum( mock.sentinel.cluster_handle, mock.sentinel.object_type) self._mock_run.assert_called_once_with( self._clusapi.ClusterOpenEnumEx, mock.sentinel.cluster_handle, mock.sentinel.object_type, None, **self._clusapi_utils._open_handle_check_flags) self.assertEqual(self._mock_run.return_value, handle) def test_open_cluster_group(self): self._mock_ctypes() handle = self._clusapi_utils.open_cluster_group( mock.sentinel.cluster_handle, mock.sentinel.group_name) self._mock_run.assert_called_once_with( self._clusapi.OpenClusterGroup, mock.sentinel.cluster_handle, self._ctypes.c_wchar_p(mock.sentinel.group_name), **self._clusapi_utils._open_handle_check_flags) self.assertEqual(self._mock_run.return_value, handle) def test_open_cluster_node(self): self._mock_ctypes() handle = self._clusapi_utils.open_cluster_node( mock.sentinel.cluster_handle, mock.sentinel.node_name) self._mock_run.assert_called_once_with( self._clusapi.OpenClusterNode, mock.sentinel.cluster_handle, self._ctypes.c_wchar_p(mock.sentinel.node_name), **self._clusapi_utils._open_handle_check_flags) self.assertEqual(self._mock_run.return_value, handle) def test_open_cluster_resource(self): self._mock_ctypes() handle = self._clusapi_utils.open_cluster_resource( mock.sentinel.cluster_handle, mock.sentinel.resource_name) self._mock_run.assert_called_once_with( self._clusapi.OpenClusterResource, mock.sentinel.cluster_handle, self._ctypes.c_wchar_p(mock.sentinel.resource_name), **self._clusapi_utils._open_handle_check_flags) self.assertEqual(self._mock_run.return_value, handle) def test_close_cluster(self): self._clusapi_utils.close_cluster(mock.sentinel.handle) self._clusapi.CloseCluster.assert_called_once_with( mock.sentinel.handle) def test_close_cluster_group(self): self._clusapi_utils.close_cluster_group(mock.sentinel.handle) self._clusapi.CloseClusterGroup.assert_called_once_with( mock.sentinel.handle) def test_close_cluster_node(self): self._clusapi_utils.close_cluster_node(mock.sentinel.handle) self._clusapi.CloseClusterNode.assert_called_once_with( mock.sentinel.handle) def test_close_cluster_resource(self): self._clusapi_utils.close_cluster_resource(mock.sentinel.handle) self._clusapi.CloseClusterResource.assert_called_once_with( mock.sentinel.handle) def test_close_cluster_enum(self): self._clusapi_utils.close_cluster_enum(mock.sentinel.handle) self._clusapi.ClusterCloseEnumEx.assert_called_once_with( mock.sentinel.handle) def test_online_cluster_group(self): self._clusapi_utils.online_cluster_group(mock.sentinel.group_handle, mock.sentinel.dest_handle) self._mock_run.assert_called_once_with( self._clusapi.OnlineClusterGroup, mock.sentinel.group_handle, mock.sentinel.dest_handle) def test_destroy_cluster_group(self): self._clusapi_utils.destroy_cluster_group(mock.sentinel.group_handle) self._mock_run.assert_called_once_with( self._clusapi.DestroyClusterGroup, mock.sentinel.group_handle) def test_offline_cluster_group(self): self._clusapi_utils.offline_cluster_group(mock.sentinel.group_handle) self._mock_run.assert_called_once_with( self._clusapi.OfflineClusterGroup, mock.sentinel.group_handle) @ddt.data(0, w_const.ERROR_IO_PENDING) def test_cancel_cluster_group_operation(self, cancel_ret_val): self._mock_run.return_value = cancel_ret_val expected_ret_val = cancel_ret_val != w_const.ERROR_IO_PENDING ret_val = self._clusapi_utils.cancel_cluster_group_operation( mock.sentinel.group_handle) self.assertEqual(expected_ret_val, ret_val) self._mock_run.assert_called_once_with( self._clusapi.CancelClusterGroupOperation, mock.sentinel.group_handle, 0, ignored_error_codes=[w_const.ERROR_IO_PENDING]) @ddt.data(mock.sentinel.prop_list, None) def test_move_cluster_group(self, prop_list): self._mock_ctypes() expected_prop_list_arg = ( self._ctypes.byref(prop_list) if prop_list else None) expected_prop_list_sz = ( self._ctypes.sizeof(prop_list) if prop_list else 0) self._clusapi_utils.move_cluster_group( mock.sentinel.group_handle, mock.sentinel.dest_node_handle, mock.sentinel.move_flags, prop_list) self._mock_run.assert_called_once_with( self._clusapi.MoveClusterGroupEx, mock.sentinel.group_handle, mock.sentinel.dest_node_handle, mock.sentinel.move_flags, expected_prop_list_arg, expected_prop_list_sz, ignored_error_codes=[w_const.ERROR_IO_PENDING]) def test_get_cluster_group_state(self): owner_node = 'fake owner node' def fake_get_state(inst, group_handle, node_name_buff, node_name_len, error_ret_vals, error_on_nonzero_ret_val, ret_val_is_err_code): self.assertEqual(mock.sentinel.group_handle, group_handle) # Those arguments would not normally get to the ClusApi # function, instead being used by the helper invoking # it and catching errors. For convenience, we validate # those arguments at this point. self.assertEqual([constants.CLUSTER_GROUP_STATE_UNKNOWN], error_ret_vals) self.assertFalse(error_on_nonzero_ret_val) self.assertFalse(ret_val_is_err_code) node_name_len_arg = ctypes.cast( node_name_len, wintypes.PDWORD).contents self.assertEqual(w_const.MAX_PATH, node_name_len_arg.value) node_name_arg = ctypes.cast( node_name_buff, ctypes.POINTER( ctypes.c_wchar * w_const.MAX_PATH)).contents node_name_arg.value = owner_node return mock.sentinel.group_state self._mock_run.side_effect = fake_get_state state_info = self._clusapi_utils.get_cluster_group_state( mock.sentinel.group_handle) expected_state_info = dict(state=mock.sentinel.group_state, owner_node=owner_node) self.assertEqual(expected_state_info, state_info) @ddt.data({'notif_filters': (clusapi_def.NOTIFY_FILTER_AND_TYPE * 2)(), 'exp_notif_filters_len': 2}, {'notif_filters': clusapi_def.NOTIFY_FILTER_AND_TYPE(), 'notif_port_h': mock.sentinel.notif_port_h, 'notif_key': mock.sentinel.notif_key}) @ddt.unpack def test_create_cluster_notify_port(self, notif_filters, exp_notif_filters_len=1, notif_port_h=None, notif_key=None): self._mock_ctypes() self._ctypes.Array = ctypes.Array self._clusapi_utils.create_cluster_notify_port_v2( mock.sentinel.cluster_handle, notif_filters, notif_port_h, notif_key) exp_notif_key_p = self._ctypes.byref(notif_key) if notif_key else None exp_notif_port_h = notif_port_h or w_const.INVALID_HANDLE_VALUE self._mock_run.assert_called_once_with( self._clusapi.CreateClusterNotifyPortV2, exp_notif_port_h, mock.sentinel.cluster_handle, self._ctypes.byref(notif_filters), self._ctypes.c_ulong(exp_notif_filters_len), exp_notif_key_p, **self._clusapi_utils._open_handle_check_flags) def test_close_cluster_notify_port(self): self._clusapi_utils.close_cluster_notify_port(mock.sentinel.handle) self._clusapi.CloseClusterNotifyPort.assert_called_once_with( mock.sentinel.handle) def test_get_cluster_notify_v2(self): fake_notif_key = 1 fake_notif_port_h = 2 fake_notif_type = 3 fake_filter_flags = 4 fake_clus_obj_name = 'fake-changed-clus-object' fake_event_buff = 'fake-event-buff' fake_obj_type = 'fake-object-type' fake_obj_id = 'fake-obj-id' fake_parent_id = 'fake-parent-id' notif_key = ctypes.c_ulong(fake_notif_key) requested_buff_sz = 1024 def fake_get_cluster_notify(func, notif_port_h, pp_notif_key, p_filter_and_type, p_buff, p_buff_sz, p_obj_id_buff, p_obj_id_buff_sz, p_parent_id_buff, p_parent_id_buff_sz, p_obj_name_buff, p_obj_name_buff_sz, p_obj_type, p_obj_type_sz, timeout_ms): self.assertEqual(self._clusapi.GetClusterNotifyV2, func) self.assertEqual(fake_notif_port_h, notif_port_h) obj_name_buff_sz = ctypes.cast( p_obj_name_buff_sz, wintypes.PDWORD).contents buff_sz = ctypes.cast( p_buff_sz, wintypes.PDWORD).contents obj_type_sz = ctypes.cast( p_obj_type_sz, wintypes.PDWORD).contents obj_id_sz = ctypes.cast( p_obj_id_buff_sz, wintypes.PDWORD).contents parent_id_buff_sz = ctypes.cast( p_parent_id_buff_sz, wintypes.PDWORD).contents # We'll just request the tested method to pass us # a buffer this large. if (buff_sz.value < requested_buff_sz or obj_name_buff_sz.value < requested_buff_sz or parent_id_buff_sz.value < requested_buff_sz or obj_type_sz.value < requested_buff_sz or obj_id_sz.value < requested_buff_sz): buff_sz.value = requested_buff_sz obj_name_buff_sz.value = requested_buff_sz parent_id_buff_sz.value = requested_buff_sz obj_type_sz.value = requested_buff_sz obj_id_sz.value = requested_buff_sz raise exceptions.ClusterWin32Exception( error_code=w_const.ERROR_MORE_DATA, func_name='GetClusterNotifyV2', error_message='error more data') pp_notif_key = ctypes.cast(pp_notif_key, ctypes.c_void_p) p_notif_key = ctypes.c_void_p.from_address(pp_notif_key.value) p_notif_key.value = ctypes.addressof(notif_key) filter_and_type = ctypes.cast( p_filter_and_type, ctypes.POINTER(clusapi_def.NOTIFY_FILTER_AND_TYPE)).contents filter_and_type.dwObjectType = fake_notif_type filter_and_type.FilterFlags = fake_filter_flags def set_wchar_buff(p_wchar_buff, wchar_buff_sz, value): wchar_buff = ctypes.cast( p_wchar_buff, ctypes.POINTER( ctypes.c_wchar * (wchar_buff_sz // ctypes.sizeof(ctypes.c_wchar)))) wchar_buff = wchar_buff.contents ctypes.memset(wchar_buff, 0, wchar_buff_sz) wchar_buff.value = value return wchar_buff set_wchar_buff(p_obj_name_buff, requested_buff_sz, fake_clus_obj_name) set_wchar_buff(p_buff, requested_buff_sz, fake_event_buff) set_wchar_buff(p_parent_id_buff, requested_buff_sz, fake_parent_id) set_wchar_buff(p_obj_type, requested_buff_sz, fake_obj_type) set_wchar_buff(p_obj_id_buff, requested_buff_sz, fake_obj_id) self.assertEqual(mock.sentinel.timeout_ms, timeout_ms) self._mock_run.side_effect = fake_get_cluster_notify event = self._clusapi_utils.get_cluster_notify_v2( fake_notif_port_h, mock.sentinel.timeout_ms) w_event_buff = ctypes.cast( event['buff'], ctypes.POINTER( ctypes.c_wchar * (requested_buff_sz // ctypes.sizeof(ctypes.c_wchar)))) w_event_buff = w_event_buff.contents[:] event['buff'] = w_event_buff.split('\x00')[0] expected_event = dict(cluster_object_name=fake_clus_obj_name, object_id=fake_obj_id, object_type=fake_notif_type, object_type_str=fake_obj_type, filter_flags=fake_filter_flags, parent_id=fake_parent_id, buff=fake_event_buff, buff_sz=requested_buff_sz, notif_key=fake_notif_key) self.assertEqual(expected_event, event) def _get_fake_prop_list(self): syntax = w_const.CLUSPROP_SYNTAX_LIST_VALUE_DWORD migr_type = wintypes.DWORD(self._LIVE_MIGRATION_TYPE) prop_entries = [ self._clusapi_utils.get_property_list_entry( w_const.CLUS_RESTYPE_NAME_VM, syntax, migr_type), self._clusapi_utils.get_property_list_entry( w_const.CLUS_RESTYPE_NAME_VM_CONFIG, syntax, migr_type), self._clusapi_utils.get_property_list_entry( w_const.CLUSREG_NAME_GRP_STATUS_INFORMATION, w_const.CLUSPROP_SYNTAX_LIST_VALUE_ULARGE_INTEGER, ctypes.c_ulonglong(w_const. CLUSGRP_STATUS_WAITING_IN_QUEUE_FOR_MOVE)), # noqa self._clusapi_utils.get_property_list_entry( w_const.CLUSREG_NAME_GRP_TYPE, w_const.CLUSPROP_SYNTAX_LIST_VALUE_DWORD, ctypes.c_ulong(w_const.ClusGroupTypeVirtualMachine)), ] prop_list = self._clusapi_utils.get_property_list(prop_entries) return prop_list def test_get_prop_list_entry_p_not_found(self): prop_list = self._get_fake_prop_list() self.assertRaises(exceptions.ClusterPropertyListEntryNotFound, self._clusapi_utils.get_prop_list_entry_p, ctypes.byref(prop_list), ctypes.sizeof(prop_list), 'InexistentProperty') def test_get_prop_list_entry_p_parsing_error(self): prop_list = self._get_fake_prop_list() prop_entry_name_len_addr = ctypes.addressof( prop_list.entries_buff) + ctypes.sizeof(ctypes.c_ulong) prop_entry_name_len = ctypes.c_ulong.from_address( prop_entry_name_len_addr) prop_entry_name_len.value = ctypes.sizeof(prop_list) self.assertRaises(exceptions.ClusterPropertyListParsingError, self._clusapi_utils.get_prop_list_entry_p, ctypes.byref(prop_list), ctypes.sizeof(prop_list), w_const.CLUS_RESTYPE_NAME_VM) def test_get_prop_list_entry_p(self): prop_list = self._get_fake_prop_list() prop_entry = self._clusapi_utils.get_prop_list_entry_p( ctypes.byref(prop_list), ctypes.sizeof(prop_list), w_const.CLUS_RESTYPE_NAME_VM_CONFIG) self.assertEqual( w_const.CLUSPROP_SYNTAX_LIST_VALUE_DWORD, prop_entry['syntax']) self.assertEqual( ctypes.sizeof(ctypes.c_ulong), prop_entry['length']) val = ctypes.c_ulong.from_address(prop_entry['val_p'].value).value self.assertEqual(self._LIVE_MIGRATION_TYPE, val) def test_cluster_group_control(self): fake_out_buff = 'fake-event-buff' requested_buff_sz = 1024 def fake_cluster_group_ctrl(func, group_handle, node_handle, control_code, in_buff_p, in_buff_sz, out_buff_p, out_buff_sz, requested_buff_sz_p): self.assertEqual(self._clusapi.ClusterGroupControl, func) self.assertEqual(mock.sentinel.group_handle, group_handle) self.assertEqual(mock.sentinel.node_handle, node_handle) self.assertEqual(mock.sentinel.control_code, control_code) self.assertEqual(mock.sentinel.in_buff_p, in_buff_p) self.assertEqual(mock.sentinel.in_buff_sz, in_buff_sz) req_buff_sz = ctypes.cast( requested_buff_sz_p, wintypes.PDWORD).contents req_buff_sz.value = requested_buff_sz # We'll just request the tested method to pass us # a buffer this large. if (out_buff_sz.value < requested_buff_sz): raise exceptions.ClusterWin32Exception( error_code=w_const.ERROR_MORE_DATA, func_name='ClusterGroupControl', error_message='error more data') out_buff = ctypes.cast( out_buff_p, ctypes.POINTER( ctypes.c_wchar * (requested_buff_sz // ctypes.sizeof(ctypes.c_wchar)))) out_buff = out_buff.contents out_buff.value = fake_out_buff self._mock_run.side_effect = fake_cluster_group_ctrl out_buff, out_buff_sz = self._clusapi_utils.cluster_group_control( mock.sentinel.group_handle, mock.sentinel.control_code, mock.sentinel.node_handle, mock.sentinel.in_buff_p, mock.sentinel.in_buff_sz) self.assertEqual(requested_buff_sz, out_buff_sz) wp_out_buff = ctypes.cast( out_buff, ctypes.POINTER(ctypes.c_wchar * requested_buff_sz)) self.assertEqual(fake_out_buff, wp_out_buff.contents[:len(fake_out_buff)]) def test_get_cluster_group_status_info(self): prop_list = self._get_fake_prop_list() status_info = self._clusapi_utils.get_cluster_group_status_info( ctypes.byref(prop_list), ctypes.sizeof(prop_list)) self.assertEqual( w_const.CLUSGRP_STATUS_WAITING_IN_QUEUE_FOR_MOVE, status_info) def test_get_cluster_group_type(self): prop_list = self._get_fake_prop_list() status_info = self._clusapi_utils.get_cluster_group_type( ctypes.byref(prop_list), ctypes.sizeof(prop_list)) self.assertEqual( w_const.ClusGroupTypeVirtualMachine, status_info) def test_cluster_get_enum_count(self): ret_val = self._clusapi_utils.cluster_get_enum_count( mock.sentinel.enum_handle) self.assertEqual(self._mock_run.return_value, ret_val) self._mock_run.assert_called_once_with( self._clusapi.ClusterGetEnumCountEx, mock.sentinel.enum_handle, error_on_nonzero_ret_val=False, ret_val_is_err_code=False) def test_cluster_enum(self): obj_id = 'fake_obj_id' obj_id_wchar_p = ctypes.c_wchar_p(obj_id) requested_buff_sz = 1024 def fake_cluster_enum(func, enum_handle, index, buff_p, buff_sz_p, ignored_error_codes=tuple()): self.assertEqual(self._clusapi.ClusterEnumEx, func) self.assertEqual(mock.sentinel.enum_handle, enum_handle) self.assertEqual(mock.sentinel.index, index) buff_sz = ctypes.cast( buff_sz_p, wintypes.PDWORD).contents # We'll just request the tested method to pass us # a buffer this large. if (buff_sz.value < requested_buff_sz): buff_sz.value = requested_buff_sz if w_const.ERROR_MORE_DATA not in ignored_error_codes: raise exceptions.ClusterWin32Exception( error_code=w_const.ERROR_MORE_DATA) return item = ctypes.cast( buff_p, clusapi_def.PCLUSTER_ENUM_ITEM).contents item.lpszId = obj_id_wchar_p item.cbId = len(obj_id) self._mock_run.side_effect = fake_cluster_enum item = self._clusapi_utils.cluster_enum( mock.sentinel.enum_handle, mock.sentinel.index) self.assertEqual(obj_id, item.lpszId) @ddt.ddt class TestClusterContextManager(test_base.OsWinBaseTestCase): _autospec_classes = [_clusapi_utils.ClusApiUtils] def setUp(self): super(TestClusterContextManager, self).setUp() self._cmgr = _clusapi_utils.ClusterContextManager() self._clusapi_utils = self._cmgr._clusapi_utils @ddt.data(None, mock.sentinel.cluster_name) def test_open_cluster(self, cluster_name): with self._cmgr.open_cluster(cluster_name) as f: self._clusapi_utils.open_cluster.assert_called_once_with( cluster_name) self.assertEqual(f, self._clusapi_utils.open_cluster.return_value) self._clusapi_utils.close_cluster.assert_called_once_with( self._clusapi_utils.open_cluster.return_value) def test_open_cluster_group(self): with self._cmgr.open_cluster_group(mock.sentinel.group_name) as f: self._clusapi_utils.open_cluster.assert_called_once_with(None) self._clusapi_utils.open_cluster_group.assert_called_once_with( self._clusapi_utils.open_cluster.return_value, mock.sentinel.group_name) self.assertEqual( f, self._clusapi_utils.open_cluster_group.return_value) self._clusapi_utils.close_cluster_group.assert_called_once_with( self._clusapi_utils.open_cluster_group.return_value) self._clusapi_utils.close_cluster.assert_called_once_with( self._clusapi_utils.open_cluster.return_value) def test_open_missing_cluster_group(self): exc = exceptions.ClusterWin32Exception( func_name='OpenClusterGroup', message='expected exception', error_code=w_const.ERROR_GROUP_NOT_FOUND) self._clusapi_utils.open_cluster_group.side_effect = exc self.assertRaises( exceptions.ClusterObjectNotFound, self._cmgr.open_cluster_group(mock.sentinel.group_name).__enter__) def test_open_cluster_group_with_handle(self): with self._cmgr.open_cluster_group( mock.sentinel.group_name, cluster_handle=mock.sentinel.cluster_handle) as f: self._clusapi_utils.open_cluster.assert_not_called() self._clusapi_utils.open_cluster_group.assert_called_once_with( mock.sentinel.cluster_handle, mock.sentinel.group_name) self.assertEqual( f, self._clusapi_utils.open_cluster_group.return_value) self._clusapi_utils.close_cluster_group.assert_called_once_with( self._clusapi_utils.open_cluster_group.return_value) # If we pass our own handle, we don't want the tested method to # close it. self._clusapi_utils.close_cluster.assert_not_called() def test_open_cluster_resource(self): with self._cmgr.open_cluster_resource(mock.sentinel.res_name) as f: self._clusapi_utils.open_cluster.assert_called_once_with(None) self._clusapi_utils.open_cluster_resource.assert_called_once_with( self._clusapi_utils.open_cluster.return_value, mock.sentinel.res_name) self.assertEqual( f, self._clusapi_utils.open_cluster_resource.return_value) self._clusapi_utils.close_cluster_resource.assert_called_once_with( self._clusapi_utils.open_cluster_resource.return_value) self._clusapi_utils.close_cluster.assert_called_once_with( self._clusapi_utils.open_cluster.return_value) def test_open_cluster_node(self): with self._cmgr.open_cluster_node(mock.sentinel.node_name) as f: self._clusapi_utils.open_cluster.assert_called_once_with(None) self._clusapi_utils.open_cluster_node.assert_called_once_with( self._clusapi_utils.open_cluster.return_value, mock.sentinel.node_name) self.assertEqual( f, self._clusapi_utils.open_cluster_node.return_value) self._clusapi_utils.close_cluster_node.assert_called_once_with( self._clusapi_utils.open_cluster_node.return_value) self._clusapi_utils.close_cluster.assert_called_once_with( self._clusapi_utils.open_cluster.return_value) def test_open_cluster_enum(self): with self._cmgr.open_cluster_enum(mock.sentinel.object_type) as f: self._clusapi_utils.open_cluster.assert_called_once_with(None) self._clusapi_utils.open_cluster_enum.assert_called_once_with( self._clusapi_utils.open_cluster.return_value, mock.sentinel.object_type) self.assertEqual( f, self._clusapi_utils.open_cluster_enum.return_value) self._clusapi_utils.close_cluster_enum.assert_called_once_with( self._clusapi_utils.open_cluster_enum.return_value) self._clusapi_utils.close_cluster.assert_called_once_with( self._clusapi_utils.open_cluster.return_value) def test_invalid_handle_type(self): self.assertRaises(exceptions.Invalid, self._cmgr._open(handle_type=None).__enter__) self.assertRaises(exceptions.Invalid, self._cmgr._close, mock.sentinel.handle, handle_type=None)