1# (c) 2018, NetApp, Inc 2# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) 3 4''' unit tests ONTAP Ansible module: na_ontap_volume_clone''' 5 6from __future__ import print_function 7import json 8import pytest 9 10from units.compat import unittest 11from units.compat.mock import patch 12from ansible.module_utils import basic 13from ansible.module_utils._text import to_bytes 14import ansible.module_utils.netapp as netapp_utils 15 16from ansible.modules.storage.netapp.na_ontap_volume_clone \ 17 import NetAppONTAPVolumeClone as my_module 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 pass 32 33 34class AnsibleFailJson(Exception): 35 """Exception class to be raised by module.fail_json and caught by the test case""" 36 pass 37 38 39def exit_json(*args, **kwargs): # pylint: disable=unused-argument 40 """function to patch over exit_json; package return data into an exception""" 41 if 'changed' not in kwargs: 42 kwargs['changed'] = False 43 raise AnsibleExitJson(kwargs) 44 45 46def fail_json(*args, **kwargs): # pylint: disable=unused-argument 47 """function to patch over fail_json; package return data into an exception""" 48 kwargs['failed'] = True 49 raise AnsibleFailJson(kwargs) 50 51 52class MockONTAPConnection(object): 53 ''' mock server connection to ONTAP host ''' 54 55 def __init__(self, kind=None): 56 ''' save arguments ''' 57 self.type = kind 58 self.xml_in = None 59 self.xml_out = None 60 61 def invoke_successfully(self, xml, enable_tunneling): # pylint: disable=unused-argument 62 ''' mock invoke_successfully returning xml data ''' 63 self.xml_in = xml 64 if self.type == 'volume_clone': 65 xml = self.build_volume_clone_info() 66 elif self.type == 'volume_clone_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_volume_clone_info(): 73 ''' build xml data for volume-clone-info ''' 74 xml = netapp_utils.zapi.NaElement('xml') 75 data = {'attributes': {'volume-clone-info': {'volume': 'ansible', 76 'parent-volume': 'ansible'}}} 77 xml.translate_struct(data) 78 return xml 79 80 81class TestMyModule(unittest.TestCase): 82 ''' a group of related Unit Tests ''' 83 84 def setUp(self): 85 self.mock_module_helper = patch.multiple(basic.AnsibleModule, 86 exit_json=exit_json, 87 fail_json=fail_json) 88 self.mock_module_helper.start() 89 self.addCleanup(self.mock_module_helper.stop) 90 self.server = MockONTAPConnection() 91 self.onbox = False 92 93 def set_default_args(self): 94 if self.onbox: 95 hostname = '10.10.10.10' 96 username = 'username' 97 password = 'password' 98 vserver = 'ansible' 99 volume = 'ansible' 100 parent_volume = 'ansible' 101 else: 102 hostname = '10.10.10.10' 103 username = 'username' 104 password = 'password' 105 vserver = 'ansible' 106 volume = 'ansible' 107 parent_volume = 'ansible' 108 return dict({ 109 'hostname': hostname, 110 'username': username, 111 'password': password, 112 'vserver': vserver, 113 'volume': volume, 114 'parent_volume': parent_volume 115 }) 116 117 def test_module_fail_when_required_args_missing(self): 118 ''' required arguments are reported as errors ''' 119 with pytest.raises(AnsibleFailJson) as exc: 120 set_module_args({}) 121 my_module() 122 print('Info: %s' % exc.value.args[0]['msg']) 123 124 def test_ensure_get_called(self): 125 ''' test get_volume_clone() for non-existent volume clone''' 126 set_module_args(self.set_default_args()) 127 my_obj = my_module() 128 my_obj.server = self.server 129 assert my_obj.get_volume_clone() is None 130 131 def test_ensure_get_called_existing(self): 132 ''' test get_volume_clone() for existing volume clone''' 133 set_module_args(self.set_default_args()) 134 my_obj = my_module() 135 my_obj.server = MockONTAPConnection(kind='volume_clone') 136 assert my_obj.get_volume_clone() 137 138 @patch('ansible.modules.storage.netapp.na_ontap_volume_clone.NetAppONTAPVolumeClone.create_volume_clone') 139 def test_successful_create(self, create_volume_clone): 140 ''' creating volume_clone and testing idempotency ''' 141 module_args = { 142 'parent_snapshot': 'abc', 143 'parent_vserver': 'abc', 144 'volume_type': 'dp', 145 'qos_policy_group_name': 'abc', 146 'junction_path': 'abc', 147 'uid': '1', 148 'gid': '1' 149 } 150 module_args.update(self.set_default_args()) 151 set_module_args(module_args) 152 my_obj = my_module() 153 if not self.onbox: 154 my_obj.server = self.server 155 with pytest.raises(AnsibleExitJson) as exc: 156 my_obj.apply() 157 assert exc.value.args[0]['changed'] 158 create_volume_clone.assert_called_with() 159 # to reset na_helper from remembering the previous 'changed' value 160 my_obj = my_module() 161 if not self.onbox: 162 my_obj.server = MockONTAPConnection('volume_clone') 163 with pytest.raises(AnsibleExitJson) as exc: 164 my_obj.apply() 165 assert not exc.value.args[0]['changed'] 166 167 def test_if_all_methods_catch_exception(self): 168 module_args = {} 169 module_args.update(self.set_default_args()) 170 set_module_args(module_args) 171 my_obj = my_module() 172 if not self.onbox: 173 my_obj.server = MockONTAPConnection('volume_clone_fail') 174 with pytest.raises(AnsibleFailJson) as exc: 175 my_obj.get_volume_clone() 176 assert 'Error fetching volume clone information ' in exc.value.args[0]['msg'] 177 with pytest.raises(AnsibleFailJson) as exc: 178 my_obj.create_volume_clone() 179 assert 'Error creating volume clone: ' in exc.value.args[0]['msg'] 180