wbrawner/ansible
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactors eos_command module (#20739)

Browse source 
* eos_command now uses network_cli connection plugin
* adds unit test cases for eos_command module
This commit is contained in:
Peter Sprygada 2017-01-27 08:45:13 -05:00 committed by GitHub
parent 57660abf33
commit c8dcc7f5d6
3 changed files with 259 additions and 100 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

210
lib/ansible/modules/network/eos/eos_command.py
View file

@ -16,9 +16,11 @@
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
#
ANSIBLE_METADATA = {'status': ['preview'],
ANSIBLE_METADATA = {
'status': ['preview'],
'supported_by': 'core',
'version': '1.0'}
'version': '1.0'
}
DOCUMENTATION = """
---
@ -31,7 +33,7 @@ description:
read from the device. This 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: eos
extends_documentation_fragment: eapi
options:
commands:
description:
@ -83,32 +85,20 @@ options:
"""
EXAMPLES = """
# Note: examples below use the following provider dict to handle
# transport and authentication to the node.
vars:
cli:
host: "{{ inventory_hostname }}"
username: admin
password: admin
transport: cli
- name: run show version on remote devices
eos_command:
commands: show version
provider: "{{ cli }}"
- name: run show version and check to see if output contains Arista
eos_command:
commands: show version
wait_for: result[0] contains Arista
provider: "{{ cli }}"
- name: run multiple commands on remote nodes
eos_command:
commands:
- show version
- show interfaces
provider: "{{ cli }}"
- name: run multiple commands and evaluate the output
eos_command:
@ -118,69 +108,90 @@ vars:
wait_for:
- result[0] contains Arista
- result[1] contains Loopback0
provider: "{{ cli }}"
- name: run commands and specify the output format
eos_command:
commands:
- command: show version
output: json
provider: "{{ cli }}"
"""
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
import ansible.module_utils.eos
from functools import partial
from ansible.module_utils.basic import get_exception
from ansible.module_utils.network import NetworkModule, NetworkError
from ansible.module_utils.netcli import CommandRunner
from ansible.module_utils.netcli import AddCommandError, AddConditionError
from ansible.module_utils.netcli import FailedConditionsError
from ansible.module_utils.netcli import FailedConditionalError
from ansible.module_utils import eos
from ansible.module_utils import eapi
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.local import LocalAnsibleModule
from ansible.module_utils.six import string_types
from ansible.module_utils.netcli import Conditional
from ansible.module_utils.network_common import ComplexList
SHARED_LIB = 'eos'
VALID_KEYS = ['command', 'output', 'prompt', 'response']
def get_ansible_module():
if SHARED_LIB == 'eos':
return LocalAnsibleModule
return AnsibleModule
def invoke(name, *args, **kwargs):
obj = globals().get(SHARED_LIB)
func = getattr(obj, name)
return func(*args, **kwargs)
run_commands = partial(invoke, 'run_commands')
def check_args(module, warnings):
if SHARED_LIB == 'eapi':
eapi.check_args(module)
def to_lines(stdout):
lines = list()
for item in stdout:
if isinstance(item, string_types):
item = str(item).split('\n')
yield item
lines.append(item)
return lines
def parse_commands(module):
for cmd in module.params['commands']:
if isinstance(cmd, string_types):
cmd = dict(command=cmd, output=None)
elif 'command' not in cmd:
module.fail_json(msg='command keyword argument is required')
elif cmd.get('output') not in [None, 'text', 'json']:
module.fail_json(msg='invalid output specified for command')
elif not set(cmd.keys()).issubset(VALID_KEYS):
module.fail_json(msg='unknown command keyword specified. Valid '
'values are %s' % ', '.join(VALID_KEYS))
yield cmd
def parse_commands(module, warnings):
cast = ComplexList(dict(
command=dict(key=True),
output=dict(),
prompt=dict(),
response=dict()
))
commands = cast(module.params['commands'])
for index, item in enumerate(commands):
if module.check_mode and not item['command'].startswith('show'):
warnings.append(
'Only show commands are supported when using check_mode, not '
'executing %s' % item['command']
)
return commands
def to_cli(obj):
cmd = obj['command']
if obj.get('output') == 'json':
cmd += ' | json'
return cmd
def main():
spec = dict(
"""entry point for module execution
"""
argument_spec = dict(
# { command: <str>, output: <str>, prompt: <str>, response: <str> }
commands=dict(type='list', required=True),
@ -191,70 +202,59 @@ def main():
interval=dict(default=1, type='int')
)
module = NetworkModule(argument_spec=spec,
supports_check_mode=True)
argument_spec.update(eapi.eapi_argument_spec)
commands = list(parse_commands(module))
conditionals = module.params['wait_for'] or list()
cls = get_ansible_module()
module = cls(argument_spec=argument_spec, supports_check_mode=True)
warnings = list()
check_args(module, warnings)
runner = CommandRunner(module)
for cmd in commands:
if module.check_mode and not cmd['command'].startswith('show'):
warnings.append('only show commands are supported when using '
'check mode, not executing `%s`' % cmd['command'])
else:
if cmd['command'].startswith('conf'):
module.fail_json(msg='eos_command does not support running '
'config mode commands. Please use '
'eos_config instead')
try:
runner.add_command(**cmd)
except AddCommandError:
exc = get_exception()
warnings.append('duplicate command detected: %s' % cmd)
try:
for item in conditionals:
runner.add_conditional(item)
except AddConditionError:
exc = get_exception()
module.fail_json(msg=str(exc), condition=exc.condition)
runner.retries = module.params['retries']
runner.interval = module.params['interval']
runner.match = module.params['match']
try:
runner.run()
except FailedConditionsError:
exc = get_exception()
module.fail_json(msg=str(exc), failed_conditions=exc.failed_conditions)
except FailedConditionalError:
exc = get_exception()
module.fail_json(msg=str(exc), failed_conditional=exc.failed_conditional)
except NetworkError:
exc = get_exception()
module.fail_json(msg=str(exc), **exc.kwargs)
result = dict(changed=False, stdout=list())
for cmd in commands:
try:
output = runner.get_command(cmd['command'], cmd.get('output'))
except ValueError:
output = 'command not executed due to check_mode, see warnings'
result['stdout'].append(output)
result = {'changed': False}
commands = parse_commands(module, warnings)
if warnings:
result['warnings'] = warnings
result['stdout_lines'] = list(to_lines(result['stdout']))
wait_for = module.params['wait_for'] or list()
conditionals = [Conditional(c) for c in wait_for]
retries = module.params['retries']
interval = module.params['interval']
match = module.params['match']
commands = [to_cli(c) for c in commands]
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': to_lines(responses)
})
module.exit_json(**result)
if __name__ == '__main__':
SHARED_LIB = 'eapi'
main()

15
test/units/modules/network/eos/fixtures/eos_command_show_version.txt Normal file
View file

@ -0,0 +1,15 @@
Arista vEOS
Hardware version:
Serial number:
System MAC address: 0050.564a.009e
Software image version: 4.15.4F
Architecture: i386
Internal build version: 4.15.4F-2923910.4154F
Internal build ID: d8a3c846-c735-4766-93cd-82bb7427da51
Uptime: 30 weeks, 3 days, 18 hours and 28 minutes
Total memory: 1897592 kB
Free memory: 63424 kB

144
test/units/modules/network/eos/test_eos_command.py Normal file
View file

@ -0,0 +1,144 @@
# (c) 2016 Red Hat 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 <http://www.gnu.org/licenses/>.
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import os
import json
from ansible.compat.tests import unittest
from ansible.compat.tests.mock import patch, MagicMock
from ansible.errors import AnsibleModuleExit
from ansible.modules.network.eos import eos_command
from ansible.module_utils import basic
from ansible.module_utils._text import to_bytes
def set_module_args(args):
args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
basic._ANSIBLE_ARGS = to_bytes(args)
fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures')
fixture_data = {}
def load_fixture(name):
path = os.path.join(fixture_path, name)
if path in fixture_data:
return fixture_data[path]
with open(path) as f:
data = f.read()
try:
data = json.loads(data)
except:
pass
fixture_data[path] = data
return data
class test_EosCommandModule(unittest.TestCase):
def setUp(self):
self.mock_run_commands = patch('ansible.modules.network.eos.eos_command.run_commands')
self.run_commands = self.mock_run_commands.start()
def tearDown(self):
self.mock_run_commands.stop()
def execute_module(self, failed=False, changed=False):
def load_from_file(*args, **kwargs):
module, commands = args
output = list()
for item in commands:
try:
obj = json.loads(item)
command = obj['command']
except ValueError:
command = item
filename = str(command).replace(' ', '_')
filename = 'eos_command_%s.txt' % filename
output.append(load_fixture(filename))
return output
self.run_commands.side_effect = load_from_file
with self.assertRaises(AnsibleModuleExit) as exc:
eos_command.main()
result = exc.exception.result
if failed:
self.assertTrue(result.get('failed'))
else:
self.assertEqual(result.get('changed'), changed, result)
return result
def test_eos_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('Arista'))
def test_eos_command_multiple(self):
set_module_args(dict(commands=['show version', 'show version']))
result = self.execute_module()
self.assertEqual(len(result['stdout']), 2)
self.assertTrue(result['stdout'][0].startswith('Arista'))
def test_eos_command_wait_for(self):
wait_for = 'result[0] contains "Arista vEOS"'
set_module_args(dict(commands=['show version'], wait_for=wait_for))
self.execute_module()
def test_eos_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_eos_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_eos_command_match_any(self):
wait_for = ['result[0] contains "Arista"',
'result[0] contains "test string"']
set_module_args(dict(commands=['show version'], wait_for=wait_for, match='any'))
self.execute_module()
def test_eos_command_match_all(self):
wait_for = ['result[0] contains "Arista"',
'result[0] contains "Software image"']
set_module_args(dict(commands=['show version'], wait_for=wait_for, match='all'))
self.execute_module()
def test_eos_command_match_all_failure(self):
wait_for = ['result[0] contains "Arista"',
'result[0] contains "test string"']
commands = ['show version', 'show version']
set_module_args(dict(commands=commands, wait_for=wait_for, match='all'))
self.execute_module(failed=True)