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

updates ios_command to use network_cli plugin (#19992)

Browse source 
* refactors ios_command to use network_cli
* adds unit test cases for ios_command
This commit is contained in:
Peter Sprygada 2017-01-06 17:06:40 -05:00 committed by GitHub
parent c980b52d33
commit 7d3366acc0
3 changed files with 252 additions and 61 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

125
lib/ansible/modules/network/ios/ios_command.py
View file

@ -16,9 +16,11 @@
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
#
ANSIBLE_METADATA = {'status': ['preview'],
'supported_by': 'core',
'version': '1.0'}
ANSIBLE_METADATA = {
'status': ['preview'],
'supported_by': 'core',
'version': '1.0'
}
DOCUMENTATION = """
---
@ -143,14 +145,15 @@ failed_conditions:
type: list
sample: ['...', '...']
"""
import ansible.module_utils.ios
from ansible.module_utils.basic import get_exception
from ansible.module_utils.netcli import CommandRunner
from ansible.module_utils.netcli import AddCommandError, FailedConditionsError
from ansible.module_utils.network import NetworkModule, NetworkError
import time
from ansible.module_utils.local import LocalAnsibleModule
from ansible.module_utils.ios import run_commands
from ansible.module_utils.network_common import ComplexList
from ansible.module_utils.netcli import Conditional
from ansible.module_utils.six import string_types
VALID_KEYS = ['command', 'prompt', 'response']
VALID_KEYS = ['command', 'output']
def to_lines(stdout):
for item in stdout:
@ -158,15 +161,27 @@ def to_lines(stdout):
item = str(item).split('\n')
yield item
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 not set(cmd.keys()).issubset(VALID_KEYS):
module.fail_json(msg='unknown keyword specified')
yield cmd
def parse_commands(module, warnings):
command = ComplexList(dict(
command=dict(key=True),
prompt=dict(),
response=dict()
))
commands = command(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']
)
elif item['command'].startswith('conf'):
module.fail_json(
msg='ios_command does not support running config mode '
'commands. Please use ios_config instead'
)
commands[index] = module.jsonify(item)
return commands
def main():
spec = dict(
@ -180,59 +195,47 @@ def main():
interval=dict(default=1, type='int')
)
module = NetworkModule(argument_spec=spec,
connect_on_load=False,
supports_check_mode=True)
commands = list(parse_commands(module))
conditionals = module.params['wait_for'] or list()
module = LocalAnsibleModule(argument_spec=spec,
supports_check_mode=True)
warnings = list()
commands = parse_commands(module, warnings)
runner = CommandRunner(module)
wait_for = module.params['wait_for'] or list()
conditionals = [Conditional(c) for c in wait_for]
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='ios_command does not support running '
'config mode commands. Please use '
'ios_config instead')
try:
runner.add_command(**cmd)
except AddCommandError:
exc = get_exception()
warnings.append('duplicate command detected: %s' % cmd)
retries = module.params['retries']
interval = module.params['interval']
match = module.params['match']
for item in conditionals:
runner.add_conditional(item)
while retries > 0:
responses = run_commands(module, commands)
runner.retries = module.params['retries']
runner.interval = module.params['interval']
runner.match = module.params['match']
for item in list(conditionals):
if item(responses):
if match == 'any':
conditionals = list()
break
conditionals.remove(item)
try:
runner.run()
except FailedConditionsError:
exc = get_exception()
module.fail_json(msg=str(exc), failed_conditions=exc.failed_conditions)
except NetworkError:
exc = get_exception()
module.fail_json(msg=str(exc))
if not conditionals:
break
result = dict(changed=False, stdout=list())
time.sleep(interval)
retries -= 1
for cmd in commands:
try:
output = runner.get_command(cmd['command'])
except ValueError:
output = 'command not executed due to check_mode, see warnings'
result['stdout'].append(output)
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['warnings'] = warnings
result['stdout_lines'] = list(to_lines(result['stdout']))
result = {
'changed': False,
'stdout': responses,
'warnings': warnings,
'stdout_lines': list(to_lines(responses))
}
module.exit_json(**result)

45
test/units/modules/network/ios/fixtures/show_version Normal file
View file

@ -0,0 +1,45 @@
Cisco IOS Software, IOSv Software (VIOS-ADVENTERPRISEK9-M), Version 15.6(1)T, RELEASE SOFTWARE (fc1)
Technical Support: http://www.cisco.com/techsupport
Copyright (c) 1986-2015 by Cisco Systems, Inc.
Compiled Fri 20-Nov-15 13:39 by prod_rel_team
ROM: Bootstrap program is IOSv
ios01 uptime is 7 weeks, 5 days, 11 hours, 14 minutes
System returned to ROM by reload
System image file is "flash0:/vios-adventerprisek9-m"
Last reload reason: Unknown reason
This product contains cryptographic features and is subject to United
States and local country laws governing import, export, transfer and
use. Delivery of Cisco cryptographic products does not imply
third-party authority to import, export, distribute or use encryption.
Importers, exporters, distributors and users are responsible for
compliance with U.S. and local country laws. By using this product you
agree to comply with applicable laws and regulations. If you are unable
to comply with U.S. and local laws, return this product immediately.
A summary of U.S. laws governing Cisco cryptographic products may be found at:
http://www.cisco.com/wwl/export/crypto/tool/stqrg.html
If you require further assistance please contact us by sending email to
export@cisco.com.
Cisco IOSv (revision 1.0) with with 472441K/50176K bytes of memory.
Processor board ID 99I10YFMUCJ3JEZMV4DQB
3 Gigabit Ethernet interfaces
DRAM configuration is 72 bits wide with parity disabled.
256K bytes of non-volatile configuration memory.
2097152K bytes of ATA System CompactFlash 0 (Read/Write)
0K bytes of ATA CompactFlash 1 (Read/Write)
0K bytes of ATA CompactFlash 2 (Read/Write)
10080K bytes of ATA CompactFlash 3 (Read/Write)
Configuration register is 0x0

143
test/units/modules/network/ios/test_ios_command.py Normal file
View file

@ -0,0 +1,143 @@
# (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.ios import ios_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_iosCommandModule(unittest.TestCase):
def setUp(self):
self.mock_run_commands = patch('ansible.modules.network.ios.ios_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(' ', '_')
output.append(load_fixture(filename))
return output
self.run_commands.side_effect = load_from_file
with self.assertRaises(AnsibleModuleExit) as exc:
ios_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_ios_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('Cisco IOS Software'))
def test_ios_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('Cisco IOS Software'))
def test_ios_command_wait_for(self):
wait_for = 'result[0] contains "Cisco IOS"'
set_module_args(dict(commands=['show version'], wait_for=wait_for))
self.execute_module()
def test_ios_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_ios_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_ios_command_match_any(self):
wait_for = ['result[0] contains "Cisco IOS"',
'result[0] contains "test string"']
set_module_args(dict(commands=['show version'], wait_for=wait_for, match='any'))
self.execute_module()
def test_ios_command_match_all(self):
wait_for = ['result[0] contains "Cisco IOS"',
'result[0] contains "IOSv Software"']
set_module_args(dict(commands=['show version'], wait_for=wait_for, match='any'))
self.execute_module()
def test_ios_command_match_all_failure(self):
wait_for = ['result[0] contains "Cisco IOS"',
'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)