1# (c) 2021, NetApp, Inc 2# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) 3 4''' unit test for ONTAP fpolicy status Ansible module ''' 5 6from __future__ import (absolute_import, division, print_function) 7__metaclass__ = type 8import json 9import pytest 10 11from ansible.module_utils import basic 12from ansible.module_utils._text import to_bytes 13from ansible_collections.netapp.ontap.tests.unit.compat.mock import patch 14import ansible_collections.netapp.ontap.plugins.module_utils.netapp as netapp_utils 15 16from ansible_collections.netapp.ontap.plugins.modules.na_ontap_fpolicy_status \ 17 import NetAppOntapFpolicyStatus as my_module # module under test 18 19if not netapp_utils.has_netapp_lib(): 20 pytestmark = pytest.mark.skip('skipping as missing required netapp_lib') 21 22 23def set_module_args(args): 24 """prepare arguments so that they will be picked up during module creation""" 25 args = json.dumps({'ANSIBLE_MODULE_ARGS': args}) 26 basic._ANSIBLE_ARGS = to_bytes(args) # pylint: disable=protected-access 27 28 29class AnsibleExitJson(Exception): 30 """Exception class to be raised by module.exit_json and caught by the test case""" 31 32 33class AnsibleFailJson(Exception): 34 """Exception class to be raised by module.fail_json and caught by the test case""" 35 36 37def exit_json(*args, **kwargs): # pylint: disable=unused-argument 38 """function to patch over exit_json; package return data into an exception""" 39 if 'changed' not in kwargs: 40 kwargs['changed'] = False 41 raise AnsibleExitJson(kwargs) 42 43 44def fail_json(*args, **kwargs): # pylint: disable=unused-argument 45 """function to patch over fail_json; package return data into an exception""" 46 kwargs['failed'] = True 47 raise AnsibleFailJson(kwargs) 48 49 50class MockONTAPConnection(): 51 ''' mock server connection to ONTAP host ''' 52 53 def __init__(self, kind=None): 54 ''' save arguments ''' 55 self.type = kind 56 self.xml_in = None 57 self.xml_out = None 58 59 def invoke_successfully(self, xml, enable_tunneling): # pylint: disable=unused-argument 60 ''' mock invoke_successfully returning xml data ''' 61 self.xml_in = xml 62 if self.type == 'fpolicy_policy_enabled': 63 xml = self.build_fpolicy_status_info_enabled() 64 elif self.type == 'fpolicy_policy_disabled': 65 xml = self.build_fpolicy_status_info_disabled() 66 elif self.type == 'fpolicy_policy_fail': 67 raise netapp_utils.zapi.NaApiError(code='TEST', message="This exception is from the unit test") 68 self.xml_out = xml 69 return xml 70 71 @staticmethod 72 def build_fpolicy_status_info_enabled(): 73 ''' build xml data for fpolicy-policy-status-info ''' 74 xml = netapp_utils.zapi.NaElement('xml') 75 data = { 76 'attributes-list': { 77 'fpolicy-policy-status-info': { 78 'vserver': 'svm1', 79 'policy-name': 'fPolicy1', 80 'status': 'true' 81 } 82 } 83 } 84 xml.translate_struct(data) 85 return xml 86 87 @staticmethod 88 def build_fpolicy_status_info_disabled(): 89 ''' build xml data for fpolicy-policy-status-info ''' 90 xml = netapp_utils.zapi.NaElement('xml') 91 data = { 92 'attributes-list': { 93 'fpolicy-policy-status-info': { 94 'vserver': 'svm1', 95 'policy-name': 'fPolicy1', 96 'status': 'false' 97 } 98 } 99 } 100 xml.translate_struct(data) 101 return xml 102 103 104def default_args(): 105 args = { 106 'vserver': 'svm1', 107 'policy_name': 'fPolicy1', 108 'sequence_number': '10', 109 'hostname': '10.10.10.10', 110 'username': 'username', 111 'password': 'password', 112 'use_rest': 'always' 113 } 114 return args 115 116 117# REST API canned responses when mocking send_request 118SRR = { 119 # common responses 120 'is_rest': (200, dict(version=dict(generation=9, major=9, minor=0, full='dummy')), None), 121 'is_rest_9_8': (200, dict(version=dict(generation=9, major=8, minor=0, full='dummy')), None), 122 'is_zapi': (400, {}, "Unreachable"), 123 'empty_good': (200, {}, None), 124 'zero_record': (200, dict(records=[], num_records=0), None), 125 # 'one_record_uuid': (200, dict(records=[dict(uuid='a1b2c3')], num_records=1), None), 126 'end_of_sequence': (500, None, "Unexpected call to send_request"), 127 'generic_error': (400, None, "Expected error"), 128 'uuid': (200, { 129 'records': [{ 130 'uuid': '56ab5d21' 131 }], 132 'num_records': 1 133 }, None), 134 'fpolicy_status_info_enabled': (200, { 135 'records': [{ 136 'svm': { 137 'uuid': '56ab5d21', 138 'name': 'svm1' 139 }, 140 'policies': [{ 141 'name': 'fPolicy1', 142 'enabled': True, 143 'priority': 10 144 }] 145 }], 146 'num_records': 1 147 }, None), 148 'fpolicy_status_info_disabled': (200, { 149 'records': [{ 150 'svm': { 151 'uuid': '56ab5d21', 152 'name': 'svm1' 153 }, 154 'policies': [{ 155 'name': 'fPolicy1', 156 'enabled': False 157 }] 158 }], 159 'num_records': 1 160 }, None) 161 162} 163 164 165# using pytest natively, without unittest.TestCase 166@pytest.fixture 167def patch_ansible(): 168 with patch.multiple(basic.AnsibleModule, 169 exit_json=exit_json, 170 fail_json=fail_json) as mocks: 171 yield mocks 172 173 174def get_fpolicy_status_mock_object(cx_type='zapi', kind=None): 175 fpolicy_status_obj = my_module() 176 if cx_type == 'zapi': 177 if kind is None: 178 fpolicy_status_obj.server = MockONTAPConnection() 179 else: 180 fpolicy_status_obj.server = MockONTAPConnection(kind=kind) 181 return fpolicy_status_obj 182 183 184def test_module_fail_when_required_args_missing(patch_ansible): 185 ''' required arguments are reported as errors ''' 186 with pytest.raises(AnsibleFailJson) as exc: 187 set_module_args({}) 188 my_module() 189 print('Info: %s' % exc.value.args[0]['msg']) 190 191 192def test_ensure_get_called(patch_ansible): 193 ''' test get_fpolicy_policy_status for non-existent fPolicy''' 194 args = dict(default_args()) 195 args['use_rest'] = 'never' 196 set_module_args(args) 197 print('starting') 198 my_obj = my_module() 199 print('use_rest:', my_obj.use_rest) 200 my_obj.server = MockONTAPConnection('fpolicy_policy_enabled') 201 assert my_obj.get_fpolicy_policy_status is not None 202 203 204def test_rest_missing_arguments(patch_ansible): # pylint: disable=redefined-outer-name,unused-argument 205 ''' enable fpolicy ''' 206 args = dict(default_args()) 207 del args['hostname'] 208 set_module_args(args) 209 with pytest.raises(AnsibleFailJson) as exc: 210 my_module() 211 msg = 'missing required arguments: hostname' 212 assert exc.value.args[0]['msg'] == msg 213 214 215@patch('ansible_collections.netapp.ontap.plugins.modules.na_ontap_fpolicy_status.NetAppOntapFpolicyStatus.enable_fpolicy_policy') 216def test_successful_enable(self, patch_ansible): 217 ''' Enable fPolicy and test idempotency ''' 218 args = dict(default_args()) 219 args['use_rest'] = 'never' 220 set_module_args(args) 221 my_obj = my_module() 222 my_obj.server = MockONTAPConnection('fpolicy_policy_disabled') 223 with patch.object(my_module, 'enable_fpolicy_policy', wraps=my_obj.enable_fpolicy_policy) as mock_enable: 224 with pytest.raises(AnsibleExitJson) as exc: 225 my_obj.apply() 226 print('Enable: ' + repr(exc.value)) 227 assert exc.value.args[0]['changed'] 228 mock_enable.assert_called_with() 229 # test idempotency 230 args = dict(default_args()) 231 args['use_rest'] = 'never' 232 set_module_args(args) 233 my_obj = my_module() 234 my_obj.server = MockONTAPConnection('fpolicy_policy_enabled') 235 with pytest.raises(AnsibleExitJson) as exc: 236 my_obj.apply() 237 print('Enable: ' + repr(exc.value)) 238 assert not exc.value.args[0]['changed'] 239 240 241@patch('ansible_collections.netapp.ontap.plugins.modules.na_ontap_fpolicy_status.NetAppOntapFpolicyStatus.disable_fpolicy_policy') 242def test_successful_disable(self, patch_ansible): 243 ''' Disable fPolicy and test idempotency ''' 244 args = dict(default_args()) 245 args['use_rest'] = 'never' 246 args['state'] = 'absent' 247 set_module_args(args) 248 my_obj = my_module() 249 my_obj.server = MockONTAPConnection('fpolicy_policy_enabled') 250 with patch.object(my_module, 'disable_fpolicy_policy', wraps=my_obj.disable_fpolicy_policy) as mock_disable: 251 with pytest.raises(AnsibleExitJson) as exc: 252 my_obj.apply() 253 print('Enable: ' + repr(exc.value)) 254 assert exc.value.args[0]['changed'] 255 mock_disable.assert_called_with() 256 # test idempotency 257 args = dict(default_args()) 258 args['use_rest'] = 'never' 259 args['state'] = 'absent' 260 set_module_args(args) 261 my_obj = my_module() 262 my_obj.server = MockONTAPConnection('fpolicy_policy_disabled') 263 with pytest.raises(AnsibleExitJson) as exc: 264 my_obj.apply() 265 print('Enable: ' + repr(exc.value)) 266 assert not exc.value.args[0]['changed'] 267 268 269def test_if_all_methods_catch_exception(patch_ansible): 270 args = dict(default_args()) 271 args['use_rest'] = 'never' 272 set_module_args(args) 273 my_obj = my_module() 274 my_obj.server = MockONTAPConnection('fpolicy_policy_fail') 275 with pytest.raises(AnsibleFailJson) as exc: 276 my_obj.enable_fpolicy_policy() 277 print(str(exc.value.args[0]['msg'])) 278 assert 'Error enabling fPolicy policy ' in exc.value.args[0]['msg'] 279 with pytest.raises(AnsibleFailJson) as exc: 280 my_obj.disable_fpolicy_policy() 281 assert 'Error disabling fPolicy policy ' in exc.value.args[0]['msg'] 282 283 284@patch('ansible_collections.netapp.ontap.plugins.module_utils.netapp.OntapRestAPI.send_request') 285def test_rest_enable(mock_request, patch_ansible): # pylint: disable=redefined-outer-name,unused-argument 286 ''' enable fPolicy policy ''' 287 args = dict(default_args()) 288 set_module_args(args) 289 mock_request.side_effect = [ 290 SRR['is_rest'], 291 SRR['uuid'], # get 292 SRR['fpolicy_status_info_disabled'], # get 293 SRR['empty_good'], # patch 294 SRR['end_of_sequence'] 295 ] 296 my_obj = my_module() 297 with pytest.raises(AnsibleExitJson) as exc: 298 my_obj.apply() 299 assert exc.value.args[0]['changed'] is True 300 print(mock_request.mock_calls) 301 assert len(mock_request.mock_calls) == 4 302 303 304@patch('ansible_collections.netapp.ontap.plugins.module_utils.netapp.OntapRestAPI.send_request') 305def test_rest_disable(mock_request, patch_ansible): # pylint: disable=redefined-outer-name,unused-argument 306 ''' disable fPolicy policy ''' 307 args = dict(default_args()) 308 args['state'] = 'absent' 309 set_module_args(args) 310 mock_request.side_effect = [ 311 SRR['is_rest'], 312 SRR['uuid'], # get 313 SRR['fpolicy_status_info_enabled'], # get 314 SRR['empty_good'], # patch 315 SRR['end_of_sequence'] 316 ] 317 my_obj = my_module() 318 with pytest.raises(AnsibleExitJson) as exc: 319 my_obj.apply() 320 assert exc.value.args[0]['changed'] is True 321 print(mock_request.mock_calls) 322 assert len(mock_request.mock_calls) == 4 323