summaryrefslogtreecommitdiffstats
path: root/collections-debian-merged/ansible_collections/netapp/elementsw/tests
diff options
context:
space:
mode:
Diffstat (limited to 'collections-debian-merged/ansible_collections/netapp/elementsw/tests')
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/compat/__init__.py0
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/compat/builtins.py33
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/compat/mock.py122
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/compat/unittest.py38
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_access_group.py175
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_access_group_volumes.py245
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_account.py137
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster.py228
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster_config.py157
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster_snmp.py176
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_info.py328
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_initiators.py201
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_network_interfaces.py293
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_nodes.py324
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_qos_policy.py300
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_template.py138
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_vlan.py343
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_volume.py364
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules_utils/test_netapp_module.py149
-rw-r--r--collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/requirements.txt2
20 files changed, 3753 insertions, 0 deletions
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/compat/__init__.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/compat/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/compat/__init__.py
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/compat/builtins.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/compat/builtins.py
new file mode 100644
index 00000000..f60ee678
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/compat/builtins.py
@@ -0,0 +1,33 @@
+# (c) 2014, Toshio Kuratomi <tkuratomi@ansible.com>
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+#
+# Compat for python2.7
+#
+
+# One unittest needs to import builtins via __import__() so we need to have
+# the string that represents it
+try:
+ import __builtin__
+except ImportError:
+ BUILTINS = 'builtins'
+else:
+ BUILTINS = '__builtin__'
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/compat/mock.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/compat/mock.py
new file mode 100644
index 00000000..0972cd2e
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/compat/mock.py
@@ -0,0 +1,122 @@
+# (c) 2014, Toshio Kuratomi <tkuratomi@ansible.com>
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+'''
+Compat module for Python3.x's unittest.mock module
+'''
+import sys
+
+# Python 2.7
+
+# Note: Could use the pypi mock library on python3.x as well as python2.x. It
+# is the same as the python3 stdlib mock library
+
+try:
+ # Allow wildcard import because we really do want to import all of mock's
+ # symbols into this compat shim
+ # pylint: disable=wildcard-import,unused-wildcard-import
+ from unittest.mock import *
+except ImportError:
+ # Python 2
+ # pylint: disable=wildcard-import,unused-wildcard-import
+ try:
+ from mock import *
+ except ImportError:
+ print('You need the mock library installed on python2.x to run tests')
+
+
+# Prior to 3.4.4, mock_open cannot handle binary read_data
+if sys.version_info >= (3,) and sys.version_info < (3, 4, 4):
+ file_spec = None
+
+ def _iterate_read_data(read_data):
+ # Helper for mock_open:
+ # Retrieve lines from read_data via a generator so that separate calls to
+ # readline, read, and readlines are properly interleaved
+ sep = b'\n' if isinstance(read_data, bytes) else '\n'
+ data_as_list = [l + sep for l in read_data.split(sep)]
+
+ if data_as_list[-1] == sep:
+ # If the last line ended in a newline, the list comprehension will have an
+ # extra entry that's just a newline. Remove this.
+ data_as_list = data_as_list[:-1]
+ else:
+ # If there wasn't an extra newline by itself, then the file being
+ # emulated doesn't have a newline to end the last line remove the
+ # newline that our naive format() added
+ data_as_list[-1] = data_as_list[-1][:-1]
+
+ for line in data_as_list:
+ yield line
+
+ def mock_open(mock=None, read_data=''):
+ """
+ A helper function to create a mock to replace the use of `open`. It works
+ for `open` called directly or used as a context manager.
+
+ The `mock` argument is the mock object to configure. If `None` (the
+ default) then a `MagicMock` will be created for you, with the API limited
+ to methods or attributes available on standard file handles.
+
+ `read_data` is a string for the `read` methoddline`, and `readlines` of the
+ file handle to return. This is an empty string by default.
+ """
+ def _readlines_side_effect(*args, **kwargs):
+ if handle.readlines.return_value is not None:
+ return handle.readlines.return_value
+ return list(_data)
+
+ def _read_side_effect(*args, **kwargs):
+ if handle.read.return_value is not None:
+ return handle.read.return_value
+ return type(read_data)().join(_data)
+
+ def _readline_side_effect():
+ if handle.readline.return_value is not None:
+ while True:
+ yield handle.readline.return_value
+ for line in _data:
+ yield line
+
+ global file_spec
+ if file_spec is None:
+ import _io
+ file_spec = list(set(dir(_io.TextIOWrapper)).union(set(dir(_io.BytesIO))))
+
+ if mock is None:
+ mock = MagicMock(name='open', spec=open)
+
+ handle = MagicMock(spec=file_spec)
+ handle.__enter__.return_value = handle
+
+ _data = _iterate_read_data(read_data)
+
+ handle.write.return_value = None
+ handle.read.return_value = None
+ handle.readline.return_value = None
+ handle.readlines.return_value = None
+
+ handle.read.side_effect = _read_side_effect
+ handle.readline.side_effect = _readline_side_effect()
+ handle.readlines.side_effect = _readlines_side_effect
+
+ mock.return_value = handle
+ return mock
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/compat/unittest.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/compat/unittest.py
new file mode 100644
index 00000000..98f08ad6
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/compat/unittest.py
@@ -0,0 +1,38 @@
+# (c) 2014, Toshio Kuratomi <tkuratomi@ansible.com>
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+'''
+Compat module for Python2.7's unittest module
+'''
+
+import sys
+
+# Allow wildcard import because we really do want to import all of
+# unittests's symbols into this compat shim
+# pylint: disable=wildcard-import,unused-wildcard-import
+if sys.version_info < (2, 7):
+ try:
+ # Need unittest2 on python2.6
+ from unittest2 import *
+ except ImportError:
+ print('You need unittest2 installed on python2.6.x to run tests')
+else:
+ from unittest import *
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_access_group.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_access_group.py
new file mode 100644
index 00000000..0bd1e255
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_access_group.py
@@ -0,0 +1,175 @@
+''' unit test for Ansible module: na_elementsw_account.py '''
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+import json
+import pytest
+
+from ansible_collections.netapp.elementsw.tests.unit.compat import unittest
+from ansible_collections.netapp.elementsw.tests.unit.compat.mock import patch
+from ansible.module_utils import basic
+from ansible.module_utils._text import to_bytes
+import ansible_collections.netapp.elementsw.plugins.module_utils.netapp as netapp_utils
+
+if not netapp_utils.has_sf_sdk():
+ pytestmark = pytest.mark.skip('skipping as missing required SolidFire Python SDK')
+
+from ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_access_group \
+ import ElementSWAccessGroup as my_module # module under test
+
+
+def set_module_args(args):
+ """prepare arguments so that they will be picked up during module creation"""
+ args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
+ basic._ANSIBLE_ARGS = to_bytes(args) # pylint: disable=protected-access
+
+
+class AnsibleExitJson(Exception):
+ """Exception class to be raised by module.exit_json and caught by the test case"""
+ pass
+
+
+class AnsibleFailJson(Exception):
+ """Exception class to be raised by module.fail_json and caught by the test case"""
+ pass
+
+
+def exit_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over exit_json; package return data into an exception"""
+ if 'changed' not in kwargs:
+ kwargs['changed'] = False
+ raise AnsibleExitJson(kwargs)
+
+
+def fail_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over fail_json; package return data into an exception"""
+ kwargs['failed'] = True
+ raise AnsibleFailJson(kwargs)
+
+
+ADD_ERROR = 'some_error_in_add_access_group'
+
+
+class MockSFConnection(object):
+ ''' mock connection to ElementSW host '''
+
+ class Bunch(object): # pylint: disable=too-few-public-methods
+ ''' create object with arbitrary attributes '''
+ def __init__(self, **kw):
+ ''' called with (k1=v1, k2=v2), creates obj.k1, obj.k2 with values v1, v2 '''
+ setattr(self, '__dict__', kw)
+
+ def __init__(self, force_error=False, where=None):
+ ''' save arguments '''
+ self.force_error = force_error
+ self.where = where
+
+ def list_volume_access_groups(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' build access_group list: access_groups.name, access_groups.account_id '''
+ access_groups = list()
+ access_group_list = self.Bunch(volume_access_groups=access_groups)
+ return access_group_list
+
+ def create_volume_access_group(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' We don't check the return code, but could force an exception '''
+ if self.force_error and 'add' in self.where:
+ # The module does not check for a specific exception :(
+ raise OSError(ADD_ERROR)
+
+ def get_account_by_name(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' returns account_id '''
+ if self.force_error and 'account_id' in self.where:
+ account_id = None
+ else:
+ account_id = 1
+ print('account_id', account_id)
+ account = self.Bunch(account_id=account_id)
+ result = self.Bunch(account=account)
+ return result
+
+
+class TestMyModule(unittest.TestCase):
+ ''' a group of related Unit Tests '''
+
+ def setUp(self):
+ self.mock_module_helper = patch.multiple(basic.AnsibleModule,
+ exit_json=exit_json,
+ fail_json=fail_json)
+ self.mock_module_helper.start()
+ self.addCleanup(self.mock_module_helper.stop)
+
+ def test_module_fail_when_required_args_missing(self):
+ ''' required arguments are reported as errors '''
+ with pytest.raises(AnsibleFailJson) as exc:
+ set_module_args({})
+ my_module()
+ print('Info: %s' % exc.value.args[0]['msg'])
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_ensure_command_called(self, mock_create_sf_connection):
+ ''' a more interesting test '''
+ set_module_args({
+ 'state': 'present',
+ 'name': 'element_groupname',
+ 'account_id': 'element_account_id',
+ 'hostname': 'hostname',
+ 'username': 'username',
+ 'password': 'password',
+ })
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ # It may not be a good idea to start with apply
+ # More atomic methods can be easier to mock
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_check_error_reporting_on_add_exception(self, mock_create_sf_connection):
+ ''' a more interesting test '''
+ set_module_args({
+ 'state': 'present',
+ 'name': 'element_groupname',
+ 'account_id': 'element_account_id',
+ 'hostname': 'hostname',
+ 'username': 'username',
+ 'password': 'password',
+ })
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(force_error=True, where=['add'])
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ # It may not be a good idea to start with apply
+ # More atomic methods can be easier to mock
+ # apply() is calling list_accounts() and add_account()
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = 'Error creating volume access group element_groupname: %s' % ADD_ERROR
+ assert exc.value.args[0]['msg'] == message
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_check_error_reporting_on_invalid_account_id(self, mock_create_sf_connection):
+ ''' a more interesting test '''
+ set_module_args({
+ 'state': 'present',
+ 'name': 'element_groupname',
+ 'account_id': 'element_account_id',
+ 'volumes': ['volume1'],
+ 'hostname': 'hostname',
+ 'username': 'username',
+ 'password': 'password',
+ })
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(force_error=True, where=['account_id'])
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ # It may not be a good idea to start with apply
+ # More atomic methods can be easier to mock
+ # apply() is calling list_accounts() and add_account()
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = 'Error: Specified account id "%s" does not exist.' % 'element_account_id'
+ assert exc.value.args[0]['msg'] == message
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_access_group_volumes.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_access_group_volumes.py
new file mode 100644
index 00000000..fb78ad78
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_access_group_volumes.py
@@ -0,0 +1,245 @@
+''' unit test for Ansible module: na_elementsw_access_group_volumes.py '''
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+import json
+import pytest
+
+from ansible_collections.netapp.elementsw.tests.unit.compat import unittest
+from ansible_collections.netapp.elementsw.tests.unit.compat.mock import patch
+from ansible.module_utils import basic
+from ansible.module_utils._text import to_bytes
+import ansible_collections.netapp.elementsw.plugins.module_utils.netapp as netapp_utils
+
+if not netapp_utils.has_sf_sdk():
+ pytestmark = pytest.mark.skip('skipping as missing required SolidFire Python SDK')
+
+from ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_access_group_volumes \
+ import ElementSWAccessGroupVolumes as my_module # module under test
+
+
+def set_module_args(args):
+ """prepare arguments so that they will be picked up during module creation"""
+ args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
+ basic._ANSIBLE_ARGS = to_bytes(args) # pylint: disable=protected-access
+
+
+class AnsibleExitJson(Exception):
+ """Exception class to be raised by module.exit_json and caught by the test case"""
+ pass
+
+
+class AnsibleFailJson(Exception):
+ """Exception class to be raised by module.fail_json and caught by the test case"""
+ pass
+
+
+def exit_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over exit_json; package return data into an exception"""
+ if 'changed' not in kwargs:
+ kwargs['changed'] = False
+ raise AnsibleExitJson(kwargs)
+
+
+def fail_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over fail_json; package return data into an exception"""
+ kwargs['failed'] = True
+ raise AnsibleFailJson(kwargs)
+
+
+MODIFY_ERROR = 'some_error_in_modify_access_group'
+
+VOLUME_ID = 777
+
+
+class MockSFConnection(object):
+ ''' mock connection to ElementSW host '''
+
+ class Bunch(object): # pylint: disable=too-few-public-methods
+ ''' create object with arbitrary attributes '''
+ def __init__(self, **kw):
+ ''' called with (k1=v1, k2=v2), creates obj.k1, obj.k2 with values v1, v2 '''
+ setattr(self, '__dict__', kw)
+
+ def __init__(self, force_error=False, where=None, volume_id=None):
+ ''' save arguments '''
+ self.force_error = force_error
+ self.where = where
+ self.volume_id = volume_id
+
+ def list_volume_access_groups(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' build access_group list: access_groups.name, access_groups.account_id '''
+ group_name = 'element_groupname'
+ if self.volume_id is None:
+ volume_list = list()
+ else:
+ volume_list = [self.volume_id]
+ access_group = self.Bunch(name=group_name, volume_access_group_id=888, volumes=volume_list)
+ access_groups = [access_group]
+ access_group_list = self.Bunch(volume_access_groups=access_groups)
+ return access_group_list
+
+ def list_volumes_for_account(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' build volume list: volume.name, volume.id '''
+ volume = self.Bunch(name='element_volumename', volume_id=VOLUME_ID, delete_time='')
+ volumes = [volume]
+ volume_list = self.Bunch(volumes=volumes)
+ return volume_list
+
+ def modify_volume_access_group(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' We don't check the return code, but could force an exception '''
+ if self.force_error and 'modify_exception' in self.where:
+ # The module does not check for a specific exception :(
+ raise OSError(MODIFY_ERROR)
+
+ def get_account_by_name(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' returns account_id '''
+ if self.force_error and 'get_account_id' in self.where:
+ account_id = None
+ else:
+ account_id = 1
+ print('account_id', account_id)
+ account = self.Bunch(account_id=account_id)
+ result = self.Bunch(account=account)
+ return result
+
+
+class TestMyModule(unittest.TestCase):
+ ''' a group of related Unit Tests '''
+
+ ARGS = {
+ 'state': 'present',
+ 'access_group': 'element_groupname',
+ 'volumes': 'element_volumename',
+ 'account_id': 'element_account_id',
+ 'hostname': 'hostname',
+ 'username': 'username',
+ 'password': 'password',
+ }
+
+ def setUp(self):
+ self.mock_module_helper = patch.multiple(basic.AnsibleModule,
+ exit_json=exit_json,
+ fail_json=fail_json)
+ self.mock_module_helper.start()
+ self.addCleanup(self.mock_module_helper.stop)
+
+ def test_module_fail_when_required_args_missing(self):
+ ''' required arguments are reported as errors '''
+ with pytest.raises(AnsibleFailJson) as exc:
+ set_module_args({})
+ my_module()
+ print('Info: %s' % exc.value.args[0]['msg'])
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_add_volume(self, mock_create_sf_connection):
+ ''' adding a volume '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_add_volume_idempotent(self, mock_create_sf_connection):
+ ''' adding a volume that is already in the access group '''
+ args = dict(self.ARGS)
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(volume_id=VOLUME_ID)
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_remove_volume(self, mock_create_sf_connection):
+ ''' removing a volume that is in the access group '''
+ args = dict(self.ARGS)
+ args['state'] = 'absent'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(volume_id=VOLUME_ID)
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_remove_volume_idempotent(self, mock_create_sf_connection):
+ ''' removing a volume that is not in the access group '''
+ args = dict(self.ARGS)
+ args['state'] = 'absent'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_check_error_reporting_on_modify_exception(self, mock_create_sf_connection):
+ ''' modify does not return anything but can raise an exception '''
+ args = dict(self.ARGS)
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(force_error=True, where=['modify_exception'])
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = 'Error updating volume access group element_groupname: %s' % MODIFY_ERROR
+ assert exc.value.args[0]['msg'] == message
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_check_error_reporting_on_invalid_volume_name(self, mock_create_sf_connection):
+ ''' report error if volume does not exist '''
+ args = dict(self.ARGS)
+ args['volumes'] = ['volume1']
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = 'Error: Specified volume %s does not exist' % 'volume1'
+ assert exc.value.args[0]['msg'] == message
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_check_error_reporting_on_invalid_account_group_name(self, mock_create_sf_connection):
+ ''' report error if access group does not exist '''
+ args = dict(self.ARGS)
+ args['access_group'] = 'something_else'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = 'Error: Specified access group "%s" does not exist for account id: %s.' % ('something_else', 'element_account_id')
+ assert exc.value.args[0]['msg'] == message
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_check_error_reporting_on_invalid_account_id(self, mock_create_sf_connection):
+ ''' report error if account id is not found '''
+ args = dict(self.ARGS)
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(force_error=True, where='get_account_id')
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = 'Error: Specified account id "%s" does not exist.' % 'element_account_id'
+ assert exc.value.args[0]['msg'] == message
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_account.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_account.py
new file mode 100644
index 00000000..8075ba5c
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_account.py
@@ -0,0 +1,137 @@
+''' unit test for Ansible module: na_elementsw_account.py '''
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+import json
+import pytest
+
+from ansible_collections.netapp.elementsw.tests.unit.compat import unittest
+from ansible_collections.netapp.elementsw.tests.unit.compat.mock import patch
+from ansible.module_utils import basic
+from ansible.module_utils._text import to_bytes
+import ansible_collections.netapp.elementsw.plugins.module_utils.netapp as netapp_utils
+
+if not netapp_utils.has_sf_sdk():
+ pytestmark = pytest.mark.skip('skipping as missing required SolidFire Python SDK')
+
+from ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_account \
+ import ElementSWAccount as my_module # module under test
+
+
+def set_module_args(args):
+ """prepare arguments so that they will be picked up during module creation"""
+ args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
+ basic._ANSIBLE_ARGS = to_bytes(args) # pylint: disable=protected-access
+
+
+class AnsibleExitJson(Exception):
+ """Exception class to be raised by module.exit_json and caught by the test case"""
+ pass
+
+
+class AnsibleFailJson(Exception):
+ """Exception class to be raised by module.fail_json and caught by the test case"""
+ pass
+
+
+def exit_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over exit_json; package return data into an exception"""
+ if 'changed' not in kwargs:
+ kwargs['changed'] = False
+ raise AnsibleExitJson(kwargs)
+
+
+def fail_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over fail_json; package return data into an exception"""
+ kwargs['failed'] = True
+ raise AnsibleFailJson(kwargs)
+
+
+ADD_ERROR = 'some_error_in_add_account'
+
+
+class MockSFConnection(object):
+ ''' mock connection to ElementSW host '''
+
+ class Bunch(object): # pylint: disable=too-few-public-methods
+ ''' create object with arbitrary attributes '''
+ def __init__(self, **kw):
+ ''' called with (k1=v1, k2=v2), creates obj.k1, obj.k2 with values v1, v2 '''
+ setattr(self, '__dict__', kw)
+
+ def __init__(self, force_error=False, where=None):
+ ''' save arguments '''
+ self.force_error = force_error
+ self.where = where
+
+ def list_accounts(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' build account list: account.username, account.account_id '''
+ accounts = list()
+ account_list = self.Bunch(accounts=accounts)
+ return account_list
+
+ def add_account(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' We don't check the return code, but could force an exception '''
+ if self.force_error and 'add' in self.where:
+ # The module does not check for a specific exception :(
+ raise OSError(ADD_ERROR)
+
+
+class TestMyModule(unittest.TestCase):
+ ''' a group of related Unit Tests '''
+
+ def setUp(self):
+ self.mock_module_helper = patch.multiple(basic.AnsibleModule,
+ exit_json=exit_json,
+ fail_json=fail_json)
+ self.mock_module_helper.start()
+ self.addCleanup(self.mock_module_helper.stop)
+
+ def test_module_fail_when_required_args_missing(self):
+ ''' required arguments are reported as errors '''
+ with pytest.raises(AnsibleFailJson) as exc:
+ set_module_args({})
+ my_module()
+ print('Info: %s' % exc.value.args[0]['msg'])
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_ensure_command_called(self, mock_create_sf_connection):
+ ''' a more interesting test '''
+ set_module_args({
+ 'state': 'present',
+ 'element_username': 'element_username',
+ 'hostname': 'hostname',
+ 'username': 'username',
+ 'password': 'password',
+ })
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ # It may not be a good idea to start with apply
+ # More atomic methods can be easier to mock
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_check_error_reporting_on_add_exception(self, mock_create_sf_connection):
+ ''' a more interesting test '''
+ set_module_args({
+ 'state': 'present',
+ 'element_username': 'element_username',
+ 'hostname': 'hostname',
+ 'username': 'username',
+ 'password': 'password',
+ })
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(force_error=True, where=['add'])
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ # It may not be a good idea to start with apply
+ # More atomic methods can be easier to mock
+ # apply() is calling list_accounts() and add_account()
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = 'Error creating account element_username: %s' % ADD_ERROR
+ assert exc.value.args[0]['msg'] == message
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster.py
new file mode 100644
index 00000000..6624f374
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster.py
@@ -0,0 +1,228 @@
+''' unit test for Ansible module: na_elementsw_cluster.py '''
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+import inspect
+import json
+import pytest
+
+from ansible.module_utils import basic
+from ansible.module_utils._text import to_bytes
+from ansible_collections.netapp.elementsw.tests.unit.compat import unittest
+from ansible_collections.netapp.elementsw.tests.unit.compat.mock import patch
+import ansible_collections.netapp.elementsw.plugins.module_utils.netapp as netapp_utils
+
+if not netapp_utils.has_sf_sdk():
+ pytestmark = pytest.mark.skip('skipping as missing required SolidFire Python SDK')
+
+from ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_cluster \
+ import ElementSWCluster as my_module # module under test
+
+
+def set_module_args(args):
+ """prepare arguments so that they will be picked up during module creation"""
+ args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
+ basic._ANSIBLE_ARGS = to_bytes(args) # pylint: disable=protected-access
+
+
+class AnsibleExitJson(Exception):
+ """Exception class to be raised by module.exit_json and caught by the test case"""
+
+
+class AnsibleFailJson(Exception):
+ """Exception class to be raised by module.fail_json and caught by the test case"""
+
+
+def exit_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over exit_json; package return data into an exception"""
+ if 'changed' not in kwargs:
+ kwargs['changed'] = False
+ raise AnsibleExitJson(kwargs)
+
+
+def fail_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over fail_json; package return data into an exception"""
+ kwargs['failed'] = True
+ raise AnsibleFailJson(kwargs)
+
+
+NODE_ID1 = 777
+NODE_ID2 = 888
+NODE_ID3 = 999
+
+
+class MockSFConnection(object):
+ ''' mock connection to ElementSW host '''
+
+ class Bunch(object): # pylint: disable=too-few-public-methods
+ ''' create object with arbitrary attributes '''
+ def __init__(self, **kw):
+ ''' called with (k1=v1, k2=v2), creates obj.k1, obj.k2 with values v1, v2 '''
+ setattr(self, '__dict__', kw)
+
+ def __repr__(self):
+ results = dict()
+ for key, value in vars(self).items():
+ results[key] = repr(value)
+ return repr(results)
+
+ def __init__(self, force_error=False, where=None, nodes=None):
+ ''' save arguments '''
+ self.force_error = force_error
+ self.where = where
+ self.nodes = nodes
+ self._port = 442
+ self.called = list()
+
+ def record(self, args, kwargs):
+ name = inspect.stack()[1][3] # caller function name
+ print('%s: , args: %s, kwargs: %s' % (name, args, kwargs))
+ self.called.append(name)
+
+ def create_cluster(self, *args, **kwargs): # pylint: disable=unused-argument
+ self.record(repr(args), repr(kwargs))
+
+ def send_request(self, *args, **kwargs): # pylint: disable=unused-argument
+ self.record(repr(args), repr(kwargs))
+
+ def get_config(self, *args, **kwargs): # pylint: disable=unused-argument
+ self.record(repr(args), repr(kwargs))
+ if self.force_error and self.where == 'get_config_exception':
+ raise ConnectionError
+ if self.nodes is not None:
+ nodes = ['%d:%s' % (i, node) for i, node in enumerate(self.nodes)]
+ else:
+ nodes = list()
+ cluster = self.Bunch(ensemble=nodes, cluster='cl_name')
+ config = self.Bunch(cluster=cluster)
+ return self.Bunch(config=config)
+
+
+class TestMyModule(unittest.TestCase):
+ ''' a group of related Unit Tests '''
+
+ ARGS = {
+ # 'state': 'present',
+ 'management_virtual_ip': '10.10.10.10',
+ 'storage_virtual_ip': '10.10.10.11',
+ 'nodes': [NODE_ID1, NODE_ID2],
+ 'hostname': 'hostname',
+ 'username': 'username',
+ 'password': 'password',
+ }
+
+ def setUp(self):
+ self.mock_module_helper = patch.multiple(basic.AnsibleModule,
+ exit_json=exit_json,
+ fail_json=fail_json)
+ self.mock_module_helper.start()
+ self.addCleanup(self.mock_module_helper.stop)
+
+ def test_module_fail_when_required_args_missing(self):
+ ''' required arguments are reported as errors '''
+ with pytest.raises(AnsibleFailJson) as exc:
+ set_module_args({})
+ my_module()
+ print('Info: %s' % exc.value.args[0]['msg'])
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_create(self, mock_create_sf_connection):
+ ''' create cluster basic '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(force_error=True, where='get_config_exception')
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+ msg = 'created'
+ assert msg in exc.value.args[0]['msg']
+ assert 'create_cluster' in my_obj.sfe_node.called
+ assert 'send_request' not in my_obj.sfe_node.called
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_create_extra_parms(self, mock_create_sf_connection):
+ ''' force a direct call to send_request '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ args['order_number'] = '12345'
+ args['serial_number'] = '54321'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(force_error=True, where='get_config_exception')
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+ assert 'send_request' in my_obj.sfe_node.called
+ assert 'create_cluster' not in my_obj.sfe_node.called
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_create_idempotent(self, mock_create_sf_connection):
+ ''' cluster already exists with same nodes '''
+ args = dict(self.ARGS)
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(nodes=[NODE_ID1, NODE_ID2])
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+ assert 'send_request' not in my_obj.sfe_node.called
+ assert 'create_cluster' not in my_obj.sfe_node.called
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_create_idempotent_extra_nodes(self, mock_create_sf_connection):
+ ''' cluster already exists with more nodes '''
+ args = dict(self.ARGS)
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(nodes=[NODE_ID1, NODE_ID2, NODE_ID3])
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ msg = 'Error: found existing cluster with more nodes in ensemble.'
+ assert msg in exc.value.args[0]['msg']
+ assert 'send_request' not in my_obj.sfe_node.called
+ assert 'create_cluster' not in my_obj.sfe_node.called
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_create_idempotent_extra_nodes_ok(self, mock_create_sf_connection):
+ ''' cluster already exists with more nodes but we're OK with a superset '''
+ args = dict(self.ARGS)
+ args['fail_if_cluster_already_exists_with_larger_ensemble'] = False
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(nodes=[NODE_ID1, NODE_ID2, NODE_ID3])
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+ msg = 'cluster already exists'
+ assert msg in exc.value.args[0]['msg']
+ assert 'send_request' not in my_obj.sfe_node.called
+ assert 'create_cluster' not in my_obj.sfe_node.called
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_create_idempotent_missing_nodes(self, mock_create_sf_connection):
+ ''' cluster already exists with fewer nodes.
+ Since not every node is lister in the ensemble, we can't tell if it's an error or not '''
+ args = dict(self.ARGS)
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(nodes=[NODE_ID1])
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+ msg = 'cluster already exists'
+ assert msg in exc.value.args[0]['msg']
+ assert 'send_request' not in my_obj.sfe_node.called
+ assert 'create_cluster' not in my_obj.sfe_node.called
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster_config.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster_config.py
new file mode 100644
index 00000000..79f461cc
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster_config.py
@@ -0,0 +1,157 @@
+''' unit test for Ansible module: na_elementsw_cluster_config.py '''
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+import json
+import pytest
+
+from ansible_collections.netapp.elementsw.tests.unit.compat import unittest
+from ansible_collections.netapp.elementsw.tests.unit.compat.mock import patch
+from ansible.module_utils import basic
+from ansible.module_utils._text import to_bytes
+import ansible_collections.netapp.elementsw.plugins.module_utils.netapp as netapp_utils
+
+if not netapp_utils.has_sf_sdk():
+ pytestmark = pytest.mark.skip('skipping as missing required SolidFire Python SDK')
+
+from ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_cluster_config \
+ import ElementSWClusterConfig as my_module # module under test
+
+
+def set_module_args(args):
+ """prepare arguments so that they will be picked up during module creation"""
+ args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
+ basic._ANSIBLE_ARGS = to_bytes(args) # pylint: disable=protected-access
+
+
+class AnsibleExitJson(Exception):
+ """Exception class to be raised by module.exit_json and caught by the test case"""
+ pass
+
+
+class AnsibleFailJson(Exception):
+ """Exception class to be raised by module.fail_json and caught by the test case"""
+ pass
+
+
+def exit_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over exit_json; package return data into an exception"""
+ if 'changed' not in kwargs:
+ kwargs['changed'] = False
+ raise AnsibleExitJson(kwargs)
+
+
+def fail_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over fail_json; package return data into an exception"""
+ kwargs['failed'] = True
+ raise AnsibleFailJson(kwargs)
+
+
+GET_ERROR = 'some_error_in_get_ntp_info'
+
+
+class MockSFConnection(object):
+ ''' mock connection to ElementSW host '''
+
+ class Bunch(object): # pylint: disable=too-few-public-methods
+ ''' create object with arbitrary attributes '''
+
+ def __init__(self, **kw):
+ ''' called with (k1=v1, k2=v2), creates obj.k1, obj.k2 with values v1, v2 '''
+ setattr(self, '__dict__', kw)
+
+ def __init__(self, force_error=False, where=None):
+ ''' save arguments '''
+ self.force_error = force_error
+ self.where = where
+
+
+class TestMyModule(unittest.TestCase):
+ ''' a group of related Unit Tests '''
+
+ def setUp(self):
+ self.mock_module_helper = patch.multiple(basic.AnsibleModule,
+ exit_json=exit_json,
+ fail_json=fail_json)
+ self.mock_module_helper.start()
+ self.addCleanup(self.mock_module_helper.stop)
+
+ def set_default_args(self):
+ return dict({
+ 'hostname': '10.253.168.129',
+ 'username': 'namburu',
+ 'password': 'SFlab1234',
+ })
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_module_fail_when_required_args_missing(self, mock_create_sf_connection):
+ ''' required arguments are reported as errors '''
+ with pytest.raises(AnsibleFailJson) as exc:
+ set_module_args({})
+ my_module()
+ print('Info: %s' % exc.value.args[0]['msg'])
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_ensure_setup_ntp_info_called(self, mock_create_sf_connection):
+ ''' test if setup_ntp_info is called '''
+ module_args = {}
+ module_args.update(self.set_default_args())
+ ntp_dict = {'set_ntp_info': {'broadcastclient': None,
+ 'ntp_servers': ['1.1.1.1']}}
+ module_args.update(ntp_dict)
+ set_module_args(module_args)
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print('Info: test_setup_ntp_info: %s' % repr(exc.value))
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_ensure_set_encryption_at_rest_called(self, mock_create_sf_connection):
+ ''' test if set_encryption_at_rest is called '''
+ module_args = {}
+ module_args.update(self.set_default_args())
+ module_args.update({'encryption_at_rest': 'present'})
+ set_module_args(module_args)
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print('Info: test_set_encryption_at_rest enable: %s' % repr(exc.value))
+ assert not exc.value.args[0]['changed']
+ module_args.update({'encryption_at_rest': 'absent'})
+ set_module_args(module_args)
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print('Info: test_set_encryption_at_rest disable: %s' % repr(exc.value))
+ assert not exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_ensure_enable_feature_called(self, mock_create_sf_connection):
+ ''' test if enable_feature for vvols is called '''
+ module_args = {}
+ module_args.update(self.set_default_args())
+ module_args.update({'enable_virtual_volumes': True})
+ set_module_args(module_args)
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print('Info: test_enable_feature: %s' % repr(exc.value))
+ assert not exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_ensure_set_cluster_full_threshold_called(self, mock_create_sf_connection):
+ ''' test if set_cluster_full threshold is called '''
+ module_args = {}
+ module_args.update(self.set_default_args())
+ cluster_mod_dict = \
+ {'modify_cluster_full_threshold': {'stage2_aware_threshold': 2,
+ 'stage3_block_threshold_percent': 2,
+ 'max_metadata_over_provision_factor': 2}}
+ module_args.update(cluster_mod_dict)
+ set_module_args(module_args)
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print('Info: test_set_cluster_full_threshold: %s' % repr(exc.value))
+ assert exc.value.args[0]['changed']
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster_snmp.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster_snmp.py
new file mode 100644
index 00000000..9236daa0
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster_snmp.py
@@ -0,0 +1,176 @@
+''' unit test for Ansible module: na_elementsw_cluster_snmp.py '''
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+import json
+import pytest
+
+from ansible_collections.netapp.elementsw.tests.unit.compat import unittest
+from ansible_collections.netapp.elementsw.tests.unit.compat.mock import patch
+from ansible.module_utils import basic
+from ansible.module_utils._text import to_bytes
+import ansible_collections.netapp.elementsw.plugins.module_utils.netapp as netapp_utils
+
+if not netapp_utils.has_sf_sdk():
+ pytestmark = pytest.mark.skip('skipping as missing required SolidFire Python SDK')
+
+from ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_cluster_snmp \
+ import ElementSWClusterSnmp as my_module # module under test
+
+
+def set_module_args(args):
+ """prepare arguments so that they will be picked up during module creation"""
+ args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
+ basic._ANSIBLE_ARGS = to_bytes(args) # pylint: disable=protected-access
+
+
+class AnsibleExitJson(Exception):
+ """Exception class to be raised by module.exit_json and caught by the test case"""
+ pass
+
+
+class AnsibleFailJson(Exception):
+ """Exception class to be raised by module.fail_json and caught by the test case"""
+ pass
+
+
+def exit_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over exit_json; package return data into an exception"""
+ if 'changed' not in kwargs:
+ kwargs['changed'] = False
+ raise AnsibleExitJson(kwargs)
+
+
+def fail_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over fail_json; package return data into an exception"""
+ kwargs['failed'] = True
+ raise AnsibleFailJson(kwargs)
+
+
+GET_ERROR = 'some_error_in_get_snmp_info'
+
+
+class MockSFConnection(object):
+ ''' mock connection to ElementSW host '''
+
+ class Bunch(object): # pylint: disable=too-few-public-methods
+ ''' create object with arbitrary attributes '''
+ def __init__(self, **kw):
+ ''' called with (k1=v1, k2=v2), creates obj.k1, obj.k2 with values v1, v2 '''
+ setattr(self, '__dict__', kw)
+
+ def __init__(self, force_error=False, where=None):
+ ''' save arguments '''
+ self.force_error = force_error
+ self.where = where
+
+
+class TestMyModule(unittest.TestCase):
+ ''' a group of related Unit Tests '''
+
+ def setUp(self):
+ self.mock_module_helper = patch.multiple(basic.AnsibleModule,
+ exit_json=exit_json,
+ fail_json=fail_json)
+ self.mock_module_helper.start()
+ self.addCleanup(self.mock_module_helper.stop)
+
+ def set_default_args(self):
+ return dict({
+ 'hostname': '10.117.78.131',
+ 'username': 'admin',
+ 'password': 'netapp1!',
+ })
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_module_fail_when_required_args_missing(self, mock_create_sf_connection):
+ ''' required arguments are reported as errors '''
+ with pytest.raises(AnsibleFailJson) as exc:
+ set_module_args({})
+ my_module()
+ print('Info: %s' % exc.value)
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_ensure_enable_snmp_called(self, mock_create_sf_connection):
+ ''' test if enable_snmp is called '''
+ module_args = {}
+ module_args.update(self.set_default_args())
+ module_args.update({'snmp_v3_enabled': True,
+ 'state': 'present'})
+ module_args.update({'usm_users': {'access': 'rouser',
+ 'name': 'TestUser',
+ 'password': 'ChangeMe@123',
+ 'passphrase': 'ChangeMe@123',
+ 'secLevel': 'auth', }})
+
+ module_args.update({'networks': {'access': 'ro',
+ 'cidr': 24,
+ 'community': 'TestNetwork',
+ 'network': '192.168.0.1', }})
+ set_module_args(module_args)
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print('Info: test_if_enable_snmp_called: %s' % repr(exc.value))
+ assert exc.value
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_ensure_configure_snmp_from_version_3_TO_version_2_called(self, mock_create_sf_connection):
+ ''' test if configure snmp from version_3 to version_2'''
+ module_args = {}
+ module_args.update(self.set_default_args())
+ module_args.update({'snmp_v3_enabled': False,
+ 'state': 'present'})
+ module_args.update({'usm_users': {'access': 'rouser',
+ 'name': 'TestUser',
+ 'password': 'ChangeMe@123',
+ 'passphrase': 'ChangeMe@123',
+ 'secLevel': 'auth', }})
+
+ module_args.update({'networks': {'access': 'ro',
+ 'cidr': 24,
+ 'community': 'TestNetwork',
+ 'network': '192.168.0.1', }})
+ set_module_args(module_args)
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print('Info: test_ensure_configure_snmp_from_version_3_TO_version_2_called: %s' % repr(exc.value))
+ assert exc.value
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_ensure_configure_snmp_from_version_2_TO_version_3_called(self, mock_create_sf_connection):
+ ''' test if configure snmp from version_2 to version_3'''
+ module_args = {}
+ module_args.update(self.set_default_args())
+ module_args.update({'snmp_v3_enabled': True,
+ 'state': 'present'})
+ module_args.update({'usm_users': {'access': 'rouser',
+ 'name': 'TestUser_sample',
+ 'password': 'ChangeMe@123',
+ 'passphrase': 'ChangeMe@123',
+ 'secLevel': 'auth', }})
+
+ module_args.update({'networks': {'access': 'ro',
+ 'cidr': 24,
+ 'community': 'TestNetwork',
+ 'network': '192.168.0.1', }})
+ set_module_args(module_args)
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print('Info: test_ensure_configure_snmp_from_version_2_TO_version_3_called: %s' % repr(exc.value))
+ assert exc.value
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_ensure_disable_snmp_called(self, mock_create_sf_connection):
+ ''' test if disable_snmp is called '''
+ module_args = {}
+ module_args.update(self.set_default_args())
+ module_args.update({'state': 'absent'})
+ set_module_args(module_args)
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print('Info: test_if_disable_snmp_called: %s' % repr(exc.value))
+ assert exc.value
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_info.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_info.py
new file mode 100644
index 00000000..696043d6
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_info.py
@@ -0,0 +1,328 @@
+''' unit tests for Ansible module: na_elementsw_info.py '''
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+import inspect
+import json
+import pytest
+
+from ansible.module_utils import basic
+from ansible.module_utils._text import to_bytes
+from ansible_collections.netapp.elementsw.tests.unit.compat import unittest
+from ansible_collections.netapp.elementsw.tests.unit.compat.mock import patch
+import ansible_collections.netapp.elementsw.plugins.module_utils.netapp as netapp_utils
+
+if not netapp_utils.has_sf_sdk():
+ pytestmark = pytest.mark.skip('skipping as missing required SolidFire Python SDK')
+
+from ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_info \
+ import ElementSWInfo as my_module # module under test
+
+
+def set_module_args(args):
+ """prepare arguments so that they will be picked up during module creation"""
+ args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
+ basic._ANSIBLE_ARGS = to_bytes(args) # pylint: disable=protected-access
+
+
+class AnsibleExitJson(Exception):
+ """Exception class to be raised by module.exit_json and caught by the test case"""
+
+
+class AnsibleFailJson(Exception):
+ """Exception class to be raised by module.fail_json and caught by the test case"""
+
+
+def exit_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over exit_json; package return data into an exception"""
+ if 'changed' not in kwargs:
+ kwargs['changed'] = False
+ raise AnsibleExitJson(kwargs)
+
+
+def fail_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over fail_json; package return data into an exception"""
+ kwargs['failed'] = True
+ raise AnsibleFailJson(kwargs)
+
+
+NODE_ID1 = 777
+NODE_ID2 = 888
+NODE_ID3 = 999
+
+
+class MockSFConnection(object):
+ ''' mock connection to ElementSW host '''
+
+ class Bunch(object): # pylint: disable=too-few-public-methods
+ ''' create object with arbitrary attributes '''
+ def __init__(self, **kw):
+ ''' called with (k1=v1, k2=v2), creates obj.k1, obj.k2 with values v1, v2 '''
+ setattr(self, '__dict__', kw)
+
+ def __repr__(self):
+ results = dict()
+ for key, value in vars(self).items():
+ results[key] = repr(value)
+ return repr(results)
+
+ def to_json(self):
+ return json.loads(json.dumps(self, default=lambda x: x.__dict__))
+
+ def __init__(self, force_error=False, where=None):
+ ''' save arguments '''
+ self.force_error = force_error
+ self.where = where
+ self.nodes = [NODE_ID1, NODE_ID2, NODE_ID3]
+ self._port = 442
+ self.called = list()
+ if force_error and where == 'cx':
+ raise netapp_utils.solidfire.common.ApiConnectionError('testme')
+
+ def record(self, args, kwargs):
+ name = inspect.stack()[1][3] # caller function name
+ print('%s: , args: %s, kwargs: %s' % (name, args, kwargs))
+ self.called.append(name)
+
+ def list_accounts(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' build account list: account.username, account.account_id '''
+ self.record(repr(args), repr(kwargs))
+ accounts = list()
+ accounts.append({'username': 'user1'})
+ account_list = self.Bunch(accounts=accounts)
+ return account_list
+
+ def get_config(self, *args, **kwargs): # pylint: disable=unused-argument
+ self.record(repr(args), repr(kwargs))
+ if self.force_error and self.where == 'get_config_exception':
+ raise ConnectionError
+ if self.nodes is not None:
+ nodes = ['%d:%s' % (i, node) for i, node in enumerate(self.nodes)]
+ else:
+ nodes = list()
+ cluster = self.Bunch(ensemble=nodes, cluster='cl_name')
+ config = self.Bunch(cluster=cluster)
+ return self.Bunch(config=config)
+
+
+class TestMyModule(unittest.TestCase):
+ ''' a group of related Unit Tests '''
+
+ ARGS = {
+ # 'state': 'present',
+ 'hostname': 'hostname',
+ 'username': 'username',
+ 'password': 'password',
+ }
+
+ def setUp(self):
+ self.mock_module_helper = patch.multiple(basic.AnsibleModule,
+ exit_json=exit_json,
+ fail_json=fail_json)
+ self.mock_module_helper.start()
+ self.addCleanup(self.mock_module_helper.stop)
+
+ def test_module_fail_when_required_args_missing(self):
+ ''' required arguments are reported as errors '''
+ with pytest.raises(AnsibleFailJson) as exc:
+ set_module_args({})
+ my_module()
+ print('Info: %s' % exc.value.args[0]['msg'])
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_info_all_default(self, mock_create_sf_connection):
+ ''' gather all by default '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+ assert 'cluster_accounts' in exc.value.args[0]['info']
+ assert 'node_config' in exc.value.args[0]['info']
+ username = exc.value.args[0]['info']['cluster_accounts']['accounts'][0]['username']
+ assert username == 'user1'
+ assert 'list_accounts' in my_obj.sfe_node.called
+ assert 'get_config' in my_obj.sfe_node.called
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_info_all_all(self, mock_create_sf_connection):
+ ''' gather all explictly '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ args['gather_subsets'] = ['all']
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+ assert 'list_accounts' in my_obj.sfe_node.called
+ assert 'get_config' in my_obj.sfe_node.called
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_info_all_clusters(self, mock_create_sf_connection):
+ ''' gather all cluster scoped subsets '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ args['gather_subsets'] = ['all_clusters']
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+ assert 'cluster_accounts' in exc.value.args[0]['info']
+ accounts = exc.value.args[0]['info']['cluster_accounts']
+ print('accounts: >>%s<<' % accounts, type(accounts))
+ print(my_obj.sfe_node.called)
+ assert 'list_accounts' in my_obj.sfe_node.called
+ assert 'get_config' not in my_obj.sfe_node.called
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_info_all_nodes(self, mock_create_sf_connection):
+ ''' gather all node scoped subsets '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ args['gather_subsets'] = ['all_nodes']
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+ assert 'node_config' in exc.value.args[0]['info']
+ config = exc.value.args[0]['info']['node_config']
+ print('config: >>%s<<' % config, type(config))
+ print(my_obj.sfe_node.called)
+ assert 'list_accounts' not in my_obj.sfe_node.called
+ assert 'get_config' in my_obj.sfe_node.called
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_info_all_nodes_not_alone(self, mock_create_sf_connection):
+ ''' gather all node scoped subsets but fail as another subset is present '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ args['gather_subsets'] = ['all_nodes', 'dummy']
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ msg = 'no other subset is allowed'
+ assert msg in exc.value.args[0]['msg']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_info_filter_success(self, mock_create_sf_connection):
+ ''' filter on key, value - succesful match '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ args['gather_subsets'] = ['cluster_accounts']
+ args['filter'] = dict(username='user1')
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ username = exc.value.args[0]['info']['cluster_accounts']['accounts'][0]['username']
+ assert username == 'user1'
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_info_filter_bad_key(self, mock_create_sf_connection):
+ ''' filter on key, value - key not found '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ args['gather_subsets'] = ['cluster_accounts']
+ args['filter'] = dict(bad_key='user1')
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ msg = 'Error: key bad_key not found in'
+ assert msg in exc.value.args[0]['msg']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_info_filter_bad_key_ignored(self, mock_create_sf_connection):
+ ''' filter on key, value - key not found - ignore error '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ args['gather_subsets'] = ['cluster_accounts']
+ args['filter'] = dict(bad_key='user1')
+ args['fail_on_key_not_found'] = False
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['info']['cluster_accounts']['accounts'] == list()
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_info_filter_record_not_found(self, mock_create_sf_connection):
+ ''' filter on key, value - no match '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ args['gather_subsets'] = ['cluster_accounts']
+ args['filter'] = dict(bad_key='user1')
+ args['fail_on_key_not_found'] = False
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['info']['cluster_accounts']['accounts'] == list()
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_info_filter_record_not_found_error(self, mock_create_sf_connection):
+ ''' filter on key, value - no match - force error on empty '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ args['gather_subsets'] = ['cluster_accounts']
+ args['filter'] = dict(username='user111')
+ args['fail_on_record_not_found'] = True
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ msg = 'Error: no match for'
+ assert msg in exc.value.args[0]['msg']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_connection_error(self, mock_create_sf_connection):
+ ''' filter on key, value - no match - force error on empty '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ set_module_args(args)
+ # force a connection exception
+ mock_create_sf_connection.side_effect = netapp_utils.solidfire.common.ApiConnectionError('testme')
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_module()
+ print(exc.value.args[0])
+ msg = 'Failed to create connection for hostname:442'
+ assert msg in exc.value.args[0]['msg']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_other_connection_error(self, mock_create_sf_connection):
+ ''' filter on key, value - no match - force error on empty '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ set_module_args(args)
+ # force a connection exception
+ mock_create_sf_connection.side_effect = KeyError('testme')
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_module()
+ print(exc.value.args[0])
+ msg = 'Failed to connect for hostname:442'
+ assert msg in exc.value.args[0]['msg']
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_initiators.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_initiators.py
new file mode 100644
index 00000000..ee5ff85d
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_initiators.py
@@ -0,0 +1,201 @@
+''' unit test for Ansible module: na_elementsw_initiators.py '''
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+import json
+import pytest
+
+from ansible_collections.netapp.elementsw.tests.unit.compat import unittest
+from ansible_collections.netapp.elementsw.tests.unit.compat.mock import patch
+from ansible.module_utils import basic
+from ansible.module_utils._text import to_bytes
+import ansible_collections.netapp.elementsw.plugins.module_utils.netapp as netapp_utils
+
+if not netapp_utils.has_sf_sdk():
+ pytestmark = pytest.mark.skip('skipping as missing required SolidFire Python SDK')
+
+from ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_initiators \
+ import ElementSWInitiators as my_module # module under test
+
+
+def set_module_args(args):
+ """prepare arguments so that they will be picked up during module creation"""
+ args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
+ basic._ANSIBLE_ARGS = to_bytes(args) # pylint: disable=protected-access
+
+
+class AnsibleExitJson(Exception):
+ """Exception class to be raised by module.exit_json and caught by the test case"""
+ pass
+
+
+class AnsibleFailJson(Exception):
+ """Exception class to be raised by module.fail_json and caught by the test case"""
+ pass
+
+
+def exit_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over exit_json; package return data into an exception"""
+ if 'changed' not in kwargs:
+ kwargs['changed'] = False
+ raise AnsibleExitJson(kwargs)
+
+
+def fail_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over fail_json; package return data into an exception"""
+ kwargs['failed'] = True
+ raise AnsibleFailJson(kwargs)
+
+
+class MockSFConnection(object):
+ ''' mock connection to ElementSW host '''
+
+ class Bunch(object): # pylint: disable=too-few-public-methods
+ ''' create object with arbitrary attributes '''
+ def __init__(self, **kw):
+ ''' called with (k1=v1, k2=v2), creates obj.k1, obj.k2 with values v1, v2 '''
+ setattr(self, '__dict__', kw)
+
+ class Initiator(object):
+ def __init__(self, entries):
+ self.__dict__.update(entries)
+
+ def list_initiators(self):
+ ''' build initiator Obj '''
+ initiator = self.Bunch(
+ initiator_name="a",
+ initiator_id=13,
+ alias="a2",
+ # Note: 'config-mgmt' and 'event-source' are added for telemetry
+ attributes={'key': 'value', 'config-mgmt': 'ansible', 'event-source': 'na_elementsw_initiators'},
+ volume_access_groups=[1]
+ )
+ initiators = self.Bunch(
+ initiators=[initiator]
+ )
+ return initiators
+
+ def create_initiators(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' mock method '''
+ pass
+
+ def delete_initiators(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' mock method '''
+ pass
+
+ def modify_initiators(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' mock method '''
+ pass
+
+
+class TestMyModule(unittest.TestCase):
+ ''' a group of related Unit Tests '''
+
+ def setUp(self):
+ self.mock_module_helper = patch.multiple(basic.AnsibleModule,
+ exit_json=exit_json,
+ fail_json=fail_json)
+ self.mock_module_helper.start()
+ self.addCleanup(self.mock_module_helper.stop)
+
+ def set_default_args(self):
+ return dict({
+ 'hostname': '10.253.168.129',
+ 'username': 'namburu',
+ 'password': 'SFlab1234',
+ })
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_module_fail_when_required_args_missing(self, mock_create_sf_connection):
+ ''' required arguments are reported as errors '''
+ with pytest.raises(AnsibleFailJson) as exc:
+ set_module_args({})
+ my_module()
+ print('Info: %s' % exc.value.args[0]['msg'])
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_create_initiator(self, mock_create_sf_connection):
+ ''' test if create initiator is called '''
+ module_args = {}
+ module_args.update(self.set_default_args())
+ initiator_dict = {
+ "state": "present",
+ "initiators": [{
+ "name": "newinitiator1",
+ "alias": "newinitiator1alias",
+ "attributes": {"key1": "value1"}
+ }]
+ }
+ module_args.update(initiator_dict)
+ set_module_args(module_args)
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print('Info: test_create_initiators: %s' % repr(exc.value))
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_delete_initiator(self, mock_create_sf_connection):
+ ''' test if delete initiator is called '''
+ module_args = {}
+ module_args.update(self.set_default_args())
+ initiator_dict = {
+ "state": "absent",
+ "initiators": [{
+ "name": "a"
+ }]
+ }
+ module_args.update(initiator_dict)
+ set_module_args(module_args)
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print('Info: test_delete_initiators: %s' % repr(exc.value))
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_modify_initiator(self, mock_create_sf_connection):
+ ''' test if modify initiator is called '''
+ module_args = {}
+ module_args.update(self.set_default_args())
+ initiator_dict = {
+ "state": "present",
+ "initiators": [{
+ "name": "a",
+ "alias": "a3",
+ "attributes": {"key": "value"}
+ }]
+ }
+ module_args.update(initiator_dict)
+ set_module_args(module_args)
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print('Info: test_modify_initiators: %s' % repr(exc.value))
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_modify_initiator_idempotent(self, mock_create_sf_connection):
+ ''' test if modify initiator is called '''
+ module_args = {}
+ module_args.update(self.set_default_args())
+ initiator_dict = {
+ "state": "present",
+ "initiators": [{
+ "name": "a",
+ "alias": "a2",
+ "attributes": {"key": "value"},
+ "volume_access_group_id": 1
+ }]
+ }
+ module_args.update(initiator_dict)
+ set_module_args(module_args)
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print('Info: test_modify_initiators: %s' % repr(exc.value))
+ assert not exc.value.args[0]['changed']
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_network_interfaces.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_network_interfaces.py
new file mode 100644
index 00000000..5364a4e7
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_network_interfaces.py
@@ -0,0 +1,293 @@
+''' unit tests for Ansible module: na_elementsw_info.py '''
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+import inspect
+import json
+import pytest
+
+from ansible.module_utils import basic
+from ansible.module_utils._text import to_bytes
+from ansible_collections.netapp.elementsw.tests.unit.compat import unittest
+from ansible_collections.netapp.elementsw.tests.unit.compat.mock import patch
+import ansible_collections.netapp.elementsw.plugins.module_utils.netapp as netapp_utils
+
+if not netapp_utils.has_sf_sdk():
+ pytestmark = pytest.mark.skip('skipping as missing required SolidFire Python SDK')
+
+from ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_network_interfaces \
+ import ElementSWNetworkInterfaces as my_module # module under test
+
+
+def set_module_args(args):
+ """prepare arguments so that they will be picked up during module creation"""
+ args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
+ basic._ANSIBLE_ARGS = to_bytes(args) # pylint: disable=protected-access
+
+
+class AnsibleExitJson(Exception):
+ """Exception class to be raised by module.exit_json and caught by the test case"""
+
+
+class AnsibleFailJson(Exception):
+ """Exception class to be raised by module.fail_json and caught by the test case"""
+
+
+def exit_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over exit_json; package return data into an exception"""
+ if 'changed' not in kwargs:
+ kwargs['changed'] = False
+ raise AnsibleExitJson(kwargs)
+
+
+def fail_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over fail_json; package return data into an exception"""
+ kwargs['failed'] = True
+ raise AnsibleFailJson(kwargs)
+
+
+NODE_ID1 = 777
+NODE_ID2 = 888
+NODE_ID3 = 999
+
+MAPPING = dict(
+ bond_mode='bond-mode',
+ bond_lacp_rate='bond-lacp_rate',
+ dns_nameservers='dns-nameservers',
+ dns_search='dns-search',
+ virtual_network_tag='virtualNetworkTag',
+)
+
+
+def mapkey(key):
+ if key in MAPPING:
+ return MAPPING[key]
+ return key
+
+
+class MockSFConnection(object):
+ ''' mock connection to ElementSW host '''
+
+ class Bunch(object): # pylint: disable=too-few-public-methods
+ ''' create object with arbitrary attributes '''
+ def __init__(self, **kw):
+ ''' called with (k1=v1, k2=v2), creates obj.k1, obj.k2 with values v1, v2 '''
+ setattr(self, '__dict__', kw)
+
+ def __repr__(self):
+ results = dict()
+ for key, value in vars(self).items():
+ results[key] = repr(value)
+ return repr(results)
+
+ def to_json(self):
+ return json.loads(json.dumps(self, default=lambda x: x.__dict__))
+
+ def __init__(self, force_error=False, where=None):
+ ''' save arguments '''
+ self.force_error = force_error
+ self.where = where
+ # self._port = 442
+ self.called = list()
+ self.set_network_config_args = dict()
+ if force_error and where == 'cx':
+ raise netapp_utils.solidfire.common.ApiConnectionError('testme')
+
+ def record(self, args, kwargs): # pylint: disable=unused-argument
+ name = inspect.stack()[1][3] # caller function name
+ # print('%s: , args: %s, kwargs: %s' % (name, args, kwargs))
+ self.called.append(name)
+
+ def set_network_config(self, *args, **kwargs): # pylint: disable=unused-argument
+ self.record(repr(args), repr(kwargs))
+ print('network:', kwargs['network'].to_json())
+ self.set_network_config_args = kwargs['network'].to_json()
+
+
+class TestMyModule(unittest.TestCase):
+ ''' a group of related Unit Tests '''
+
+ DEPRECATED_ARGS = {
+ 'ip_address_1g': 'ip_address_1g',
+ 'subnet_1g': 'subnet_1g',
+ 'gateway_address_1g': 'gateway_address_1g',
+ 'mtu_1g': 'mtu_1g', # make sure the use a value != from default
+ 'bond_mode_1g': 'ALB', # make sure the use a value != from default
+ 'lacp_1g': 'Fast', # make sure the use a value != from default
+ 'ip_address_10g': 'ip_address_10g',
+ 'subnet_10g': 'subnet_10g',
+ 'gateway_address_10g': 'gateway_address_10g',
+ 'mtu_10g': 'mtu_10g', # make sure the use a value != from default
+ 'bond_mode_10g': 'LACP', # make sure the use a value != from default
+ 'lacp_10g': 'Fast', # make sure the use a value != from default
+ 'method': 'static',
+ 'dns_nameservers': 'dns_nameservers',
+ 'dns_search_domains': 'dns_search_domains',
+ 'virtual_network_tag': 'virtual_network_tag',
+ 'hostname': 'hostname',
+ 'username': 'username',
+ 'password': 'password',
+ }
+
+ ARGS = {
+ 'bond_1g': {
+ 'address': '10.10.10.10',
+ 'netmask': '255.255.255.0',
+ 'gateway': '10.10.10.1',
+ 'mtu': '1500',
+ 'bond_mode': 'ActivePassive',
+ 'dns_nameservers': ['dns_nameservers'],
+ 'dns_search': ['dns_search_domains'],
+ 'virtual_network_tag': 'virtual_network_tag',
+ },
+ 'bond_10g': {
+ 'bond_mode': 'LACP',
+ 'bond_lacp_rate': 'Fast',
+ },
+ 'hostname': 'hostname',
+ 'username': 'username',
+ 'password': 'password',
+ }
+
+ def setUp(self):
+ self.mock_module_helper = patch.multiple(basic.AnsibleModule,
+ exit_json=exit_json,
+ fail_json=fail_json)
+ self.mock_module_helper.start()
+ self.addCleanup(self.mock_module_helper.stop)
+
+ def test_module_fail_when_required_args_missing(self):
+ ''' required arguments are reported as errors '''
+ with pytest.raises(AnsibleFailJson) as exc:
+ set_module_args({})
+ my_module()
+ print('Info: %s' % exc.value.args[0]['msg'])
+
+ def test_deprecated_nothing(self):
+ ''' deprecated without 1g or 10g options '''
+ args = dict(self.DEPRECATED_ARGS) # deep copy as other tests can modify args
+ for key in list(args):
+ if '1g' in key or '10g' in key:
+ del args[key]
+ set_module_args(args)
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_module()
+ msg = 'Please use the new bond_1g or bond_10g options to configure the bond interfaces.'
+ assert msg in exc.value.args[0]['msg']
+ msg = 'This module cannot set or change "method"'
+ assert msg in exc.value.args[0]['msg']
+
+ def test_deprecated_all(self):
+ ''' deprecated with all options '''
+ args = dict(self.DEPRECATED_ARGS) # deep copy as other tests can modify args
+ set_module_args(args)
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_module()
+ msg = 'Please use the new bond_1g and bond_10g options to configure the bond interfaces.'
+ assert msg in exc.value.args[0]['msg']
+ msg = 'This module cannot set or change "method"'
+ assert msg in exc.value.args[0]['msg']
+
+ def test_deprecated_1g_only(self):
+ ''' deprecated with 1g options only '''
+ args = dict(self.DEPRECATED_ARGS) # deep copy as other tests can modify args
+ for key in list(args):
+ if '10g' in key:
+ del args[key]
+ set_module_args(args)
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_module()
+ msg = 'Please use the new bond_1g option to configure the bond 1G interface.'
+ assert msg in exc.value.args[0]['msg']
+ msg = 'This module cannot set or change "method"'
+ assert msg in exc.value.args[0]['msg']
+
+ def test_deprecated_10g_only(self):
+ ''' deprecated with 10g options only '''
+ args = dict(self.DEPRECATED_ARGS) # deep copy as other tests can modify args
+ for key in list(args):
+ if '1g' in key:
+ del args[key]
+ set_module_args(args)
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_module()
+ msg = 'Please use the new bond_10g option to configure the bond 10G interface.'
+ assert msg in exc.value.args[0]['msg']
+ msg = 'This module cannot set or change "method"'
+ assert msg in exc.value.args[0]['msg']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_modify_nothing(self, mock_create_sf_connection):
+ ''' modify without 1g or 10g options '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ for key in list(args):
+ if '1g' in key or '10g' in key:
+ del args[key]
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ print('LN:', my_obj.module.params)
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+ assert len(my_obj.sfe.set_network_config_args) == 0
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_modify_all(self, mock_create_sf_connection):
+ ''' modify with all options '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+ assert 'Bond1G' in my_obj.sfe.set_network_config_args
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_modify_1g_only(self, mock_create_sf_connection):
+ ''' modify with 1g options only '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ for key in list(args):
+ if '10g' in key:
+ del args[key]
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+ assert 'Bond1G' in my_obj.sfe.set_network_config_args
+ assert 'Bond10G' not in my_obj.sfe.set_network_config_args
+ print(my_obj.sfe.set_network_config_args['Bond1G'])
+ for key in args['bond_1g']:
+ if key != 'bond_lacp_rate':
+ assert my_obj.sfe.set_network_config_args['Bond1G'][mapkey(key)] == args['bond_1g'][key]
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_modify_10g_only(self, mock_create_sf_connection):
+ ''' modify with 10g options only '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ for key in list(args):
+ if '1g' in key:
+ del args[key]
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+ assert 'Bond1G' not in my_obj.sfe.set_network_config_args
+ assert 'Bond10G' in my_obj.sfe.set_network_config_args
+ assert my_obj.sfe.set_network_config_args['Bond10G']['bond-lacp_rate'] == args['bond_10g']['bond_lacp_rate']
+ for key in args['bond_10g']:
+ assert my_obj.sfe.set_network_config_args['Bond10G'][mapkey(key)] == args['bond_10g'][key]
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_nodes.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_nodes.py
new file mode 100644
index 00000000..3e163d00
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_nodes.py
@@ -0,0 +1,324 @@
+''' unit test for Ansible module: na_elementsw_node.py '''
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+import json
+import pytest
+
+from ansible.module_utils import basic
+from ansible.module_utils._text import to_bytes
+from ansible_collections.netapp.elementsw.tests.unit.compat import unittest
+from ansible_collections.netapp.elementsw.tests.unit.compat.mock import patch
+import ansible_collections.netapp.elementsw.plugins.module_utils.netapp as netapp_utils
+
+if not netapp_utils.has_sf_sdk():
+ pytestmark = pytest.mark.skip('skipping as missing required SolidFire Python SDK')
+
+from ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_node \
+ import ElementSWNode as my_module # module under test
+
+
+def set_module_args(args):
+ """prepare arguments so that they will be picked up during module creation"""
+ args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
+ basic._ANSIBLE_ARGS = to_bytes(args) # pylint: disable=protected-access
+
+
+class AnsibleExitJson(Exception):
+ """Exception class to be raised by module.exit_json and caught by the test case"""
+
+
+class AnsibleFailJson(Exception):
+ """Exception class to be raised by module.fail_json and caught by the test case"""
+
+
+def exit_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over exit_json; package return data into an exception"""
+ if 'changed' not in kwargs:
+ kwargs['changed'] = False
+ raise AnsibleExitJson(kwargs)
+
+
+def fail_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over fail_json; package return data into an exception"""
+ kwargs['failed'] = True
+ raise AnsibleFailJson(kwargs)
+
+
+MODIFY_ERROR = 'some_error_in_modify_access_group'
+
+NODE_ID1 = 777
+NODE_ID2 = 888
+NODE_NAME1 = 'node_name1'
+NODE_NAME2 = 'node_name2'
+
+
+class MockSFConnection(object):
+ ''' mock connection to ElementSW host '''
+
+ class Bunch(object): # pylint: disable=too-few-public-methods
+ ''' create object with arbitrary attributes '''
+ def __init__(self, **kw):
+ ''' called with (k1=v1, k2=v2), creates obj.k1, obj.k2 with values v1, v2 '''
+ setattr(self, '__dict__', kw)
+
+ def __init__(self, force_error=False, where=None, node_id=None, cluster_name='', node_state='Pending'):
+ ''' save arguments '''
+ self.force_error = force_error
+ self.where = where
+ self.node_id = node_id
+ self.cluster_name = cluster_name
+ self.node_state = node_state
+
+ def list_all_nodes(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' build access_group list: access_groups.name, access_groups.account_id '''
+ nodes = list()
+ pending_nodes = list()
+ active_pending_nodes = list()
+ if self.node_id is None:
+ node_list = list()
+ else:
+ node_list = [self.node_id]
+ attrs1 = dict(mip='10.10.10.101', name=NODE_NAME1, node_id=NODE_ID1)
+ attrs2 = dict(mip='10.10.10.101', name=NODE_NAME2, node_id=NODE_ID2)
+ if self.where == 'pending':
+ attrs1['pending_node_id'] = NODE_ID1
+ attrs2['pending_node_id'] = NODE_ID2
+ node1 = self.Bunch(**attrs1)
+ node2 = self.Bunch(**attrs2)
+ if self.where == 'nodes':
+ nodes = [node1, node2]
+ elif self.where == 'pending':
+ pending_nodes = [node1, node2]
+ elif self.where == 'active_pending':
+ active_pending_nodes = [node1, node2]
+ node_list = self.Bunch(nodes=nodes, pending_nodes=pending_nodes, pending_active_nodes=active_pending_nodes)
+ return node_list
+
+ def add_nodes(self, *args, **kwargs): # pylint: disable=unused-argument
+ print('adding_node: ', repr(args), repr(kwargs))
+
+ def remove_nodes(self, *args, **kwargs): # pylint: disable=unused-argument
+ print('adding_node: ', repr(args), repr(kwargs))
+
+ def get_cluster_config(self, *args, **kwargs): # pylint: disable=unused-argument
+ print('get_cluster_config: ', repr(args), repr(kwargs))
+ cluster = self.Bunch(cluster=self.cluster_name, state=self.node_state)
+ return self.Bunch(cluster=cluster)
+
+ def set_cluster_config(self, *args, **kwargs): # pylint: disable=unused-argument
+ print('set_cluster_config: ', repr(args), repr(kwargs))
+
+ def list_drives(self, *args, **kwargs): # pylint: disable=unused-argument
+ print('list_drives: ', repr(args), repr(kwargs))
+ drive = self.Bunch(node_id=self.node_id, status="active")
+ return self.Bunch(drives=[drive])
+
+
+class TestMyModule(unittest.TestCase):
+ ''' a group of related Unit Tests '''
+
+ ARGS = {
+ 'state': 'present',
+ 'node_ids': [NODE_ID1, NODE_ID2],
+ 'hostname': 'hostname',
+ 'username': 'username',
+ 'password': 'password',
+ }
+
+ def setUp(self):
+ self.mock_module_helper = patch.multiple(basic.AnsibleModule,
+ exit_json=exit_json,
+ fail_json=fail_json)
+ self.mock_module_helper.start()
+ self.addCleanup(self.mock_module_helper.stop)
+
+ def test_module_fail_when_required_args_missing(self):
+ ''' required arguments are reported as errors '''
+ with pytest.raises(AnsibleFailJson) as exc:
+ set_module_args({})
+ my_module()
+ print('Info: %s' % exc.value.args[0]['msg'])
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_add_node_fail_not_pending(self, mock_create_sf_connection):
+ ''' adding a node - fails as these nodes are unknown '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ msg = 'nodes not in pending or active lists'
+ assert msg in exc.value.args[0]['msg']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_add_node(self, mock_create_sf_connection):
+ ''' adding a node '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(where='pending')
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_add_node_idempotent(self, mock_create_sf_connection):
+ ''' adding a node that is already in the cluster '''
+ args = dict(self.ARGS)
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(where='nodes')
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_remove_node(self, mock_create_sf_connection):
+ ''' removing a node that is in the cluster '''
+ args = dict(self.ARGS)
+ args['state'] = 'absent'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(where='nodes')
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_remove_node_idempotent(self, mock_create_sf_connection):
+ ''' removing a node that is not in the cluster '''
+ args = dict(self.ARGS)
+ args['state'] = 'absent'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_remove_node_with_active_drive(self, mock_create_sf_connection):
+ ''' removing a node that is in the cluster but still associated with a drive '''
+ args = dict(self.ARGS)
+ args['state'] = 'absent'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(node_id=NODE_ID1, where='nodes')
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ msg = 'Error deleting node %s: node has active drives' % NODE_NAME1
+ assert msg in exc.value.args[0]['msg']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_set_cluster_name_only(self, mock_create_sf_connection):
+ ''' set cluster name without adding the node '''
+ args = dict(self.ARGS)
+ args['preset_only'] = True
+ args['cluster_name'] = 'cluster_name'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+ message = 'List of updated nodes with cluster_name:'
+ assert message in exc.value.args[0]['msg']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_set_cluster_name_only_idempotent(self, mock_create_sf_connection):
+ ''' set cluster name without adding the node - name already set '''
+ args = dict(self.ARGS)
+ args['preset_only'] = True
+ args['cluster_name'] = 'cluster_name'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(cluster_name=args['cluster_name'])
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+ message = ''
+ assert message == exc.value.args[0]['msg']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_set_cluster_name_and_add(self, mock_create_sf_connection):
+ ''' set cluster name and add the node '''
+ args = dict(self.ARGS)
+ args['cluster_name'] = 'cluster_name'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(where='pending')
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+ message = 'List of updated nodes with cluster_name:'
+ assert message in exc.value.args[0]['msg']
+ message = 'List of added nodes: '
+ assert message in exc.value.args[0]['msg']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_set_cluster_name_and_add_idempotent(self, mock_create_sf_connection):
+ ''' set cluster name and add the node '''
+ args = dict(self.ARGS)
+ args['cluster_name'] = 'cluster_name'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(where='nodes', cluster_name=args['cluster_name'])
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+ message = ''
+ assert message == exc.value.args[0]['msg']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_set_cluster_name_already_active_no_change(self, mock_create_sf_connection):
+ ''' set cluster name fails because node state is 'Active' '''
+ args = dict(self.ARGS)
+ args['cluster_name'] = 'cluster_name'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(where='nodes', cluster_name=args['cluster_name'], node_state='Active')
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+ message = ''
+ assert message == exc.value.args[0]['msg']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_set_cluster_name_already_active_change_not_allowed(self, mock_create_sf_connection):
+ ''' set cluster name fails because node state is 'Active' '''
+ args = dict(self.ARGS)
+ args['cluster_name'] = 'new_cluster_name'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(where='nodes', cluster_name='old_cluster_name', node_state='Active')
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = "Error updating cluster name for node %s, already in 'Active' state" % NODE_ID1
+ assert message == exc.value.args[0]['msg']
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_qos_policy.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_qos_policy.py
new file mode 100644
index 00000000..83ac3711
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_qos_policy.py
@@ -0,0 +1,300 @@
+''' unit test for Ansible module: na_elementsw_qos_policy.py '''
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+import json
+import pytest
+
+from ansible.module_utils import basic
+from ansible.module_utils._text import to_bytes
+from ansible_collections.netapp.elementsw.tests.unit.compat import unittest
+from ansible_collections.netapp.elementsw.tests.unit.compat.mock import patch
+import ansible_collections.netapp.elementsw.plugins.module_utils.netapp as netapp_utils
+
+if not netapp_utils.has_sf_sdk():
+ pytestmark = pytest.mark.skip('skipping as missing required SolidFire Python SDK')
+
+from ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_qos_policy \
+ import ElementSWQosPolicy as my_module # module under test
+
+
+def set_module_args(args):
+ """prepare arguments so that they will be picked up during module creation"""
+ args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
+ basic._ANSIBLE_ARGS = to_bytes(args) # pylint: disable=protected-access
+
+
+class AnsibleExitJson(Exception):
+ """Exception class to be raised by module.exit_json and caught by the test case"""
+
+
+class AnsibleFailJson(Exception):
+ """Exception class to be raised by module.fail_json and caught by the test case"""
+
+
+def exit_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over exit_json; package return data into an exception"""
+ if 'changed' not in kwargs:
+ kwargs['changed'] = False
+ raise AnsibleExitJson(kwargs)
+
+
+def fail_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over fail_json; package return data into an exception"""
+ kwargs['failed'] = True
+ raise AnsibleFailJson(kwargs)
+
+
+CREATE_ERROR = 'create', 'some_error_in_create_qos_policy'
+MODIFY_ERROR = 'modify', 'some_error_in_modify_qos_policy'
+DELETE_ERROR = 'delete', 'some_error_in_delete_qos_policy'
+
+POLICY_ID = 888
+POLICY_NAME = 'element_qos_policy_name'
+
+
+class MockSFConnection(object):
+ ''' mock connection to ElementSW host '''
+
+ class Bunch(object): # pylint: disable=too-few-public-methods
+ ''' create object with arbitrary attributes '''
+ def __init__(self, **kw):
+ ''' called with (k1=v1, k2=v2), creates obj.k1, obj.k2 with values v1, v2 '''
+ setattr(self, '__dict__', kw)
+
+ def __init__(self, force_error=False, where=None, qos_policy_name=None):
+ ''' save arguments '''
+ self.force_error = force_error
+ self.where = where
+ self.policy_name = qos_policy_name
+
+ def list_qos_policies(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' build qos_policy list: qos_policy.name, qos_policy.account_id '''
+ if self.policy_name:
+ qos_policy_name = self.policy_name
+ else:
+ qos_policy_name = POLICY_NAME
+ qos = self.Bunch(min_iops=1000, max_iops=20000, burst_iops=20000)
+ qos_policy = self.Bunch(name=qos_policy_name, qos_policy_id=POLICY_ID, qos=qos)
+ qos_policies = [qos_policy]
+ qos_policy_list = self.Bunch(qos_policies=qos_policies)
+ return qos_policy_list
+
+ def create_qos_policy(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' We don't check the return code, but could force an exception '''
+ if self.force_error and 'create_exception' in self.where:
+ raise netapp_utils.solidfire.common.ApiServerError(*CREATE_ERROR)
+
+ def modify_qos_policy(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' We don't check the return code, but could force an exception '''
+ if self.force_error and 'modify_exception' in self.where:
+ raise netapp_utils.solidfire.common.ApiServerError(*MODIFY_ERROR)
+
+ def delete_qos_policy(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' We don't check the return code, but could force an exception '''
+ if self.force_error and 'delete_exception' in self.where:
+ raise netapp_utils.solidfire.common.ApiServerError(*DELETE_ERROR)
+
+
+class TestMyModule(unittest.TestCase):
+ ''' a group of related Unit Tests '''
+
+ ARGS = {
+ 'state': 'present',
+ 'name': 'element_qos_policy_name',
+ 'qos': {'minIOPS': 1000, 'maxIOPS': 20000, 'burstIOPS': 20000},
+ 'hostname': 'hostname',
+ 'username': 'username',
+ 'password': 'password',
+ }
+
+ def setUp(self):
+ self.mock_module_helper = patch.multiple(basic.AnsibleModule,
+ exit_json=exit_json,
+ fail_json=fail_json)
+ self.mock_module_helper.start()
+ self.addCleanup(self.mock_module_helper.stop)
+
+ def test_module_fail_when_required_args_missing(self):
+ ''' required arguments are reported as errors '''
+ with pytest.raises(AnsibleFailJson) as exc:
+ set_module_args({})
+ my_module()
+ print('Info: %s' % exc.value.args[0]['msg'])
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_add_qos_policy(self, mock_create_sf_connection):
+ ''' adding a qos_policy '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ args['name'] += '_1' # new name to force a create
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_add_qos_policy_idempotent(self, mock_create_sf_connection):
+ ''' adding a qos_policy '''
+ args = dict(self.ARGS)
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_delete_qos_policy(self, mock_create_sf_connection):
+ ''' removing a qos policy '''
+ args = dict(self.ARGS)
+ args['state'] = 'absent'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_delete_qos_policy_idempotent(self, mock_create_sf_connection):
+ ''' removing a qos policy '''
+ args = dict(self.ARGS)
+ args['state'] = 'absent'
+ args['name'] += '_1' # new name to force idempotency
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_modify_qos_policy(self, mock_create_sf_connection):
+ ''' modifying a qos policy '''
+ args = dict(self.ARGS)
+ args['qos'] = {'minIOPS': 2000}
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_rename_qos_policy(self, mock_create_sf_connection):
+ ''' renaming a qos policy '''
+ args = dict(self.ARGS)
+ args['from_name'] = args['name']
+ args['name'] = 'a_new_name'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_rename_modify_qos_policy_idempotent(self, mock_create_sf_connection):
+ ''' renaming a qos policy '''
+ args = dict(self.ARGS)
+ args['from_name'] = 'some_older_name'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_create_qos_policy_exception(self, mock_create_sf_connection):
+ ''' creating a qos policy can raise an exception '''
+ args = dict(self.ARGS)
+ args['name'] += '_1' # new name to force a create
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(force_error=True, where=['create_exception'])
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = 'Error creating qos policy: %s' % POLICY_NAME
+ assert exc.value.args[0]['msg'].startswith(message)
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_modify_qos_policy_exception(self, mock_create_sf_connection):
+ ''' modifying a qos policy can raise an exception '''
+ args = dict(self.ARGS)
+ args['qos'] = {'minIOPS': 2000}
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(force_error=True, where=['modify_exception'])
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = 'Error updating qos policy: %s' % POLICY_NAME
+ assert exc.value.args[0]['msg'].startswith(message)
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_delete_qos_policy_exception(self, mock_create_sf_connection):
+ ''' deleting a qos policy can raise an exception '''
+ args = dict(self.ARGS)
+ args['state'] = 'absent'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(force_error=True, where=['delete_exception'])
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = 'Error deleting qos policy: %s' % POLICY_NAME
+ assert exc.value.args[0]['msg'].startswith(message)
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_check_error_reporting_on_missing_qos_option(self, mock_create_sf_connection):
+ ''' report error if qos option is not given on create '''
+ args = dict(self.ARGS)
+ args['name'] += '_1' # new name to force a create
+ args.pop('qos')
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = "Error creating qos policy: %s, 'qos:' option is required" % args['name']
+ assert exc.value.args[0]['msg'] == message
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_check_error_reporting_on_missing_from_name_policy(self, mock_create_sf_connection):
+ ''' report error if qos policy to rename does not exist '''
+ args = dict(self.ARGS)
+ args['name'] += '_1' # new name to force a create
+ args['from_name'] = 'something_not_likely_to_exist'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = "Error renaming qos policy, no existing policy with name/id: %s" % args['from_name']
+ assert exc.value.args[0]['msg'] == message
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_template.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_template.py
new file mode 100644
index 00000000..7dc6e2d6
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_template.py
@@ -0,0 +1,138 @@
+''' unit test for Ansible module: na_elementsw_account.py '''
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+import json
+import pytest
+
+from ansible_collections.netapp.elementsw.tests.unit.compat import unittest
+from ansible_collections.netapp.elementsw.tests.unit.compat.mock import patch
+from ansible.module_utils import basic
+from ansible.module_utils._text import to_bytes
+import ansible_collections.netapp.elementsw.plugins.module_utils.netapp as netapp_utils
+
+if not netapp_utils.has_sf_sdk():
+ pytestmark = pytest.mark.skip('skipping as missing required SolidFire Python SDK')
+
+from ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_account \
+ import ElementSWAccount as my_module # module under test
+
+
+def set_module_args(args):
+ """prepare arguments so that they will be picked up during module creation"""
+ args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
+ basic._ANSIBLE_ARGS = to_bytes(args) # pylint: disable=protected-access
+
+
+class AnsibleExitJson(Exception):
+ """Exception class to be raised by module.exit_json and caught by the test case"""
+ pass
+
+
+class AnsibleFailJson(Exception):
+ """Exception class to be raised by module.fail_json and caught by the test case"""
+ pass
+
+
+def exit_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over exit_json; package return data into an exception"""
+ if 'changed' not in kwargs:
+ kwargs['changed'] = False
+ raise AnsibleExitJson(kwargs)
+
+
+def fail_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over fail_json; package return data into an exception"""
+ kwargs['failed'] = True
+ raise AnsibleFailJson(kwargs)
+
+
+ADD_ERROR = 'some_error_in_add_account'
+
+
+class MockSFConnection(object):
+ ''' mock connection to ElementSW host '''
+
+ class Bunch(object): # pylint: disable=too-few-public-methods
+ ''' create object with arbitrary attributes '''
+ def __init__(self, **kw):
+ ''' called with (k1=v1, k2=v2), creates obj.k1, obj.k2 with values v1, v2 '''
+ setattr(self, '__dict__', kw)
+
+ def __init__(self, force_error=False, where=None):
+ ''' save arguments '''
+ self.force_error = force_error
+ self.where = where
+
+# TODO: replace list_accounts and add_account as needed
+ def list_accounts(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' build account list: account.username, account.account_id '''
+ accounts = list()
+ account_list = self.Bunch(accounts=accounts)
+ return account_list
+
+ def add_account(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' We don't check the return code, but could force an exception '''
+ if self.force_error and 'add' in self.where:
+ # The module does not check for a specific exception :(
+ raise OSError(ADD_ERROR)
+
+
+class TestMyModule(unittest.TestCase):
+ ''' a group of related Unit Tests '''
+
+ def setUp(self):
+ self.mock_module_helper = patch.multiple(basic.AnsibleModule,
+ exit_json=exit_json,
+ fail_json=fail_json)
+ self.mock_module_helper.start()
+ self.addCleanup(self.mock_module_helper.stop)
+
+ def test_module_fail_when_required_args_missing(self):
+ ''' required arguments are reported as errors '''
+ with pytest.raises(AnsibleFailJson) as exc:
+ set_module_args({})
+ my_module()
+ print('Info: %s' % exc.value.args[0]['msg'])
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_ensure_command_called(self, mock_create_sf_connection):
+ ''' a more interesting test '''
+ set_module_args({
+ 'state': 'present',
+ 'element_username': 'element_username',
+ 'hostname': 'hostname',
+ 'username': 'username',
+ 'password': 'password',
+ })
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ # It may not be a good idea to start with apply
+ # More atomic methods can be easier to mock
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_check_error_reporting_on_add_exception(self, mock_create_sf_connection):
+ ''' a more interesting test '''
+ set_module_args({
+ 'state': 'present',
+ 'element_username': 'element_username',
+ 'hostname': 'hostname',
+ 'username': 'username',
+ 'password': 'password',
+ })
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(force_error=True, where=['add'])
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ # It may not be a good idea to start with apply
+ # More atomic methods can be easier to mock
+ # apply() is calling list_accounts() and add_account()
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = 'Error creating account element_username: %s' % ADD_ERROR
+ assert exc.value.args[0]['msg'] == message
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_vlan.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_vlan.py
new file mode 100644
index 00000000..e2dc51f7
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_vlan.py
@@ -0,0 +1,343 @@
+''' unit test for Ansible module: na_elementsw_account.py '''
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+import json
+import pytest
+
+from ansible_collections.netapp.elementsw.tests.unit.compat import unittest
+from ansible_collections.netapp.elementsw.tests.unit.compat.mock import patch, Mock
+from ansible.module_utils import basic
+from ansible.module_utils._text import to_bytes
+import ansible_collections.netapp.elementsw.plugins.module_utils.netapp as netapp_utils
+
+if not netapp_utils.has_sf_sdk():
+ pytestmark = pytest.mark.skip('skipping as missing required SolidFire Python SDK')
+
+from ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_vlan \
+ import ElementSWVlan as vlan # module under test
+
+
+def set_module_args(args):
+ """prepare arguments so that they will be picked up during module creation"""
+ args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
+ basic._ANSIBLE_ARGS = to_bytes(args) # pylint: disable=protected-access
+
+
+class AnsibleExitJson(Exception):
+ """Exception class to be raised by module.exit_json and caught by the test case"""
+ pass
+
+
+class AnsibleFailJson(Exception):
+ """Exception class to be raised by module.fail_json and caught by the test case"""
+ pass
+
+
+def exit_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over exit_json; package return data into an exception"""
+ if 'changed' not in kwargs:
+ kwargs['changed'] = False
+ raise AnsibleExitJson(kwargs)
+
+
+def fail_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over fail_json; package return data into an exception"""
+ kwargs['failed'] = True
+ raise AnsibleFailJson(kwargs)
+
+
+ADD_ERROR = 'some_error_in_add_account'
+
+
+class MockSFConnection(object):
+ ''' mock connection to ElementSW host '''
+
+ class Bunch(object): # pylint: disable=too-few-public-methods
+ ''' create object with arbitrary attributes '''
+ def __init__(self, **kw):
+ ''' called with (k1=v1, k2=v2), creates obj.k1, obj.k2 with values v1, v2 '''
+ setattr(self, '__dict__', kw)
+
+ class Vlan(object):
+ def __init__(self, entries):
+ self.__dict__.update(entries)
+
+ def __init__(self, force_error=False, where=None):
+ ''' save arguments '''
+ self.force_error = force_error
+ self.where = where
+
+ def list_virtual_networks(self, virtual_network_tag=None): # pylint: disable=unused-argument
+ ''' list of vlans '''
+ if virtual_network_tag == '1':
+ add1 = self.Bunch(
+ start='2.2.2.2',
+ size=4
+ )
+ add2 = self.Bunch(
+ start='3.3.3.3',
+ size=4
+ )
+ vlan = self.Bunch(
+ attributes={'key': 'value', 'config-mgmt': 'ansible', 'event-source': 'na_elementsw_vlan'},
+ name="test",
+ address_blocks=[
+ add1,
+ add2
+ ],
+ svip='192.168.1.2',
+ gateway='0.0.0.0',
+ netmask='255.255.248.0',
+ namespace=False
+ )
+ vlans = self.Bunch(
+ virtual_networks=[vlan]
+ )
+ else:
+ vlans = self.Bunch(
+ virtual_networks=[]
+ )
+ return vlans
+
+ def add_virtual_network(self, virtual_network_tag=None, **create): # pylint: disable=unused-argument
+ ''' We don't check the return code, but could force an exception '''
+ if self.force_error and 'add' in self.where:
+ # The module does not check for a specific exception :(
+ raise OSError(ADD_ERROR)
+
+ def remove_virtual_network(self, virtual_network_tag=None): # pylint: disable=unused-argument
+ ''' We don't check the return code, but could force an exception '''
+ if self.force_error and 'remove' in self.where:
+ # The module does not check for a specific exception :(
+ raise OSError(ADD_ERROR)
+
+ def modify_virtual_network(self, virtual_network_tag=None, **modify): # pylint: disable=unused-argument
+ ''' We don't check the return code, but could force an exception '''
+ if self.force_error and 'modify' in self.where:
+ # The module does not check for a specific exception :(
+ raise OSError(ADD_ERROR)
+
+
+class TestMyModule(unittest.TestCase):
+ ''' a group of related Unit Tests '''
+
+ def setUp(self):
+ self.mock_module_helper = patch.multiple(basic.AnsibleModule,
+ exit_json=exit_json,
+ fail_json=fail_json)
+ self.mock_module_helper.start()
+ self.addCleanup(self.mock_module_helper.stop)
+
+ def test_module_fail_when_required_args_missing(self):
+ ''' required arguments are reported as errors '''
+ with pytest.raises(AnsibleFailJson) as exc:
+ set_module_args({})
+ vlan()
+ print('Info: %s' % exc.value.args[0]['msg'])
+
+ def mock_args(self):
+ args = {
+ 'state': 'present',
+ 'name': 'test',
+ 'vlan_tag': 1,
+ 'address_blocks': [
+ {'start': '192.168.1.2', 'size': 5}
+ ],
+ 'hostname': 'hostname',
+ 'username': 'username',
+ 'password': 'password',
+ 'netmask': '255.255.248.0',
+ 'gateway': '0.0.0.0',
+ 'namespace': False,
+ 'svip': '192.168.1.2'
+ }
+ return dict(args)
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp_elementsw_module.NaElementSWModule.set_element_attributes')
+ def test_successful_create(self, mock_set_attributes, mock_create_sf_connection):
+ ''' successful create'''
+ mock_set_attributes.return_value = {'key': 'new_value'}
+ data = self.mock_args()
+ data['vlan_tag'] = '3'
+ set_module_args(data)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = vlan()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_successful_delete(self, mock_create_sf_connection):
+ ''' successful delete'''
+ data = self.mock_args()
+ data['state'] = 'absent'
+ set_module_args(data)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = vlan()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_successful_modify(self, mock_create_sf_connection):
+ ''' successful modify'''
+ data = self.mock_args()
+ data['svip'] = '3.4.5.6'
+ set_module_args(data)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = vlan()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ @patch('ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_vlan.ElementSWVlan.get_network_details')
+ def test_successful_modify_address_blocks_same_length(self, mock_get, mock_create_sf_connection):
+ ''' successful modify'''
+ mock_get.return_value = {
+ 'address_blocks': [
+ {'start': '10.10.10.20', 'size': 5},
+ {'start': '10.10.10.40', 'size': 5}
+ ]
+ }
+ data = self.mock_args()
+ data['address_blocks'] = [{'start': '10.10.10.20', 'size': 5},
+ {'start': '10.20.10.50', 'size': 5}]
+ set_module_args(data)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = vlan()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ @patch('ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_vlan.ElementSWVlan.get_network_details')
+ def test_successful_modify_address_blocks_different_length_1(self, mock_get, mock_create_sf_connection):
+ ''' successful modify'''
+ mock_get.return_value = {
+ 'address_blocks': [
+ {'start': '10.10.10.20', 'size': 5},
+ {'start': '10.20.10.30', 'size': 5}
+ ]
+ }
+ data = self.mock_args()
+ data['address_blocks'] = [{'start': '10.10.10.20', 'size': 5},
+ {'start': '10.20.10.30', 'size': 5},
+ {'start': '10.20.10.50', 'size': 5}]
+ set_module_args(data)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = vlan()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ @patch('ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_vlan.ElementSWVlan.get_network_details')
+ def test_successful_modify_address_blocks_different_length_2(self, mock_get, mock_create_sf_connection):
+ ''' successful modify'''
+ mock_get.return_value = {
+ 'address_blocks': [
+ {'start': '10.10.10.20', 'size': 5},
+ {'start': '10.20.10.30', 'size': 5},
+ {'start': '10.20.10.40', 'size': 5}
+ ]
+ }
+ data = self.mock_args()
+ data['address_blocks'] = [{'start': '10.10.10.20', 'size': 5},
+ {'start': '10.20.10.40', 'size': 5},
+ {'start': '10.20.10.30', 'size': 5}]
+ set_module_args(data)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = vlan()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ assert not exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ @patch('ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_vlan.ElementSWVlan.get_network_details')
+ def test_successful_modify_address_blocks_different_length_3(self, mock_get, mock_create_sf_connection):
+ ''' successful modify'''
+ mock_get.return_value = {
+ 'address_blocks': [
+ {'start': '10.10.10.20', 'size': 5},
+ {'start': '10.10.10.30', 'size': 5},
+ {'start': '10.20.10.40', 'size': 5}
+ ]
+ }
+ data = self.mock_args()
+ data['address_blocks'] = [{'start': '10.10.10.20', 'size': 5},
+ {'start': '10.20.10.40', 'size': 5},
+ {'start': '10.20.10.30', 'size': 5}]
+ set_module_args(data)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = vlan()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_helper_validate_keys(self, mock_create_sf_connection):
+ '''test validate_keys()'''
+ data = self.mock_args()
+ del data['svip']
+ set_module_args(data)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = vlan()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.validate_keys()
+ msg = "One or more required fields ['address_blocks', 'svip', 'netmask', 'name'] for creating VLAN is missing"
+ assert exc.value.args[0]['msg'] == msg
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_successful_modify_idempotent(self, mock_create_sf_connection):
+ ''' successful modify'''
+ data = self.mock_args()
+ data['address_blocks'] = [{'start': '2.2.2.2', 'size': 4},
+ {'start': '3.3.3.3', 'size': 4}]
+ set_module_args(data)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = vlan()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ assert not exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_successful_modify_attribute_value(self, mock_create_sf_connection):
+ ''' successful modify'''
+ data = self.mock_args()
+ data['address_blocks'] = [{'start': '2.2.2.2', 'size': 4},
+ {'start': '3.3.3.3', 'size': 4}]
+ data['attributes'] = {'key': 'value2'}
+ set_module_args(data)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = vlan()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_successful_modify_attribute_key(self, mock_create_sf_connection):
+ ''' successful modify'''
+ data = self.mock_args()
+ data['address_blocks'] = [{'start': '2.2.2.2', 'size': 4},
+ {'start': '3.3.3.3', 'size': 4}]
+ data['attributes'] = {'key2': 'value2'}
+ set_module_args(data)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = vlan()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ assert exc.value.args[0]['changed']
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_volume.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_volume.py
new file mode 100644
index 00000000..926dda90
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_volume.py
@@ -0,0 +1,364 @@
+''' unit test for Ansible module: na_elementsw_volume.py '''
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+import json
+import pytest
+
+from ansible.module_utils import basic
+from ansible.module_utils._text import to_bytes
+from ansible_collections.netapp.elementsw.tests.unit.compat import unittest
+from ansible_collections.netapp.elementsw.tests.unit.compat.mock import patch
+import ansible_collections.netapp.elementsw.plugins.module_utils.netapp as netapp_utils
+
+if not netapp_utils.has_sf_sdk():
+ pytestmark = pytest.mark.skip('skipping as missing required SolidFire Python SDK')
+
+from ansible_collections.netapp.elementsw.plugins.modules.na_elementsw_volume \
+ import ElementSWVolume as my_module # module under test
+
+
+def set_module_args(args):
+ """prepare arguments so that they will be picked up during module creation"""
+ args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
+ basic._ANSIBLE_ARGS = to_bytes(args) # pylint: disable=protected-access
+
+
+class AnsibleExitJson(Exception):
+ """Exception class to be raised by module.exit_json and caught by the test case"""
+
+
+class AnsibleFailJson(Exception):
+ """Exception class to be raised by module.fail_json and caught by the test case"""
+
+
+def exit_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over exit_json; package return data into an exception"""
+ if 'changed' not in kwargs:
+ kwargs['changed'] = False
+ raise AnsibleExitJson(kwargs)
+
+
+def fail_json(*args, **kwargs): # pylint: disable=unused-argument
+ """function to patch over fail_json; package return data into an exception"""
+ kwargs['failed'] = True
+ raise AnsibleFailJson(kwargs)
+
+
+CREATE_ERROR = 'create', 'some_error_in_create_volume'
+MODIFY_ERROR = 'modify', 'some_error_in_modify_volume'
+DELETE_ERROR = 'delete', 'some_error_in_delete_volume'
+
+POLICY_ID = 888
+POLICY_NAME = 'element_qos_policy_name'
+VOLUME_ID = 777
+VOLUME_NAME = 'element_volume_name'
+
+
+class MockSFConnection(object):
+ ''' mock connection to ElementSW host '''
+
+ class Bunch(object): # pylint: disable=too-few-public-methods
+ ''' create object with arbitrary attributes '''
+ def __init__(self, **kw):
+ ''' called with (k1=v1, k2=v2), creates obj.k1, obj.k2 with values v1, v2 '''
+ setattr(self, '__dict__', kw)
+
+ def __init__(self, force_error=False, where=None, with_qos_policy_id=True):
+ ''' save arguments '''
+ self.force_error = force_error
+ self.where = where
+ self.with_qos_policy_id = with_qos_policy_id
+
+ def list_qos_policies(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' build qos_policy list '''
+ qos_policy_name = POLICY_NAME
+ qos = self.Bunch(min_iops=1000, max_iops=20000, burst_iops=20000)
+ qos_policy = self.Bunch(name=qos_policy_name, qos_policy_id=POLICY_ID, qos=qos)
+ qos_policy_1 = self.Bunch(name=qos_policy_name + '_1', qos_policy_id=POLICY_ID + 1, qos=qos)
+ qos_policies = [qos_policy, qos_policy_1]
+ qos_policy_list = self.Bunch(qos_policies=qos_policies)
+ return qos_policy_list
+
+ def list_volumes_for_account(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' build volume list: volume.name, volume.id '''
+ volume = self.Bunch(name=VOLUME_NAME, volume_id=VOLUME_ID, delete_time='')
+ volumes = [volume]
+ volume_list = self.Bunch(volumes=volumes)
+ return volume_list
+
+ def list_volumes(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' build volume details: volume.name, volume.id '''
+ if self.with_qos_policy_id:
+ qos_policy_id = POLICY_ID
+ else:
+ qos_policy_id = None
+ qos = self.Bunch(min_iops=1000, max_iops=20000, burst_iops=20000)
+ volume = self.Bunch(name=VOLUME_NAME, volume_id=VOLUME_ID, delete_time='', access='rw',
+ account_id=1, qos=qos, qos_policy_id=qos_policy_id, total_size=1000000000,
+ attributes={'config-mgmt': 'ansible', 'event-source': 'na_elementsw_volume'}
+ )
+ volumes = [volume]
+ volume_list = self.Bunch(volumes=volumes)
+ return volume_list
+
+ def get_account_by_name(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' returns account_id '''
+ if self.force_error and 'get_account_id' in self.where:
+ account_id = None
+ else:
+ account_id = 1
+ account = self.Bunch(account_id=account_id)
+ result = self.Bunch(account=account)
+ return result
+
+ def create_volume(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' We don't check the return code, but could force an exception '''
+ if self.force_error and 'create_exception' in self.where:
+ raise netapp_utils.solidfire.common.ApiServerError(*CREATE_ERROR)
+
+ def modify_volume(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' We don't check the return code, but could force an exception '''
+ print("modify: %s, %s " % (repr(args), repr(kwargs)))
+ if self.force_error and 'modify_exception' in self.where:
+ raise netapp_utils.solidfire.common.ApiServerError(*MODIFY_ERROR)
+
+ def delete_volume(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' We don't check the return code, but could force an exception '''
+ if self.force_error and 'delete_exception' in self.where:
+ raise netapp_utils.solidfire.common.ApiServerError(*DELETE_ERROR)
+
+ def purge_deleted_volume(self, *args, **kwargs): # pylint: disable=unused-argument
+ ''' We don't check the return code, but could force an exception '''
+ if self.force_error and 'delete_exception' in self.where:
+ raise netapp_utils.solidfire.common.ApiServerError(*DELETE_ERROR)
+
+
+class TestMyModule(unittest.TestCase):
+ ''' a group of related Unit Tests '''
+
+ ARGS = {
+ 'state': 'present',
+ 'name': VOLUME_NAME,
+ 'account_id': 'element_account_id',
+ 'qos': {'minIOPS': 1000, 'maxIOPS': 20000, 'burstIOPS': 20000},
+ 'qos_policy_name': POLICY_NAME,
+ 'size': 1,
+ 'enable512e': True,
+ 'hostname': 'hostname',
+ 'username': 'username',
+ 'password': 'password',
+ }
+
+ def setUp(self):
+ self.mock_module_helper = patch.multiple(basic.AnsibleModule,
+ exit_json=exit_json,
+ fail_json=fail_json)
+ self.mock_module_helper.start()
+ self.addCleanup(self.mock_module_helper.stop)
+
+ def test_module_fail_when_required_args_missing(self):
+ ''' required arguments are reported as errors '''
+ with pytest.raises(AnsibleFailJson) as exc:
+ set_module_args({})
+ my_module()
+ print('Info: %s' % exc.value.args[0]['msg'])
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_add_volume(self, mock_create_sf_connection):
+ ''' adding a volume '''
+ args = dict(self.ARGS) # deep copy as other tests can modify args
+ args['name'] += '_1' # new name to force a create
+ args.pop('qos') # parameters are mutually exclusive: qos|qos_policy_name
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_add_or_modify_volume_idempotent_qos_policy(self, mock_create_sf_connection):
+ ''' adding a volume '''
+ args = dict(self.ARGS)
+ args.pop('qos') # parameters are mutually exclusive: qos|qos_policy_name
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_add_or_modify_volume_idempotent_qos(self, mock_create_sf_connection):
+ ''' adding a volume '''
+ args = dict(self.ARGS)
+ args.pop('qos_policy_name') # parameters are mutually exclusive: qos|qos_policy_name
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(with_qos_policy_id=False)
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_delete_volume(self, mock_create_sf_connection):
+ ''' removing a volume '''
+ args = dict(self.ARGS)
+ args['state'] = 'absent'
+ args.pop('qos') # parameters are mutually exclusive: qos|qos_policy_name
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_delete_volume_idempotent(self, mock_create_sf_connection):
+ ''' removing a volume '''
+ args = dict(self.ARGS)
+ args['state'] = 'absent'
+ args['name'] += '_1' # new name to force idempotency
+ args.pop('qos') # parameters are mutually exclusive: qos|qos_policy_name
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert not exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_modify_volume_qos(self, mock_create_sf_connection):
+ ''' modifying a volume '''
+ args = dict(self.ARGS)
+ args['qos'] = {'minIOPS': 2000}
+ args.pop('qos_policy_name') # parameters are mutually exclusive: qos|qos_policy_name
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(with_qos_policy_id=False)
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_modify_volume_qos_policy_to_qos(self, mock_create_sf_connection):
+ ''' modifying a volume '''
+ args = dict(self.ARGS)
+ args['qos'] = {'minIOPS': 2000}
+ args.pop('qos_policy_name') # parameters are mutually exclusive: qos|qos_policy_name
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_modify_volume_qos_policy(self, mock_create_sf_connection):
+ ''' modifying a volume '''
+ args = dict(self.ARGS)
+ args['qos_policy_name'] += '_1'
+ args.pop('qos') # parameters are mutually exclusive: qos|qos_policy_name
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_modify_volume_qos_to_qos_policy(self, mock_create_sf_connection):
+ ''' modifying a volume '''
+ args = dict(self.ARGS)
+ args.pop('qos') # parameters are mutually exclusive: qos|qos_policy_name
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(with_qos_policy_id=False)
+ my_obj = my_module()
+ with pytest.raises(AnsibleExitJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ assert exc.value.args[0]['changed']
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_create_volume_exception(self, mock_create_sf_connection):
+ ''' creating a volume can raise an exception '''
+ args = dict(self.ARGS)
+ args['name'] += '_1' # new name to force a create
+ args.pop('qos') # parameters are mutually exclusive: qos|qos_policy_name
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(force_error=True, where=['create_exception'])
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = 'Error provisioning volume: %s' % args['name']
+ assert exc.value.args[0]['msg'].startswith(message)
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_modify_volume_exception(self, mock_create_sf_connection):
+ ''' modifying a volume can raise an exception '''
+ args = dict(self.ARGS)
+ args['qos'] = {'minIOPS': 2000}
+ args.pop('qos_policy_name') # parameters are mutually exclusive: qos|qos_policy_name
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(force_error=True, where=['modify_exception'])
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = 'Error updating volume: %s' % VOLUME_ID
+ assert exc.value.args[0]['msg'].startswith(message)
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_delete_volume_exception(self, mock_create_sf_connection):
+ ''' deleting a volume can raise an exception '''
+ args = dict(self.ARGS)
+ args['state'] = 'absent'
+ args.pop('qos') # parameters are mutually exclusive: qos|qos_policy_name
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection(force_error=True, where=['delete_exception'])
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = 'Error deleting volume: %s' % VOLUME_ID
+ assert exc.value.args[0]['msg'].startswith(message)
+
+ @patch('ansible_collections.netapp.elementsw.plugins.module_utils.netapp.create_sf_connection')
+ def test_check_error_reporting_on_non_existent_qos_policy(self, mock_create_sf_connection):
+ ''' report error if qos option is not given on create '''
+ args = dict(self.ARGS)
+ args['name'] += '_1' # new name to force a create
+ args.pop('qos')
+ args['qos_policy_name'] += '_2'
+ set_module_args(args)
+ # my_obj.sfe will be assigned a MockSFConnection object:
+ mock_create_sf_connection.return_value = MockSFConnection()
+ my_obj = my_module()
+ with pytest.raises(AnsibleFailJson) as exc:
+ my_obj.apply()
+ print(exc.value.args[0])
+ message = "Cannot find qos policy with name/id: %s" % args['qos_policy_name']
+ assert exc.value.args[0]['msg'] == message
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules_utils/test_netapp_module.py b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules_utils/test_netapp_module.py
new file mode 100644
index 00000000..171a7bae
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/plugins/modules_utils/test_netapp_module.py
@@ -0,0 +1,149 @@
+# Copyright (c) 2018 NetApp
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+''' unit tests for module_utils netapp_module.py '''
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+from ansible_collections.netapp.elementsw.tests.unit.compat import unittest
+from ansible_collections.netapp.elementsw.plugins.module_utils.netapp_module import NetAppModule as na_helper
+
+
+class TestMyModule(unittest.TestCase):
+ ''' a group of related Unit Tests '''
+
+ def test_get_cd_action_create(self):
+ ''' validate cd_action for create '''
+ current = None
+ desired = {'state': 'present'}
+ my_obj = na_helper()
+ result = my_obj.get_cd_action(current, desired)
+ assert result == 'create'
+
+ def test_get_cd_action_delete(self):
+ ''' validate cd_action for delete '''
+ current = {'state': 'absent'}
+ desired = {'state': 'absent'}
+ my_obj = na_helper()
+ result = my_obj.get_cd_action(current, desired)
+ assert result == 'delete'
+
+ def test_get_cd_action(self):
+ ''' validate cd_action for returning None '''
+ current = None
+ desired = {'state': 'absent'}
+ my_obj = na_helper()
+ result = my_obj.get_cd_action(current, desired)
+ assert result is None
+
+ def test_get_modified_attributes_for_no_data(self):
+ ''' validate modified attributes when current is None '''
+ current = None
+ desired = {'name': 'test'}
+ my_obj = na_helper()
+ result = my_obj.get_modified_attributes(current, desired)
+ assert result == {}
+
+ def test_get_modified_attributes(self):
+ ''' validate modified attributes '''
+ current = {'name': ['test', 'abcd', 'xyz', 'pqr'], 'state': 'present'}
+ desired = {'name': ['abcd', 'abc', 'xyz', 'pqr'], 'state': 'absent'}
+ my_obj = na_helper()
+ result = my_obj.get_modified_attributes(current, desired)
+ assert result == desired
+
+ def test_get_modified_attributes_for_intersecting_mixed_list(self):
+ ''' validate modified attributes for list diff '''
+ current = {'name': [2, 'four', 'six', 8]}
+ desired = {'name': ['a', 8, 'ab', 'four', 'abcd']}
+ my_obj = na_helper()
+ result = my_obj.get_modified_attributes(current, desired, True)
+ assert result == {'name': ['a', 'ab', 'abcd']}
+
+ def test_get_modified_attributes_for_intersecting_list(self):
+ ''' validate modified attributes for list diff '''
+ current = {'name': ['two', 'four', 'six', 'eight']}
+ desired = {'name': ['a', 'six', 'ab', 'four', 'abc']}
+ my_obj = na_helper()
+ result = my_obj.get_modified_attributes(current, desired, True)
+ assert result == {'name': ['a', 'ab', 'abc']}
+
+ def test_get_modified_attributes_for_nonintersecting_list(self):
+ ''' validate modified attributes for list diff '''
+ current = {'name': ['two', 'four', 'six', 'eight']}
+ desired = {'name': ['a', 'ab', 'abd']}
+ my_obj = na_helper()
+ result = my_obj.get_modified_attributes(current, desired, True)
+ assert result == {'name': ['a', 'ab', 'abd']}
+
+ def test_get_modified_attributes_for_list_of_dicts_no_data(self):
+ ''' validate modified attributes for list diff '''
+ current = None
+ desired = {'address_blocks': [{'start': '10.20.10.40', 'size': 5}]}
+ my_obj = na_helper()
+ result = my_obj.get_modified_attributes(current, desired, True)
+ assert result == {}
+
+ def test_get_modified_attributes_for_intersecting_list_of_dicts(self):
+ ''' validate modified attributes for list diff '''
+ current = {'address_blocks': [{'start': '10.10.10.23', 'size': 5}, {'start': '10.10.10.30', 'size': 5}]}
+ desired = {'address_blocks': [{'start': '10.10.10.23', 'size': 5}, {'start': '10.10.10.30', 'size': 5}, {'start': '10.20.10.40', 'size': 5}]}
+ my_obj = na_helper()
+ result = my_obj.get_modified_attributes(current, desired, True)
+ assert result == {'address_blocks': [{'start': '10.20.10.40', 'size': 5}]}
+
+ def test_get_modified_attributes_for_nonintersecting_list_of_dicts(self):
+ ''' validate modified attributes for list diff '''
+ current = {'address_blocks': [{'start': '10.10.10.23', 'size': 5}, {'start': '10.10.10.30', 'size': 5}]}
+ desired = {'address_blocks': [{'start': '10.20.10.23', 'size': 5}, {'start': '10.20.10.30', 'size': 5}, {'start': '10.20.10.40', 'size': 5}]}
+ my_obj = na_helper()
+ result = my_obj.get_modified_attributes(current, desired, True)
+ assert result == {'address_blocks': [{'start': '10.20.10.23', 'size': 5}, {'start': '10.20.10.30', 'size': 5}, {'start': '10.20.10.40', 'size': 5}]}
+
+ def test_get_modified_attributes_for_list_diff(self):
+ ''' validate modified attributes for list diff '''
+ current = {'name': ['test', 'abcd'], 'state': 'present'}
+ desired = {'name': ['abcd', 'abc'], 'state': 'present'}
+ my_obj = na_helper()
+ result = my_obj.get_modified_attributes(current, desired, True)
+ assert result == {'name': ['abc']}
+
+ def test_get_modified_attributes_for_no_change(self):
+ ''' validate modified attributes for same data in current and desired '''
+ current = {'name': 'test'}
+ desired = {'name': 'test'}
+ my_obj = na_helper()
+ result = my_obj.get_modified_attributes(current, desired)
+ assert result == {}
+
+ def test_is_rename_action_for_empty_input(self):
+ ''' validate rename action for input None '''
+ source = None
+ target = None
+ my_obj = na_helper()
+ result = my_obj.is_rename_action(source, target)
+ assert result == source
+
+ def test_is_rename_action_for_no_source(self):
+ ''' validate rename action when source is None '''
+ source = None
+ target = 'test2'
+ my_obj = na_helper()
+ result = my_obj.is_rename_action(source, target)
+ assert result is False
+
+ def test_is_rename_action_for_no_target(self):
+ ''' validate rename action when target is None '''
+ source = 'test2'
+ target = None
+ my_obj = na_helper()
+ result = my_obj.is_rename_action(source, target)
+ assert result is True
+
+ def test_is_rename_action(self):
+ ''' validate rename action '''
+ source = 'test'
+ target = 'test2'
+ my_obj = na_helper()
+ result = my_obj.is_rename_action(source, target)
+ assert result is False
diff --git a/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/requirements.txt b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/requirements.txt
new file mode 100644
index 00000000..65a0c585
--- /dev/null
+++ b/collections-debian-merged/ansible_collections/netapp/elementsw/tests/unit/requirements.txt
@@ -0,0 +1,2 @@
+solidfire-sdk-python
+unittest2 ; python_version < '2.7'