updates iosxr modules to support socket (#21231)

* updates all iosxr modules to support persistent socket
* adds iosxr action plugin to connect to device
* adds exec_command() to iosxr shared module
* fixes iosxr_config and iosxr_template local action
* update all unit test cases
* adds base test module for iosxr module testing
pull/21441/merge
Peter Sprygada 8 years ago committed by GitHub
parent 635e3fe9ee
commit eb1453a366

@ -26,9 +26,29 @@
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
# USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#
from ansible.module_utils.basic import env_fallback
from ansible.module_utils.network_common import to_list
from ansible.module_utils.connection import exec_command
_DEVICE_CONFIGS = {}
iosxr_argument_spec = {
'host': dict(),
'port': dict(type='int'),
'username': dict(fallback=(env_fallback, ['ANSIBLE_NET_USERNAME'])),
'password': dict(fallback=(env_fallback, ['ANSIBLE_NET_PASSWORD']), no_log=True),
'ssh_keyfile': dict(fallback=(env_fallback, ['ANSIBLE_NET_SSH_KEYFILE']), type='path'),
'timeout': dict(type='int', default=10),
'provider': dict(type='dict')
}
def check_args(module, warnings):
provider = module.params['provider'] or {}
for key in iosxr_argument_spec:
if key != 'provider' and module.params[key]:
warnings.append('argument %s has been deprecated and will be '
'removed in a future version' % key)
def get_config(module, flags=[]):
cmd = 'show running-config '
cmd += ' '.join(flags)
@ -37,7 +57,7 @@ def get_config(module, flags=[]):
try:
return _DEVICE_CONFIGS[cmd]
except KeyError:
rc, out, err = module.exec_command(cmd)
rc, out, err = exec_command(module, cmd)
if rc != 0:
module.fail_json(msg='unable to retrieve current config', stderr=err)
cfg = str(out).strip()
@ -45,38 +65,35 @@ def get_config(module, flags=[]):
return cfg
def run_commands(module, commands, check_rc=True):
assert isinstance(commands, list), 'commands must be a list'
responses = list()
for cmd in commands:
rc, out, err = module.exec_command(cmd)
for cmd in to_list(commands):
rc, out, err = exec_command(module, cmd)
if check_rc and rc != 0:
module.fail_json(msg=err, rc=rc)
responses.append(out)
return responses
def load_config(module, commands, commit=False, replace=False, comment=None):
assert isinstance(commands, list), 'commands must be a list'
rc, out, err = module.exec_command('configure terminal')
rc, out, err = exec_command(module, 'configure terminal')
if rc != 0:
module.fail_json(msg='unable to enter configuration mode', err=err)
failed = False
for command in commands:
for command in to_list(commands):
if command == 'end':
pass
rc, out, err = module.exec_command(command)
rc, out, err = exec_command(module, command)
if rc != 0:
failed = True
break
if failed:
module.exec_command('abort')
exec_command(module, 'abort')
module.fail_json(msg=err, commands=commands, rc=rc)
rc, diff, err = module.exec_command('show commit changes diff')
rc, diff, err = exec_command(module, 'show commit changes diff')
if commit:
cmd = 'commit'
if comment:
@ -84,6 +101,6 @@ def load_config(module, commands, commit=False, replace=False, comment=None):
else:
cmd = 'abort'
diff = None
module.exec_command(cmd)
exec_command(module, cmd)
return diff

@ -1,168 +0,0 @@
# This code is part of Ansible, but is an independent component.
# This particular file snippet, and this file snippet only, is BSD licensed.
# Modules you write using this snippet, which is embedded dynamically by Ansible
# still belong to the author of the module, and may assign their own license
# to the complete work.
#
# (c) 2017 Red Hat, Inc
#
# Redistribution and use in source and binary forms, with or without modification,
# are permitted provided that the following conditions are met:
#
# * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
# IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
# USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#
import re
from ansible.module_utils.shell import CliBase
from ansible.module_utils.basic import env_fallback, get_exception
from ansible.module_utils.network_common import to_list
from ansible.module_utils.netcli import Command
from ansible.module_utils.six import iteritems
from ansible.module_utils.network import NetworkError
_DEVICE_CONFIGS = {}
_DEVICE_CONNECTION = None
iosxr_cli_argument_spec = {
'host': dict(),
'port': dict(type='int'),
'username': dict(fallback=(env_fallback, ['ANSIBLE_NET_USERNAME'])),
'password': dict(fallback=(env_fallback, ['ANSIBLE_NET_PASSWORD']), no_log=True),
'authorize': dict(default=False, fallback=(env_fallback, ['ANSIBLE_NET_AUTHORIZE']), type='bool'),
'auth_pass': dict(no_log=True, fallback=(env_fallback, ['ANSIBLE_NET_AUTH_PASS'])),
'timeout': dict(type='int', default=10),
'provider': dict(type='dict'),
# deprecated in Ansible 2.3
'transport': dict(),
}
def check_args(module):
provider = module.params['provider'] or {}
for key in ('host', 'username', 'password'):
if not module.params[key] and not provider.get(key):
module.fail_json(msg='missing required argument %s' % key)
class Cli(CliBase):
CLI_PROMPTS_RE = [
re.compile(r"[\r\n]?[\w+\-\.:\/\[\]]+(?:\([^\)]+\)){,3}(?:>|#) ?$"),
re.compile(r"\[\w+\@[\w\-\.]+(?: [^\]])\] ?[>#\$] ?$")
]
CLI_ERRORS_RE = [
re.compile(r"% ?Error"),
re.compile(r"% ?Bad secret"),
re.compile(r"invalid input", re.I),
re.compile(r"(?:incomplete|ambiguous) command", re.I),
re.compile(r"connection timed out", re.I),
re.compile(r"[^\r\n]+ not found", re.I),
re.compile(r"'[^']' +returned error code: ?\d+"),
]
NET_PASSWD_RE = re.compile(r"[\r\n]?password: $", re.I)
def __init__(self, module):
self._module = module
super(Cli, self).__init__()
provider = self._module.params.get('provider') or dict()
for key, value in iteritems(provider):
if key in nxos_cli_argument_spec:
if self._module.params.get(key) is None and value is not None:
self._module.params[key] = value
try:
self.connect()
except NetworkError:
exc = get_exception()
self._module.fail_json(msg=str(exc))
def connect(self, params, **kwargs):
super(Cli, self).connect(params, kickstart=False, **kwargs)
self.shell.send(['terminal length 0', 'terminal exec prompt no-timestamp'])
def connection(module):
global _DEVICE_CONNECTION
if not _DEVICE_CONNECTION:
cli = Cli(module)
_DEVICE_CONNECTION = cli
return _DEVICE_CONNECTION
def get_config(module, flags=[]):
cmd = 'show running-config '
cmd += ' '.join(flags)
cmd = cmd.strip()
try:
return _DEVICE_CONFIGS[cmd]
except KeyError:
conn = connection(module)
rc, out, err = conn.exec_command(cmd)
if rc != 0:
module.fail_json(msg='unable to retrieve current config', stderr=err)
cfg = str(out).strip()
_DEVICE_CONFIGS[cmd] = cfg
return cfg
def run_commands(module, commands, check_rc=True):
responses = list()
for cmd in to_list(commands):
conn = connection(module)
rc, out, err = conn.exec_command(cmd)
if check_rc and rc != 0:
module.fail_json(msg=err, rc=rc)
responses.append(out)
return responses
def load_config(module, commands, commit=False, replace=False, comment=None):
rc, out, err = conn.exec_command('configure terminal')
if rc != 0:
module.fail_json(msg='unable to enter configuration mode', err=err)
failed = False
for command in to_list(commands):
if command == 'end':
pass
conn = connection(module)
rc, out, err = conn.exec_command(command)
if rc != 0:
failed = True
break
if failed:
conn.exec_command('abort')
module.fail_json(msg=err, commands=commands, rc=rc)
rc, diff, err = conn.exec_command('show commit changes diff')
if commit:
cmd = 'commit'
if comment:
cmd += ' comment {0}'.format(comment)
else:
cmd = 'abort'
diff = None
conn.exec_command(cmd)
return diff

@ -34,7 +34,6 @@ description:
commands that are not already configured. The config source can
be a set of commands or a template.
deprecated: Deprecated in 2.2. Use M(iosxr_config) instead.
extends_documentation_fragment: iosxr
options:
src:
description:
@ -99,39 +98,11 @@ updates:
returned: always
type: list
sample: ['...', '...']
start:
description: The time the job started
returned: always
type: str
sample: "2016-11-16 10:38:15.126146"
end:
description: The time the job ended
returned: always
type: str
sample: "2016-11-16 10:38:25.595612"
delta:
description: The time elapsed to perform all operations
returned: always
type: str
sample: "0:00:10.469466"
"""
from ansible.module_utils.local import LocalAnsibleModule
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.netcfg import NetworkConfig, dumps
from ansible.module_utils.iosxr import get_config, load_config
from ansible.module_utils.network import NET_TRANSPORT_ARGS, _transitional_argument_spec
def check_args(module):
warnings = list()
for key in NET_TRANSPORT_ARGS:
if module.params[key]:
warnings.append(
'network provider arguments are no longer supported. Please '
'use connection: network_cli for the task'
)
break
return warnings
from ansible.module_utils.iosxr import iosxr_argument_spec, check_args
def main():
@ -145,17 +116,16 @@ def main():
config=dict(),
)
# Removed the use of provider arguments in 2.3 due to network_cli
# connection plugin. To be removed in 2.5
argument_spec.update(_transitional_argument_spec())
argument_spec.update(iosxr_argument_spec)
mutually_exclusive = [('config', 'backup'), ('config', 'force')]
module = LocalAnsibleModule(argument_spec=argument_spec,
module = AnsibleModule(argument_spec=argument_spec,
mutually_exclusive=mutually_exclusive,
supports_check_mode=True)
warnings = check_args(module)
warnings = list()
check_args(module, warnings)
result = dict(changed=False, warnings=warnings)
@ -178,6 +148,7 @@ def main():
result['changed'] = not module.check_mode
result['updates'] = commands
result['commands'] = commands
module.exit_json(**result)

@ -126,31 +126,15 @@ failed_conditions:
returned: failed
type: list
sample: ['...', '...']
start:
description: The time the job started
returned: always
type: str
sample: "2016-11-16 10:38:15.126146"
end:
description: The time the job ended
returned: always
type: str
sample: "2016-11-16 10:38:25.595612"
delta:
description: The time elapsed to perform all operations
returned: always
type: str
sample: "0:00:10.469466"
"""
import time
from ansible.module_utils.local import LocalAnsibleModule
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.iosxr 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', 'output', 'prompt', 'response']
from ansible.module_utils.iosxr import iosxr_argument_spec, check_args
def to_lines(stdout):
for item in stdout:
@ -182,7 +166,6 @@ def parse_commands(module, warnings):
def main():
spec = dict(
# { command: <str>, output: <str>, prompt: <str>, response: <str> }
commands=dict(type='list', required=True),
wait_for=dict(type='list', aliases=['waitfor']),
@ -192,10 +175,14 @@ def main():
interval=dict(default=1, type='int')
)
module = LocalAnsibleModule(argument_spec=spec,
spec.update(iosxr_argument_spec)
module = AnsibleModule(argument_spec=spec,
supports_check_mode=True)
warnings = list()
check_args(module, warnings)
commands = parse_commands(module, warnings)
wait_for = module.params['wait_for'] or list()

@ -31,7 +31,6 @@ description:
for segmenting configuration into sections. This module provides
an implementation for working with IOS XR configuration sections in
a deterministic way.
extends_documentation_fragment: iosxr
options:
lines:
description:
@ -177,26 +176,11 @@ backup_path:
returned: when backup is yes
type: path
sample: /playbooks/ansible/backup/iosxr01.2016-07-16@22:28:34
start:
description: The time the job started
returned: always
type: str
sample: "2016-11-16 10:38:15.126146"
end:
description: The time the job ended
returned: always
type: str
sample: "2016-11-16 10:38:25.595612"
delta:
description: The time elapsed to perform all operations
returned: always
type: str
sample: "0:00:10.469466"
"""
from ansible.module_utils.local import LocalAnsibleModule
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.netcfg import NetworkConfig, dumps
from ansible.module_utils.iosxr import load_config,get_config
from ansible.module_utils.network import NET_TRANSPORT_ARGS, _transitional_argument_spec
from ansible.module_utils.iosxr import iosxr_argument_spec, check_args
DEFAULT_COMMIT_COMMENT = 'configured by iosxr_config'
@ -210,14 +194,6 @@ def check_args(module, warnings):
'match=none instead. This argument will be '
'removed in the future')
for key in NET_TRANSPORT_ARGS:
if module.params[key]:
warnings.append(
'network provider arguments are no longer supported. Please '
'use connection: network_cli for the task'
)
break
def get_running_config(module):
contents = module.params['config']
if not contents:
@ -261,7 +237,7 @@ def run(module, result):
if module.params['after']:
commands.extend(module.params['after'])
result['updates'] = commands
result['commands'] = commands
diff = load_config(module, commands, not check_mode,
replace_config, comment)
@ -293,7 +269,7 @@ def main():
comment=dict(default=DEFAULT_COMMIT_COMMENT),
)
argument_spec.update(_transitional_argument_spec())
argument_spec.update(iosxr_argument_spec)
mutually_exclusive = [('lines', 'src')]
@ -302,10 +278,10 @@ def main():
('replace', 'block', ['lines']),
('replace', 'config', ['src'])]
module = LocalAnsibleModule(argument_spec=argument_spec,
mutually_exclusive=mutually_exclusive,
required_if=required_if,
supports_check_mode=True)
module = AnsibleModule(argument_spec=argument_spec,
mutually_exclusive=mutually_exclusive,
required_if=required_if,
supports_check_mode=True)
if module.params['force'] is True:
module.params['match'] = 'none'

@ -31,7 +31,6 @@ description:
base network fact keys with C(ansible_net_<fact>). The facts
module will always collect a base set of facts from the device
and can enable or disable collection of additional facts.
extends_documentation_fragment: iosxr
options:
gather_subset:
description:
@ -122,9 +121,10 @@ ansible_net_neighbors:
import re
from ansible.module_utils.iosxr import run_commands
from ansible.module_utils.local import LocalAnsibleModule
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.six import iteritems
from ansible.module_utils.six.moves import zip
from ansible.module_utils.iosxr import iosxr_argument_spec, check_args
class FactsBase(object):
@ -359,7 +359,13 @@ def main():
gather_subset=dict(default=['!config'], type='list')
)
module = LocalAnsibleModule(argument_spec=spec, supports_check_mode=True)
spec.update(iosxr_argument_spec)
module = AnsibleModule(argument_spec=spec,
supports_check_mode=True)
warnings = list()
check_args(module, warnings)
gather_subset = module.params['gather_subset']
@ -416,7 +422,7 @@ def main():
key = 'ansible_net_%s' % key
ansible_facts[key] = value
module.exit_json(ansible_facts=ansible_facts)
module.exit_json(ansible_facts=ansible_facts, warnings=warnings)
if __name__ == '__main__':

@ -126,26 +126,12 @@ commands:
sample:
- hostname iosxr01
- ip domain-name eng.ansible.com
start:
description: The time the job started
returned: always
type: str
sample: "2016-11-16 10:38:15.126146"
end:
description: The time the job ended
returned: always
type: str
sample: "2016-11-16 10:38:25.595612"
delta:
description: The time elapsed to perform all operations
returned: always
type: str
sample: "0:00:10.469466"
"""
import re
from ansible.module_utils.local import LocalAnsibleModule
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.iosxr import get_config, load_config
from ansible.module_utils.iosxr import iosxr_argument_spec, check_args
def diff_list(want, have):
adds = set(want).difference(have)
@ -254,11 +240,15 @@ def main():
state=dict(choices=['present', 'absent'], default='present')
)
module = LocalAnsibleModule(argument_spec=argument_spec,
supports_check_mode=True)
argument_spec.update(iosxr_argument_spec)
module = AnsibleModule(argument_spec=argument_spec,
supports_check_mode=True)
warnings = list()
check_args(module, warnings)
result = {'changed': False}
result = {'changed': False, 'warnings': warnings}
want = map_params_to_obj(module)
have = map_config_to_obj(module)

@ -0,0 +1,88 @@
#
# (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/>.
#
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import os
import sys
import copy
from ansible.plugins.action.normal import ActionModule as _ActionModule
from ansible.module_utils._text import to_bytes
from ansible.utils.path import unfrackpath
from ansible.plugins import connection_loader
from ansible.compat.six import iteritems
from ansible.module_utils.iosxr import iosxr_argument_spec
from ansible.module_utils.basic import AnsibleFallbackNotFound
class ActionModule(_ActionModule):
def run(self, tmp=None, task_vars=None):
provider = self.load_provider()
pc = copy.deepcopy(self._play_context)
pc.connection = 'network_cli'
pc.network_os = 'iosxr'
pc.port = provider['port'] or self._play_context.port or 22
pc.remote_user = provider['username'] or self._play_context.connection_user
pc.password = provider['password'] or self._play_context.password
socket_path = self._get_socket_path(pc)
if not os.path.exists(socket_path):
# start the connection if it isn't started
connection = self._shared_loader_obj.connection_loader.get('persistent', pc, sys.stdin)
version = connection.exec_command('EXEC: show version')
task_vars['ansible_socket'] = socket_path
return super(ActionModule, self).run(tmp, task_vars)
def _get_socket_path(self, play_context):
ssh = connection_loader.get('ssh', class_only=True)
cp = ssh._create_control_path(play_context.remote_addr, play_context.port, play_context.remote_user)
path = unfrackpath("$HOME/.ansible/pc")
return cp % dict(directory=path)
def load_provider(self):
provider = self._task.args.get('provider', {})
for key, value in iteritems(iosxr_argument_spec):
if key != 'provider' and key not in provider:
if key in self._task.args:
provider[key] = self._task.args[key]
elif 'fallback' in value:
provider[key] = self._fallback(value['fallback'])
elif key not in provider:
provider[key] = None
return provider
def _fallback(self, fallback):
strategy = fallback[0]
args = []
kwargs = {}
for item in fallback[1:]:
if isinstance(item, dict):
kwargs = item
else:
args = item
try:
return strategy(*args, **kwargs)
except AnsibleFallbackNotFound:
pass

@ -1,5 +1,5 @@
#
# Copyright 2015 Peter Sprygada <psprygada@ansible.com>
# (c) 2017, Red Hat, Inc.
#
# This file is part of Ansible
#
@ -19,10 +19,94 @@
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from ansible.plugins.action import ActionBase
from ansible.plugins.action.net_config import ActionModule as NetActionModule
import os
import re
import time
import glob
class ActionModule(NetActionModule, ActionBase):
pass
from ansible.plugins.action.iosxr import ActionModule as _ActionModule
from ansible.module_utils._text import to_text
from ansible.module_utils.six.moves.urllib.parse import urlsplit
from ansible.utils.vars import merge_hash
PRIVATE_KEYS_RE = re.compile('__.+__')
class ActionModule(_ActionModule):
def run(self, tmp=None, task_vars=None):
if self._task.args.get('src'):
try:
self._handle_template()
except ValueError as exc:
return dict(failed=True, msg=exc.message)
result = super(ActionModule, self).run(tmp, task_vars)
if self._task.args.get('backup') and result.get('__backup__'):
# User requested backup and no error occurred in module.
# NOTE: If there is a parameter error, _backup key may not be in results.
filepath = self._write_backup(task_vars['inventory_hostname'],
result['__backup__'])
result['backup_path'] = filepath
# strip out any keys that have two leading and two trailing
# underscore characters
for key in result.keys():
if PRIVATE_KEYS_RE.match(key):
del result[key]
return result
def _get_working_path(self):
cwd = self._loader.get_basedir()
if self._task._role is not None:
cwd = self._task._role._role_path
return cwd
def _write_backup(self, host, contents):
backup_path = self._get_working_path() + '/backup'
if not os.path.exists(backup_path):
os.mkdir(backup_path)
for fn in glob.glob('%s/%s*' % (backup_path, host)):
os.remove(fn)
tstamp = time.strftime("%Y-%m-%d@%H:%M:%S", time.localtime(time.time()))
filename = '%s/%s_config.%s' % (backup_path, host, tstamp)
open(filename, 'w').write(contents)
return filename
def _handle_template(self):
src = self._task.args.get('src')
working_path = self._get_working_path()
if os.path.isabs(src) or urlsplit('src').scheme:
source = src
else:
source = self._loader.path_dwim_relative(working_path, 'templates', src)
if not source:
source = self._loader.path_dwim_relative(working_path, src)
if not os.path.exists(source):
raise ValueError('path specified in src not found')
try:
with open(source, 'r') as f:
template_data = to_text(f.read())
except IOError:
return dict(failed=True, msg='unable to load src file')
# Create a template search path in the following order:
# [working_path, self_role_path, dependent_role_paths, dirname(source)]
searchpath = [working_path]
if self._task._role is not None:
searchpath.append(self._task._role._role_path)
if hasattr(self._task, "_block:"):
dep_chain = self._task._block.get_dep_chain()
if dep_chain is not None:
for role in dep_chain:
searchpath.append(role._role_path)
searchpath.append(os.path.dirname(source))
self._templar.environment.loader.searchpath = searchpath
self._task.args['src'] = self._templar.template(template_data)

@ -19,10 +19,85 @@
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from ansible.plugins.action import ActionBase
from ansible.plugins.action.net_template import ActionModule as NetActionModule
import os
import time
import glob
import urlparse
class ActionModule(NetActionModule, ActionBase):
pass
from ansible.module_utils._text import to_text
from ansible.plugins.action.iosxr import ActionModule as _ActionModule
class ActionModule(_ActionModule):
def run(self, tmp=None, task_vars=None):
try:
self._handle_template()
except (ValueError, AttributeError) as exc:
return dict(failed=True, msg=exc.message)
result = super(ActionModule, self).run(tmp, task_vars)
if self._task.args.get('backup') and result.get('__backup__'):
# User requested backup and no error occurred in module.
# NOTE: If there is a parameter error, __backup__ key may not be in results.
self._write_backup(task_vars['inventory_hostname'], result['__backup__'])
if '__backup__' in result:
del result['__backup__']
return result
def _get_working_path(self):
cwd = self._loader.get_basedir()
if self._task._role is not None:
cwd = self._task._role._role_path
return cwd
def _write_backup(self, host, contents):
backup_path = self._get_working_path() + '/backup'
if not os.path.exists(backup_path):
os.mkdir(backup_path)
for fn in glob.glob('%s/%s*' % (backup_path, host)):
os.remove(fn)
tstamp = time.strftime("%Y-%m-%d@%H:%M:%S", time.localtime(time.time()))
filename = '%s/%s_config.%s' % (backup_path, host, tstamp)
open(filename, 'w').write(contents)
def _handle_template(self):
src = self._task.args.get('src')
if not src:
raise ValueError('missing required arguments: src')
working_path = self._get_working_path()
if os.path.isabs(src) or urlparse.urlsplit(src).scheme:
source = src
else:
source = self._loader.path_dwim_relative(working_path, 'templates', src)
if not source:
source = self._loader.path_dwim_relative(working_path, src)
if not os.path.exists(source):
return
try:
with open(source, 'r') as f:
template_data = to_text(f.read())
except IOError:
return dict(failed=True, msg='unable to load src file')
# Create a template search path in the following order:
# [working_path, self_role_path, dependent_role_paths, dirname(source)]
searchpath = [working_path]
if self._task._role is not None:
searchpath.append(self._task._role._role_path)
if hasattr(self._task, "_block:"):
dep_chain = self._task._block.get_dep_chain()
if dep_chain is not None:
for role in dep_chain:
searchpath.append(role._role_path)
searchpath.append(os.path.dirname(source))
self._templar.environment.loader.searchpath = searchpath
self._task.args['src'] = self._templar.template(template_data)

@ -1,74 +0,0 @@
#
# (c) 2015, Peter Sprygada <psprygada@ansible.com>
#
# 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/>.
class ModuleDocFragment(object):
# Standard files documentation fragment
DOCUMENTATION = """
options:
host:
description:
- Specifies the DNS host name or address for connecting to the remote
device over the specified transport. The value of host is used as
the destination address for the transport.
required: true
port:
description:
- Specifies the port to use when building the connection to the remote.
device.
required: false
default: 22
username:
description:
- Configures the username to use to authenticate the connection to
the remote device. This value is used to authenticate
the SSH session. If the value is not specified in the task, the
value of environment variable C(ANSIBLE_NET_USERNAME) will be used instead.
required: false
password:
description:
- Specifies the password to use to authenticate the connection to
the remote device. This value is used to authenticate
the SSH session. If the value is not specified in the task, the
value of environment variable C(ANSIBLE_NET_PASSWORD) will be used instead.
required: false
default: null
timeout:
description:
- Specifies the timeout in seconds for communicating with the network device
for either connecting or sending commands. If the timeout is
exceeded before the operation is completed, the module will error.
require: false
default: 10
ssh_keyfile:
description:
- Specifies the SSH key to use to authenticate the connection to
the remote device. This value is the path to the
key used to authenticate the SSH session. If the value is not specified
in the task, the value of environment variable C(ANSIBLE_NET_SSH_KEYFILE)
will be used instead.
required: false
provider:
description:
- Convenience method that allows all I(iosxr) arguments to be passed as
a dict object. All constraints (required, choices, etc) must be
met either by individual arguments or values in this dict.
required: false
default: null
"""

@ -0,0 +1,113 @@
# (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
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 AnsibleExitJson(Exception):
pass
class AnsibleFailJson(Exception):
pass
class TestIosxrModule(unittest.TestCase):
def execute_module(self, failed=False, changed=False, commands=None,
sort=True, defaults=False):
self.load_fixtures(commands)
if failed:
result = self.failed()
self.assertTrue(result['failed'], result)
else:
result = self.changed(changed)
self.assertEqual(result['changed'], changed, result)
if commands:
if sort:
self.assertEqual(sorted(commands), sorted(result['commands']), result['commands'])
else:
self.assertEqual(commands, result['commands'], result['commands'])
return result
def failed(self):
def fail_json(*args, **kwargs):
kwargs['failed'] = True
raise AnsibleFailJson(kwargs)
with patch.object(basic.AnsibleModule, 'fail_json', fail_json):
with self.assertRaises(AnsibleFailJson) as exc:
self.module.main()
result = exc.exception.args[0]
self.assertTrue(result['failed'], result)
return result
def changed(self, changed=False):
def exit_json(*args, **kwargs):
if 'changed' not in kwargs:
kwargs['changed'] = False
raise AnsibleExitJson(kwargs)
with patch.object(basic.AnsibleModule, 'exit_json', exit_json):
with self.assertRaises(AnsibleExitJson) as exc:
self.module.main()
result = exc.exception.args[0]
self.assertEqual(result['changed'], changed, result)
return result
def load_fixtures(self, commands=None):
pass

@ -19,43 +19,16 @@
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.compat.tests.mock import patch
from ansible.modules.network.iosxr import iosxr_command
from ansible.module_utils import basic
from ansible.module_utils._text import to_bytes
from .iosxr_module import TestIosxrModule, load_fixture, set_module_args
def set_module_args(args):
args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
basic._ANSIBLE_ARGS = to_bytes(args)
class TestIosxrCommandModule(TestIosxrModule):
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 TestIosxrCommandModule(unittest.TestCase):
module = iosxr_command
def setUp(self):
self.mock_run_commands = patch('ansible.modules.network.iosxr.iosxr_command.run_commands')
@ -64,7 +37,7 @@ class TestIosxrCommandModule(unittest.TestCase):
def tearDown(self):
self.mock_run_commands.stop()
def execute_module(self, failed=False, changed=False):
def load_fixtures(self, commands=None):
def load_from_file(*args, **kwargs):
module, commands = args
@ -82,18 +55,6 @@ class TestIosxrCommandModule(unittest.TestCase):
self.run_commands.side_effect = load_from_file
with self.assertRaises(AnsibleModuleExit) as exc:
iosxr_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_iosxr_command_simple(self):
set_module_args(dict(commands=['show version']))
result = self.execute_module()

@ -20,78 +20,30 @@
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.compat.tests.mock import patch
from ansible.modules.network.iosxr import iosxr_config
from ansible.module_utils import basic
from ansible.module_utils._text import to_bytes
from .iosxr_module import TestIosxrModule, load_fixture, set_module_args
class TestIosxrConfigModule(TestIosxrModule):
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 TestIosxrConfigModule(unittest.TestCase):
module = iosxr_config
def setUp(self):
self.patcher_get_config = patch('ansible.modules.network.iosxr.iosxr_config.get_config')
self.mock_get_config = self.patcher_get_config.start()
self.patcher_exec_command = patch('ansible.modules.network.iosxr.iosxr_config.LocalAnsibleModule.exec_command')
self.patcher_exec_command = patch('ansible.modules.network.iosxr.iosxr_config.load_config')
self.mock_exec_command = self.patcher_exec_command.start()
def tearDown(self):
self.patcher_get_config.stop()
self.patcher_exec_command.stop()
def execute_module(self, failed=False, changed=False, commands=None,
sort=True):
def load_fixtures(self, commands=None):
config_file = 'iosxr_config_config.cfg'
self.mock_get_config.return_value = load_fixture(config_file)
self.mock_exec_command.return_value = (0, 'dummy diff', None)
with self.assertRaises(AnsibleModuleExit) as exc:
iosxr_config.main()
result = exc.exception.result
if failed:
self.assertTrue(result['failed'], result)
else:
self.assertEqual(result.get('changed'), changed, result)
if commands:
if sort:
self.assertEqual(sorted(commands), sorted(result['updates']), result['updates'])
else:
self.assertEqual(commands, result['updates'], result['updates'])
return result
self.mock_exec_command.return_value = 'dummy diff'
def test_iosxr_config_unchanged(self):
src = load_fixture('iosxr_config_config.cfg')

@ -19,45 +19,16 @@
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
from ansible.errors import AnsibleModuleExit
from .iosxr_module import TestIosxrModule, load_fixture, set_module_args
from ansible.modules.network.iosxr import iosxr_facts
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)
class TestIosxrFacts(TestIosxrModule):
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 TestIosxrFacts(unittest.TestCase):
module = iosxr_facts
def setUp(self):
self.mock_run_commands = patch(
@ -67,7 +38,7 @@ class TestIosxrFacts(unittest.TestCase):
def tearDown(self):
self.mock_run_commands.stop()
def execute_module(self, failed=False, changed=False):
def load_fixtures(self, commands=None):
def load_from_file(*args, **kwargs):
module, commands = args
@ -87,18 +58,6 @@ class TestIosxrFacts(unittest.TestCase):
self.run_commands.side_effect = load_from_file
with self.assertRaises(AnsibleModuleExit) as exc:
iosxr_facts.main()
result = exc.exception.result
if failed:
self.assertTrue(result.get('failed'))
else:
self.assertEqual(result.get('changed'), changed, result)
return result
def test_iosxr_facts_gather_subset_default(self):
set_module_args(dict())
result = self.execute_module()

@ -21,44 +21,16 @@
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import os
import json
import ansible.module_utils.basic
from ansible.compat.tests import unittest
from ansible.compat.tests.mock import patch, MagicMock
from ansible.errors import AnsibleModuleExit
from ansible.compat.tests.mock import patch
from .iosxr_module import TestIosxrModule, load_fixture, set_module_args
from ansible.modules.network.iosxr import iosxr_system
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 TestIosxrSystemModule(TestIosxrModule):
class TestIosxrSystemModule(unittest.TestCase):
module = iosxr_system
def setUp(self):
self.mock_get_config = patch('ansible.modules.network.iosxr.iosxr_system.get_config')
@ -71,29 +43,10 @@ class TestIosxrSystemModule(unittest.TestCase):
self.mock_get_config.stop()
self.mock_load_config.stop()
def execute_module(self, failed=False, changed=False, commands=None, sort=True):
def load_fixtures(self, commands=None):
self.get_config.return_value = load_fixture('iosxr_system_config.cfg')
self.load_config.return_value = dict(diff=None, session='session')
with self.assertRaises(AnsibleModuleExit) as exc:
iosxr_system.main()
result = exc.exception.result
if failed:
self.assertTrue(result['failed'], result)
else:
self.assertEqual(result['changed'], changed, result)
if commands:
if sort:
self.assertEqual(sorted(commands), sorted(result['commands']), result['commands'])
else:
self.assertEqual(commands, result['commands'])
return result
def test_iosxr_system_hostname_changed(self):
set_module_args(dict(hostname='foo'))
commands = ['hostname foo']

@ -20,43 +20,16 @@
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.compat.tests.mock import patch
from .iosxr_module import TestIosxrModule, load_fixture, set_module_args
from ansible.modules.network.iosxr import _iosxr_template
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)
class TestIosxrTemplateModule(TestIosxrModule):
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 TestIosxrTemplateModule(unittest.TestCase):
module = _iosxr_template
def setUp(self):
self.mock_get_config = patch('ansible.modules.network.iosxr._iosxr_template.get_config')
@ -69,31 +42,11 @@ class TestIosxrTemplateModule(unittest.TestCase):
self.mock_get_config.stop()
self.mock_load_config.stop()
def execute_module(self, failed=False, changed=False, commands=None,
sort=True):
def load_fixtures(self, commands=None):
config_file = 'iosxr_template_config.cfg'
self.get_config.return_value = load_fixture(config_file)
self.load_config.return_value = None
with self.assertRaises(AnsibleModuleExit) as exc:
_iosxr_template.main()
result = exc.exception.result
if failed:
self.assertTrue(result['failed'], result)
else:
self.assertEqual(result.get('changed'), changed, result)
if commands:
if sort:
self.assertEqual(sorted(commands), sorted(result['updates']), result['updates'])
else:
self.assertEqual(commands, result['updates'], result['updates'])
return result
def test_iosxr_template_unchanged(self):
src = load_fixture('iosxr_template_config.cfg')
set_module_args(dict(src=src))

Loading…
Cancel
Save