Convert NAPALM over to use Netmiko >= 4.0.0 (#1613)

* Netmiko 4.0 prep work and some f-string conversions * Updating pyIOSXR for Netmiko 4.0.0 * IOS-XR Netmiko 4.0 support * Removing debugger statement
napalm-automation · Apr 28, 2022 · 4b8cc85 · 4b8cc85
1 parent d03d241
commit 4b8cc85
Show file tree

Hide file tree

Showing 4 changed files with 48 additions and 59 deletions.
diff --git a/napalm/ios/ios.py b/napalm/ios/ios.py
@@ -416,17 +416,13 @@ def compare_config(self):
         )
 
         if self.config_replace:
-            cmd = "show archive config differences {} {}".format(
-                base_file_full, new_file_full
-            )
-            diff = self.device.send_command_expect(cmd)
+            cmd = f"show archive config differences {base_file_full} {new_file_full}"
+            diff = self.device.send_command(cmd)
             diff = self._normalize_compare_config(diff)
         else:
             # merge
-            cmd = "show archive config incremental-diffs {} ignorecase".format(
-                new_file_full
-            )
-            diff = self.device.send_command_expect(cmd)
+            cmd = f"show archive config incremental-diffs {new_file_full} ignorecase"
+            diff = self.device.send_command(cmd)
             if "error code 5" in diff or "returned error 5" in diff:
                 diff = (
                     "You have encountered the obscure 'error 5' message. This generally "
@@ -435,8 +431,8 @@ def compare_config(self):
             elif "% Invalid" not in diff:
                 diff = self._normalize_merge_diff_incr(diff)
             else:
-                cmd = "more {}".format(new_file_full)
-                diff = self.device.send_command_expect(cmd)
+                cmd = f"more {new_file_full}"
+                diff = self.device.send_command(cmd)
                 diff = self._normalize_merge_diff(diff)
 
         return diff.strip()
@@ -455,8 +451,8 @@ def wrapper(self, *args, **kwargs):
                 else:
                     # check if the command is already in the running-config
                     cmd = "file prompt quiet"
-                    show_cmd = "show running-config | inc {}".format(cmd)
-                    output = self.device.send_command_expect(show_cmd)
+                    show_cmd = f"show running-config | inc {cmd}"
+                    output = self.device.send_command(show_cmd)
                     if cmd in output:
                         self.prompt_quiet_configured = True
                     else:
@@ -483,8 +479,8 @@ def _commit_handler(self, cmd):
         pattern1 = r"[>#{}]\s*$".format(terminating_char)
         # Handle special username removal pattern
         pattern2 = r".*all username.*confirm"
-        patterns = r"(?:{}|{})".format(pattern1, pattern2)
-        output = self.device.send_command_expect(cmd, expect_string=patterns)
+        patterns = rf"(?:{pattern1}|{pattern2})"
+        output = self.device.send_command(cmd, expect_string=patterns)
         loop_count = 50
         new_output = output
         for i in range(loop_count):
@@ -567,12 +563,12 @@ def discard_config(self):
     @_file_prompt_quiet
     def _discard_config(self):
         """Set candidate_cfg to current running-config. Erase the merge_cfg file."""
-        discard_candidate = "copy running-config {}".format(
-            self._gen_full_path(self.candidate_cfg)
+        discard_candidate = (
+            f"copy running-config {self._gen_full_path(self.candidate_cfg)}"
         )
-        discard_merge = "copy null: {}".format(self._gen_full_path(self.merge_cfg))
-        self.device.send_command_expect(discard_candidate)
-        self.device.send_command_expect(discard_merge)
+        discard_merge = f"copy null: {self._gen_full_path(self.merge_cfg)}"
+        self.device.send_command(discard_candidate)
+        self.device.send_command(discard_merge)
 
     def rollback(self):
         """Rollback configuration to filename or to self.rollback_cfg file."""
@@ -685,8 +681,8 @@ def _xfer_file(
 
             if use_scp:
                 cmd = "ip scp server enable"
-                show_cmd = "show running-config | inc {}".format(cmd)
-                output = self.device.send_command_expect(show_cmd)
+                show_cmd = f"show running-config | inc {cmd}"
+                output = self.device.send_command(show_cmd)
                 if cmd not in output:
                     msg = (
                         "SCP file transfers are not enabled. "
@@ -721,8 +717,8 @@ def _gen_full_path(self, filename, file_system=None):
     def _gen_rollback_cfg(self):
         """Save a configuration that can be used for rollback."""
         cfg_file = self._gen_full_path(self.rollback_cfg)
-        cmd = "copy running-config {}".format(cfg_file)
-        self.device.send_command_expect(cmd)
+        cmd = f"copy running-config {cfg_file}"
+        self.device.send_command(cmd)
 
     def _check_file_exists(self, cfg_file):
         """
@@ -738,9 +734,9 @@ def _check_file_exists(self, cfg_file):
 
         return boolean
         """
-        cmd = "dir {}".format(cfg_file)
-        success_pattern = "Directory of {}".format(cfg_file)
-        output = self.device.send_command_expect(cmd)
+        cmd = f"dir {cfg_file}"
+        success_pattern = f"Directory of {cfg_file}"
+        output = self.device.send_command(cmd)
         if "Error opening" in output:
             return False
         elif success_pattern in output:
@@ -892,7 +888,7 @@ def get_lldp_neighbors_detail(self, interface=""):
         lldp_interfaces = []
 
         if interface:
-            command = "show lldp neighbors {} detail".format(interface)
+            command = f"show lldp neighbors {interface} detail"
         else:
             command = "show lldp neighbors detail"
         lldp_entries = self._send_command(command)
@@ -916,7 +912,7 @@ def get_lldp_neighbors_detail(self, interface=""):
         # which is in the same sequence as the detailed output
         if not lldp_entries[0]["local_interface"]:
             if interface:
-                command = "show lldp neighbors {}".format(interface)
+                command = f"show lldp neighbors {interface}"
             else:
                 command = "show lldp neighbors"
             lldp_brief = self._send_command(command)
@@ -1577,12 +1573,12 @@ def get_bgp_neighbors(self):
                 "ipv6 unicast",
                 "ipv6 multicast",
             ]:
-                cmd_bgp_neighbor = "show bgp %s neighbors" % afi
+                cmd_bgp_neighbor = f"show bgp {afi} neighbors"
                 neighbor_output += self._send_command(cmd_bgp_neighbor).strip()
                 # trailing newline required for parsing
                 neighbor_output += "\n"
             elif afi in ["vpnv4 unicast", "vpnv6 unicast", "ipv4 mdt"]:
-                cmd_bgp_neighbor = "show bgp %s all neighbors" % afi
+                cmd_bgp_neighbor = f"show bgp {afi} all neighbors"
                 neighbor_output += self._send_command(cmd_bgp_neighbor).strip()
                 # trailing newline required for parsing
                 neighbor_output += "\n"
@@ -2320,7 +2316,7 @@ def get_arp_table(self, vrf=""):
             ]
         """
         if vrf:
-            command = "show arp vrf {} | exclude Incomplete".format(vrf)
+            command = f"show arp vrf {vrf} | exclude Incomplete"
         else:
             command = "show arp | exclude Incomplete"
 
@@ -3318,19 +3314,18 @@ def traceroute(
         if timeout:
             # Timeout should be an integer between 1 and 3600
             if isinstance(timeout, int) and 1 <= timeout <= 3600:
-                command += " timeout {}".format(str(timeout))
+                command += f" timeout {str(timeout)}"
 
-        # Calculation to leave enough time for traceroute to complete assumes send_command
-        # delay of .2 seconds.
-        max_loops = (5 * ttl * timeout) + 150
-        if max_loops < 500:  # Make sure max_loops isn't set artificially low
-            max_loops = 500
-        output = self.device.send_command(command, max_loops=max_loops)
+        read_timeout = ttl * timeout
+        # Make sure read_timeout isn't artificially low
+        if read_timeout < 100:
+            read_timeout = 100
+        output = self.device.send_command(command, read_timeout=read_timeout)
 
         # Prepare return dict
         traceroute_dict = dict()
         if re.search("Unrecognized host or address", output):
-            traceroute_dict["error"] = "unknown host %s" % destination
+            traceroute_dict["error"] = f"unknown host {destination}"
             return traceroute_dict
         else:
             traceroute_dict["success"] = dict()
@@ -3511,7 +3506,7 @@ def get_config(self, retrieve="all", full=False, sanitized=False):
             configs["startup"] = output.strip()
 
         if retrieve in ("running", "all"):
-            command = "show running-config{}".format(run_full)
+            command = f"show running-config{run_full}"
             output = self._send_command(command)
             output = re.sub(filter_pattern, "", output, flags=re.M)
             configs["running"] = output.strip()

diff --git a/napalm/pyIOSXR/iosxr.py b/napalm/pyIOSXR/iosxr.py
@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 # Copyright 2015 Netflix. All rights reserved.
 # Copyright 2016 BigWaveIT. All rights reserved.
-# Copyright 2021 Kirk Byers. All rights reserved.
+# Copyright 2022 Kirk Byers. All rights reserved.
 #
 # 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
@@ -61,8 +61,7 @@ class IOSXR(object):
 
     _XML_SHELL = "xml"
     _XML_MODE_PROMPT = r"XML>"
-    _READ_DELAY = 0.1  # at least 0.1, corresponding to 600 max loops (60s timeout)
-    _XML_MODE_DELAY = 1  # should be able to read within one second
+    _XML_MODE_READ_TIMEOUT = 10  # should be able to complete read within 10 seconds
 
     _ITERATOR_ID_ERROR_MSG = (
         "Non-supported IteratorID in response object. "
@@ -235,11 +234,9 @@ def _unlock_xml_agent(self):
             self._xml_agent_locker.release()
 
     def _send_command_timing(self, command):
-
         return self.device.send_command_timing(
             command,
-            delay_factor=self._READ_DELAY,
-            max_loops=self._XML_MODE_DELAY / self._READ_DELAY,
+            read_timeout=self._XML_MODE_READ_TIMEOUT,
             strip_prompt=False,
             strip_command=False,
         )
@@ -277,7 +274,8 @@ def _enter_xml_mode(self):
     def _send_command(
         self,
         command,
-        delay_factor=None,
+        delay_factor=None,  # noqa
+        read_timeout=None,
         start=None,
         expect_string=None,
         read_output=None,
@@ -290,29 +288,25 @@ def _send_command(
         if read_output is None:
             read_output = ""
 
-        if not delay_factor:
-            delay_factor = self._READ_DELAY
+        if not read_timeout:
+            read_timeout = self.timeout
 
         if not start:
             start = time.time()
 
         output = read_output
-
         last_read = ""
-
         if not read_output and not receive:
             # because the XML agent is able to process only one single request over the same SSH
             # session at a time first come first served
             self._lock_xml_agent(start)
             try:
-                max_loops = self.timeout / delay_factor
-                last_read = self.device.send_command_expect(
+                last_read = self.device.send_command(
                     command,
                     expect_string=expect_string,
                     strip_prompt=False,
                     strip_command=False,
-                    delay_factor=delay_factor,
-                    max_loops=max_loops,
+                    read_timeout=read_timeout,
                 )
                 output += last_read
             except IOError:
@@ -347,7 +341,7 @@ def _send_command(
                         return self._send_command(
                             command,
                             expect_string=expect_string,
-                            delay_factor=delay_factor,
+                            read_timeout=read_timeout,
                         )
         else:
             output += self.device._read_channel_timing()  # try to read some more

diff --git a/requirements.txt b/requirements.txt
@@ -8,7 +8,7 @@ jinja2
 netaddr
 pyYAML
 pyeapi>=0.8.2
-netmiko>=3.3.0,<4.0.0
+netmiko>=4.0.0
 junos-eznc>=2.2.1
 scp
 lxml>=4.3.0

diff --git a/test/pyiosxr/test_iosxr.py b/test/pyiosxr/test_iosxr.py
@@ -68,8 +68,8 @@ def find_prompt(self):
     def send_command(
         self,
         command_string,
-        delay_factor=0.1,
-        max_loops=150,
+        expect_string="",
+        read_timeout=10,
         strip_prompt=True,
         strip_command=True,
     ):