diff options
Diffstat (limited to 'ansible_collections/netapp/elementsw/tests')
20 files changed, 0 insertions, 3774 deletions
diff --git a/ansible_collections/netapp/elementsw/tests/unit/compat/__init__.py b/ansible_collections/netapp/elementsw/tests/unit/compat/__init__.py deleted file mode 100644 index e69de29bb..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/compat/__init__.py +++ /dev/null diff --git a/ansible_collections/netapp/elementsw/tests/unit/compat/builtins.py b/ansible_collections/netapp/elementsw/tests/unit/compat/builtins.py deleted file mode 100644 index f60ee6782..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/compat/builtins.py +++ /dev/null @@ -1,33 +0,0 @@ -# (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/ansible_collections/netapp/elementsw/tests/unit/compat/mock.py b/ansible_collections/netapp/elementsw/tests/unit/compat/mock.py deleted file mode 100644 index 0972cd2e8..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/compat/mock.py +++ /dev/null @@ -1,122 +0,0 @@ -# (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/ansible_collections/netapp/elementsw/tests/unit/compat/unittest.py b/ansible_collections/netapp/elementsw/tests/unit/compat/unittest.py deleted file mode 100644 index 73a20cf8c..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/compat/unittest.py +++ /dev/null @@ -1,44 +0,0 @@ -# (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 - -import pytest - -# 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') - - class TestCase: - """ skip everything """ - pytestmark = pytest.mark.skip('Skipping Unit Tests on 2.6 as unittest2 may not be available') -else: - from unittest import * diff --git a/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_access_group.py b/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_access_group.py deleted file mode 100644 index 0bd1e2550..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_access_group.py +++ /dev/null @@ -1,175 +0,0 @@ -''' 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/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_access_group_volumes.py b/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_access_group_volumes.py deleted file mode 100644 index fb78ad78a..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_access_group_volumes.py +++ /dev/null @@ -1,245 +0,0 @@ -''' 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/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_account.py b/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_account.py deleted file mode 100644 index 8075ba5c4..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_account.py +++ /dev/null @@ -1,137 +0,0 @@ -''' 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/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster.py b/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster.py deleted file mode 100644 index 6624f374d..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster.py +++ /dev/null @@ -1,228 +0,0 @@ -''' 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/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster_config.py b/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster_config.py deleted file mode 100644 index 79f461ccc..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster_config.py +++ /dev/null @@ -1,157 +0,0 @@ -''' 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/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster_snmp.py b/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster_snmp.py deleted file mode 100644 index 9236daa04..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_cluster_snmp.py +++ /dev/null @@ -1,176 +0,0 @@ -''' 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/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_info.py b/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_info.py deleted file mode 100644 index dc8fd5e23..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_info.py +++ /dev/null @@ -1,344 +0,0 @@ -''' 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 list_all_nodes(self, *args, **kwargs): # pylint: disable=unused-argument - ''' build all_node list: all_node.name, all_node.all_node_id ''' - self.record(repr(args), repr(kwargs)) - all_nodes = list() - all_nodes.append({'id': 123}) - all_node_list = self.Bunch(all_nodes=all_nodes) - return all_node_list - - def list_drives(self, *args, **kwargs): # pylint: disable=unused-argument - ''' build drive list: drive.name, drive.drive_id ''' - self.record(repr(args), repr(kwargs)) - drives = list() - drives.append({'id': 123}) - drive_list = self.Bunch(drives=drives) - return drive_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/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_initiators.py b/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_initiators.py deleted file mode 100644 index ee5ff85db..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_initiators.py +++ /dev/null @@ -1,201 +0,0 @@ -''' 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/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_network_interfaces.py b/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_network_interfaces.py deleted file mode 100644 index 5364a4e76..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_network_interfaces.py +++ /dev/null @@ -1,293 +0,0 @@ -''' 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/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_nodes.py b/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_nodes.py deleted file mode 100644 index 3e163d000..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_nodes.py +++ /dev/null @@ -1,324 +0,0 @@ -''' 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/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_qos_policy.py b/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_qos_policy.py deleted file mode 100644 index 83ac3711a..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_qos_policy.py +++ /dev/null @@ -1,300 +0,0 @@ -''' 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/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_template.py b/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_template.py deleted file mode 100644 index 7dc6e2d6b..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_template.py +++ /dev/null @@ -1,138 +0,0 @@ -''' 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/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_vlan.py b/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_vlan.py deleted file mode 100644 index e2dc51f79..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_vlan.py +++ /dev/null @@ -1,343 +0,0 @@ -''' 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/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_volume.py b/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_volume.py deleted file mode 100644 index 926dda90b..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/plugins/modules/test_na_elementsw_volume.py +++ /dev/null @@ -1,364 +0,0 @@ -''' 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/ansible_collections/netapp/elementsw/tests/unit/plugins/modules_utils/test_netapp_module.py b/ansible_collections/netapp/elementsw/tests/unit/plugins/modules_utils/test_netapp_module.py deleted file mode 100644 index 171a7bae5..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/plugins/modules_utils/test_netapp_module.py +++ /dev/null @@ -1,149 +0,0 @@ -# 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/ansible_collections/netapp/elementsw/tests/unit/requirements.txt b/ansible_collections/netapp/elementsw/tests/unit/requirements.txt deleted file mode 100644 index dde1958f1..000000000 --- a/ansible_collections/netapp/elementsw/tests/unit/requirements.txt +++ /dev/null @@ -1 +0,0 @@ -solidfire-sdk-python ; python_version >= '2.7' |