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