updates eos modules to use socket (#21197)

* updates eos modules to use persistent connection socket
* removes split eos shared module and combines into one
* adds singular eos doc frag (eos_local to be removed after module updates)
* updates unit test cases
pull/21387/head
Peter Sprygada 8 years ago committed by GitHub
parent 9937e604f5
commit 14b942f3fb

@ -1,10 +1,12 @@
#
# 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) 2016 Red Hat Inc.
# (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:
@ -25,130 +27,393 @@
# 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 os
import re
import time
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
from ansible.module_utils.urls import fetch_url
from ansible.module_utils.connection import exec_command
_DEVICE_CONFIGS = {}
_DEVICE_CONNECTION = None
def get_config(module, flags=[]):
cmd = 'show running-config '
cmd += ' '.join(flags)
cmd = cmd.strip()
try:
return _DEVICE_CONFIGS[cmd]
except KeyError:
rc, out, err = module.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
eos_argument_spec = {
'host': dict(),
'port': dict(type='int'),
def check_authorization(module):
for cmd in ['show clock', 'prompt()']:
rc, out, err = module.exec_command(cmd)
return out.endswith('#')
'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'),
def supports_sessions(module):
rc, out, err = module.exec_command('show configuration sessions')
return rc == 0
def run_commands(module, commands):
"""Run list of commands on remote device and return results
"""
responses = list()
'authorize': dict(fallback=(env_fallback, ['ANSIBLE_NET_AUTHORIZE']), type='bool'),
'auth_pass': dict(no_log=True, fallback=(env_fallback, ['ANSIBLE_NET_AUTH_PASS'])),
'use_ssl': dict(type='bool'),
'validate_certs': dict(type='bool'),
'timeout': dict(type='int'),
'provider': dict(type='dict'),
'transport': dict(choices=['cli', 'eapi'])
}
def check_args(module, warnings):
provider = module.params['provider'] or {}
for key in eos_argument_spec:
if key != ['provider', 'transport'] and module.params[key]:
warnings.append('argument %s has been deprecated and will be '
'removed in a future version' % key)
def load_params(module):
provider = module.params.get('provider') or dict()
for key, value in iteritems(provider):
if key in eos_argument_spec:
if module.params.get(key) is None and value is not None:
module.params[key] = value
def get_connection(module):
global _DEVICE_CONNECTION
if not _DEVICE_CONNECTION:
load_params(module)
if 'transport' not in module.params:
conn = Cli(module)
elif module.params['transport'] == 'eapi':
conn = Eapi(module)
else:
conn = Cli(module)
_DEVICE_CONNECTION = conn
return _DEVICE_CONNECTION
class Cli:
def __init__(self, module):
self._module = module
self._device_configs = {}
def exec_command(self, command):
if isinstance(command, dict):
command = self._module.jsonify(command)
return exec_command(self._module, command)
def check_authorization(self):
for cmd in ['show clock', 'prompt()']:
rc, out, err = self.exec_command(cmd)
return out.endswith('#')
def supports_sessions(self):
rc, out, err = self.exec_command('show configuration sessions')
return rc == 0
def get_config(self, flags=[]):
"""Retrieves the current config from the device or cache
"""
cmd = 'show running-config '
cmd += ' '.join(flags)
cmd = cmd.strip()
try:
return self._device_configs[cmd]
except KeyError:
conn = get_connection(self)
rc, out, err = self.exec_command(cmd)
if rc != 0:
self._module.fail_json(msg=err)
cfg = str(out).strip()
self._device_configs[cmd] = cfg
return cfg
for cmd in to_list(commands):
cmd = module.jsonify(cmd)
rc, out, err = module.exec_command(cmd)
def run_commands(self, commands, check_rc=True):
"""Run list of commands on remote device and return results
"""
responses = list()
for cmd in to_list(commands):
rc, out, err = self.exec_command(cmd)
if check_rc and rc != 0:
self._module.fail_json(msg=err)
try:
out = self._module.from_json(out)
except ValueError:
out = str(out).strip()
responses.append(out)
return responses
def send_config(self, commands):
multiline = False
for command in to_list(commands):
if command == 'end':
pass
if command.startswith('banner') or multiline:
multiline = True
command = self._module.jsonify({'command': command, 'sendonly': True})
elif command == 'EOF' and multiline:
multiline = False
rc, out, err = self.exec_command(command)
if rc != 0:
return (rc, out, err)
return (rc, 'ok','')
def configure(self, commands):
"""Sends configuration commands to the remote device
"""
if not check_authorization(self):
self._module.fail_json(msg='configuration operations require privilege escalation')
conn = get_connection(self)
rc, out, err = self.exec_command('configure')
if rc != 0:
module.fail_json(msg=err)
self._module.fail_json(msg='unable to enter configuration mode', output=err)
rc, out, err = send_config(self, commands)
if rc != 0:
self._module.fail_json(msg=err)
self.exec_command('end')
return {}
def load_config(self, commands, commit=False, replace=False):
"""Loads the config commands onto the remote device
"""
if not check_authorization(self):
self._module.fail_json(msg='configuration operations require privilege escalation')
use_session = os.getenv('ANSIBLE_EOS_USE_SESSIONS', True)
try:
out = module.from_json(out)
use_session = int(use_session)
except ValueError:
out = str(out).strip()
pass
responses.append(out)
return responses
if not all((bool(use_session), supports_sessions(self))):
return configure(self, commands)
def send_config(module, commands):
multiline = False
for command in to_list(commands):
if command == 'end':
pass
conn = get_connection(self)
session = 'ansible_%s' % int(time.time())
result = {'session': session}
rc, out, err = self.exec_command('configure session %s' % session)
if rc != 0:
self._module.fail_json(msg='unable to enter configuration mode', output=err)
if command.startswith('banner') or multiline:
multiline = True
command = module.jsonify({'command': command, 'sendonly': True})
elif command == 'EOF' and multiline:
multiline = False
if replace:
self.exec_command('rollback clean-config', check_rc=True)
rc, out, err = module.exec_command(command)
rc, out, err = send_config(self, commands)
if rc != 0:
return (rc, out, err)
return (rc, 'ok','')
self.exec_command('abort')
conn.fail_json(msg=err, commands=commands)
rc, out, err = self.exec_command('show session-config diffs')
if rc == 0:
result['diff'] = out.strip()
if commit:
self.exec_command('commit')
else:
self.exec_command('abort')
def configure(module, commands):
"""Sends configuration commands to the remote device
"""
if not check_authorization(module):
module.fail_json(msg='configuration operations require privilege escalation')
return result
rc, out, err = module.exec_command('configure')
if rc != 0:
module.fail_json(msg='unable to enter configuration mode', output=err)
class Eapi:
rc, out, err = send_config(module, commands)
if rc != 0:
module.fail_json(msg=err)
def __init__(self, module):
self._module = module
self._enable = None
self._session_support = None
self._device_config = {}
module.exec_command('end')
return {}
host = module.params['host']
port = module.params['port']
def load_config(module, commands, commit=False, replace=False):
"""Loads the config commands onto the remote device
"""
if not check_authorization(module):
module.fail_json(msg='configuration operations require privilege escalation')
self._module.params['url_username'] = self._module.params['username']
self._module.params['url_password'] = self._module.params['password']
use_session = os.getenv('ANSIBLE_EOS_USE_SESSIONS', True)
try:
use_session = int(use_session)
except ValueError:
pass
if module.params['use_ssl']:
proto = 'https'
if not port:
port = 443
else:
proto = 'http'
if not port:
port = 80
if not all((bool(use_session), supports_sessions(module))):
return configure(module, commands)
self._url = '%s://%s:%s/command-api' % (proto, host, port)
session = 'ansible_%s' % int(time.time())
if module.params['auth_pass']:
self._enable = {'cmd': 'enable', 'input': module.params['auth_pass']}
else:
self._enable = 'enable'
result = {'session': session}
def _request_builder(self, commands, output, reqid=None):
params = dict(version=1, cmds=commands, format=output)
return dict(jsonrpc='2.0', id=reqid, method='runCmds', params=params)
rc, out, err = module.exec_command('configure session %s' % session)
if rc != 0:
module.fail_json(msg='unable to enter configuration mode', output=err)
def send_request(self, commands, output='text'):
commands = to_list(commands)
if replace:
module.exec_command('rollback clean-config', check_rc=True)
if self._enable:
commands.insert(0, 'enable')
rc, out, err = send_config(module, commands)
if rc != 0:
module.exec_command('abort')
module.fail_json(msg=err, commands=commands)
body = self._request_builder(commands, output)
data = self._module.jsonify(body)
rc, out, err = module.exec_command('show session-config diffs')
if rc == 0:
result['diff'] = out.strip()
headers = {'Content-Type': 'application/json-rpc'}
timeout = self._module.params['timeout']
response, headers = fetch_url(
self._module, self._url, data=data, headers=headers,
method='POST', timeout=timeout
)
if headers['status'] != 200:
self._module.fail_json(**headers)
try:
data = response.read()
response = self._module.from_json(data)
except ValueError:
self._module.fail_json(msg='unable to load response from device', data=data)
if self._enable and 'result' in response:
response['result'].pop(0)
return response
def run_commands(self, commands):
"""Runs list of commands on remote device and returns results
"""
output = None
queue = list()
responses = list()
def _send(commands, output):
response = self.send_request(commands, output=output)
if 'error' in response:
err = response['error']
self._module.fail_json(msg=err['message'], code=err['code'])
return response['result']
for item in to_list(commands):
if item['output'] == 'json' and not is_json(item['command']):
item['command'] = '%s | json' % item['command']
if item['output'] == 'text' and is_json(item['command']):
item['command'] = str(item['command']).split('|')[0]
if all((output == 'json', is_text(item['command']))) or all((output =='text', is_json(item['command']))):
responses.extend(_send(queue, output))
queue = list()
output = item['output'] or 'json'
queue.append(item['command'])
if queue:
responses.extend(_send(queue, output))
for index, item in enumerate(commands):
try:
responses[index] = responses[index]['output'].strip()
except KeyError:
pass
return responses
def get_config(self, flags=[]):
"""Retrieves the current config from the device or cache
"""
cmd = 'show running-config '
cmd += ' '.join(flags)
cmd = cmd.strip()
try:
return self._device_configs[cmd]
except KeyError:
out = self.send_request(cmd)
cfg = str(out['result'][0]['output']).strip()
self._device_configs[cmd] = cfg
return cfg
def supports_sessions(self):
if self._session_support:
return self._session_support
response = self.send_request(['show configuration sessions'])
self._session_support = 'error' not in response
return self._session_support
def configure(self, commands):
"""Sends the ordered set of commands to the device
"""
cmds = ['configure terminal']
cmds.extend(commands)
responses = self.send_request(commands)
if 'error' in response:
err = response['error']
self._module.fail_json(msg=err['message'], code=err['code'])
return responses[1:]
def load_config(self, config, commit=False, replace=False):
"""Loads the configuration onto the remote devices
If the device doesn't support configuration sessions, this will
fallback to using configure() to load the commands. If that happens,
there will be no returned diff or session values
"""
if not supports_sessions():
return configure(self, commands)
session = 'ansible_%s' % int(time.time())
result = {'session': session}
commands = ['configure session %s' % session]
if replace:
commands.append('rollback clean-config')
commands.extend(config)
response = self.send_request(commands)
if 'error' in response:
commands = ['configure session %s' % session, 'abort']
self.send_request(commands)
err = response['error']
self._module.fail_json(msg=err['message'], code=err['code'])
commands = ['configure session %s' % session, 'show session-config diffs']
if commit:
commands.append('commit')
else:
commands.append('abort')
response = self.send_request(commands, output='text')
diff = response['result'][1]['output']
if diff:
result['diff'] = diff
return result
is_json = lambda x: str(x).endswith('| json')
is_text = lambda x: not str(x).endswith('| json')
def get_config(module, flags=[]):
conn = get_connection(module)
return conn.get_config(flags)
def run_commands(module, commands):
conn = get_connection(module)
return conn.run_commands(commands)
if commit:
module.exec_command('commit')
else:
module.exec_command('abort')
def load_config(module, config, commit=False, replace=False):
conn = get_connection(module)
return conn.load_config(config, commit, replace)
return result

@ -1,449 +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
import time
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
from ansible.module_utils.urls import fetch_url
_DEVICE_CONNECTION = None
eos_local_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(fallback=(env_fallback, ['ANSIBLE_NET_AUTHORIZE']), type='bool'),
'auth_pass': dict(no_log=True, fallback=(env_fallback, ['ANSIBLE_NET_AUTH_PASS'])),
'use_ssl': dict(type='bool'),
'validate_certs': dict(type='bool'),
'timeout': dict(type='int'),
'provider': dict(type='dict'),
'transport': dict(choices=['cli', 'eapi'])
}
def check_args(module, warnings):
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)
def load_params(module):
provider = module.params.get('provider') or dict()
for key, value in iteritems(provider):
if key in eos_local_argument_spec:
if module.params.get(key) is None and value is not None:
module.params[key] = value
def get_connection(module):
global _DEVICE_CONNECTION
if not _DEVICE_CONNECTION:
load_params(module)
if module.params['transport'] == 'eapi':
conn = Eapi(module)
else:
conn = Cli(module)
_DEVICE_CONNECTION = conn
return _DEVICE_CONNECTION
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"^% \w+", re.M),
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+"),
re.compile(r"[^\r\n]\/bin\/(?:ba)?sh")
]
def __init__(self, module):
self._module = module
super(Cli, self).__init__()
try:
self.connect()
except NetworkError:
exc = get_exception()
self._module.fail_json(msg=str(exc))
if module.params['authorize']:
self.authorize()
def connect(self):
super(Cli, self).connect(self._module.params, kickstart=False)
self.exec_command('terminal length 0')
def authorize(self):
passwd = self._module.params['auth_pass']
if passwd:
prompt = r"[\r\n]?Password: $"
self.exec_command(dict(command='enable', prompt=prompt, response=passwd))
else:
self.exec_command('enable')
def check_authorization(self):
for cmd in ['show clock', 'prompt()']:
rc, out, err = self.exec_command(cmd)
return out.endswith('#')
def supports_sessions(self):
conn = get_connection(self)
rc, out, err = self.exec_command('show configuration sessions')
return rc == 0
def get_config(self, flags=[]):
"""Retrieves the current config from the device or cache
"""
cmd = 'show running-config '
cmd += ' '.join(flags)
cmd = cmd.strip()
try:
return _DEVICE_CONFIGS[cmd]
except KeyError:
conn = get_connection(self)
rc, out, err = self.exec_command(cmd)
if rc != 0:
self._module.fail_json(msg=err)
cfg = str(out).strip()
_DEVICE_CONFIGS[cmd] = cfg
return cfg
def run_commands(self, commands, check_rc=True):
"""Run list of commands on remote device and return results
"""
responses = list()
for cmd in to_list(commands):
rc, out, err = self.exec_command(cmd)
if check_rc and rc != 0:
self._module.fail_json(msg=err)
try:
out = self._module.from_json(out)
except ValueError:
out = str(out).strip()
responses.append(out)
return responses
def send_config(self, commands):
multiline = False
for command in to_list(commands):
if command == 'end':
pass
if command.startswith('banner') or multiline:
multiline = True
command = self._module.jsonify({'command': command, 'sendonly': True})
elif command == 'EOF' and multiline:
multiline = False
rc, out, err = self.exec_command(command)
if rc != 0:
return (rc, out, err)
return (rc, 'ok','')
def configure(self, commands):
"""Sends configuration commands to the remote device
"""
if not check_authorization(self):
self._module.fail_json(msg='configuration operations require privilege escalation')
conn = get_connection(self)
rc, out, err = self.exec_command('configure')
if rc != 0:
self._module.fail_json(msg='unable to enter configuration mode', output=err)
rc, out, err = send_config(self, commands)
if rc != 0:
self._module.fail_json(msg=err)
self.exec_command('end')
return {}
def load_config(self, commands, commit=False, replace=False):
"""Loads the config commands onto the remote device
"""
if not check_authorization(self):
self._module.fail_json(msg='configuration operations require privilege escalation')
use_session = os.getenv('ANSIBLE_EOS_USE_SESSIONS', True)
try:
use_session = int(use_session)
except ValueError:
pass
if not all((bool(use_session), supports_sessions(self))):
return configure(self, commands)
conn = get_connection(self)
session = 'ansible_%s' % int(time.time())
result = {'session': session}
rc, out, err = self.exec_command('configure session %s' % session)
if rc != 0:
self._module.fail_json(msg='unable to enter configuration mode', output=err)
if replace:
self.exec_command('rollback clean-config', check_rc=True)
rc, out, err = send_config(self, commands)
if rc != 0:
self.exec_command('abort')
conn.fail_json(msg=err, commands=commands)
rc, out, err = self.exec_command('show session-config diffs')
if rc == 0:
result['diff'] = out.strip()
if commit:
self.exec_command('commit')
else:
self.exec_command('abort')
return result
class Eapi:
def __init__(self, module):
self._module = module
self._enable = None
self._session_support = None
self._device_config = {}
host = module.params['host']
port = module.params['port']
self._module.params['url_username'] = self._module.params['username']
self._module.params['url_password'] = self._module.params['password']
if module.params['use_ssl']:
proto = 'https'
if not port:
port = 443
else:
proto = 'http'
if not port:
port = 80
self._url = '%s://%s:%s/command-api' % (proto, host, port)
if module.params['auth_pass']:
self._enable = {'cmd': 'enable', 'input': module.params['auth_pass']}
else:
self._enable = 'enable'
def _request_builder(self, commands, output, reqid=None):
params = dict(version=1, cmds=commands, format=output)
return dict(jsonrpc='2.0', id=reqid, method='runCmds', params=params)
def send_request(self, commands, output='text'):
commands = to_list(commands)
if self._enable:
commands.insert(0, 'enable')
body = self._request_builder(commands, output)
data = self._module.jsonify(body)
headers = {'Content-Type': 'application/json-rpc'}
timeout = self._module.params['timeout']
response, headers = fetch_url(
self._module, self._url, data=data, headers=headers,
method='POST', timeout=timeout
)
if headers['status'] != 200:
self._module.fail_json(**headers)
try:
data = response.read()
response = self._module.from_json(data)
except ValueError:
self._module.fail_json(msg='unable to load response from device', data=data)
if self._enable and 'result' in response:
response['result'].pop(0)
return response
def run_commands(self, commands):
"""Runs list of commands on remote device and returns results
"""
output = None
queue = list()
responses = list()
def _send(commands, output):
response = self.send_request(commands, output=output)
if 'error' in response:
err = response['error']
self._module.fail_json(msg=err['message'], code=err['code'])
return response['result']
for item in to_list(commands):
if item['output'] == 'json' and not is_json(item['command']):
item['command'] = '%s | json' % item['command']
if item['output'] == 'text' and is_json(item['command']):
item['command'] = str(item['command']).split('|')[0]
if all((output == 'json', is_text(item['command']))) or all((output =='text', is_json(item['command']))):
responses.extend(_send(queue, output))
queue = list()
output = item['output'] or 'json'
queue.append(item['command'])
if queue:
responses.extend(_send(queue, output))
for index, item in enumerate(commands):
try:
responses[index] = responses[index]['output'].strip()
except KeyError:
pass
return responses
def get_config(self, flags=[]):
"""Retrieves the current config from the device or cache
"""
cmd = 'show running-config '
cmd += ' '.join(flags)
cmd = cmd.strip()
try:
return self._device_configs[cmd]
except KeyError:
out = self.send_request(cmd)
cfg = str(out['result'][0]['output']).strip()
self._device_configs[cmd] = cfg
return cfg
def supports_sessions(self):
if self._session_support:
return self._session_support
response = self.send_request(['show configuration sessions'])
self._session_support = 'error' not in response
return self._session_support
def configure(self, commands):
"""Sends the ordered set of commands to the device
"""
cmds = ['configure terminal']
cmds.extend(commands)
responses = self.send_request(commands)
if 'error' in response:
err = response['error']
self._module.fail_json(msg=err['message'], code=err['code'])
return responses[1:]
def load_config(self, config, commit=False, replace=False):
"""Loads the configuration onto the remote devices
If the device doesn't support configuration sessions, this will
fallback to using configure() to load the commands. If that happens,
there will be no returned diff or session values
"""
if not supports_sessions():
return configure(self, commands)
session = 'ansible_%s' % int(time.time())
result = {'session': session}
commands = ['configure session %s' % session]
if replace:
commands.append('rollback clean-config')
commands.extend(config)
response = self.send_request(commands)
if 'error' in response:
commands = ['configure session %s' % session, 'abort']
self.send_request(commands)
err = response['error']
self._module.fail_json(msg=err['message'], code=err['code'])
commands = ['configure session %s' % session, 'show session-config diffs']
if commit:
commands.append('commit')
else:
commands.append('abort')
response = self.send_request(commands, output='text')
diff = response['result'][1]['output']
if diff:
result['diff'] = diff
return result
is_json = lambda x: str(x).endswith('| json')
is_text = lambda x: not str(x).endswith('| json')
def get_config(module, flags=[]):
conn = get_connection(module)
return conn.get_config(flags)
def run_commands(module, commands):
conn = get_connection(module)
return conn.run_commands(commands)
def load_config(module, config, commit=False, replace=False):
conn = get_connection(module)
return conn.load_config(config, commit, replace)

@ -35,7 +35,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(eos_config) instead
extends_documentation_fragment: eos_local
options:
src:
description:
@ -124,33 +123,10 @@ responses:
"""
import re
from ansible.module_utils import eos
from ansible.module_utils import eos_local
from ansible.module_utils.local import LocalAnsibleModule
from ansible.module_utils.eos import load_config, get_config
from ansible.module_utils.basic import AnsibleModle
from ansible.module_utils.netcfg import NetworkConfig, dumps
SHARED_LIB = 'eos'
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)
load_config = partial(invoke, 'load_config')
get_config = partial(invoke, 'get_config')
def check_args(module):
warnings = list()
if SHARED_LIB == 'eos_local':
eos_local.check_args(module)
return warnings
def get_current_config(module):
config = module.params.get('config')
if not config and not module.params['force']:
@ -201,11 +177,9 @@ def main():
mutually_exclusive = [('config', 'backup'), ('config', 'force')]
cls = get_ansible_module()
module = cls(argument_spec=argument_spec,
mutually_exclusive=mutually_exclusive,
supports_check_mode=True)
module = AnsibleModule(argument_spec=argument_spec,
mutually_exclusive=mutually_exclusive,
supports_check_mode=True)
warnings = check_args(module)
@ -245,5 +219,4 @@ def main():
module.exit_json(**result)
if __name__ == '__main__':
SHARED_LIB = 'eos_local'
main()

@ -16,9 +16,11 @@
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
#
ANSIBLE_METADATA = {'status': ['preview'],
'supported_by': 'community',
'version': '1.0'}
ANSIBLE_METADATA = {
'status': ['preview'],
'supported_by': 'community',
'version': '1.0'
}
DOCUMENTATION = """
---
@ -30,8 +32,6 @@ description:
- This will configure both login and motd banners on remote devices
running Arista EOS. It allows playbooks to add or remote
banner text from the active running configuration.
notes:
- This module requires connection to be network_cli
options:
banner:
description:
@ -91,23 +91,8 @@ session_name:
returned: always
type: str
sample: ansible_1479315771
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.eos import load_config, run_commands
def map_obj_to_commands(updates, module):
@ -156,9 +141,9 @@ def main():
required_if = [('state', 'present', ('text',))]
module = LocalAnsibleModule(argument_spec=argument_spec,
required_if=required_if,
supports_check_mode=True)
module = AnsibleModule(argument_spec=argument_spec,
required_if=required_if,
supports_check_mode=True)
result = {'changed': False}

@ -33,7 +33,6 @@ 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_local
options:
commands:
description:
@ -125,37 +124,15 @@ failed_conditions:
"""
import time
from functools import partial
from ansible.module_utils import eos
from ansible.module_utils import eos_local
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'
from ansible.module_utils.eos import run_commands
from ansible.module_utils.eos import eos_argument_spec, check_args
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 == 'eos_local':
eos_local.check_args(module, warnings)
def to_lines(stdout):
lines = list()
for item in stdout:
@ -193,7 +170,6 @@ def main():
"""entry point for module execution
"""
argument_spec = dict(
# { command: <str>, output: <str>, prompt: <str>, response: <str> }
commands=dict(type='list', required=True),
wait_for=dict(type='list', aliases=['waitfor']),
@ -203,16 +179,15 @@ def main():
interval=dict(default=1, type='int')
)
argument_spec.update(eos_local.eos_local_argument_spec)
cls = get_ansible_module()
module = cls(argument_spec=argument_spec, supports_check_mode=True)
argument_spec.update(eos_argument_spec)
warnings = list()
check_args(module, warnings)
module = AnsibleModule(argument_spec=argument_spec,
supports_check_mode=True)
result = {'changed': False}
warnings = list()
check_args(module, warnings)
commands = parse_commands(module, warnings)
if warnings:
result['warnings'] = warnings
@ -255,5 +230,4 @@ def main():
if __name__ == '__main__':
SHARED_LIB = 'eos_local'
main()

@ -34,7 +34,6 @@ description:
an implementation for working with eos configuration sections in
a deterministic way. This module works with either CLI or eAPI
transports.
extends_documentation_fragment: eos_local
options:
lines:
description:
@ -203,61 +202,21 @@ backup_path:
returned: when backup is yes
type: path
sample: /playbooks/ansible/backup/eos_config.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 functools import partial
from ansible.module_utils import eos
from ansible.module_utils import eos_local
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.local import LocalAnsibleModule
from ansible.module_utils.netcfg import NetworkConfig, dumps
SHARED_LIB = 'eos'
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')
get_config = partial(invoke, 'get_config')
load_config = partial(invoke, 'load_config')
supports_sessions = partial(invoke, 'supports_sessions')
from ansible.module_utils.eos import get_config, load_config
from ansible.module_utils.eos import run_commands
from ansible.module_utils.eos import eos_argument_spec
from ansible.module_utils.eos import check_args as eos_check_args
def check_args(module, warnings):
if SHARED_LIB == 'eos_local':
eos_local.check_args(module)
eos_check_args(module, warnings)
if module.params['force']:
warnings.append('The force argument is deprecated, please use '
'match=none instead. This argument will be '
'removed in the future')
if not supports_sessions(module):
warnings.append('The current version of EOS on the remote device does '
'not support configuration sessions. The commit '
'argument will be ignored')
def get_candidate(module):
candidate = NetworkConfig(indent=3)
if module.params['src']:
@ -330,7 +289,7 @@ def main():
force=dict(default=False, type='bool'),
)
argument_spec.update(eos_local.eos_local_argument_spec)
argument_spec.update(eos_argument_spec)
mutually_exclusive = [('lines', 'src')]
@ -339,12 +298,10 @@ def main():
('replace', 'block', ['lines']),
('replace', 'config', ['src'])]
cls = get_ansible_module()
module = cls(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'
@ -371,5 +328,4 @@ def main():
if __name__ == '__main__':
SHARED_LIB = 'eos_local'
main()

@ -178,25 +178,10 @@ session_name:
returned: when changed is True
type: str
sample: ansible_1479315771
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.eos import run_commands, load_config
from ansible.module_utils.six import iteritems
@ -335,8 +320,8 @@ def main():
state=dict(default='started', choices=['stopped', 'started']),
)
module = LocalAnsibleModule(argument_spec=argument_spec,
supports_check_mode=True)
module = AnsibleModule(argument_spec=argument_spec,
supports_check_mode=True)
result = {'changed': False}

@ -34,7 +34,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: eos_local
options:
gather_subset:
description:
@ -135,32 +134,10 @@ ansible_net_neighbors:
"""
import re
from functools import partial
from ansible.module_utils import eos
from ansible.module_utils import eos_local
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.local import LocalAnsibleModule
from ansible.module_utils.six import iteritems
SHARED_LIB = 'eos'
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 == 'eos_local':
eos_local.check_args(module, warnings)
from ansible.module_utils.eos import run_commands
from ansible.module_utils.eos import eos_argument_spec, check_args
class FactsBase(object):
@ -335,10 +312,10 @@ def main():
gather_subset=dict(default=['!config'], type='list')
)
argument_spec.update(eos_local.eos_local_argument_spec)
argument_spec.update(eos_argument_spec)
cls = get_ansible_module()
module = cls(argument_spec=argument_spec, supports_check_mode=True)
module = AnsibleModule(argument_spec=argument_spec,
supports_check_mode=True)
warnings = list()
check_args(module, warnings)
@ -397,5 +374,4 @@ def main():
if __name__ == '__main__':
SHARED_LIB = 'eos_local'
main()

@ -137,25 +137,10 @@ session_name:
returned: when changed is True
type: str
sample: ansible_1479315771
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.network_common import ComplexList
from ansible.module_utils.eos import load_config, get_config
@ -319,8 +304,8 @@ def main():
state=dict(default='present', choices=['present', 'absent'])
)
module = LocalAnsibleModule(argument_spec=argument_spec,
supports_check_mode=True)
module = AnsibleModule(argument_spec=argument_spec,
supports_check_mode=True)
result = {'changed': False}

@ -32,8 +32,6 @@ description:
either individual usernames or the collection of usernames in the
current running config. It also supports purging usernames from the
configuration that are not explicitly defined.
notes:
- This module requires connection to be network_cli
options:
users:
description:
@ -142,27 +140,12 @@ session_name:
returned: when changed is True
type: str
sample: ansible_1479315771
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 functools import partial
from ansible.module_utils.local import LocalAnsibleModule
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.eos import get_config, load_config
from ansible.module_utils.six import iteritems
@ -333,9 +316,9 @@ def main():
mutually_exclusive = [('username', 'users')]
module = LocalAnsibleModule(argument_spec=argument_spec,
mutually_exclusive=mutually_exclusive,
supports_check_mode=True)
module = AnsibleModule(argument_spec=argument_spec,
mutually_exclusive=mutually_exclusive,
supports_check_mode=True)
result = {'changed': False}

@ -0,0 +1,112 @@
#
# (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.utils.path import unfrackpath
from ansible.plugins import connection_loader
from ansible.compat.six import iteritems
from ansible.module_utils.eos import eos_argument_spec
from ansible.module_utils.basic import AnsibleFallbackNotFound
from ansible.module_utils._text import to_bytes
class ActionModule(_ActionModule):
def run(self, tmp=None, task_vars=None):
provider = self.load_provider()
transport = provider['transport']
if not transport or 'cli' in transport:
pc = copy.deepcopy(self._play_context)
pc.connection = 'network_cli'
pc.network_os = 'eos'
pc.remote_user = provider['username'] or self._play_context.connection_user
pc.password = provider['password'] or self._play_context.password or 22
pc.become = provider['authorize'] or False
pc.become_pass = provider['auth_pass']
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)
connection.exec_command('EXEC: show version')
task_vars['ansible_socket'] = socket_path
else:
if provider['host'] is None:
self._task.args['host'] = self._play_context.remote_addr
if provider['username'] is None:
self._task.args['username'] = self._play_context.connection_user
if provider['password'] is None:
self._task.args['password'] = self._play_context.password
if provider['timeout'] is None:
self._task.args['timeout'] = self._play_context.timeout
if task_vars.get('eapi_use_ssl'):
self._task.args['use_ssl'] = task_vars['eapi_use_ssl']
if task_vars.get('eapi_validate_certs'):
self._task.args['validate_certs'] = task_vars['eapi_validate_certs']
if self._play_context.become_method == 'enable':
self._play_context.become = False
self._play_context.become_method = None
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(eos_argument_spec):
if key == 'provider':
continue
elif 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

@ -19,16 +19,95 @@
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from ansible.plugins.action.net_config import ActionModule as NetworkActionModule
import os
import re
import time
import glob
try:
from __main__ import display
except ImportError:
from ansible.utils.display import Display
display = Display()
from ansible.plugins.action.eos 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
class ActionModule(NetworkActionModule):
PRIVATE_KEYS_RE = re.compile('__.+__')
class ActionModule(_ActionModule):
def run(self, tmp=None, task_vars=None):
display.vvvvv('Using connection plugin %s' % self._play_context.connection)
return NetworkActionModule.run(self, tmp, task_vars)
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,8 +19,84 @@
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.eos 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)

@ -79,9 +79,4 @@ class TerminalModule(TerminalBase):
elif prompt.endswith('#'):
self._exec_cli_command('disable')
@staticmethod
def guess_network_os(conn):
stdin, stdout, stderr = conn.exec_command('show version')
if 'Arista' in stdout.read():
return 'eos'

@ -1,105 +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. This value applies to either I(cli) or I(eapi). The port
value will default to the appropriate transport common port if
none is provided in the task. (cli=22, http=80, https=443).
required: false
default: null
username:
description:
- Configures the username to use to authenticate the connection to
the remote device. The value of I(username) is used to authenticate
either the CLI login or the eAPI authentication depending on which
transport is used. 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 is a common argument used for either I(cli)
or I(eapi) transports. 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
ssh_keyfile:
description:
- Specifies the SSH keyfile to use to authenticate the connection to
the remote device. This argument is only used for I(cli) transports.
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
authorize:
description:
- Instructs the module to enter privileged mode on the remote device
before sending any commands. If not specified, the device will
attempt to execute all commands in non-privileged mode. If the value
is not specified in the task, the value of environment variable
C(ANSIBLE_NET_AUTHORIZE) will be used instead.
required: false
default: null
choices: ['yes', 'no']
auth_pass:
description:
- Specifies the password to use if required to enter privileged mode
on the remote device. If I(authorize) is false, then this argument
does nothing. If the value is not specified in the task, the value of
environment variable C(ANSIBLE_NET_AUTH_PASS) will be used instead.
required: false
default: null
transport:
description:
- Configures the transport connection to use when connecting to the
remote device.
required: true
choices:
- eapi
- cli
default: null
use_ssl:
description:
- Configures the I(transport) to use SSL if set to true only when the
C(transport=eapi). If the transport
argument is not eapi, this value is ignored.
required: false
default: null
choices: ['yes', 'no']
provider:
description:
- Convenience method that allows all I(eos) 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
"""

@ -1,120 +0,0 @@
#
# (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.module_utils import eos
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 TestEosModuleUtil(unittest.TestCase):
def setUp(self):
eos._DEVICE_CONFIGS = {}
def test_eos_get_config(self):
mock_module = MagicMock(name='AnsibleModule')
mock_module.exec_command.return_value = (0, ' sample config\n', '')
self.assertFalse('show running-config' in eos._DEVICE_CONFIGS)
out = eos.get_config(mock_module)
self.assertEqual(out, 'sample config')
self.assertTrue('show running-config' in eos._DEVICE_CONFIGS)
self.assertEqual(eos._DEVICE_CONFIGS['show running-config'], 'sample config')
def test_eos_get_config_cached(self):
mock_module = MagicMock(name='AnsibleModule')
mock_module.exec_command.return_value = (0, ' sample config\n', '')
eos._DEVICE_CONFIGS['show running-config'] = 'different config'
out = eos.get_config(mock_module)
self.assertEqual(out, 'different config')
self.assertFalse(mock_module.exec_command.called)
def test_eos_get_config_error(self):
mock_module = MagicMock(name='AnsibleModule')
mock_module.exec_command.return_value = (1, '', 'error')
out = eos.get_config(mock_module, 'show running_config')
self.assertTrue(mock_module.fail_json.called)
def test_eos_supports_sessions_fail(self):
mock_module = MagicMock(name='AnsibleModule')
mock_module.exec_command.return_value = (1, '', '')
self.assertFalse(eos.supports_sessions(mock_module))
mock_module.exec_command.called_with_args(['show configuration sessions'])
def test_eos_supports_sessions_pass(self):
mock_module = MagicMock(name='AnsibleModule')
mock_module.exec_command.return_value = (0, '', '')
self.assertTrue(eos.supports_sessions(mock_module))
mock_module.exec_command.called_with_args(['show configuration sessions'])
def test_eos_run_commands(self):
mock_module = MagicMock(name='AnsibleModule')
mock_module.exec_command.return_value = (0, 'stdout', '')
mock_module.from_json.side_effect = ValueError
out = eos.run_commands(mock_module, 'command')
self.assertEqual(out, ['stdout'])
def test_eos_run_commands_returns_json(self):
mock_module = MagicMock(name='AnsibleModule')
mock_module.exec_command.return_value = (0, '{"key": "value"}', '')
mock_module.from_json.return_value = json.loads('{"key": "value"}')
out = eos.run_commands(mock_module, 'command')
self.assertEqual(out, [{'key': 'value'}])
def test_eos_run_commands_check_rc_fails(self):
mock_module = MagicMock(name='AnsibleModule')
mock_module.exec_command.return_value = (1, '', 'stderr')
out = eos.run_commands(mock_module, 'command')
mock_module.fail_json.called_with_args({'msg': 'stderr'})

@ -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 TestEosModule(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

@ -17,43 +17,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.eos import eos_banner
from ansible.module_utils import basic
from ansible.module_utils._text import to_bytes
from .eos_module import TestEosModule, load_fixture, set_module_args
def set_module_args(args):
args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
basic._ANSIBLE_ARGS = to_bytes(args)
class TestEosBannerModule(TestEosModule):
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 TestEosBannerModule(unittest.TestCase):
module = eos_banner
def setUp(self):
self.mock_run_commands = patch('ansible.modules.network.eos.eos_banner.run_commands')
@ -66,29 +39,10 @@ class TestEosBannerModule(unittest.TestCase):
self.mock_run_commands.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.run_commands.return_value = load_fixture('eos_banner_show_banner.txt').strip()
self.load_config.return_value = dict(diff=None, session='session')
with self.assertRaises(AnsibleModuleExit) as exc:
eos_banner.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_eos_banner_create(self):
set_module_args(dict(banner='login', text='test\nbanner\nstring'))
commands = ['banner login', 'test', 'banner', 'string', 'EOF']

@ -19,43 +19,15 @@
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.eos import eos_command
from ansible.module_utils import basic
from ansible.module_utils._text import to_bytes
from .eos_module import TestEosModule, load_fixture, set_module_args
class TestEosCommandModule(TestEosModule):
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):
module = eos_command
def setUp(self):
self.mock_run_commands = patch('ansible.modules.network.eos.eos_command.run_commands')
@ -64,8 +36,7 @@ class test_EosCommandModule(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
output = list()
@ -83,18 +54,6 @@ class test_EosCommandModule(unittest.TestCase):
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()

@ -21,50 +21,16 @@
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import os
import sys
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.eos import eos_config
from ansible.module_utils.netcfg import NetworkConfig
from ansible.module_utils import basic
from ansible.module_utils._text import to_bytes
from .eos_module import TestEosModule, load_fixture, set_module_args
PROVIDER_ARGS = {
'host': 'localhost',
'username': 'username',
'password': 'password'
}
def set_module_args(args):
args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
basic._ANSIBLE_ARGS = to_bytes(args)
class TestEosConfigModule(TestEosModule):
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 TestEosConfigModule(unittest.TestCase):
module = eos_config
def setUp(self):
self.mock_get_config = patch('ansible.modules.network.eos.eos_config.get_config')
@ -73,40 +39,21 @@ class TestEosConfigModule(unittest.TestCase):
self.mock_load_config = patch('ansible.modules.network.eos.eos_config.load_config')
self.load_config = self.mock_load_config.start()
self.mock_supports_sessions = patch('ansible.modules.network.eos.eos_config.supports_sessions')
self.supports_sessions = self.mock_supports_sessions.start()
self.supports_sessions.return_value = True
def tearDown(self):
self.mock_get_config.stop()
self.mock_load_config.stop()
def execute_module(self, failed=False, changed=False):
def load_fixtures(self, commands=None):
self.get_config.return_value = load_fixture('eos_config_config.cfg')
self.load_config.return_value = dict(diff=None, session='session')
with self.assertRaises(AnsibleModuleExit) as exc:
eos_config.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_config_no_change(self):
args = dict(lines=['hostname localhost'])
args.update(PROVIDER_ARGS)
set_module_args(args)
result = self.execute_module()
def test_eos_config_src(self):
args = dict(src=load_fixture('eos_config_candidate.cfg'))
args.update(PROVIDER_ARGS)
set_module_args(args)
result = self.execute_module(changed=True)
@ -117,7 +64,6 @@ class TestEosConfigModule(unittest.TestCase):
def test_eos_config_lines(self):
args = dict(lines=['hostname switch01', 'ip domain-name eng.ansible.com'])
args.update(PROVIDER_ARGS)
set_module_args(args)
result = self.execute_module(changed=True)
@ -129,7 +75,6 @@ class TestEosConfigModule(unittest.TestCase):
args = dict(lines=['hostname switch01', 'ip domain-name eng.ansible.com'],
before=['before command'])
args.update(PROVIDER_ARGS)
set_module_args(args)
result = self.execute_module(changed=True)
@ -142,7 +87,6 @@ class TestEosConfigModule(unittest.TestCase):
args = dict(lines=['hostname switch01', 'ip domain-name eng.ansible.com'],
after=['after command'])
args.update(PROVIDER_ARGS)
set_module_args(args)
result = self.execute_module(changed=True)
@ -153,7 +97,6 @@ class TestEosConfigModule(unittest.TestCase):
def test_eos_config_parents(self):
args = dict(lines=['ip address 1.2.3.4/5', 'no shutdown'], parents=['interface Ethernet10'])
args.update(PROVIDER_ARGS)
set_module_args(args)
result = self.execute_module(changed=True)
@ -163,37 +106,31 @@ class TestEosConfigModule(unittest.TestCase):
def test_eos_config_src_and_lines_fails(self):
args = dict(src='foo', lines='foo')
args.update(PROVIDER_ARGS)
set_module_args(args)
result = self.execute_module(failed=True)
def test_eos_config_match_exact_requires_lines(self):
args = dict(match='exact')
args.update(PROVIDER_ARGS)
set_module_args(args)
result = self.execute_module(failed=True)
def test_eos_config_match_strict_requires_lines(self):
args = dict(match='strict')
args.update(PROVIDER_ARGS)
set_module_args(args)
result = self.execute_module(failed=True)
def test_eos_config_replace_block_requires_lines(self):
args = dict(replace='block')
args.update(PROVIDER_ARGS)
set_module_args(args)
result = self.execute_module(failed=True)
def test_eos_config_replace_config_requires_src(self):
args = dict(replace='config')
args.update(PROVIDER_ARGS)
set_module_args(args)
result = self.execute_module(failed=True)
def test_eos_config_backup_returns__backup__(self):
args = dict(backup=True)
args.update(PROVIDER_ARGS)
set_module_args(args)
result = self.execute_module()
self.assertIn('__backup__', result)

@ -21,45 +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 ansible.modules.network.eos import eos_eapi
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
from .eos_module import TestEosModule, load_fixture, set_module_args
fixture_data[path] = data
return data
class TestEosEapiModule(TestEosModule):
class TestEosEapiModule(unittest.TestCase):
module = eos_eapi
def setUp(self):
self.mock_run_commands = patch('ansible.modules.network.eos.eos_eapi.run_commands')
@ -68,54 +39,34 @@ class TestEosEapiModule(unittest.TestCase):
self.mock_load_config = patch('ansible.modules.network.eos.eos_eapi.load_config')
self.load_config = self.mock_load_config.start()
self.command_fixtures = {}
def tearDown(self):
self.mock_run_commands.stop()
self.mock_load_config.stop()
def execute_module(self, failed=False, changed=False, commands=None,
sort=True, command_fixtures={}):
def load_fixtures(self, commands=None):
def run_commands(module, commands, **kwargs):
output = list()
for cmd in commands:
output.append(load_fixture(command_fixtures[cmd]))
output.append(load_fixture(self.command_fixtures[cmd]))
return (0, output, '')
self.run_commands.side_effect = run_commands
self.load_config.return_value = dict(diff=None, session='session')
with self.assertRaises(AnsibleModuleExit) as exc:
eos_eapi.main()
result = exc.exception.result
if failed:
self.assertTrue(result.get('failed'), result)
else:
self.assertEqual(result.get('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 start_configured(self, *args, **kwargs):
command_fixtures = {
self.command_fixtures = {
'show vrf': 'eos_eapi_show_vrf.text',
'show management api http-commands | json': 'eos_eapi_show_mgmt.json'
}
kwargs['command_fixtures'] = command_fixtures
return self.execute_module(*args, **kwargs)
def start_unconfigured(self, *args, **kwargs):
command_fixtures = {
self.command_fixtures = {
'show vrf': 'eos_eapi_show_vrf.text',
'show management api http-commands | json': 'eos_eapi_show_mgmt_unconfigured.json'
}
kwargs['command_fixtures'] = command_fixtures
return self.execute_module(*args, **kwargs)
def test_eos_eapi_http_enable(self):

@ -21,44 +21,15 @@
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 ansible.modules.network.eos import eos_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
from .eos_module import TestEosModule, load_fixture, set_module_args
fixture_data[path] = data
return data
class TestEosSystemModule(TestEosModule):
class TestEosSystemModule(unittest.TestCase):
module = eos_system
def setUp(self):
self.mock_get_config = patch('ansible.modules.network.eos.eos_system.get_config')
@ -71,29 +42,10 @@ class TestEosSystemModule(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('eos_system_config.cfg')
self.load_config.return_value = dict(diff=None, session='session')
with self.assertRaises(AnsibleModuleExit) as exc:
eos_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_eos_system_hostname_changed(self):
set_module_args(dict(hostname='foo'))
commands = ['hostname foo']

@ -17,43 +17,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.eos import eos_user
from ansible.module_utils import basic
from ansible.module_utils._text import to_bytes
from .eos_module import TestEosModule, load_fixture, set_module_args
def set_module_args(args):
args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
basic._ANSIBLE_ARGS = to_bytes(args)
class TestEosUserModule(TestEosModule):
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 TestEosUserModule(unittest.TestCase):
module = eos_user
def setUp(self):
self.mock_get_config = patch('ansible.modules.network.eos.eos_user.get_config')
@ -66,29 +39,10 @@ class TestEosUserModule(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('eos_user_config.cfg')
self.load_config.return_value = dict(diff=None, session='session')
with self.assertRaises(AnsibleModuleExit) as exc:
eos_user.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_eos_user_create(self):
set_module_args(dict(username='test', nopassword=True))
commands = ['username test nopassword']
@ -101,7 +55,7 @@ class TestEosUserModule(unittest.TestCase):
def test_eos_user_password(self):
set_module_args(dict(username='ansible', password='test'))
commands = ['username ansible secret ********']
commands = ['username ansible secret test']
self.execute_module(changed=True, commands=commands)
def test_eos_user_privilege(self):
@ -130,17 +84,16 @@ class TestEosUserModule(unittest.TestCase):
def test_eos_user_update_password_changed(self):
set_module_args(dict(username='test', password='test', update_password='on_create'))
commands = ['username ******** secret ********']
commands = ['username test secret test']
self.execute_module(changed=True, commands=commands)
def test_eos_user_update_password_on_create_ok(self):
set_module_args(dict(username='ansible', password='test', update_password='on_create'))
commands = []
self.execute_module(commands=commands)
self.execute_module()
def test_eos_user_update_password_always(self):
set_module_args(dict(username='ansible', password='test', update_password='always'))
commands = ['username ansible secret ********']
commands = ['username ansible secret test']
self.execute_module(changed=True, commands=commands)

Loading…
Cancel
Save