ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
Code Releases Activity

New networking module: voss_command (#43741)

Browse source 
* new voss_command module

* Removed incorrect version_added line
This commit is contained in:
Lindsay Hill 2018-08-08 06:56:50 -07:00 committed by Ricardo Carrillo Cruz
parent 3419c75411
commit 8429f777da
11 changed files with 1003 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
.github/BOTMETA.yml vendored
View file

@ -505,6 +505,7 @@ files:
$modules/network/slxos/: $team_extreme
$modules/network/sros/: privateip
$modules/network/system/: $team_networking
$modules/network/voss/: $team_extreme
$modules/network/vyos/: Qalthos samdoran
$modules/notification/bearychat.py: tonyseek
$modules/notification/campfire.py: fabulops
@ -931,6 +932,9 @@ files:
maintainers: $team_vmware
labels: vmware
support: community
$module_utils/network/voss:
maintainers: $team_extreme
labels: networking
$module_utils/network:
maintainers: $team_networking
labels: networking
@ -1050,6 +1054,9 @@ files:
labels:
- cloud
- openstack
lib/ansible/plugins/cliconf/voss.py:
maintainers: $team_extreme
labels: networking
lib/ansible/plugins/connection/winrm.py:
maintainers: $team_windows_core
labels:
@ -1144,6 +1151,9 @@ files:
lib/ansible/plugins/terminal/sros.py:
maintainers: $team_networking
labels: networking
lib/ansible/plugins/terminal/voss.py:
maintainers: $team_extreme
labels: networking
lib/ansible/plugins/terminal/vyos.py:
maintainers: $team_networking samdoran
labels: networking

0
lib/ansible/module_utils/network/voss/__init__.py Normal file
View file

116
lib/ansible/module_utils/network/voss/voss.py Normal file
View file

@ -0,0 +1,116 @@
# 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) 2018 Extreme Networks 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 json
from ansible.module_utils._text import to_text
from ansible.module_utils.network.common.utils import to_list, ComplexList
from ansible.module_utils.connection import Connection, ConnectionError
_DEVICE_CONFIGS = {}
def get_connection(module):
if hasattr(module, '_voss_connection'):
return module._voss_connection
capabilities = get_capabilities(module)
network_api = capabilities.get('network_api')
if network_api == 'cliconf':
module._voss_connection = Connection(module._socket_path)
else:
module.fail_json(msg='Invalid connection type %s' % network_api)
return module._voss_connection
def get_capabilities(module):
if hasattr(module, '_voss_capabilities'):
return module._voss_capabilities
try:
capabilities = Connection(module._socket_path).get_capabilities()
except ConnectionError as exc:
module.fail_json(msg=to_text(exc, errors='surrogate_then_replace'))
module._voss_capabilities = json.loads(capabilities)
return module._voss_capabilities
def check_args(module, warnings):
pass
def get_defaults_flag(module):
connection = get_connection(module)
try:
out = connection.get_defaults_flag()
except ConnectionError as exc:
module.fail_json(msg=to_text(exc, errors='surrogate_then_replace'))
return to_text(out, errors='surrogate_then_replace').strip()
def get_config(module, flags=None):
flag_str = ' '.join(to_list(flags))
try:
return _DEVICE_CONFIGS[flag_str]
except KeyError:
connection = get_connection(module)
try:
out = connection.get_config(flags=flags)
except ConnectionError as exc:
module.fail_json(msg=to_text(exc, errors='surrogate_then_replace'))
cfg = to_text(out, errors='surrogate_then_replace').strip()
_DEVICE_CONFIGS[flag_str] = cfg
return cfg
def to_commands(module, commands):
spec = {
'command': dict(key=True),
'prompt': dict(),
'answer': dict()
}
transform = ComplexList(spec, module)
return transform(commands)
def run_commands(module, commands, check_rc=True):
connection = get_connection(module)
try:
out = connection.run_commands(commands=commands, check_rc=check_rc)
return out
except ConnectionError as exc:
module.fail_json(msg=to_text(exc))
def load_config(module, commands):
connection = get_connection(module)
try:
resp = connection.edit_config(commands)
return resp.get('response')
except ConnectionError as exc:
module.fail_json(msg=to_text(exc))

0
lib/ansible/modules/network/voss/__init__.py Normal file
View file

242
lib/ansible/modules/network/voss/voss_command.py Normal file
View file

@ -0,0 +1,242 @@
#!/usr/bin/python
#
# 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
ANSIBLE_METADATA = {'metadata_version': '1.1',
'status': ['preview'],
'supported_by': 'community'}
DOCUMENTATION = """
---
module: voss_command
version_added: "2.7"
author: "Lindsay Hill (@LindsayHill)"
short_description: Run commands on remote devices running Extreme VOSS
description:
- Sends arbitrary commands to an Extreme VSP device running VOSS, and
returns the results 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.
- This module does not support running commands in configuration mode.
Please use M(voss_config) to configure VOSS devices.
notes:
- Tested against VOSS 7.0.0
options:
commands:
description:
- List of commands to send to the remote VOSS device. The
resulting output from the command is returned. If the
I(wait_for) argument is provided, the module is not returned
until the condition is satisfied or the number of retries has
expired. If a command sent to the device requires answering a
prompt, it is possible to pass a dict containing I(command),
I(answer) and I(prompt). Common answers are 'y' or "\\r"
(carriage return, must be double quotes). See examples.
required: true
wait_for:
description:
- List of conditions to evaluate against the output of the
command. The task will wait for each condition to be true
before moving forward. If the conditional is not true
within the configured number of retries, the task fails.
See examples.
match:
description:
- The I(match) argument is used in conjunction with the
I(wait_for) argument to specify the match policy. Valid
values are C(all) or C(any). If the value is set to C(all)
then all conditionals in the wait_for must be satisfied. If
the value is set to C(any) then only one of the values must be
satisfied.
default: all
choices: ['any', 'all']
retries:
description:
- Specifies the number of retries a command should by tried
before it is considered failed. The command is run on the
target device every retry and evaluated against the
I(wait_for) conditions.
default: 10
interval:
description:
- Configures the interval in seconds to wait between retries
of the command. If the command does not pass the specified
conditions, the interval indicates how long to wait before
trying the command again.
default: 1
"""
EXAMPLES = r"""
tasks:
- name: run show sys software on remote devices
voss_command:
commands: show sys software
- name: run show sys software and check to see if output contains VOSS
voss_command:
commands: show sys software
wait_for: result[0] contains VOSS
- name: run multiple commands on remote nodes
voss_command:
commands:
- show sys software
- show interfaces vlan
- name: run multiple commands and evaluate the output
voss_command:
commands:
- show sys software
- show interfaces vlan
wait_for:
- result[0] contains Version
- result[1] contains Basic
- name: run command that requires answering a prompt
voss_command:
commands:
- command: 'reset'
prompt: 'Are you sure you want to reset the switch? (y/n)'
answer: 'y'
"""
RETURN = """
stdout:
description: The set of responses from the commands
returned: always apart from low level errors (such as action plugin)
type: list
sample: ['...', '...']
stdout_lines:
description: The value of stdout split into a list
returned: always apart from low level errors (such as action plugin)
type: list
sample: [['...', '...'], ['...'], ['...']]
failed_conditions:
description: The list of conditionals that have failed
returned: failed
type: list
sample: ['...', '...']
"""
import re
import time
from ansible.module_utils.network.voss.voss import run_commands
from ansible.module_utils.network.voss.voss import check_args
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.network.common.utils import ComplexList
from ansible.module_utils.network.common.parsing import Conditional
from ansible.module_utils.six import string_types
def to_lines(stdout):
for item in stdout:
if isinstance(item, string_types):
item = str(item).split('\n')
yield item
def parse_commands(module, warnings):
command = ComplexList(dict(
command=dict(key=True),
prompt=dict(),
answer=dict()
), module)
commands = command(module.params['commands'])
for item in list(commands):
configure_type = re.match(r'conf(?:\w*)(?:\s+(\w+))?', item['command'])
if module.check_mode:
if configure_type and configure_type.group(1) not in ('confirm', 'replace', 'revert', 'network'):
module.fail_json(
msg='voss_command does not support running config mode '
'commands. Please use voss_config instead'
)
if not item['command'].startswith('show'):
warnings.append(
'only show commands are supported when using check mode, not '
'executing `%s`' % item['command']
)
commands.remove(item)
return commands
def main():
"""main entry point for module execution
"""
argument_spec = dict(
commands=dict(type='list', required=True),
wait_for=dict(type='list'),
match=dict(default='all', choices=['all', 'any']),
retries=dict(default=10, type='int'),
interval=dict(default=1, type='int')
)
module = AnsibleModule(argument_spec=argument_spec,
supports_check_mode=True)
result = {'changed': False}
warnings = list()
check_args(module, warnings)
commands = parse_commands(module, warnings)
result['warnings'] = warnings
wait_for = module.params['wait_for'] or list()
conditionals = [Conditional(c) for c in wait_for]
retries = module.params['retries']
interval = module.params['interval']
match = module.params['match']
while retries > 0:
responses = run_commands(module, commands)
for item in list(conditionals):
if item(responses):
if match == 'any':
conditionals = list()
break
conditionals.remove(item)
if not conditionals:
break
time.sleep(interval)
retries -= 1
if conditionals:
failed_conditions = [item.raw for item in conditionals]
msg = 'One or more conditional statements have not been satisfied'
module.fail_json(msg=msg, failed_conditions=failed_conditions)
result.update({
'changed': False,
'stdout': responses,
'stdout_lines': list(to_lines(responses))
})
module.exit_json(**result)
if __name__ == '__main__':
main()

231
lib/ansible/plugins/cliconf/voss.py Normal file
View file

@ -0,0 +1,231 @@
#
# (c) 2018 Extreme Networks 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 collections
import re
import json
from ansible.errors import AnsibleConnectionFailure
from ansible.module_utils._text import to_text
from ansible.module_utils.network.common.config import NetworkConfig, dumps
from ansible.module_utils.network.common.utils import to_list
from ansible.plugins.cliconf import CliconfBase, enable_mode
class Cliconf(CliconfBase):
@enable_mode
def get_config(self, source='running', flags=None, format=None):
if source not in ('running', 'startup'):
return self.invalid_params("fetching configuration from %s is not supported" % source)
if format:
raise ValueError("'format' value %s is not supported for get_config" % format)
if not flags:
flags = []
if source == 'running':
cmd = 'show running-config '
else:
cmd = 'more /intflash/config.cfg'
cmd += ' '.join(to_list(flags))
cmd = cmd.strip()
return self.send_command(cmd)
def get_diff(self, candidate=None, running=None, diff_match='line', diff_ignore_lines=None, path=None, diff_replace='line'):
"""
Generate diff between candidate and running configuration. If the
remote host supports onbox diff capabilities ie. supports_onbox_diff in that case
candidate and running configurations are not required to be passed as argument.
In case if onbox diff capability is not supported candidate argument is mandatory
and running argument is optional.
:param candidate: The configuration which is expected to be present on remote host.
:param running: The base configuration which is used to generate diff.
:param diff_match: Instructs how to match the candidate configuration with current device configuration
Valid values are 'line', 'strict', 'exact', 'none'.
'line' - commands are matched line by line
'strict' - command lines are matched with respect to position
'exact' - command lines must be an equal match
'none' - will not compare the candidate configuration with the running configuration
:param diff_ignore_lines: Use this argument to specify one or more lines that should be
ignored during the diff. This is used for lines in the configuration
that are automatically updated by the system. This argument takes
a list of regular expressions or exact line matches.
:param path: The ordered set of parents that uniquely identify the section or hierarchy
the commands should be checked against. If the parents argument
is omitted, the commands are checked against the set of top
level or global commands.
:param diff_replace: Instructs on the way to perform the configuration on the device.
If the replace argument is set to I(line) then the modified lines are
pushed to the device in configuration mode. If the replace argument is
set to I(block) then the entire command block is pushed to the device in
configuration mode if any line is not correct.
:return: Configuration diff in json format.
{
'config_diff': '',
'banner_diff': {}
}
"""
diff = {}
device_operations = self.get_device_operations()
option_values = self.get_option_values()
if candidate is None and device_operations['supports_generate_diff']:
raise ValueError("candidate configuration is required to generate diff")
if diff_match not in option_values['diff_match']:
raise ValueError("'match' value %s in invalid, valid values are %s" % (diff_match, ', '.join(option_values['diff_match'])))
if diff_replace not in option_values['diff_replace']:
raise ValueError("'replace' value %s in invalid, valid values are %s" % (diff_replace, ', '.join(option_values['diff_replace'])))
# prepare candidate configuration
candidate_obj = NetworkConfig(indent=1)
candidate_obj.load(candidate)
if running and diff_match != 'none':
# running configuration
running_obj = NetworkConfig(indent=1, contents=running, ignore_lines=diff_ignore_lines)
configdiffobjs = candidate_obj.difference(running_obj, path=path, match=diff_match, replace=diff_replace)
else:
configdiffobjs = candidate_obj.items
diff['config_diff'] = dumps(configdiffobjs, 'commands') if configdiffobjs else ''
return diff
@enable_mode
def edit_config(self, candidate=None, commit=True, replace=None, comment=None):
resp = {}
operations = self.get_device_operations()
self.check_edit_config_capabiltiy(operations, candidate, commit, replace, comment)
results = []
requests = []
if commit:
self.send_command('configure terminal')
for line in to_list(candidate):
if not isinstance(line, collections.Mapping):
line = {'command': line}
cmd = line['command']
if cmd != 'end' and cmd[0] != '!':
results.append(self.send_command(**line))
requests.append(cmd)
self.send_command('end')
else:
raise ValueError('check mode is not supported')
resp['request'] = requests
resp['response'] = results
return resp
def get(self, command=None, prompt=None, answer=None, sendonly=False, output=None):
if not command:
raise ValueError('must provide value of command to execute')
if output:
raise ValueError("'output' value %s is not supported for get" % output)
return self.send_command(command=command, prompt=prompt, answer=answer, sendonly=sendonly)
def get_device_info(self):
device_info = {}
device_info['network_os'] = 'voss'
reply = self.get(command='show sys-info')
data = to_text(reply, errors='surrogate_or_strict').strip()
match = re.search(r'SysDescr\s+: \S+ \((\S+)\)', data)
if match:
device_info['network_os_version'] = match.group(1)
match = re.search(r'Chassis\s+: (\S+)', data)
if match:
device_info['network_os_model'] = match.group(1)
match = re.search(r'SysName\s+: (\S+)', data)
if match:
device_info['network_os_hostname'] = match.group(1)
return device_info
def get_device_operations(self):
return {
'supports_diff_replace': True,
'supports_commit': False,
'supports_rollback': False,
'supports_defaults': True,
'supports_onbox_diff': False,
'supports_commit_comment': False,
'supports_multiline_delimiter': False,
'supports_diff_match': True,
'supports_diff_ignore_lines': True,
'supports_generate_diff': True,
'supports_replace': False
}
def get_option_values(self):
return {
'format': ['text'],
'diff_match': ['line', 'strict', 'exact', 'none'],
'diff_replace': ['line', 'block'],
'output': []
}
def get_capabilities(self):
result = dict()
result['rpc'] = self.get_base_rpc() + ['edit_banner', 'get_diff', 'run_commands', 'get_defaults_flag']
result['network_api'] = 'cliconf'
result['device_info'] = self.get_device_info()
result['device_operations'] = self.get_device_operations()
result.update(self.get_option_values())
return json.dumps(result)
def run_commands(self, commands=None, check_rc=True):
if commands is None:
raise ValueError("'commands' value is required")
responses = list()
for cmd in to_list(commands):
if not isinstance(cmd, collections.Mapping):
cmd = {'command': cmd}
output = cmd.pop('output', None)
if output:
raise ValueError("'output' value %s is not supported for run_commands" % output)
try:
out = self.send_command(**cmd)
except AnsibleConnectionFailure as e:
if check_rc:
raise
out = getattr(e, 'err', e)
responses.append(out)
return responses
def get_defaults_flag(self):
return 'verbose'

89
lib/ansible/plugins/terminal/voss.py Normal file
View file

@ -0,0 +1,89 @@
#
# (c) 2018 Extreme Networks 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 json
import re
from ansible.errors import AnsibleConnectionFailure
from ansible.module_utils._text import to_text, to_bytes
from ansible.plugins.terminal import TerminalBase
class TerminalModule(TerminalBase):
terminal_stdout_re = [
re.compile(br"[\r\n]+[^\s#>]+(?:[>#])$", re.M)
]
terminal_stderr_re = [
re.compile(br"% ?Error"),
re.compile(br"% ?Bad secret"),
re.compile(br"[\r\n%] Bad passwords"),
re.compile(br"invalid input", re.I),
re.compile(br"(?:incomplete|ambiguous) command", re.I),
re.compile(br"connection timed out", re.I),
re.compile(br"[^\r\n]+ not found"),
re.compile(br"'[^']' +returned error code: ?\d+"),
re.compile(br"Discontiguous Subnet Mask"),
re.compile(br"Conflicting IP address"),
re.compile(br"[\r\n]Error: ?[\S]+"),
re.compile(br"[%\S] ?Informational: ?[\s]+", re.I),
re.compile(br"Command authorization failed")
]
def on_open_shell(self):
try:
self._exec_cli_command(u'terminal more disable')
except AnsibleConnectionFailure:
raise AnsibleConnectionFailure('unable to set terminal parameters')
def on_become(self, passwd=None):
if self._get_prompt().endswith(b'#'):
return
cmd = {u'command': u'enable'}
if passwd:
# Note: python-3.5 cannot combine u"" and r"" together. Thus make
# an r string and use to_text to ensure it's text on both py2 and py3.
cmd[u'prompt'] = to_text(r"[\r\n](?:Local_)?[Pp]assword: ?$", errors='surrogate_or_strict')
cmd[u'answer'] = passwd
cmd[u'prompt_retry_check'] = True
try:
self._exec_cli_command(to_bytes(json.dumps(cmd), errors='surrogate_or_strict'))
prompt = self._get_prompt()
if prompt is None or not prompt.endswith(b'#'):
raise AnsibleConnectionFailure('failed to elevate privilege to enable mode still at prompt [%s]' % prompt)
except AnsibleConnectionFailure as e:
prompt = self._get_prompt()
raise AnsibleConnectionFailure('unable to elevate privilege to enable mode, at prompt [%s] with error: %s' % (prompt, e.message))
def on_unbecome(self):
prompt = self._get_prompt()
if prompt is None:
# if prompt is None most likely the terminal is hung up at a prompt
return
if prompt.endswith(b')#'):
self._exec_cli_command(b'end')
self._exec_cli_command(b'disable')
elif prompt.endswith(b'#'):
self._exec_cli_command(b'disable')

0
test/units/modules/network/voss/__init__.py Normal file
View file

107
test/units/modules/network/voss/fixtures/show_sys-info Normal file
View file

@ -0,0 +1,107 @@
General Info :
SysDescr : VSP-4450GSX-PWR+ (7.0.0.0_B015)
SysName : VSP-4450GSX-PWR+
SysUpTime : 5 day(s), 17:13:09
SysContact : http://www.extremenetworks.com/contact/
SysLocation :
Chassis Info:
Chassis : 4450GSX-PWR+
ModelName : 4450GSX-PWR+
BrandName : Extreme Networks.
Serial# : 14JP512E0001
H/W Revision : 01
H/W Config : none
Part Number :
NumSlots : 1
NumPorts : 50
BaseMacAddr : b4:47:5e:00:00:00
MacAddrCapacity : 256
System MTU : 1950
Card Info :
Slot# CardType Serial# Part# Oper Admin Power
Status Status State
1 4450GSX-PWR+ 14JP512E0001 -- up up on
Temperature Info :
Chassis Temperature
30
Power Supply Info :
Ps#1 Status : UP
Ps#1 Type : AC
Ps#1 Description : AC-DC-54V-1000W
Ps#1 Serial Number: LBNNTMPL20180R
Ps#1 Version : --
Ps#1 Part Number : 325220-A.01
Ps#2 Status : empty
Total Power Available : 1000 watts
Total Power Usage : 127 watts
Fan Info :
Description OperStatus OperSpeed AirflowDir
Tray 1 Fan 1 up mediumSpeed left-right
Tray 1 Fan 2 up mediumSpeed left-right
Tray 1 Fan 3 up mediumSpeed left-right
LED Info :
LED#1 Label : PWR
LED#1 Status : GreenSteady
LED#2 Label : Status
LED#2 Status : GreenSteady
LED#3 Label : Rps
LED#3 Status : Off
LED#4 Label : Up
LED#4 Status : UnSupported
LED#5 Label : Down
LED#5 Status : UnSupported
LED#6 Label : Base
LED#6 Status : UnSupported
System Error Info :
Send Login Success Trap : false
Send Authentication Trap : false
Error Code : 0
Error Severity : 0
Port Lock Info :
Status : off
LockedPorts :
Message Control Info :
Action : suppress-msg
Control-Interval : 5
Max-msg-num : 5
Status : disable
Configuration Operation Info Since Boot Up:
Last Change: 0 day(s), 08:31:10 (5 day(s), 08:41:59 ago)
Last Vlan Change: 0 day(s), 08:27:35 (5 day(s), 08:45:34 ago)
Last Statistic Reset: 5 day(s), 16:56:45 (0 day(s), 00:16:24 ago)
Current Uboot Info :
----------------------------------------------------------------------------------------------------
VU-Boot 2012.04-00002-g6fb1c26 (Apr 26 2017 - 13:37:44) bld=17042617

120
test/units/modules/network/voss/test_voss_command.py Normal file
View file

@ -0,0 +1,120 @@
# (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 json
from ansible.compat.tests.mock import patch
from ansible.modules.network.voss import voss_command
from units.modules.utils import set_module_args
from .voss_module import TestVossModule, load_fixture
class TestVossCommandModule(TestVossModule):
module = voss_command
def setUp(self):
super(TestVossCommandModule, self).setUp()
self.mock_run_commands = patch('ansible.modules.network.voss.voss_command.run_commands')
self.run_commands = self.mock_run_commands.start()
def tearDown(self):
super(TestVossCommandModule, self).tearDown()
self.mock_run_commands.stop()
def load_fixtures(self, commands=None):
def load_from_file(*args, **kwargs):
module, commands = args
output = list()
for item in commands:
try:
obj = json.loads(item['command'])
command = obj['command']
except ValueError:
command = item['command']
filename = str(command).replace(' ', '_')
output.append(load_fixture(filename))
return output
self.run_commands.side_effect = load_from_file
def test_voss_command_simple(self):
set_module_args(dict(commands=['show sys-info']))
result = self.execute_module()
self.assertEqual(len(result['stdout']), 1)
self.assertTrue(result['stdout'][0].startswith('General Info'))
def test_voss_command_multiple(self):
set_module_args(dict(commands=['show sys-info', 'show sys-info']))
result = self.execute_module()
self.assertEqual(len(result['stdout']), 2)
self.assertTrue(result['stdout'][0].startswith('General Info'))
def test_voss_command_wait_for(self):
wait_for = 'result[0] contains "General Info"'
set_module_args(dict(commands=['show sys-info'], wait_for=wait_for))
self.execute_module()
def test_voss_command_wait_for_fails(self):
wait_for = 'result[0] contains "test string"'
set_module_args(dict(commands=['show sys-info'], wait_for=wait_for))
self.execute_module(failed=True)
self.assertEqual(self.run_commands.call_count, 10)
def test_voss_command_retries(self):
wait_for = 'result[0] contains "test string"'
set_module_args(dict(commands=['show sys-info'], wait_for=wait_for, retries=2))
self.execute_module(failed=True)
self.assertEqual(self.run_commands.call_count, 2)
def test_voss_command_match_any(self):
wait_for = ['result[0] contains "General Info"',
'result[0] contains "test string"']
set_module_args(dict(commands=['show sys-info'], wait_for=wait_for, match='any'))
self.execute_module()
def test_voss_command_match_all(self):
wait_for = ['result[0] contains "General Info"',
'result[0] contains "Chassis Info"']
set_module_args(dict(commands=['show sys-info'], wait_for=wait_for, match='all'))
self.execute_module()
def test_voss_command_match_all_failure(self):
wait_for = ['result[0] contains "General Info"',
'result[0] contains "test string"']
commands = ['show sys-info', 'show sys-info']
set_module_args(dict(commands=commands, wait_for=wait_for, match='all'))
self.execute_module(failed=True)
def test_voss_command_configure_error(self):
commands = ['configure terminal']
set_module_args({
'commands': commands,
'_ansible_check_mode': True,
})
result = self.execute_module(failed=True)
self.assertEqual(
result['msg'],
'voss_command does not support running config mode commands. Please use voss_config instead'
)

88
test/units/modules/network/voss/voss_module.py Normal file
View file

@ -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/>.
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import os
import json
from units.modules.utils import AnsibleExitJson, AnsibleFailJson, ModuleTestCase
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 TestVossModule(ModuleTestCase):
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 is not None:
if sort:
self.assertEqual(sorted(commands), sorted(result['commands']), result['commands'])
else:
self.assertEqual(commands, result['commands'], result['commands'])
return result
def failed(self):
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):
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