From 309f7e59a36812c7e5d1a260031affd7b147610d Mon Sep 17 00:00:00 2001 From: Chris Archibald Date: Tue, 19 Mar 2019 07:49:58 -0700 Subject: [PATCH] Add ability to run in Privileged mode to Na_ontap_command (#52343) * update command * doc update * fix review comments --- .../storage/netapp/na_ontap_command.py | 31 +++- .../storage/netapp/test_na_ontap_command.py | 169 ++++++++++++++++++ 2 files changed, 198 insertions(+), 2 deletions(-) create mode 100644 test/units/modules/storage/netapp/test_na_ontap_command.py diff --git a/lib/ansible/modules/storage/netapp/na_ontap_command.py b/lib/ansible/modules/storage/netapp/na_ontap_command.py index 38466a5101..3258b3520f 100644 --- a/lib/ansible/modules/storage/netapp/na_ontap_command.py +++ b/lib/ansible/modules/storage/netapp/na_ontap_command.py @@ -23,6 +23,13 @@ options: command: description: - a comma separated list containing the command and arguments. + required: true + privilege: + description: + - privilege level at which to run the command. + choices: ['admin', 'advanced'] + default: admin + version_added: "2.8" ''' EXAMPLES = """ @@ -39,6 +46,7 @@ EXAMPLES = """ username: "{{ admin username }}" password: "{{ admin password }}" command: ['network', 'interface', 'show'] + privilege: 'admin' """ RETURN = """ @@ -53,10 +61,14 @@ HAS_NETAPP_LIB = netapp_utils.has_netapp_lib() class NetAppONTAPCommand(object): + ''' calls a CLI command ''' + def __init__(self): self.argument_spec = netapp_utils.na_ontap_host_argument_spec() self.argument_spec.update(dict( - command=dict(required=True, type='list') + command=dict(required=True, type='list'), + privilege=dict(required=False, type='str', choices=['admin', 'advanced'], + default='admin') )) self.module = AnsibleModule( argument_spec=self.argument_spec, @@ -65,19 +77,33 @@ class NetAppONTAPCommand(object): parameters = self.module.params # set up state variables self.command = parameters['command'] + self.privilege = parameters['privilege'] 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) - def run_command(self): + def asup_log_for_cserver(self, event_name): + """ + Fetch admin vserver for the given cluster + Create and Autosupport log event with the given module name + :param event_name: Name of the event log + :return: None + """ + results = netapp_utils.get_cserver(self.server) + cserver = netapp_utils.setup_na_ontap_zapi(module=self.module, vserver=results) + netapp_utils.ems_log_event(event_name, cserver) + def run_command(self): + ''' calls the ZAPI ''' + self.asup_log_for_cserver("na_ontap_command: " + str(self.command)) command_obj = netapp_utils.zapi.NaElement("system-cli") args_obj = netapp_utils.zapi.NaElement("args") for arg in self.command: args_obj.add_new_child('arg', arg) command_obj.add_child_elem(args_obj) + command_obj.add_new_child('priv', self.privilege) try: output = self.server.invoke_successfully(command_obj, True) @@ -88,6 +114,7 @@ class NetAppONTAPCommand(object): exception=traceback.format_exc()) def apply(self): + ''' calls the command and returns raw output ''' changed = True output = self.run_command() self.module.exit_json(changed=changed, msg=output) diff --git a/test/units/modules/storage/netapp/test_na_ontap_command.py b/test/units/modules/storage/netapp/test_na_ontap_command.py new file mode 100644 index 0000000000..184ae3a41a --- /dev/null +++ b/test/units/modules/storage/netapp/test_na_ontap_command.py @@ -0,0 +1,169 @@ +# (c) 2018, NetApp, Inc +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +''' unit test for ONTAP Command Ansible module ''' + +from __future__ import print_function +import json +import pytest + +from units.compat import unittest +from units.compat.mock import patch, Mock +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_command \ + import NetAppONTAPCommand as my_module # module under test + +if not netapp_utils.has_netapp_lib(): + pytestmark = pytest.mark.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, parm1=None): + ''' save arguments ''' + self.type = kind + self.parm1 = parm1 + 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 + # print(xml.to_string()) + + if self.type == 'version': + priv = xml.get_child_content('priv') + xml = self.build_version(priv) + self.xml_out = xml + return xml + + @staticmethod + def build_version(priv): + ''' build xml data for version ''' + prefix = 'NetApp Release' + if priv == 'advanced': + prefix = '\n' + prefix + xml = netapp_utils.zapi.NaElement('results') + xml.add_new_child('cli-output', prefix) + # print(xml.to_string()) + 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(kind='version') + # whether to use a mock or a simulator + self.use_vsim = False + + 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']) + + @staticmethod + def set_default_args(vsim=False): + ''' populate hostname/username/password ''' + if vsim: + hostname = '10.193.78.219' + username = 'admin' + password = 'netapp1!' + else: + hostname = 'hostname' + username = 'username' + password = 'password' + return dict({ + 'hostname': hostname, + 'username': username, + 'password': password, + 'https': True, + 'validate_certs': False + }) + + def call_command(self, module_args, vsim=False): + ''' utility function to call apply ''' + module_args.update(self.set_default_args(vsim=vsim)) + set_module_args(module_args) + my_obj = my_module() + my_obj.asup_log_for_cserver = Mock(return_value=None) + if not vsim: + # mock the connection + my_obj.server = self.server + with pytest.raises(AnsibleExitJson) as exc: + my_obj.apply() + msg = exc.value.args[0]['msg'] + return msg + + def test_default_priv(self): + ''' make sure privilege is not required ''' + module_args = { + 'command': 'version', + } + msg = self.call_command(module_args, vsim=self.use_vsim) + needle = b'NetApp Release' + assert needle in msg + print('Version (raw): %s' % msg) + + def test_admin_priv(self): + ''' make sure admin is accepted ''' + module_args = { + 'command': 'version', + 'privilege': 'admin', + } + msg = self.call_command(module_args, vsim=self.use_vsim) + needle = b'NetApp Release' + assert needle in msg + print('Version (raw): %s' % msg) + + def test_advanced_priv(self): + ''' make sure advanced is not required ''' + module_args = { + 'command': 'version', + 'privilege': 'advanced', + } + msg = self.call_command(module_args, vsim=self.use_vsim) + # Interestingly, the ZAPI returns a slightly different response + needle = b'\nNetApp Release' + assert needle in msg + print('Version (raw): %s' % msg)