updates eos shared modules (#20738)

* eos module now uses network_cli connection plugin
* adds unit tests for eos module
* eapi support now provided by eapi module
* updates doc fragment for eapi common properties

lib/ansible/module_utils/eapi.py

@@ -0,0 +1,259 @@
+# 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 time
+
+from ansible.module_utils.basic import env_fallback
+from ansible.module_utils.urls import fetch_url
+from ansible.module_utils.network_common import to_list
+
+_DEVICE_CONNECTION = None
+_DEVICE_CONFIGS = {}
+_SESSION_SUPPORT = None
+
+eapi_argument_spec = dict(
+    host=dict(),
+    port=dict(type='int'),
+
+    url_username=dict(fallback=(env_fallback, ['ANSIBLE_NET_USERNAME']), aliases=('username',)),
+    url_password=dict(fallback=(env_fallback, ['ANSIBLE_NET_PASSWORD']), aliases=('password',), no_log=True),
+
+    authorize=dict(default=False, fallback=(env_fallback, ['ANSIBLE_NET_AUTHORIZE']), type='bool'),
+    auth_pass=dict(no_log=True, fallback=(env_fallback, ['ANSIBLE_NET_AUTH_PASS'])),
+
+    provider=dict(type='dict'),
+
+    # deprecated in Ansible 2.3
+    transport=dict(),
+
+    use_ssl=dict(type='bool', default=True),
+    validate_certs=dict(type='bool', default=True),
+    timeout=dict(default=10, type='int')
+)
+
+def check_args(module):
+    for key in ('host', 'username', 'password'):
+        if not module.params[key]:
+            module.fail_json(msg='missing required argument %s' % key)
+
+    if module.params['transport'] == 'cli':
+        module.fail_json(msg='transport: cli is no longer supported, use '
+                             'connection=network_cli instead')
+
+
+class Eapi:
+
+    def __init__(self, module):
+        self._module = module
+        self._enable = None
+
+        host = module.params['host']
+        port = module.params['port']
+
+        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:
+            module.fail_json(**headers)
+
+        try:
+            data = response.read()
+            response = self._module.from_json(data)
+        except ValueError:
+            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 connection(module):
+    global _DEVICE_CONNECTION
+    if not _DEVICE_CONNECTION:
+        _DEVICE_CONNECTION = Eapi(module)
+    return _DEVICE_CONNECTION
+
+is_json = lambda x: str(x).endswith('| json')
+is_text = lambda x: not str(x).endswith('| json')
+
+def run_commands(module, commands):
+    """Runs list of commands on remote device and returns results
+    """
+    output = None
+    queue = list()
+    responses = list()
+
+    conn = connection(module)
+
+    def _send(commands, output):
+        response = conn.send_request(commands, output=output)
+        if 'error' in response:
+            err = response['error']
+            module.fail_json(msg=err['message'], code=err['code'])
+        return response['result']
+
+    for item in to_list(commands):
+        if all((output == 'json', is_text(item))) or all((output =='text', is_json(item))):
+            responses.extend(_send(queue, output))
+            queue = list()
+
+        if is_json(item):
+            output = 'json'
+        else:
+            output = 'text'
+
+        queue.append(item)
+
+    if queue:
+        responses.extend(_send(queue, output))
+
+    for index, item in enumerate(commands):
+        if is_text(item):
+            responses[index] = responses[index]['output'].strip()
+
+    return responses
+
+def get_config(module, 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 = connection(module)
+        out = conn.send_request(cmd)
+        cfg = str(out['result'][0]['output']).strip()
+        _DEVICE_CONFIGS[cmd] = cfg
+        return cfg
+
+def supports_sessions(module):
+    global _SESSION_SUPPORT
+    if _SESSION_SUPPORT is not None:
+        return _SESSION_SUPPORT
+
+    conn = connection(module)
+    response = conn.send_request(['show configuration sessions'])
+    _SESSION_SUPPORT = 'error' not in response
+
+    return _SESSION_SUPPORT
+
+def configure(module, commands):
+    """Sends the ordered set of commands to the device
+    """
+    cmds = ['configure terminal']
+    cmds.extend(commands)
+
+    conn = connection(module)
+
+    responses = conn.send_request(commands)
+    if 'error' in response:
+        err = response['error']
+        module.fail_json(msg=err['message'], code=err['code'])
+
+    return responses[1:]
+
+def load_config(module, 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(module):
+        return configure(module, commands)
+
+    conn = connection(module)
+
+    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 = conn.send_request(commands)
+    if 'error' in response:
+        commands = ['configure session %s' % session, 'abort']
+        conn.send_request(commands)
+        err = response['error']
+        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 = conn.send_request(commands, output='text')
+    diff = response['result'][1]['output']
+    if diff:
+        result['diff'] = diff
+
+    return result

lib/ansible/module_utils/eos.py

@@ -4,7 +4,7 @@
 # still belong to the author of the module, and may assign their own license
 # to the complete work.
 #
-# Copyright (c) 2015 Peter Sprygada, <psprygada@ansible.com>
+# (c) 2016 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,314 +25,114 @@
 # 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.basic import json, get_exception
+from ansible.module_utils.network_common import to_list
-from ansible.module_utils.network import ModuleStub, NetworkError, NetworkModule
-from ansible.module_utils.network import add_argument, register_transport, to_list
-from ansible.module_utils.netcli import Command
-from ansible.module_utils.shell import CliBase
-from ansible.module_utils.urls import fetch_url, url_argument_spec
-from ansible.module_utils._text import to_native
 
-EAPI_FORMATS = ['json', 'text']
+_DEVICE_CONFIGS = {}
 
-add_argument('use_ssl', dict(default=True, type='bool'))
+def get_config(module, flags=[]):
-add_argument('validate_certs', dict(default=True, type='bool'))
+    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
 
-class EosConfigMixin(object):
+def check_authorization(module):
+    for cmd in ['show clock', 'prompt()']:
+        rc, out, err = module.exec_command(cmd)
+    return out.endswith('#')
 
-    ### Config methods ###
+def supports_sessions(module):
+    rc, out, err = module.exec_command('show configuration sessions')
+    return rc == 0
 
-    def configure(self, commands, **kwargs):
+def run_commands(module, commands):
-        cmds = ['configure terminal']
+    """Run list of commands on remote device and return results
-        cmds.extend(to_list(commands))
+    """
-        cmds.append('end')
+    responses = list()
-        responses = self.execute(cmds)
-        return responses[1:-1]
 
-    def get_config(self, include_defaults=False, **kwargs):
+    for cmd in to_list(commands):
-        cmd = 'show running-config'
+        rc, out, err = module.exec_command(cmd)
-        if include_defaults:
-            cmd += ' all'
-        return self.execute([cmd])[0]
 
-    def load_config(self, config, commit=False, replace=False):
+        if rc != 0:
-        if self.supports_sessions():
+            module.fail_json(msg=err)
-            return self.load_config_session(config, commit, replace)
-        else:
-            return self.configure(config)
-
-    def load_config_session(self, config, commit=False, replace=False):
-        """ Loads the configuration into the remote device
-        """
-        session = 'ansible_%s' % int(time.time())
-        commands = ['configure session %s' % session]
-
-        if replace:
-            commands.append('rollback clean-config')
-
-        commands.extend(config)
-
-        if commands[-1] != 'end':
-            commands.append('end')
 
         try:
-            self.execute(commands)
+           out = module.from_json(out)
-            diff = self.diff_config(session)
-            if commit:
-                self.commit_config(session)
-            else:
-                self.execute(['no configure session %s' % session])
-        except NetworkError:
-            exc = get_exception()
-            if 'timeout trying to send command' in to_native(exc):
-                # try to get control back and get out of config mode
-                if isinstance(self, Cli):
-                    self.execute(['\x03', 'end'])
-            self.abort_config(session)
-            diff = None
-            raise
-
-        return diff
-
-    def save_config(self):
-        self.execute(['copy running-config startup-config'])
-
-    def diff_config(self, session):
-        commands = ['configure session %s' % session,
-                    'show session-config diffs',
-                    'end']
-
-        if isinstance(self, Eapi):
-            response = self.execute(commands, output='text')
-            response[-2] = response[-2].get('output').strip()
-        else:
-            response = self.execute(commands)
-
-        return response[-2]
-
-    def commit_config(self, session):
-        commands = ['configure session %s' % session, 'commit']
-        self.execute(commands)
-
-    def abort_config(self, session):
-        commands = ['configure session %s' % session, 'abort']
-        self.execute(commands)
-
-    def supports_sessions(self):
-        try:
-            if isinstance(self, Eapi):
-                self.execute(['show configuration sessions'], output='text')
-            else:
-                self.execute('show configuration sessions')
-            return True
-        except NetworkError:
-            return False
-
-
-
-class Eapi(EosConfigMixin):
-
-    def __init__(self):
-        self.url = None
-        self.url_args = ModuleStub(url_argument_spec(), self._error)
-        self.enable = None
-        self.default_output = 'json'
-        self._connected = False
-
-    def _error(self, msg):
-        raise NetworkError(msg, url=self.url)
-
-    def _get_body(self, commands, output, reqid=None):
-        """Create a valid eAPI JSON-RPC request message
-        """
-        if output not in EAPI_FORMATS:
-            msg = 'invalid format, received %s, expected one of %s' % \
-                    (output, ', '.join(EAPI_FORMATS))
-            self._error(msg=msg)
-
-        params = dict(version=1, cmds=commands, format=output)
-        return dict(jsonrpc='2.0', id=reqid, method='runCmds', params=params)
-
-    def connect(self, params, **kwargs):
-        host = params['host']
-        port = params['port']
-
-        # sets the module_utils/urls.py req parameters
-        self.url_args.params['url_username'] = params['username']
-        self.url_args.params['url_password'] = params['password']
-        self.url_args.params['validate_certs'] = params['validate_certs']
-        self.url_args.params['timeout'] = params['timeout']
-
-        if 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)
-        self._connected = True
-
-    def disconnect(self, **kwargs):
-        self.url = None
-        self._connected = False
-
-    def authorize(self, params, **kwargs):
-        if params.get('auth_pass'):
-            passwd = params['auth_pass']
-            self.enable = dict(cmd='enable', input=passwd)
-        else:
-            self.enable = 'enable'
-
-    ### Command methods ###
-
-    def execute(self, commands, output='json', **kwargs):
-        """Send commands to the device.
-        """
-        if self.url is None:
-            raise NetworkError('Not connected to endpoint.')
-
-        if self.enable is not None:
-            commands.insert(0, self.enable)
-
-        body = self._get_body(commands, output)
-        data = json.dumps(body)
-
-        headers = {'Content-Type': 'application/json-rpc'}
-        timeout = self.url_args.params['timeout']
-
-        response, headers = fetch_url(
-            self.url_args, self.url, data=data, headers=headers,
-            method='POST', timeout=timeout
-        )
-
-        if headers['status'] != 200:
-            raise NetworkError(**headers)
-
-        try:
-            response = json.loads(response.read())
         except ValueError:
-            raise NetworkError('unable to load response from device')
+            out = str(out).strip()
 
-        if 'error' in response:
+        responses.append(out)
-            err = response['error']
+    return responses
-            raise NetworkError(
-                msg=err['message'], code=err['code'], data=err['data'],
-                commands=commands
-            )
 
-        if self.enable:
+def configure(module, commands):
-            response['result'].pop(0)
+    """Sends configuration commands to the remote device
+    """
+    if not check_authorization(module):
+        module.fail_json(msg='configuration operations require privilege escalation')
 
-        return response['result']
+    rc, out, err = module.exec_command('configure')
+    if rc != 0:
+        module.fail_json(msg='unable to enter configuration mode', output=err)
 
-    def run_commands(self, commands, **kwargs):
+    for cmd in to_list(commands):
-        output = None
+        if cmd == 'end':
-        cmds = list()
+            continue
-        responses = list()
+        rc, out, err = module.exec_command(cmd)
+        if rc != 0:
+            module.fail_json(msg=err)
 
-        for cmd in commands:
+    module.exec_command('end')
-            if output and output != cmd.output:
-                responses.extend(self.execute(cmds, output=output))
-                cmds = list()
 
-            output = cmd.output
+def load_config(module, commands, commit=False, replace=False):
-            cmds.append(str(cmd))
+    """Loads the config commands onto the remote device
+    """
+    if not check_authorization(module):
+        module.fail_json(msg='configuration operations require privilege escalation')
 
-        if cmds:
+    if not supports_sessions(module):
-            responses.extend(self.execute(cmds, output=output))
+        return configure(commands)
 
-        for index, cmd in enumerate(commands):
+    session = 'ansible_%s' % int(time.time())
-            if cmd.output == 'text':
-                responses[index] = responses[index].get('output')
 
-        return responses
+    result = {'session': session}
 
-    ### Config methods ###
+    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 get_config(self, include_defaults=False):
+    if replace:
-        cmd = 'show running-config'
+        module.exec_command('rollback clean-config', check_rc=True)
-        if include_defaults:
-            cmd += ' all'
-        return self.execute([cmd], output='text')[0]['output']
 
-Eapi = register_transport('eapi')(Eapi)
+    failed = False
+    for command in to_list(commands):
+        if command == 'end':
+            pass
 
+        rc, out, err = module.exec_command(command)
+        if rc != 0:
+            failed = True
+            break
 
-class Cli(EosConfigMixin, CliBase):
+    rc, out, err = module.exec_command('show session-config diffs')
+    if rc == 0:
+        result['diff'] = out
 
-    CLI_PROMPTS_RE = [
+    if failed:
-        re.compile(r"[\r\n]?[\w+\-\.:\/\[\]]+(?:\([^\)]+\)){,3}(?:>|#) ?$"),
+        module.exec_command('abort')
-        re.compile(r"\[\w+\@[\w\-\.]+(?: [^\]])\] ?[>#\$] ?$")
+        module.fail_json(msg=err, commands=commands)
-    ]
+    elif commit:
+        module.exec_command('commit')
+    else:
+        module.exec_command('abort')
 
-    CLI_ERRORS_RE = [
+    return result
-        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")
-    ]
-
-    NET_PASSWD_RE = re.compile(r"[\r\n]?password: $", re.I)
-
-    def connect(self, params, **kwargs):
-        super(Cli, self).connect(params, kickstart=False, **kwargs)
-        self.shell.send('terminal length 0')
-
-    def authorize(self, params, **kwargs):
-        passwd = params['auth_pass']
-        if passwd:
-            self.execute(Command('enable', prompt=self.NET_PASSWD_RE, response=passwd))
-        else:
-            self.execute('enable')
-
-    ### Command methods ###
-
-    def run_commands(self, commands):
-        cmds = list(prepare_commands(commands))
-        responses = self.execute(cmds)
-        for index, cmd in enumerate(commands):
-            if cmd.output == 'json':
-                try:
-                    responses[index] = json.loads(responses[index])
-                except ValueError:
-                    raise NetworkError(
-                        msg='unable to load response from device',
-                        response=responses[index],
-                        responses=responses
-                    )
-        return responses
-
-Cli = register_transport('cli', default=True)(Cli)
-
-
-def prepare_config(commands):
-    commands = to_list(commands)
-    commands.insert(0, 'configure terminal')
-    commands.append('end')
-    return commands
-
-
-def prepare_commands(commands):
-    jsonify = lambda x: '%s | json' % x
-    for item in to_list(commands):
-        if item.output == 'json':
-            cmd = jsonify(item)
-        elif item.command.endswith('| json'):
-            item.output = 'json'
-            cmd = str(item)
-        else:
-            cmd = str(item)
-        yield cmd

lib/ansible/utils/module_docs_fragments/eapi.py

@@ -0,0 +1,97 @@
+#
+# (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.  The port value will default to the appropriate transport
+        common port if none is provided in the task.  (http=80, https=443).
+    required: false
+    default: null
+  url_username:
+    description:
+      - Configures the username to use to authenticate the connection to
+        the remote device.  This value is used to authenticate
+        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
+    default: null
+    aliases: ['username']
+  url_password:
+    description:
+      - Specifies the password to use to authenticate the connection to
+        the remote device.  This is a common argument used for the I(eapi)
+        transport. 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
+    aliases: ['password']
+  timeout:
+    description:
+      - Specifies the timeout in seconds for communicating with the network device
+        for either connecting or sending commands.  If the timeout is
+        exceeded before the operation is completed, the module will error.
+    require: false
+    default: 10
+  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: no
+    choices: ['true', 'false']
+  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: none
+  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: yes
+    choices: ['true', 'false']
+  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
+"""

test/units/module_utils/test_eos.py

@@ -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 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'})
+