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 scope 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_scope \ 17 import NetAppOntapFpolicyScope 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_scope': 63 xml = self.build_fpolicy_scope_info() 64 elif self.type == 'fpolicy_scope_fail': 65 raise netapp_utils.zapi.NaApiError(code='TEST', message="This exception is from the unit test") 66 self.xml_out = xml 67 return xml 68 69 @staticmethod 70 def build_fpolicy_scope_info(): 71 ''' build xml data for fpolicy-policy-info ''' 72 xml = netapp_utils.zapi.NaElement('xml') 73 data = { 74 'attributes-list': { 75 'fpolicy-scope-config': { 76 'vserver': 'svm1', 77 'policy-name': 'policy1', 78 'export-policies-to-exclude': [ 79 {'string': 'export1'} 80 ], 81 'is-file-extension-check-on-directories-enabled': True, 82 'is-monitoring-of-objects-with-no-extension-enabled': False 83 } 84 } 85 } 86 xml.translate_struct(data) 87 return xml 88 89 90def default_args(): 91 args = { 92 'vserver': 'svm1', 93 'name': 'policy1', 94 'export_policies_to_exclude': 'export1', 95 'hostname': '10.10.10.10', 96 'username': 'username', 97 'password': 'password', 98 'use_rest': 'always' 99 } 100 return args 101 102 103# REST API canned responses when mocking send_request 104SRR = { 105 # common responses 106 'is_rest': (200, dict(version=dict(generation=9, major=9, minor=0, full='dummy')), None), 107 'is_rest_9_8': (200, dict(version=dict(generation=9, major=8, minor=0, full='dummy')), None), 108 'is_zapi': (400, {}, "Unreachable"), 109 'empty_good': (200, {}, None), 110 'zero_record': (200, dict(records=[], num_records=0), None), 111 'one_record_uuid': (200, dict(records=[dict(uuid='a1b2c3')], num_records=1), None), 112 'end_of_sequence': (500, None, "Unexpected call to send_request"), 113 'generic_error': (400, None, "Expected error"), 114 'one_fpolicy_scope_record': (200, { 115 "records": [{ 116 'vserver': 'svm1', 117 'policy_name': 'policy1', 118 'export_policies_to_exclude': ['export1'], 119 'is_file_extension_check_on_directories_enabled': True, 120 'is_monitoring_of_objects_with_no_extension_enabled': False 121 }], 122 'num_records': 1 123 }, None) 124} 125 126 127# using pytest natively, without unittest.TestCase 128@pytest.fixture 129def patch_ansible(): 130 with patch.multiple(basic.AnsibleModule, 131 exit_json=exit_json, 132 fail_json=fail_json) as mocks: 133 yield mocks 134 135 136def get_fpolicy_scope_mock_object(cx_type='zapi', kind=None): 137 fpolicy_scope_obj = my_module() 138 if cx_type == 'zapi': 139 if kind is None: 140 fpolicy_scope_obj.server = MockONTAPConnection() 141 else: 142 fpolicy_scope_obj.server = MockONTAPConnection(kind=kind) 143 return fpolicy_scope_obj 144 145 146def test_module_fail_when_required_args_missing(patch_ansible): 147 ''' required arguments are reported as errors ''' 148 with pytest.raises(AnsibleFailJson) as exc: 149 set_module_args({}) 150 my_module() 151 print('Info: %s' % exc.value.args[0]['msg']) 152 153 154def test_ensure_get_called(patch_ansible): 155 ''' test get_fpolicy_scope for non-existent policy''' 156 args = dict(default_args()) 157 args['use_rest'] = 'never' 158 set_module_args(args) 159 print('starting') 160 my_obj = my_module() 161 print('use_rest:', my_obj.use_rest) 162 my_obj.server = MockONTAPConnection() 163 assert my_obj.get_fpolicy_scope is not None 164 165 166def test_rest_missing_arguments(patch_ansible): # pylint: disable=redefined-outer-name,unused-argument 167 ''' create fpolicy scope ''' 168 args = dict(default_args()) 169 del args['hostname'] 170 set_module_args(args) 171 with pytest.raises(AnsibleFailJson) as exc: 172 my_module() 173 msg = 'missing required arguments: hostname' 174 assert exc.value.args[0]['msg'] == msg 175 176 177@patch('ansible_collections.netapp.ontap.plugins.modules.na_ontap_fpolicy_scope.NetAppOntapFpolicyScope.create_fpolicy_scope') 178def test_successful_create(self, patch_ansible): 179 ''' creating fpolicy_scope and test idempotency ''' 180 args = dict(default_args()) 181 args['use_rest'] = 'never' 182 set_module_args(args) 183 my_obj = my_module() 184 my_obj.server = MockONTAPConnection() 185 with patch.object(my_module, 'create_fpolicy_scope', wraps=my_obj.create_fpolicy_scope) as mock_create: 186 with pytest.raises(AnsibleExitJson) as exc: 187 my_obj.apply() 188 print('Create: ' + repr(exc.value)) 189 assert exc.value.args[0]['changed'] 190 mock_create.assert_called_with() 191 # test idempotency 192 args = dict(default_args()) 193 args['use_rest'] = 'never' 194 set_module_args(args) 195 my_obj = my_module() 196 my_obj.server = MockONTAPConnection('fpolicy_scope') 197 with pytest.raises(AnsibleExitJson) as exc: 198 my_obj.apply() 199 print('Create: ' + repr(exc.value)) 200 assert not exc.value.args[0]['changed'] 201 202 203@patch('ansible_collections.netapp.ontap.plugins.modules.na_ontap_fpolicy_scope.NetAppOntapFpolicyScope.delete_fpolicy_scope') 204def test_successful_delete(self, patch_ansible): 205 ''' delete fpolicy_scope and test idempotency ''' 206 args = dict(default_args()) 207 args['use_rest'] = 'never' 208 args['state'] = 'absent' 209 set_module_args(args) 210 my_obj = my_module() 211 my_obj.server = MockONTAPConnection('fpolicy_scope') 212 with patch.object(my_module, 'delete_fpolicy_scope', wraps=my_obj.delete_fpolicy_scope) as mock_delete: 213 with pytest.raises(AnsibleExitJson) as exc: 214 my_obj.apply() 215 print('Delete: ' + repr(exc.value)) 216 assert exc.value.args[0]['changed'] 217 mock_delete.assert_called_with() 218 # test idempotency 219 args = dict(default_args()) 220 args['use_rest'] = 'never' 221 args['state'] = 'absent' 222 set_module_args(args) 223 my_obj = my_module() 224 my_obj.server = MockONTAPConnection() 225 with pytest.raises(AnsibleExitJson) as exc: 226 my_obj.apply() 227 print('Delete: ' + repr(exc.value)) 228 assert not exc.value.args[0]['changed'] 229 230 231@patch('ansible_collections.netapp.ontap.plugins.modules.na_ontap_fpolicy_scope.NetAppOntapFpolicyScope.modify_fpolicy_scope') 232def test_successful_modify(self, patch_ansible): 233 ''' modifying fpolicy_scope and testing idempotency ''' 234 args = dict(default_args()) 235 args['use_rest'] = 'never' 236 args['export_policies_to_exclude'] = 'export1,export2' 237 set_module_args(args) 238 my_obj = my_module() 239 my_obj.server = MockONTAPConnection('fpolicy_scope') 240 with patch.object(my_module, 'modify_fpolicy_scope', wraps=my_obj.modify_fpolicy_scope) as mock_modify: 241 with pytest.raises(AnsibleExitJson) as exc: 242 my_obj.apply() 243 print('Modify: ' + repr(exc.value)) 244 assert exc.value.args[0]['changed'] 245 mock_modify.assert_called_with({'export_policies_to_exclude': ['export1', 'export2']}) 246 # test idempotency 247 args = dict(default_args()) 248 args['use_rest'] = 'never' 249 set_module_args(args) 250 my_obj = my_module() 251 my_obj.server = MockONTAPConnection('fpolicy_scope') 252 with pytest.raises(AnsibleExitJson) as exc: 253 my_obj.apply() 254 print('Modify: ' + repr(exc.value)) 255 print(exc.value.args[0]['changed']) 256 assert not exc.value.args[0]['changed'] 257 258 259def test_if_all_methods_catch_exception(patch_ansible): 260 args = dict(default_args()) 261 args['use_rest'] = 'never' 262 set_module_args(args) 263 my_obj = my_module() 264 my_obj.server = MockONTAPConnection('fpolicy_scope_fail') 265 with pytest.raises(AnsibleFailJson) as exc: 266 my_obj.create_fpolicy_scope() 267 assert 'Error creating fPolicy policy scope ' in exc.value.args[0]['msg'] 268 with pytest.raises(AnsibleFailJson) as exc: 269 my_obj.delete_fpolicy_scope() 270 assert 'Error deleting fPolicy policy scope ' in exc.value.args[0]['msg'] 271 with pytest.raises(AnsibleFailJson) as exc: 272 my_obj.modify_fpolicy_scope(modify={}) 273 assert 'Error modifying fPolicy policy scope ' in exc.value.args[0]['msg'] 274 275 276@patch('ansible_collections.netapp.ontap.plugins.module_utils.netapp.OntapRestAPI.send_request') 277def test_rest_create(mock_request, patch_ansible): # pylint: disable=redefined-outer-name,unused-argument 278 ''' create fpolicy scope ''' 279 args = dict(default_args()) 280 set_module_args(args) 281 mock_request.side_effect = [ 282 SRR['is_rest'], 283 SRR['zero_record'], # get 284 SRR['empty_good'], # post 285 SRR['end_of_sequence'] 286 ] 287 my_obj = my_module() 288 with pytest.raises(AnsibleExitJson) as exc: 289 my_obj.apply() 290 assert exc.value.args[0]['changed'] is True 291 print(mock_request.mock_calls) 292 assert len(mock_request.mock_calls) == 3 293 294 295@patch('ansible_collections.netapp.ontap.plugins.module_utils.netapp.OntapRestAPI.send_request') 296def test_rest_create_no_action(mock_request, patch_ansible): # pylint: disable=redefined-outer-name,unused-argument 297 ''' create fpolicy scope idempotent ''' 298 args = dict(default_args()) 299 set_module_args(args) 300 mock_request.side_effect = [ 301 SRR['is_rest'], 302 SRR['one_fpolicy_scope_record'], # get 303 SRR['end_of_sequence'] 304 ] 305 my_obj = my_module() 306 with pytest.raises(AnsibleExitJson) as exc: 307 my_obj.apply() 308 assert exc.value.args[0]['changed'] is False 309 print(mock_request.mock_calls) 310 assert len(mock_request.mock_calls) == 2 311 312 313@patch('ansible_collections.netapp.ontap.plugins.module_utils.netapp.OntapRestAPI.send_request') 314def test_rest_delete_no_action(mock_request, patch_ansible): # pylint: disable=redefined-outer-name,unused-argument 315 ''' delete fpolicy scope ''' 316 args = dict(default_args()) 317 args['state'] = 'absent' 318 set_module_args(args) 319 mock_request.side_effect = [ 320 SRR['is_rest'], 321 SRR['zero_record'], # get 322 SRR['end_of_sequence'] 323 ] 324 my_obj = my_module() 325 with pytest.raises(AnsibleExitJson) as exc: 326 my_obj.apply() 327 assert exc.value.args[0]['changed'] is False 328 print(mock_request.mock_calls) 329 assert len(mock_request.mock_calls) == 2 330 331 332@patch('ansible_collections.netapp.ontap.plugins.module_utils.netapp.OntapRestAPI.send_request') 333def test_rest_delete(mock_request, patch_ansible): # pylint: disable=redefined-outer-name,unused-argument 334 ''' delete fpolicy scope ''' 335 args = dict(default_args()) 336 args['state'] = 'absent' 337 set_module_args(args) 338 mock_request.side_effect = [ 339 SRR['is_rest'], 340 SRR['one_fpolicy_scope_record'], # get 341 SRR['empty_good'], # delete 342 SRR['end_of_sequence'] 343 ] 344 my_obj = my_module() 345 with pytest.raises(AnsibleExitJson) as exc: 346 my_obj.apply() 347 assert exc.value.args[0]['changed'] is True 348 print(mock_request.mock_calls) 349 assert len(mock_request.mock_calls) == 3 350 351 352@patch('ansible_collections.netapp.ontap.plugins.module_utils.netapp.OntapRestAPI.send_request') 353def test_rest_modify_no_action(mock_request, patch_ansible): # pylint: disable=redefined-outer-name,unused-argument 354 ''' modify fpolicy scope ''' 355 args = dict(default_args()) 356 set_module_args(args) 357 mock_request.side_effect = [ 358 SRR['is_rest'], 359 SRR['one_fpolicy_scope_record'], # get 360 SRR['end_of_sequence'] 361 ] 362 my_obj = my_module() 363 with pytest.raises(AnsibleExitJson) as exc: 364 my_obj.apply() 365 assert exc.value.args[0]['changed'] is False 366 print(mock_request.mock_calls) 367 assert len(mock_request.mock_calls) == 2 368 369 370@patch('ansible_collections.netapp.ontap.plugins.module_utils.netapp.OntapRestAPI.send_request') 371def test_rest_modify_prepopulate(mock_request, patch_ansible): # pylint: disable=redefined-outer-name,unused-argument 372 ''' modify fpolicy scope ''' 373 args = dict(default_args()) 374 args['export_policies_to_exclude'] = 'export1,export2' 375 set_module_args(args) 376 mock_request.side_effect = [ 377 SRR['is_rest'], 378 SRR['one_fpolicy_scope_record'], # get 379 SRR['empty_good'], # patch 380 SRR['end_of_sequence'] 381 ] 382 my_obj = my_module() 383 with pytest.raises(AnsibleExitJson) as exc: 384 my_obj.apply() 385 assert exc.value.args[0]['changed'] is True 386 print(mock_request.mock_calls) 387 assert len(mock_request.mock_calls) == 3 388