From 177a4fb3ec00c86867e11677d70453fcbdf3e4ba Mon Sep 17 00:00:00 2001 From: Anil Kumar Muraleedharan Date: Tue, 14 Nov 2017 20:39:29 +0530 Subject: [PATCH] Adding enos_command module and unit test (#32782) * Adding git_command module and its UT file * Changing Author Name and removing 2 blank lines * Removing blank lines * Adding enos_config and its UT files * Removing config module as I am allowed to have only module per PR * Work on Ganesh's Review comments * John Review Comments on enos_command.py * Review comments of John --- .../modules/network/enos/enos_command.py | 219 ++++++++++++++++++ .../modules/network/enos/test_enos_command.py | 103 ++++++++ 2 files changed, 322 insertions(+) create mode 100644 lib/ansible/modules/network/enos/enos_command.py create mode 100644 test/units/modules/network/enos/test_enos_command.py diff --git a/lib/ansible/modules/network/enos/enos_command.py b/lib/ansible/modules/network/enos/enos_command.py new file mode 100644 index 00000000000..55904e56510 --- /dev/null +++ b/lib/ansible/modules/network/enos/enos_command.py @@ -0,0 +1,219 @@ +#!/usr/bin/python +# +# Copyright (C) 2017 Lenovo, 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': 'community'} + + +DOCUMENTATION = """ +--- +module: enos_command +version_added: "2.5" +author: "Anil Kumar Muraleedharan (@amuraleedhar)" +short_description: Run arbitrary commands on Lenovo ENOS devices +description: + - Sends arbitrary commands to an ENOS node and returns the results + read from the device. The C(enos_command) module includes an + argument that will cause the module to wait for a specific condition + before returning or timing out if the condition is not met. +extends_documentation_fragment: enos +options: + commands: + description: + - List of commands to send to the remote device over the + configured provider. The resulting output from the command + is returned. If the I(wait_for) argument is provided, the + module is not returned until the condition is satisfied or + the number of retires as expired. + required: true + wait_for: + description: + - List of conditions to evaluate against the output of the + command. The task will wait for each condition to be true + before moving forward. If the conditional is not true + within the configured number of retries, the task fails. + See examples. + required: false + default: null + match: + description: + - The I(match) argument is used in conjunction with the + I(wait_for) argument to specify the match policy. Valid + values are C(all) or C(any). If the value is set to C(all) + then all conditionals in the wait_for must be satisfied. If + the value is set to C(any) then only one of the values must be + satisfied. + required: false + default: all + choices: ['any', 'all'] + retries: + description: + - Specifies the number of retries a command should by tried + before it is considered failed. The command is run on the + target device every retry and evaluated against the + I(wait_for) conditions. + required: false + default: 10 + interval: + description: + - Configures the interval in seconds to wait between retries + of the command. If the command does not pass the specified + conditions, the interval indicates how long to wait before + trying the command again. + required: false + default: 1 +""" + +EXAMPLES = """ +# Note: examples below use the following provider dict to handle +# transport and authentication to the node. +--- +vars: + cli: + host: "{{ inventory_hostname }}" + port: 22 + username: admin + password: admin + timeout: 30 + +--- +- name: test contains operator + enos_command: + commands: + - show version + - show system memory + wait_for: + - "result[0] contains 'Lenovo'" + - "result[1] contains 'MemFree'" + provider: "{{ cli }}" + register: result + +- assert: + that: + - "result.changed == false" + - "result.stdout is defined" + +- name: get output for single command + enos_command: + commands: ['show version'] + provider: "{{ cli }}" + register: result + +- assert: + that: + - "result.changed == false" + - "result.stdout is defined" + +- name: get output for multiple commands + enos_command: + commands: + - show version + - show interface information + provider: "{{ cli }}" + register: result + +- assert: + that: + - "result.changed == false" + - "result.stdout is defined" + - "result.stdout | length == 2" +""" + +RETURN = """ +stdout: + description: the set of responses from the commands + returned: always + type: list + sample: ['...', '...'] +stdout_lines: + description: The value of stdout split into a list + returned: always + type: list + sample: [['...', '...'], ['...'], ['...']] +failed_conditions: + description: the conditionals that failed + returned: failed + type: list + sample: ['...', '...'] +""" + +import time + +from ansible.module_utils.basic import AnsibleModule +from ansible.module_utils.enos import run_commands, check_args +from ansible.module_utils.enos import enos_argument_spec +from ansible.module_utils.netcli import Conditional +from ansible.module_utils.six import string_types + + +def to_lines(stdout): + for item in stdout: + if isinstance(item, string_types): + item = str(item).split('\n') + yield item + + +def main(): + spec = dict( + # { command: , prompt: , response: } + commands=dict(type='list', required=True), + + wait_for=dict(type='list'), + match=dict(default='all', choices=['all', 'any']), + + retries=dict(default=10, type='int'), + interval=dict(default=1, type='int') + ) + + spec.update(enos_argument_spec) + + module = AnsibleModule(argument_spec=spec, supports_check_mode=True) + result = {'changed': False} + + wait_for = module.params['wait_for'] or list() + conditionals = [Conditional(c) for c in wait_for] + + commands = module.params['commands'] + retries = module.params['retries'] + interval = module.params['interval'] + match = module.params['match'] + + while retries > 0: + responses = run_commands(module, commands) + + for item in list(conditionals): + if item(responses): + if match == 'any': + conditionals = list() + break + conditionals.remove(item) + + if not conditionals: + break + + time.sleep(interval) + retries -= 1 + + if conditionals: + failed_conditions = [item.raw for item in conditionals] + msg = 'One or more conditional statements have not be satisfied' + module.fail_json(msg=msg, failed_conditions=failed_conditions) + + result.update({ + 'changed': False, + 'stdout': responses, + 'stdout_lines': list(to_lines(responses)) + }) + + module.exit_json(**result) + + +if __name__ == '__main__': + main() diff --git a/test/units/modules/network/enos/test_enos_command.py b/test/units/modules/network/enos/test_enos_command.py new file mode 100644 index 00000000000..99e7403d2bf --- /dev/null +++ b/test/units/modules/network/enos/test_enos_command.py @@ -0,0 +1,103 @@ +# Copyright (C) 2017 Lenovo, Inc. +# +# 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 . + +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +import json + +from ansible.compat.tests.mock import patch +from ansible.modules.network.enos import enos_command +from .enos_module import TestEnosModule, load_fixture, set_module_args + + +class TestEnosCommandModule(TestEnosModule): + + module = enos_command + + def setUp(self): + self.mock_run_commands = patch('ansible.modules.network.enos.enos_command.run_commands') + self.run_commands = self.mock_run_commands.start() + + def tearDown(self): + self.mock_run_commands.stop() + + def load_fixtures(self, commands=None): + + def load_from_file(*args, **kwargs): + module, commands = args + output = list() + + for item in commands: + try: + command = item + except ValueError: + command = 'show version' + filename = str(command).replace(' ', '_') + output.append(load_fixture(filename)) + return output + + self.run_commands.side_effect = load_from_file + + def test_enos_command_simple(self): + set_module_args(dict(commands=['show version'])) + result = self.execute_module() + self.assertEqual(len(result['stdout']), 1) + self.assertTrue(result['stdout'][0].startswith('System Information')) + + def test_enos_command_multiple(self): + set_module_args(dict(commands=['show version', 'show run'])) + result = self.execute_module() + self.assertEqual(len(result['stdout']), 2) + self.assertTrue(result['stdout'][0].startswith('System Information')) + + def test_enos_command_wait_for(self): + wait_for = 'result[0] contains "System Information"' + set_module_args(dict(commands=['show version'], wait_for=wait_for)) + self.execute_module() + + def test_enos_command_wait_for_fails(self): + wait_for = 'result[0] contains "test string"' + set_module_args(dict(commands=['show version'], wait_for=wait_for)) + self.execute_module(failed=True) + self.assertEqual(self.run_commands.call_count, 10) + + def test_enos_command_retries(self): + wait_for = 'result[0] contains "test string"' + set_module_args(dict(commands=['show version'], wait_for=wait_for, retries=2)) + self.execute_module(failed=True) + self.assertEqual(self.run_commands.call_count, 2) + + def test_enos_command_match_any(self): + wait_for = ['result[0] contains "System Information"', + 'result[0] contains "test string"'] + set_module_args(dict(commands=['show version'], wait_for=wait_for, match='any')) + self.execute_module() + + def test_enos_command_match_all(self): + wait_for = ['result[0] contains "System Information"', + 'result[0] contains "Lenovo"'] + set_module_args(dict(commands=['show version'], wait_for=wait_for, match='all')) + self.execute_module() + + def test_enos_command_match_all_failure(self): + wait_for = ['result[0] contains "Lenovo ENOS"', + 'result[0] contains "test string"'] + commands = ['show version', 'show run'] + set_module_args(dict(commands=commands, wait_for=wait_for, match='all')) + self.execute_module(failed=True)