Add Support for Mellanox switches: first module: mlnxos_command (#33121)

* Add Support for Mellanox switches: first module: mlnxos_command

Signed-off-by: Samer Deeb <samerd@mellanox.com>

* Add cliconf support for mlnxos

Signed-off-by: Samer Deeb <samerd@mellanox.com>

* 1- Fix short description, 2- remove waitfor

Signed-off-by: Samer Deeb <samerd@mellanox.com>

* remove usage of check_args

Signed-off-by: Samer Deeb <samerd@mellanox.com>
This commit is contained in:
Samer Deeb 2017-11-27 12:55:08 -08:00 committed by John R Barker
parent bb38e34fc5
commit cbf28c20cb
13 changed files with 869 additions and 1 deletions

View file

@ -30,6 +30,7 @@ The following is a list of module_utils files and a general description. The mod
- junos.py - Definitions and helper functions for modules that manage Junos networking devices
- known_hosts.py - utilities for working with known_hosts file
- manageiq.py - Functions and utilities for modules that work with ManageIQ platform and its resources.
- mlnxos.py - Definitions and helper functions for modules that manage Mellanox MLNX-OS networking devices
- mysql.py - Allows modules to connect to a MySQL instance
- netapp.py - Functions and utilities for modules that work with the NetApp storage platforms.
- netcfg.py - Configuration utility functions for use by networking modules

View file

@ -1293,7 +1293,7 @@ MERGE_MULTIPLE_CLI_TAGS:
version_added: "2.3"
NETWORK_GROUP_MODULES:
name: Network module families
default: [eos, nxos, ios, iosxr, junos, enos, ce, vyos, sros, dellos9, dellos10, dellos6, asa, aruba, aireos, bigip, ironware]
default: [eos, nxos, ios, iosxr, junos, enos, ce, vyos, sros, dellos9, dellos10, dellos6, asa, aruba, aireos, bigip, ironware, mlnxos]
description: 'TODO: write it'
env: [{name: NETWORK_GROUP_MODULES}]
ini:

View file

@ -0,0 +1,87 @@
# -*- coding: utf-8 -*-
#
# (c) 2017, Ansible by Red Hat, inc
#
# This file is part of Ansible by Red Hat
#
# 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 ansible.module_utils._text import to_text
from ansible.module_utils.basic import env_fallback
from ansible.module_utils.connection import Connection
from ansible.module_utils.network_common import to_list, EntityCollection
_DEVICE_CONFIGS = {}
_CONNECTION = None
mlnxos_provider_spec = {
'host': dict(),
'port': dict(type='int'),
'username': dict(fallback=(env_fallback,
['ANSIBLE_NET_USERNAME'])),
'password': dict(fallback=(env_fallback,
['ANSIBLE_NET_PASSWORD']), no_log=True),
'ssh_keyfile': dict(fallback=(env_fallback,
['ANSIBLE_NET_SSH_KEYFILE']), type='path'),
'authorize': dict(fallback=(env_fallback,
['ANSIBLE_NET_AUTHORIZE']), type='bool'),
'auth_pass': dict(fallback=(env_fallback,
['ANSIBLE_NET_AUTH_PASS']), no_log=True),
'timeout': dict(type='int')
}
mlnxos_argument_spec = {
'provider': dict(type='dict', options=mlnxos_provider_spec),
}
command_spec = {
'command': dict(key=True),
'prompt': dict(),
'answer': dict()
}
def get_provider_argspec():
return mlnxos_provider_spec
def get_connection(module):
global _CONNECTION
if _CONNECTION:
return _CONNECTION
_CONNECTION = Connection(module._socket_path)
return _CONNECTION
def to_commands(module, commands):
if not isinstance(commands, list):
raise AssertionError('argument must be of type <list>')
transform = EntityCollection(module, command_spec)
commands = transform(commands)
return commands
def run_commands(module, commands, check_rc=True):
connection = get_connection(module)
commands = to_commands(module, to_list(commands))
responses = list()
for cmd in commands:
out = connection.get(**cmd)
responses.append(to_text(out, errors='surrogate_then_replace'))
return responses

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': 'network'}
DOCUMENTATION = """
---
module: mlnxos_command
extends_documentation_fragment: mlnxos
version_added: "2.5"
author: "Samer Deeb (@samerd)"
short_description: Run commands on remote devices running Mellanox MLNX-OS
description:
- >-
Sends arbitrary commands to an mlnxos node 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(mlnxos_config) to configure Mellanox MLNX-OS devices.
notes:
- tested on Mellanox OS 3.6.4000
options:
commands:
description:
- >-
List of commands to send to the remote mlnxos device over the
configured provider. 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.
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.
required: false
default: null
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.
required: false
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.
required: false
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.
required: false
default: 1
"""
EXAMPLES = """
tasks:
- name: run show version on remote devices
mlnxos_command:
commands: show version
- name: run show version and check to see if output contains MLNXOS
mlnxos_command:
commands: show version
wait_for: result[0] contains MLNXOS
- name: run multiple commands on remote nodes
mlnxos_command:
commands:
- show version
- show interfaces
- name: run multiple commands and evaluate the output
mlnxos_command:
commands:
- show version
- show interfaces
wait_for:
- result[0] contains MLNXOS
- result[1] contains mgmt1
"""
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 time
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.netcli import Conditional
from ansible.module_utils.network_common import ComplexList
from ansible.module_utils.six import string_types
from ansible.module_utils.mlnxos import mlnxos_argument_spec, run_commands
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):
if module.check_mode and not item['command'].startswith('show'):
warnings.append(
'only show commands are supported when using check mode, not '
'executing `%s`' % item['command']
)
commands.remove(item)
elif item['command'].startswith('conf'):
module.fail_json(
msg='mlnxos_command does not support running config mode '
'commands. Please use mlnxos_config instead'
)
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')
)
argument_spec.update(mlnxos_argument_spec)
module = AnsibleModule(argument_spec=argument_spec,
supports_check_mode=True)
result = {'changed': False}
warnings = list()
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 be 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()

View file

@ -0,0 +1,87 @@
#
# (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 copy
import sys
from ansible import constants as C
from ansible.module_utils.network_common import load_provider
from ansible.plugins.action.normal import ActionModule as _ActionModule
from ansible.utils.display import Display
from ansible.module_utils.mlnxos import mlnxos_provider_spec
try:
from __main__ import display
except ImportError:
display = Display()
class ActionModule(_ActionModule):
def run(self, tmp=None, task_vars=None):
if self._play_context.connection != 'local':
return dict(
failed=True,
msg='invalid connection specified, expected connection=local, '
'got %s' % self._play_context.connection
)
provider = load_provider(mlnxos_provider_spec, self._task.args)
pc = copy.deepcopy(self._play_context)
pc.connection = 'network_cli'
pc.network_os = 'mlnxos'
pc.remote_addr = provider['host'] or self._play_context.remote_addr
pc.port = int(provider['port'] or self._play_context.port or 22)
pc.remote_user = provider['username'] or \
self._play_context.connection_user
pc.password = provider['password'] or self._play_context.password
pc.private_key_file = provider['ssh_keyfile'] or \
self._play_context.private_key_file
pc.timeout = int(provider['timeout'] or C.PERSISTENT_COMMAND_TIMEOUT)
pc.become = provider['authorize'] or False
pc.become_pass = provider['auth_pass']
display.vvv('using connection plugin %s' %
pc.connection, pc.remote_addr)
connection = self._shared_loader_obj.connection_loader.get(
'persistent', pc, sys.stdin)
socket_path = connection.run()
display.vvvv('socket_path: %s' % socket_path, pc.remote_addr)
if not socket_path:
return {'failed': True,
'msg': 'unable to open shell. Please see: '
'https://docs.ansible.com/ansible/'
'network_debug_troubleshooting.html#'
'unable-to-open-shell'}
task_vars['ansible_socket'] = socket_path
if self._play_context.become_method == 'enable':
self._play_context.become = False
self._play_context.become_method = None
result = super(ActionModule, self).run(tmp, task_vars)
return result

View file

@ -0,0 +1,70 @@
#
# (c) 2017 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 json
from itertools import chain
from ansible.module_utils._text import to_text
from ansible.module_utils.network_common import to_list
from ansible.plugins.cliconf import CliconfBase, enable_mode
class Cliconf(CliconfBase):
def get_device_info(self):
device_info = {}
reply = self.get(b'show version | json-print')
data = json.loads(reply)
device_info['network_os'] = data['Product name']
device_info['network_os_version'] = data['Version summary']
device_info['network_os_model'] = data['Product model']
reply = self.get(b'show version | include Hostname')
data = to_text(reply, errors='surrogate_or_strict').strip()
hostname = data.split(':')[1]
hostname = hostname.strip()
device_info['network_os_hostname'] = hostname
return device_info
@enable_mode
def get_config(self, source='running'):
if source not in ('running',):
return self.invalid_params("fetching configuration from %s is not supported" % source)
cmd = b'show running-config'
return self.send_command(cmd)
@enable_mode
def edit_config(self, command):
for cmd in chain([b'configure terminal'], to_list(command), [b'exit']):
self.send_command(cmd)
def get(self, *args, **kwargs):
return self.send_command(*args, **kwargs)
def get_capabilities(self):
result = {}
result['rpc'] = self.get_base_rpc()
result['network_api'] = 'cliconf'
result['device_info'] = self.get_device_info()
return json.dumps(result)

View file

@ -0,0 +1,80 @@
#
# (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 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"(?P<prompt>(.*)( > | # )\Z)"),
]
terminal_stderr_re = [
re.compile(br"\A%|\r\n%|\n%"),
]
init_commands = [b'no cli session paging enable', ]
def on_open_shell(self):
try:
for cmd in self.init_commands:
self._exec_cli_command(cmd)
except AnsibleConnectionFailure:
raise AnsibleConnectionFailure('unable to set terminal parameters')
def on_authorize(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]?password: $",
errors='surrogate_or_strict')
cmd[u'answer'] = passwd
try:
self._exec_cli_command(to_bytes(json.dumps(cmd),
errors='surrogate_or_strict'))
except AnsibleConnectionFailure:
raise AnsibleConnectionFailure(
'unable to elevate privilege to enable mode')
def on_deauthorize(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 b'(config' in prompt:
self._exec_cli_command(b'exit')
self._exec_cli_command(b'disable')
elif prompt.endswith(b'#'):
self._exec_cli_command(b'disable')

View file

@ -0,0 +1,80 @@
#
# 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:
provider:
description:
- A dict object containing connection details.
default: null
suboptions:
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.
default: 22
username:
description:
- Configures the username to use to authenticate the connection to
the remote device. This value is used to authenticate
the SSH session. If the value is not specified in the task, the
value of environment variable C(ANSIBLE_NET_USERNAME) will be used instead.
password:
description:
- Specifies the password to use to authenticate the connection to
the remote device. This value is used to authenticate
the SSH session. If the value is not specified in the task, the
value of environment variable C(ANSIBLE_NET_PASSWORD) will be used instead.
default: null
timeout:
description:
- Specifies the timeout in seconds for communicating with the network device
for either connecting or sending commands. If the timeout is
exceeded before the operation is completed, the module will error.
default: 10
ssh_keyfile:
description:
- Specifies the SSH key to use to authenticate the connection to
the remote device. This value is the path to the
key used to authenticate the SSH session. If the value is not specified
in the task, the value of environment variable C(ANSIBLE_NET_SSH_KEYFILE)
will be used instead.
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.
default: no
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.
default: none
"""

View file

@ -0,0 +1,19 @@
Product name: MLNX-OS
Product release: 3.6.5000
Build ID: #1-dev
Build date: 2017-11-10 18:14:32
Target arch: x86_64
Target hw: x86_64
Built by: jenkins@cc45f26cd083
Version summary: X86_64 3.6.5000 2017-11-10 18:14:32 x86_64
Product model: x86onie
Host ID: 248A073D505C
System serial num: \"MT1632X00205\"
System UUID: 0b19d6d0-5eca-11e6-8000-7cfe90fadc40
Uptime: 1d 16h 31m 43.856s
CPU load averages: 0.06 / 0.12 / 0.13
Number of CPUs: 4
System memory: 2597 MB used / 5213 MB free / 7810 MB total
Swap: 0 MB used / 0 MB free / 0 MB total

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 json
import os
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 TestMlnxosModule(ModuleTestCase):
def execute_module(self, failed=False, changed=False, commands=None, inputs=None, sort=True, defaults=False, transport='cli'):
self.load_fixtures(commands, transport=transport)
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, transport='cli'):
pass

View file

@ -0,0 +1,114 @@
# (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.mlnxos import mlnxos_command
from units.modules.utils import set_module_args
from .mlnxos_module import TestMlnxosModule, load_fixture
class TestMlnxosCommandModule(TestMlnxosModule):
module = mlnxos_command
def setUp(self):
super(TestMlnxosCommandModule, self).setUp()
self.mock_run_commands = patch(
'ansible.modules.network.mlnxos.mlnxos_command.run_commands')
self.run_commands = self.mock_run_commands.start()
def tearDown(self):
super(TestMlnxosCommandModule, self).tearDown()
self.mock_run_commands.stop()
def load_fixtures(self, commands=None, transport='cli'):
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(' ', '_')
filename = 'mlnxos_command_%s.txt' % filename
output.append(load_fixture(filename))
return output
self.run_commands.side_effect = load_from_file
def test_mlnxos_command_simple(self):
set_module_args(dict(commands=['show version']))
result = self.execute_module()
self.assertEqual(len(result['stdout']), 1)
self.assertTrue(result['stdout'][0].startswith('Product name'))
def test_mlnxos_command_multiple(self):
set_module_args(dict(commands=['show version', 'show version']))
result = self.execute_module()
self.assertEqual(len(result['stdout']), 2)
self.assertTrue(result['stdout'][0].startswith('Product name'))
def test_mlnxos_command_wait_for(self):
wait_for = 'result[0] contains "MLNX"'
set_module_args(dict(commands=['show version'], wait_for=wait_for))
self.execute_module()
def test_mlnxos_command_wait_for_fails(self):
wait_for = 'result[0] contains "test string"'
set_module_args(dict(commands=['show version'], wait_for=wait_for))
self.execute_module(failed=True)
self.assertEqual(self.run_commands.call_count, 10)
def test_mlnxos_command_retries(self):
wait_for = 'result[0] contains "test string"'
set_module_args(
dict(commands=['show version'], wait_for=wait_for, retries=2))
self.execute_module(failed=True)
self.assertEqual(self.run_commands.call_count, 2)
def test_mlnxos_command_match_any(self):
wait_for = ['result[0] contains "MLNX"',
'result[0] contains "test string"']
set_module_args(dict(
commands=['show version'],
wait_for=wait_for,
match='any'))
self.execute_module()
def test_mlnxos_command_match_all(self):
wait_for = ['result[0] contains "MLNX"',
'result[0] contains "Version summary"']
set_module_args(
dict(commands=['show version'], wait_for=wait_for, match='all'))
self.execute_module()
def test_mlnxos_command_match_all_failure(self):
wait_for = ['result[0] contains "MLNX"',
'result[0] contains "test string"']
commands = ['show version', 'show version']
set_module_args(
dict(commands=commands, wait_for=wait_for, match='all'))
self.execute_module(failed=True)