1''' unit test for Ansible module: na_elementsw_volume.py ''' 2 3from __future__ import absolute_import, division, print_function 4__metaclass__ = type 5 6import json 7import pytest 8 9from ansible.module_utils import basic 10from ansible.module_utils._text import to_bytes 11from ansible_collections.netapp.elementsw.tests.unit.compat import unittest 12from ansible_collections.netapp.elementsw.tests.unit.compat.mock import patch 13import ansible_collections.netapp.elementsw.plugins.module_utils.netapp as netapp_utils 14 15if not netapp_utils.has_sf_sdk(): 16 pytestmark = pytest.mark.skip('skipping as missing required SolidFire Python SDK') 17 18from ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_volume \ 19 import ElementSWVolume as my_module # module under test 20 21 22def set_module_args(args): 23 """prepare arguments so that they will be picked up during module creation""" 24 args = json.dumps({'ANSIBLE_MODULE_ARGS': args}) 25 basic._ANSIBLE_ARGS = to_bytes(args) # pylint: disable=protected-access 26 27 28class AnsibleExitJson(Exception): 29 """Exception class to be raised by module.exit_json and caught by the test case""" 30 31 32class AnsibleFailJson(Exception): 33 """Exception class to be raised by module.fail_json and caught by the test case""" 34 35 36def exit_json(*args, **kwargs): # pylint: disable=unused-argument 37 """function to patch over exit_json; package return data into an exception""" 38 if 'changed' not in kwargs: 39 kwargs['changed'] = False 40 raise AnsibleExitJson(kwargs) 41 42 43def fail_json(*args, **kwargs): # pylint: disable=unused-argument 44 """function to patch over fail_json; package return data into an exception""" 45 kwargs['failed'] = True 46 raise AnsibleFailJson(kwargs) 47 48 49CREATE_ERROR = 'create', 'some_error_in_create_volume' 50MODIFY_ERROR = 'modify', 'some_error_in_modify_volume' 51DELETE_ERROR = 'delete', 'some_error_in_delete_volume' 52 53POLICY_ID = 888 54POLICY_NAME = 'element_qos_policy_name' 55VOLUME_ID = 777 56VOLUME_NAME = 'element_volume_name' 57 58 59class MockSFConnection(object): 60 ''' mock connection to ElementSW host ''' 61 62 class Bunch(object): # pylint: disable=too-few-public-methods 63 ''' create object with arbitrary attributes ''' 64 def __init__(self, **kw): 65 ''' called with (k1=v1, k2=v2), creates obj.k1, obj.k2 with values v1, v2 ''' 66 setattr(self, '__dict__', kw) 67 68 def __init__(self, force_error=False, where=None, with_qos_policy_id=True): 69 ''' save arguments ''' 70 self.force_error = force_error 71 self.where = where 72 self.with_qos_policy_id = with_qos_policy_id 73 74 def list_qos_policies(self, *args, **kwargs): # pylint: disable=unused-argument 75 ''' build qos_policy list ''' 76 qos_policy_name = POLICY_NAME 77 qos = self.Bunch(min_iops=1000, max_iops=20000, burst_iops=20000) 78 qos_policy = self.Bunch(name=qos_policy_name, qos_policy_id=POLICY_ID, qos=qos) 79 qos_policy_1 = self.Bunch(name=qos_policy_name + '_1', qos_policy_id=POLICY_ID + 1, qos=qos) 80 qos_policies = [qos_policy, qos_policy_1] 81 qos_policy_list = self.Bunch(qos_policies=qos_policies) 82 return qos_policy_list 83 84 def list_volumes_for_account(self, *args, **kwargs): # pylint: disable=unused-argument 85 ''' build volume list: volume.name, volume.id ''' 86 volume = self.Bunch(name=VOLUME_NAME, volume_id=VOLUME_ID, delete_time='') 87 volumes = [volume] 88 volume_list = self.Bunch(volumes=volumes) 89 return volume_list 90 91 def list_volumes(self, *args, **kwargs): # pylint: disable=unused-argument 92 ''' build volume details: volume.name, volume.id ''' 93 if self.with_qos_policy_id: 94 qos_policy_id = POLICY_ID 95 else: 96 qos_policy_id = None 97 qos = self.Bunch(min_iops=1000, max_iops=20000, burst_iops=20000) 98 volume = self.Bunch(name=VOLUME_NAME, volume_id=VOLUME_ID, delete_time='', access='rw', 99 account_id=1, qos=qos, qos_policy_id=qos_policy_id, total_size=1000000000, 100 attributes={'config-mgmt': 'ansible', 'event-source': 'na_elementsw_volume'} 101 ) 102 volumes = [volume] 103 volume_list = self.Bunch(volumes=volumes) 104 return volume_list 105 106 def get_account_by_name(self, *args, **kwargs): # pylint: disable=unused-argument 107 ''' returns account_id ''' 108 if self.force_error and 'get_account_id' in self.where: 109 account_id = None 110 else: 111 account_id = 1 112 account = self.Bunch(account_id=account_id) 113 result = self.Bunch(account=account) 114 return result 115 116 def create_volume(self, *args, **kwargs): # pylint: disable=unused-argument 117 ''' We don't check the return code, but could force an exception ''' 118 if self.force_error and 'create_exception' in self.where: 119 raise netapp_utils.solidfire.common.ApiServerError(*CREATE_ERROR) 120 121 def modify_volume(self, *args, **kwargs): # pylint: disable=unused-argument 122 ''' We don't check the return code, but could force an exception ''' 123 print("modify: %s, %s " % (repr(args), repr(kwargs))) 124 if self.force_error and 'modify_exception' in self.where: 125 raise netapp_utils.solidfire.common.ApiServerError(*MODIFY_ERROR) 126 127 def delete_volume(self, *args, **kwargs): # pylint: disable=unused-argument 128 ''' We don't check the return code, but could force an exception ''' 129 if self.force_error and 'delete_exception' in self.where: 130 raise netapp_utils.solidfire.common.ApiServerError(*DELETE_ERROR) 131 132 def purge_deleted_volume(self, *args, **kwargs): # pylint: disable=unused-argument 133 ''' We don't check the return code, but could force an exception ''' 134 if self.force_error and 'delete_exception' in self.where: 135 raise netapp_utils.solidfire.common.ApiServerError(*DELETE_ERROR) 136 137 138class TestMyModule(unittest.TestCase): 139 ''' a group of related Unit Tests ''' 140 141 ARGS = { 142 'state': 'present', 143 'name': VOLUME_NAME, 144 'account_id': 'element_account_id', 145 'qos': {'minIOPS': 1000, 'maxIOPS': 20000, 'burstIOPS': 20000}, 146 'qos_policy_name': POLICY_NAME, 147 'size': 1, 148 'enable512e': True, 149 'hostname': 'hostname', 150 'username': 'username', 151 'password': 'password', 152 } 153 154 def setUp(self): 155 self.mock_module_helper = patch.multiple(basic.AnsibleModule, 156 exit_json=exit_json, 157 fail_json=fail_json) 158 self.mock_module_helper.start() 159 self.addCleanup(self.mock_module_helper.stop) 160 161 def test_module_fail_when_required_args_missing(self): 162 ''' required arguments are reported as errors ''' 163 with pytest.raises(AnsibleFailJson) as exc: 164 set_module_args({}) 165 my_module() 166 print('Info: %s' % exc.value.args[0]['msg']) 167 168 @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection') 169 def test_add_volume(self, mock_create_sf_connection): 170 ''' adding a volume ''' 171 args = dict(self.ARGS) # deep copy as other tests can modify args 172 args['name'] += '_1' # new name to force a create 173 args.pop('qos') # parameters are mutually exclusive: qos|qos_policy_name 174 set_module_args(args) 175 # my_obj.sfe will be assigned a MockSFConnection object: 176 mock_create_sf_connection.return_value = MockSFConnection() 177 my_obj = my_module() 178 with pytest.raises(AnsibleExitJson) as exc: 179 my_obj.apply() 180 print(exc.value.args[0]) 181 assert exc.value.args[0]['changed'] 182 183 @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection') 184 def test_add_or_modify_volume_idempotent_qos_policy(self, mock_create_sf_connection): 185 ''' adding a volume ''' 186 args = dict(self.ARGS) 187 args.pop('qos') # parameters are mutually exclusive: qos|qos_policy_name 188 set_module_args(args) 189 # my_obj.sfe will be assigned a MockSFConnection object: 190 mock_create_sf_connection.return_value = MockSFConnection() 191 my_obj = my_module() 192 with pytest.raises(AnsibleExitJson) as exc: 193 my_obj.apply() 194 print(exc.value.args[0]) 195 assert not exc.value.args[0]['changed'] 196 197 @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection') 198 def test_add_or_modify_volume_idempotent_qos(self, mock_create_sf_connection): 199 ''' adding a volume ''' 200 args = dict(self.ARGS) 201 args.pop('qos_policy_name') # parameters are mutually exclusive: qos|qos_policy_name 202 set_module_args(args) 203 # my_obj.sfe will be assigned a MockSFConnection object: 204 mock_create_sf_connection.return_value = MockSFConnection(with_qos_policy_id=False) 205 my_obj = my_module() 206 with pytest.raises(AnsibleExitJson) as exc: 207 my_obj.apply() 208 print(exc.value.args[0]) 209 assert not exc.value.args[0]['changed'] 210 211 @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection') 212 def test_delete_volume(self, mock_create_sf_connection): 213 ''' removing a volume ''' 214 args = dict(self.ARGS) 215 args['state'] = 'absent' 216 args.pop('qos') # parameters are mutually exclusive: qos|qos_policy_name 217 set_module_args(args) 218 # my_obj.sfe will be assigned a MockSFConnection object: 219 mock_create_sf_connection.return_value = MockSFConnection() 220 my_obj = my_module() 221 with pytest.raises(AnsibleExitJson) as exc: 222 my_obj.apply() 223 print(exc.value.args[0]) 224 assert exc.value.args[0]['changed'] 225 226 @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection') 227 def test_delete_volume_idempotent(self, mock_create_sf_connection): 228 ''' removing a volume ''' 229 args = dict(self.ARGS) 230 args['state'] = 'absent' 231 args['name'] += '_1' # new name to force idempotency 232 args.pop('qos') # parameters are mutually exclusive: qos|qos_policy_name 233 set_module_args(args) 234 # my_obj.sfe will be assigned a MockSFConnection object: 235 mock_create_sf_connection.return_value = MockSFConnection() 236 my_obj = my_module() 237 with pytest.raises(AnsibleExitJson) as exc: 238 my_obj.apply() 239 print(exc.value.args[0]) 240 assert not exc.value.args[0]['changed'] 241 242 @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection') 243 def test_modify_volume_qos(self, mock_create_sf_connection): 244 ''' modifying a volume ''' 245 args = dict(self.ARGS) 246 args['qos'] = {'minIOPS': 2000} 247 args.pop('qos_policy_name') # parameters are mutually exclusive: qos|qos_policy_name 248 set_module_args(args) 249 # my_obj.sfe will be assigned a MockSFConnection object: 250 mock_create_sf_connection.return_value = MockSFConnection(with_qos_policy_id=False) 251 my_obj = my_module() 252 with pytest.raises(AnsibleExitJson) as exc: 253 my_obj.apply() 254 print(exc.value.args[0]) 255 assert exc.value.args[0]['changed'] 256 257 @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection') 258 def test_modify_volume_qos_policy_to_qos(self, mock_create_sf_connection): 259 ''' modifying a volume ''' 260 args = dict(self.ARGS) 261 args['qos'] = {'minIOPS': 2000} 262 args.pop('qos_policy_name') # parameters are mutually exclusive: qos|qos_policy_name 263 set_module_args(args) 264 # my_obj.sfe will be assigned a MockSFConnection object: 265 mock_create_sf_connection.return_value = MockSFConnection() 266 my_obj = my_module() 267 with pytest.raises(AnsibleExitJson) as exc: 268 my_obj.apply() 269 print(exc.value.args[0]) 270 assert exc.value.args[0]['changed'] 271 272 @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection') 273 def test_modify_volume_qos_policy(self, mock_create_sf_connection): 274 ''' modifying a volume ''' 275 args = dict(self.ARGS) 276 args['qos_policy_name'] += '_1' 277 args.pop('qos') # parameters are mutually exclusive: qos|qos_policy_name 278 set_module_args(args) 279 # my_obj.sfe will be assigned a MockSFConnection object: 280 mock_create_sf_connection.return_value = MockSFConnection() 281 my_obj = my_module() 282 with pytest.raises(AnsibleExitJson) as exc: 283 my_obj.apply() 284 print(exc.value.args[0]) 285 assert exc.value.args[0]['changed'] 286 287 @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection') 288 def test_modify_volume_qos_to_qos_policy(self, mock_create_sf_connection): 289 ''' modifying a volume ''' 290 args = dict(self.ARGS) 291 args.pop('qos') # parameters are mutually exclusive: qos|qos_policy_name 292 set_module_args(args) 293 # my_obj.sfe will be assigned a MockSFConnection object: 294 mock_create_sf_connection.return_value = MockSFConnection(with_qos_policy_id=False) 295 my_obj = my_module() 296 with pytest.raises(AnsibleExitJson) as exc: 297 my_obj.apply() 298 print(exc.value.args[0]) 299 assert exc.value.args[0]['changed'] 300 301 @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection') 302 def test_create_volume_exception(self, mock_create_sf_connection): 303 ''' creating a volume can raise an exception ''' 304 args = dict(self.ARGS) 305 args['name'] += '_1' # new name to force a create 306 args.pop('qos') # parameters are mutually exclusive: qos|qos_policy_name 307 set_module_args(args) 308 # my_obj.sfe will be assigned a MockSFConnection object: 309 mock_create_sf_connection.return_value = MockSFConnection(force_error=True, where=['create_exception']) 310 my_obj = my_module() 311 with pytest.raises(AnsibleFailJson) as exc: 312 my_obj.apply() 313 print(exc.value.args[0]) 314 message = 'Error provisioning volume: %s' % args['name'] 315 assert exc.value.args[0]['msg'].startswith(message) 316 317 @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection') 318 def test_modify_volume_exception(self, mock_create_sf_connection): 319 ''' modifying a volume can raise an exception ''' 320 args = dict(self.ARGS) 321 args['qos'] = {'minIOPS': 2000} 322 args.pop('qos_policy_name') # parameters are mutually exclusive: qos|qos_policy_name 323 set_module_args(args) 324 # my_obj.sfe will be assigned a MockSFConnection object: 325 mock_create_sf_connection.return_value = MockSFConnection(force_error=True, where=['modify_exception']) 326 my_obj = my_module() 327 with pytest.raises(AnsibleFailJson) as exc: 328 my_obj.apply() 329 print(exc.value.args[0]) 330 message = 'Error updating volume: %s' % VOLUME_ID 331 assert exc.value.args[0]['msg'].startswith(message) 332 333 @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection') 334 def test_delete_volume_exception(self, mock_create_sf_connection): 335 ''' deleting a volume can raise an exception ''' 336 args = dict(self.ARGS) 337 args['state'] = 'absent' 338 args.pop('qos') # parameters are mutually exclusive: qos|qos_policy_name 339 set_module_args(args) 340 # my_obj.sfe will be assigned a MockSFConnection object: 341 mock_create_sf_connection.return_value = MockSFConnection(force_error=True, where=['delete_exception']) 342 my_obj = my_module() 343 with pytest.raises(AnsibleFailJson) as exc: 344 my_obj.apply() 345 print(exc.value.args[0]) 346 message = 'Error deleting volume: %s' % VOLUME_ID 347 assert exc.value.args[0]['msg'].startswith(message) 348 349 @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection') 350 def test_check_error_reporting_on_non_existent_qos_policy(self, mock_create_sf_connection): 351 ''' report error if qos option is not given on create ''' 352 args = dict(self.ARGS) 353 args['name'] += '_1' # new name to force a create 354 args.pop('qos') 355 args['qos_policy_name'] += '_2' 356 set_module_args(args) 357 # my_obj.sfe will be assigned a MockSFConnection object: 358 mock_create_sf_connection.return_value = MockSFConnection() 359 my_obj = my_module() 360 with pytest.raises(AnsibleFailJson) as exc: 361 my_obj.apply() 362 print(exc.value.args[0]) 363 message = "Cannot find qos policy with name/id: %s" % args['qos_policy_name'] 364 assert exc.value.args[0]['msg'] == message 365