Repository: napalm-automation/napalm
Pull request #673 (2018-03-13 04:17:56)

Unify netmiko argument processing across napalm drivers
.

coveralls: 
[![Coverage Status](https://coveralls.io/builds/15942136/badge)](https://coveralls.io/builds/15942136)

Coverage increased (+0.04%) to 78.898% when pulling **0535b678b16b29c2bf1dd5bee8290562ddb31ff3 on ktbyers:netmiko_restructure** into **602e25fc7f70c231d216f5b7ca7957b2174317ba on napalm-automation:develop**.

dbarrosop: YES, PLEASE! :D

Patch:
diff --git a/napalm/base/netmiko_helpers.py b/napalm/base/netmiko_helpers.py
new file mode 100644
index 00000000..a75922b5
--- /dev/null
+++ b/napalm/base/netmiko_helpers.py
@@ -0,0 +1,47 @@
+# The contents of this file are licensed under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+from __future__ import unicode_literals
+import inspect
+from netmiko import BaseConnection
+
+
+def netmiko_args(optional_args):
+    """Check for Netmiko arguments that were passed in as NAPALM optional arguments.
+
+    Return a dictionary of these optional args  that will be passed into the Netmiko
+    ConnectHandler call.
+    """
+    netmiko_args, _, _, netmiko_defaults = inspect.getargspec(BaseConnection.__init__)
+
+    check_self = netmiko_args.pop(0)
+    if check_self != 'self':
+        raise ValueError("Error processing Netmiko arguments")
+
+    netmiko_argument_map = dict(zip(netmiko_args, netmiko_defaults))
+
+    # Netmiko arguments that are integrated into NAPALM already
+    netmiko_filter = ['ip', 'host', 'username', 'password', 'device_type', 'timeout']
+
+    # Filter out all of the arguments that are integrated into NAPALM
+    for k in netmiko_filter:
+        netmiko_argument_map.pop(k)
+
+    # Check if any of these arguments were passed in as NAPALM optional_args
+    netmiko_optional_args = {}
+    for k, v in netmiko_argument_map.items():
+        try:
+            netmiko_optional_args[k] = optional_args[k]
+        except KeyError:
+            pass
+
+    # Return these arguments for use with establishing Netmiko SSH connection
+    return netmiko_optional_args
diff --git a/napalm/ios/ios.py b/napalm/ios/ios.py
index d4c5bb5c..ba990a74 100644
--- a/napalm/ios/ios.py
+++ b/napalm/ios/ios.py
@@ -27,6 +27,7 @@ import copy
 
 from netmiko import ConnectHandler, FileTransfer, InLineTransfer
 from napalm.base.base import NetworkDriver
+from napalm.base.netmiko_helpers import netmiko_args
 from napalm.base.exceptions import ReplaceConfigException, MergeConfigException, \
             ConnectionClosedException, CommandErrorException
 
@@ -97,36 +98,14 @@ class IOSDriver(NetworkDriver):
         # Control automatic toggling of 'file prompt quiet' for file operations
         self.auto_file_prompt = optional_args.get('auto_file_prompt', True)
 
-        # Netmiko possible arguments
-        netmiko_argument_map = {
-            'port': None,
-            'secret': '',
-            'verbose': False,
-            'keepalive': 30,
-            'global_delay_factor': 1,
-            'use_keys': False,
-            'key_file': None,
-            'ssh_strict': False,
-            'system_host_keys': False,
-            'alt_host_keys': False,
-            'alt_key_file': '',
-            'ssh_config_file': None,
-            'allow_agent': False,
-        }
-
-        # Build dict of any optional Netmiko args
-        self.netmiko_optional_args = {}
-        for k, v in netmiko_argument_map.items():
-            try:
-                self.netmiko_optional_args[k] = optional_args[k]
-            except KeyError:
-                pass
+        self.netmiko_optional_args = netmiko_args(optional_args)
 
+        # Set the default port if not set
         default_port = {
             'ssh': 22,
             'telnet': 23
         }
-        self.port = optional_args.get('port', default_port[self.transport])
+        self.netmiko_optional_args.setdefault('port', default_port[self.transport])
 
         self.device = None
         self.config_replace = False
diff --git a/napalm/iosxr/iosxr.py b/napalm/iosxr/iosxr.py
index 44f717b6..9c79c9ed 100644
--- a/napalm/iosxr/iosxr.py
+++ b/napalm/iosxr/iosxr.py
@@ -26,8 +26,6 @@ from lxml import etree as ETREE
 from netaddr import IPAddress  # needed for traceroute, to check IP version
 from netaddr.core import AddrFormatError
 
-from netmiko import __version__ as netmiko_version
-
 from pyIOSXR import IOSXR
 from pyIOSXR.exceptions import ConnectError
 from pyIOSXR.exceptions import TimeoutError
@@ -35,6 +33,7 @@ from pyIOSXR.exceptions import InvalidInputError
 
 # import NAPALM base
 import napalm.base.helpers
+from napalm.base.netmiko_helpers import netmiko_args
 from napalm.iosxr import constants as C
 from napalm.base.base import NetworkDriver
 from napalm.base.utils import py23_compat
@@ -46,7 +45,6 @@ from napalm.base.utils.py23_compat import text_type
 
 
 class IOSXRDriver(NetworkDriver):
-
     """IOS-XR driver class: inherits NetworkDriver from napalm.base."""
 
     def __init__(self, hostname, username, password, timeout=60, optional_args=None):
@@ -58,35 +56,14 @@ class IOSXRDriver(NetworkDriver):
         self.replace = False
         if optional_args is None:
             optional_args = {}
-        self.port = optional_args.get('port', 22)
         self.lock_on_connect = optional_args.get('config_lock', False)
-        # Netmiko possible arguments
-        netmiko_argument_map = {
-            'keepalive': 30,
-            'verbose': False,
-            'global_delay_factor': 1,
-            'use_keys': False,
-            'key_file': None,
-            'ssh_strict': False,
-            'system_host_keys': False,
-            'alt_host_keys': False,
-            'alt_key_file': '',
-            'ssh_config_file': None
-        }
-        fields = netmiko_version.split('.')
-        fields = [int(x) for x in fields]
-        maj_ver, min_ver, bug_fix = fields
-        if maj_ver >= 2:
-            netmiko_argument_map['allow_agent'] = False
-        elif maj_ver == 1 and min_ver >= 1:
-            netmiko_argument_map['allow_agent'] = False
-        # Build dict of any optional Netmiko args
-        self.netmiko_optional_args = {}
-        for k, v in netmiko_argument_map.items():
-            try:
-                self.netmiko_optional_args[k] = optional_args[k]
-            except KeyError:
-                pass
+
+        self.netmiko_optional_args = netmiko_args(optional_args)
+        try:
+            self.port = self.netmiko_optional_args.pop('port')
+        except KeyError:
+            self.port = 22
+
         self.device = IOSXR(hostname,
                             username,
                             password,
diff --git a/napalm/nxos/nxos.py b/napalm/nxos/nxos.py
index f323ceeb..b5f26237 100644
--- a/napalm/nxos/nxos.py
+++ b/napalm/nxos/nxos.py
@@ -189,12 +189,8 @@ class NXOSDriver(NXOSDriverBase):
         self.replace_file = None
         self.merge_candidate = ''
 
-        if optional_args is None:
-            optional_args = {}
-
         # nxos_protocol is there for backwards compatibility, transport is the preferred method
         self.transport = optional_args.get('transport', optional_args.get('nxos_protocol', 'https'))
-
         if self.transport == 'https':
             self.port = optional_args.get('port', 443)
         elif self.transport == 'http':
diff --git a/napalm/nxos_ssh/nxos_ssh.py b/napalm/nxos_ssh/nxos_ssh.py
index 656bb114..f86e5f4a 100644
--- a/napalm/nxos_ssh/nxos_ssh.py
+++ b/napalm/nxos_ssh/nxos_ssh.py
@@ -35,6 +35,7 @@ from netmiko.ssh_exception import NetMikoTimeoutException
 
 # import NAPALM Base
 import napalm.base.helpers
+from napalm.base.netmiko_helpers import netmiko_args
 from napalm.base.utils import py23_compat
 from napalm.base.exceptions import ConnectionException
 from napalm.base.exceptions import MergeConfigException
@@ -373,41 +374,13 @@ class NXOSSSHDriver(NXOSDriverBase):
         self.username = username
         self.password = password
         self.timeout = timeout
-        self.up = False
         self.replace = True
         self.loaded = False
-        self.fc = None
         self.changed = False
         self.replace_file = None
         self.merge_candidate = ''
-
-        if optional_args is None:
-            optional_args = {}
-
-        # Netmiko possible arguments
-        netmiko_argument_map = {
-            'port': None,
-            'verbose': False,
-            'timeout': self.timeout,
-            'global_delay_factor': 1,
-            'use_keys': False,
-            'key_file': None,
-            'ssh_strict': False,
-            'system_host_keys': False,
-            'alt_host_keys': False,
-            'alt_key_file': '',
-            'ssh_config_file': None,
-            'allow_agent': False
-        }
-
-        # Build dict of any optional Netmiko args
-        self.netmiko_optional_args = {
-            k: optional_args.get(k, v)
-            for k, v in netmiko_argument_map.items()
-        }
-
-        self.port = optional_args.get('port', 22)
-        self.sudo_pwd = optional_args.get('sudo_pwd', self.password)
+        self.netmiko_optional_args = netmiko_args(optional_args)
+        self.device = None
 
     def open(self):
         try:


Test patch:
diff --git a/test/base/test_helpers.py b/test/base/test_helpers.py
index 1f9ab15a..87df4885 100644
--- a/test/base/test_helpers.py
+++ b/test/base/test_helpers.py
@@ -38,6 +38,7 @@ except ImportError:
 
 # NAPALM base
 import napalm.base.helpers
+from napalm.base.netmiko_helpers import netmiko_args
 import napalm.base.exceptions
 from napalm.base.base import NetworkDriver
 from napalm.base.utils.string_parsers import convert_uptime_string_seconds
@@ -440,6 +441,89 @@ class TestBaseHelpers(unittest.TestCase):
                          addl_name_map={"loop": "Loopback"},
                          addl_reverse_map={"Loopback": "lo"}), "lo10")
 
+    def test_netmiko_arguments(self):
+        """Test the netmiko argument processing."""
+        self.assertEqual(netmiko_args(optional_args={}), {})
+
+        test_case = {'secret': 'whatever'}
+        self.assertEqual(netmiko_args(test_case), test_case)
+
+        test_case = {
+            'secret': 'whatever',
+            'use_keys': True,
+        }
+        self.assertEqual(netmiko_args(test_case), test_case)
+
+        test_case = {
+            'secret': 'whatever',
+            'use_keys': True,
+            'ssh_config_file': '~/.ssh/config',
+        }
+        self.assertEqual(netmiko_args(test_case), test_case)
+
+        test_case = {
+            'secret': 'whatever',
+            'transport': 'telnet',
+        }
+        self.assertEqual(netmiko_args(test_case), {'secret': 'whatever'})
+
+        test_case = {
+            'secret': 'whatever',
+            'transport': 'telnet',
+            'port': 8022,
+        }
+        self.assertEqual(netmiko_args(test_case), {'secret': 'whatever', 'port': 8022})
+
+        test_case = {
+            'secret': '',
+            'port': None,
+            'verbose': False,
+            'global_delay_factor': 1,
+            'use_keys': False,
+            'key_file': None,
+            'allow_agent': False,
+            'ssh_strict': False,
+            'system_host_keys': False,
+            'alt_host_keys': False,
+            'alt_key_file': '',
+            'ssh_config_file': None,
+            'session_timeout': 60,
+            'blocking_timeout': 8,
+            'keepalive': 0,
+            'default_enter': None,
+            'response_return': None,
+            'serial_settings': None
+        }
+        self.assertEqual(netmiko_args(test_case), test_case)
+
+        test_case = {
+            'inline_transfer': True,
+            'transport': 'ssh',
+            'secret': '',
+            'port': None,
+            'verbose': False,
+            'global_delay_factor': 1,
+            'use_keys': False,
+            'key_file': None,
+            'allow_agent': False,
+            'ssh_strict': False,
+            'system_host_keys': False,
+            'alt_host_keys': False,
+            'alt_key_file': '',
+            'ssh_config_file': None,
+            'session_timeout': 60,
+            'blocking_timeout': 8,
+            'keepalive': 0,
+            'default_enter': None,
+            'response_return': None,
+            'serial_settings': None
+        }
+        result_dict = {}
+        result_dict.update(test_case)
+        result_dict.pop('inline_transfer')
+        result_dict.pop('transport')
+        self.assertEqual(netmiko_args(test_case), result_dict)
+
 
 class FakeNetworkDriver(NetworkDriver):