diff options
Diffstat (limited to 'collections-debian-merged/ansible_collections/netapp/elementsw/tests')
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' |