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_nvme_snapshot''' 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_snapshot \ 17 import NetAppOntapSnapshot 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 == 'snapshot': 65 xml = self.build_snapshot_info() 66 elif self.type == 'snapshot_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_snapshot_info(): 73 ''' build xml data for snapshot-info ''' 74 xml = netapp_utils.zapi.NaElement('xml') 75 data = {'num-records': 1, 76 'attributes-list': {'snapshot-info': {'comment': 'new comment', 77 'name': 'ansible', 78 'snapmirror-label': 'label12'}}} 79 xml.translate_struct(data) 80 return xml 81 82 83class TestMyModule(unittest.TestCase): 84 ''' a group of related Unit Tests ''' 85 86 def setUp(self): 87 self.mock_module_helper = patch.multiple(basic.AnsibleModule, 88 exit_json=exit_json, 89 fail_json=fail_json) 90 self.mock_module_helper.start() 91 self.addCleanup(self.mock_module_helper.stop) 92 self.server = MockONTAPConnection() 93 self.onbox = False 94 95 def set_default_args(self): 96 if self.onbox: 97 hostname = '10.193.75.3' 98 username = 'admin' 99 password = 'netapp1!' 100 vserver = 'ansible' 101 volume = 'ansible' 102 snapshot = 'ansible' 103 comment = 'new comment' 104 snapmirror_label = 'label12' 105 else: 106 hostname = 'hostname' 107 username = 'username' 108 password = 'password' 109 vserver = 'vserver' 110 volume = 'ansible' 111 snapshot = 'ansible' 112 comment = 'new comment' 113 snapmirror_label = 'label12' 114 return dict({ 115 'hostname': hostname, 116 'username': username, 117 'password': password, 118 'vserver': vserver, 119 'volume': volume, 120 'snapshot': snapshot, 121 'comment': comment, 122 'snapmirror_label': snapmirror_label 123 }) 124 125 def test_module_fail_when_required_args_missing(self): 126 ''' required arguments are reported as errors ''' 127 with pytest.raises(AnsibleFailJson) as exc: 128 set_module_args({}) 129 my_module() 130 print('Info: %s' % exc.value.args[0]['msg']) 131 132 def test_ensure_get_called(self): 133 ''' test get_snapshot() for non-existent snapshot''' 134 set_module_args(self.set_default_args()) 135 my_obj = my_module() 136 my_obj.server = self.server 137 assert my_obj.get_snapshot() is None 138 139 def test_ensure_get_called_existing(self): 140 ''' test get_snapshot() for existing snapshot''' 141 set_module_args(self.set_default_args()) 142 my_obj = my_module() 143 my_obj.server = MockONTAPConnection(kind='snapshot') 144 assert my_obj.get_snapshot() 145 146 @patch('ansible.modules.storage.netapp.na_ontap_snapshot.NetAppOntapSnapshot.create_snapshot') 147 def test_successful_create(self, create_snapshot): 148 ''' creating snapshot and testing idempotency ''' 149 set_module_args(self.set_default_args()) 150 my_obj = my_module() 151 if not self.onbox: 152 my_obj.server = self.server 153 with pytest.raises(AnsibleExitJson) as exc: 154 my_obj.apply() 155 assert exc.value.args[0]['changed'] 156 create_snapshot.assert_called_with() 157 # to reset na_helper from remembering the previous 'changed' value 158 my_obj = my_module() 159 if not self.onbox: 160 my_obj.server = MockONTAPConnection('snapshot') 161 with pytest.raises(AnsibleExitJson) as exc: 162 my_obj.apply() 163 assert not exc.value.args[0]['changed'] 164 165 @patch('ansible.modules.storage.netapp.na_ontap_snapshot.NetAppOntapSnapshot.modify_snapshot') 166 def test_successful_modify(self, modify_snapshot): 167 ''' modifying snapshot and testing idempotency ''' 168 data = self.set_default_args() 169 data['comment'] = 'adding comment' 170 data['snapmirror_label'] = 'label22' 171 set_module_args(data) 172 my_obj = my_module() 173 if not self.onbox: 174 my_obj.server = MockONTAPConnection('snapshot') 175 with pytest.raises(AnsibleExitJson) as exc: 176 my_obj.apply() 177 assert exc.value.args[0]['changed'] 178 modify_snapshot.assert_called_with() 179 # to reset na_helper from remembering the previous 'changed' value 180 data['comment'] = 'new comment' 181 data['snapmirror_label'] = 'label12' 182 set_module_args(data) 183 my_obj = my_module() 184 if not self.onbox: 185 my_obj.server = MockONTAPConnection('snapshot') 186 with pytest.raises(AnsibleExitJson) as exc: 187 my_obj.apply() 188 assert not exc.value.args[0]['changed'] 189 190 @patch('ansible.modules.storage.netapp.na_ontap_snapshot.NetAppOntapSnapshot.delete_snapshot') 191 def test_successful_delete(self, delete_snapshot): 192 ''' deleting snapshot and testing idempotency ''' 193 data = self.set_default_args() 194 data['state'] = 'absent' 195 set_module_args(data) 196 my_obj = my_module() 197 if not self.onbox: 198 my_obj.server = MockONTAPConnection('snapshot') 199 with pytest.raises(AnsibleExitJson) as exc: 200 my_obj.apply() 201 assert exc.value.args[0]['changed'] 202 delete_snapshot.assert_called_with() 203 # to reset na_helper from remembering the previous 'changed' value 204 my_obj = my_module() 205 if not self.onbox: 206 my_obj.server = self.server 207 with pytest.raises(AnsibleExitJson) as exc: 208 my_obj.apply() 209 assert not exc.value.args[0]['changed'] 210 211 def test_if_all_methods_catch_exception(self): 212 module_args = {} 213 module_args.update(self.set_default_args()) 214 set_module_args(module_args) 215 my_obj = my_module() 216 if not self.onbox: 217 my_obj.server = MockONTAPConnection('snapshot_fail') 218 with pytest.raises(AnsibleFailJson) as exc: 219 my_obj.create_snapshot() 220 assert 'Error creating snapshot ansible:' in exc.value.args[0]['msg'] 221 with pytest.raises(AnsibleFailJson) as exc: 222 my_obj.delete_snapshot() 223 assert 'Error deleting snapshot ansible:' in exc.value.args[0]['msg'] 224 with pytest.raises(AnsibleFailJson) as exc: 225 my_obj.modify_snapshot() 226 assert 'Error modifying snapshot ansible:' in exc.value.args[0]['msg'] 227