cmueller/ansible
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

New Module: na_ontap_nvme (#51182)

Browse source 
* changes to clusteR

* Revert "changes to clusteR"

This reverts commit 33ee1b71e4bc8435fb315762a871f8c4cb6c5f80.

* Add new module nvme
This commit is contained in:
Chris Archibald 2019-02-13 12:36:58 -08:00 committed by John R Barker
parent 1b5575c669
commit 6a06e9f485
2 changed files with 425 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

209
lib/ansible/modules/storage/netapp/na_ontap_nvme.py Normal file
View file

@ -0,0 +1,209 @@
#!/usr/bin/python
# (c) 2019, NetApp, Inc
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
ANSIBLE_METADATA = {'metadata_version': '1.1',
'status': ['preview'],
'supported_by': 'certified'}
DOCUMENTATION = '''
author: NetApp Ansible Team (@carchi8py) (ng-ansibleteam@netapp.com)
description:
- Create/Delete NVMe Service
extends_documentation_fragment:
- netapp.na_ontap
module: na_ontap_nvme
options:
state:
choices: ['present', 'absent']
description:
- Whether the specified NVMe should exist or not.
default: present
vserver:
description:
- Name of the vserver to use.
required: true
status_admin:
description:
- Whether the status of NVMe should be up or down
type: bool
short_description: "NetApp ONTAP Manage NVMe Service"
version_added: "2.8"
'''
EXAMPLES = """
- name: Create NVMe
na_ontap_nvme:
state: present
status_admin: False
vserver: "{{ vserver }}"
hostname: "{{ hostname }}"
username: "{{ username }}"
password: "{{ password }}"
- name: Modify NVMe
na_ontap_nvme:
state: present
status_admin: True
vserver: "{{ vserver }}"
hostname: "{{ hostname }}"
username: "{{ username }}"
password: "{{ password }}"
- name: Delete NVMe
na_ontap_nvme:
state: absent
vserver: "{{ vserver }}"
hostname: "{{ hostname }}"
username: "{{ username }}"
password: "{{ password }}"
"""
RETURN = """
"""
import traceback
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils._text import to_native
import ansible.module_utils.netapp as netapp_utils
from ansible.module_utils.netapp_module import NetAppModule
HAS_NETAPP_LIB = netapp_utils.has_netapp_lib()
class NetAppONTAPNVMe(object):
"""
Class with NVMe service methods
"""
def __init__(self):
self.argument_spec = netapp_utils.na_ontap_host_argument_spec()
self.argument_spec.update(dict(
state=dict(required=False, type='str', choices=['present', 'absent'], default='present'),
vserver=dict(required=True, type='str'),
status_admin=dict(required=False, type=bool)
))
self.module = AnsibleModule(
argument_spec=self.argument_spec,
supports_check_mode=True
)
self.na_helper = NetAppModule()
self.parameters = self.na_helper.set_parameters(self.module.params)
if HAS_NETAPP_LIB is False:
self.module.fail_json(msg="the python NetApp-Lib module is required")
else:
self.server = netapp_utils.setup_na_ontap_zapi(module=self.module, vserver=self.parameters['vserver'])
def get_nvme(self):
"""
Get current nvme details
:return: dict if nvme exists, None otherwise
"""
nvme_get = netapp_utils.zapi.NaElement('nvme-get-iter')
query = {
'query': {
'nvme-target-service-info': {
'vserver': self.parameters['vserver']
}
}
}
nvme_get.translate_struct(query)
try:
result = self.server.invoke_successfully(nvme_get, enable_tunneling=False)
except netapp_utils.zapi.NaApiError as error:
self.module.fail_json(msg='Error fetching nvme info: %s' % to_native(error),
exception=traceback.format_exc())
if result.get_child_by_name('num-records') and int(result.get_child_content('num-records')) >= 1:
attributes_list = result.get_child_by_name('attributes-list')
nvme_info = attributes_list.get_child_by_name('nvme-target-service-info')
return_value = {'status_admin': nvme_info.get_child_content('is-available')}
return return_value
return None
def create_nvme(self):
"""
Create NVMe service
"""
nvme_create = netapp_utils.zapi.NaElement('nvme-create')
if self.parameters.get('status_admin') is not None:
options = {'is-available': self.parameters['status_admin']}
nvme_create.translate_struct(options)
try:
self.server.invoke_successfully(nvme_create, enable_tunneling=True)
except netapp_utils.zapi.NaApiError as error:
self.module.fail_json(msg='Error creating nvme for vserver %s: %s'
% (self.parameters['vserver'], to_native(error)),
exception=traceback.format_exc())
def delete_nvme(self):
"""
Delete NVMe service
"""
nvme_delete = netapp_utils.zapi.NaElement('nvme-delete')
try:
self.server.invoke_successfully(nvme_delete, enable_tunneling=True)
except netapp_utils.zapi.NaApiError as error:
self.module.fail_json(msg='Error deleting nvme for vserver %s: %s'
% (self.parameters['vserver'], to_native(error)),
exception=traceback.format_exc())
def modify_nvme(self, status=None):
"""
Modify NVMe service
"""
if status is None:
status = self.parameters['status_admin']
options = {'is-available': status}
nvme_modify = netapp_utils.zapi.NaElement('nvme-modify')
nvme_modify.translate_struct(options)
try:
self.server.invoke_successfully(nvme_modify, enable_tunneling=True)
except netapp_utils.zapi.NaApiError as error:
self.module.fail_json(msg='Error modifying nvme for vserver %s: %s'
% (self.parameters['vserver'], to_native(error)),
exception=traceback.format_exc())
def apply(self):
"""
Apply action to NVMe service
"""
netapp_utils.ems_log_event("na_ontap_nvme", self.server)
current = self.get_nvme()
cd_action = self.na_helper.get_cd_action(current, self.parameters)
if self.parameters.get('status_admin') is not None:
self.parameters['status_admin'] = self.na_helper.get_value_for_bool(False, self.parameters['status_admin'])
if cd_action is None and self.parameters['state'] == 'present':
modify = self.na_helper.get_modified_attributes(current, self.parameters)
if self.na_helper.changed:
if self.module.check_mode:
pass
else:
if cd_action == 'create':
self.create_nvme()
elif cd_action == 'delete':
# NVMe status_admin needs to be down before deleting it
self.modify_nvme('false')
self.delete_nvme()
elif modify:
self.modify_nvme()
self.module.exit_json(changed=self.na_helper.changed)
def main():
"""Execute action"""
community_obj = NetAppONTAPNVMe()
community_obj.apply()
if __name__ == '__main__':
main()

216
test/units/modules/storage/netapp/test_na_ontap_nvme.py Normal file
View file

@ -0,0 +1,216 @@
# (c) 2018, NetApp, Inc
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
''' unit tests ONTAP Ansible module: na_ontap_nvme'''
from __future__ import print_function
import json
import pytest
from units.compat import unittest
from units.compat.mock import patch
from ansible.module_utils import basic
from ansible.module_utils._text import to_bytes
import ansible.module_utils.netapp as netapp_utils
from ansible.modules.storage.netapp.na_ontap_nvme \
import NetAppONTAPNVMe as my_module
if not netapp_utils.has_netapp_lib():
pytestmark = pytest.skip('skipping as missing required netapp_lib')
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 MockONTAPConnection(object):
''' mock server connection to ONTAP host '''
def __init__(self, kind=None):
''' save arguments '''
self.type = kind
self.xml_in = None
self.xml_out = None
def invoke_successfully(self, xml, enable_tunneling): # pylint: disable=unused-argument
''' mock invoke_successfully returning xml data '''
self.xml_in = xml
if self.type == 'nvme':
xml = self.build_nvme_info()
elif self.type == 'nvme_fail':
raise netapp_utils.zapi.NaApiError(code='TEST', message="This exception is from the unit test")
self.xml_out = xml
return xml
@staticmethod
def build_nvme_info():
''' build xml data for nvme-info '''
xml = netapp_utils.zapi.NaElement('xml')
data = {'num-records': 1,
'attributes-list': [{'nvme-target-service-info': {'is-available': 'true'}}]}
xml.translate_struct(data)
return xml
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)
self.server = MockONTAPConnection()
self.onbox = False
def set_default_args(self):
if self.onbox:
hostname = '10.193.75.3'
username = 'admin'
password = 'netapp1!'
vserver = 'ansible'
status_admin = True
else:
hostname = 'hostname'
username = 'username'
password = 'password'
vserver = 'vserver'
status_admin = True
return dict({
'hostname': hostname,
'username': username,
'password': password,
'vserver': vserver,
'status_admin': status_admin
})
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_ensure_get_called(self):
''' test get_nvme() for non-existent nvme'''
set_module_args(self.set_default_args())
my_obj = my_module()
my_obj.server = self.server
assert my_obj.get_nvme() is None
def test_ensure_get_called_existing(self):
''' test get_nvme() for existing nvme'''
set_module_args(self.set_default_args())
my_obj = my_module()
my_obj.server = MockONTAPConnection(kind='nvme')
assert my_obj.get_nvme()
@patch('ansible.modules.storage.netapp.na_ontap_nvme.NetAppONTAPNVMe.create_nvme')
def test_successful_create(self, create_nvme):
''' creating nvme and testing idempotency '''
set_module_args(self.set_default_args())
my_obj = my_module()
if not self.onbox:
my_obj.server = self.server
with pytest.raises(AnsibleExitJson) as exc:
my_obj.apply()
assert exc.value.args[0]['changed']
create_nvme.assert_called_with()
# to reset na_helper from remembering the previous 'changed' value
my_obj = my_module()
if not self.onbox:
my_obj.server = MockONTAPConnection('nvme')
with pytest.raises(AnsibleExitJson) as exc:
my_obj.apply()
assert not exc.value.args[0]['changed']
@patch('ansible.modules.storage.netapp.na_ontap_nvme.NetAppONTAPNVMe.delete_nvme')
def test_successful_delete(self, delete_nvme):
''' deleting nvme and testing idempotency '''
data = self.set_default_args()
data['state'] = 'absent'
set_module_args(data)
my_obj = my_module()
if not self.onbox:
my_obj.server = MockONTAPConnection('nvme')
with pytest.raises(AnsibleExitJson) as exc:
my_obj.apply()
assert exc.value.args[0]['changed']
delete_nvme.assert_called_with()
# to reset na_helper from remembering the previous 'changed' value
my_obj = my_module()
if not self.onbox:
my_obj.server = self.server
with pytest.raises(AnsibleExitJson) as exc:
my_obj.apply()
assert not exc.value.args[0]['changed']
@patch('ansible.modules.storage.netapp.na_ontap_nvme.NetAppONTAPNVMe.modify_nvme')
def test_successful_modify(self, modify_nvme):
''' modifying nvme and testing idempotency '''
data = self.set_default_args()
data['status_admin'] = False
set_module_args(data)
my_obj = my_module()
if not self.onbox:
my_obj.server = MockONTAPConnection('nvme')
with pytest.raises(AnsibleExitJson) as exc:
my_obj.apply()
assert exc.value.args[0]['changed']
modify_nvme.assert_called_with()
# to reset na_helper from remembering the previous 'changed' value
data = self.set_default_args()
set_module_args(data)
my_obj = my_module()
if not self.onbox:
my_obj.server = MockONTAPConnection('nvme')
with pytest.raises(AnsibleExitJson) as exc:
my_obj.apply()
assert not exc.value.args[0]['changed']
def test_if_all_methods_catch_exception(self):
module_args = {}
module_args.update(self.set_default_args())
set_module_args(module_args)
my_obj = my_module()
if not self.onbox:
my_obj.server = MockONTAPConnection('nvme_fail')
with pytest.raises(AnsibleFailJson) as exc:
my_obj.get_nvme()
assert 'Error fetching nvme info:' in exc.value.args[0]['msg']
with pytest.raises(AnsibleFailJson) as exc:
my_obj.create_nvme()
assert 'Error creating nvme' in exc.value.args[0]['msg']
with pytest.raises(AnsibleFailJson) as exc:
my_obj.delete_nvme()
assert 'Error deleting nvme' in exc.value.args[0]['msg']
with pytest.raises(AnsibleFailJson) as exc:
my_obj.modify_nvme()
assert 'Error modifying nvme' in exc.value.args[0]['msg']