Merge branch 'develop' into update_a10_driver

Author: inaba-vdom-0

.gitignore

@@ -26,6 +26,7 @@ bin/.netmiko.cfg
 
 # Virtual Environment files
 .venv/
+.venv_old/
 bin/
 lib/
 pyvenv.cfg

PLATFORMS.md

@@ -118,6 +118,7 @@
 - MRV LX
 - Nokia/Alcatel SR-OS
 - Nokia SR Linux
+- Perle IOLAN Console Server
 - Optilink EOLT 9702 (telnet only)
 - QuantaMesh
 - Rad ETX

netmiko/base_connection.py

@@ -193,7 +193,7 @@ def __init__(
         serial_settings: Optional[Dict[str, Any]] = None,
         fast_cli: bool = True,
         _legacy_mode: bool = False,
-        session_log: Optional[SessionLog] = None,
+        session_log: Optional[Union[str, io.BufferedIOBase, SessionLog]] = None,
         session_log_record_writes: bool = False,
         session_log_file_mode: str = "write",
         allow_auto_change: bool = False,

netmiko/checkpoint/checkpoint_gaia_ssh.py

@@ -1,3 +1,6 @@
+import time
+import re
+
 from netmiko.no_config import NoConfig
 from netmiko.base_connection import BaseConnection
 
@@ -8,16 +11,43 @@ class CheckPointGaiaSSH(NoConfig, BaseConnection):
     firewalls.
     """
 
+    prompt_pattern = r"[>#]"
+
     def session_preparation(self) -> None:
         """
         Prepare the session after the connection has been established.
 
         Set the base prompt for interaction ('>').
         """
-        self._test_channel_read(pattern=r"[>#]")
+        self._test_channel_read(pattern=self.prompt_pattern)
         self.set_base_prompt()
         self.disable_paging(command="set clienv rows 0")
 
+        # Clear read buffer
+        time.sleep(0.3 * self.global_delay_factor)
+        self.clear_buffer()
+
+    def command_echo_read(self, cmd: str, read_timeout: float) -> str:
+        """Check Point clish double echoes the command (at least sometimes)"""
+
+        re_cmd = re.escape(cmd)
+        pattern = rf"{self.prompt_pattern}\s{re_cmd}"
+
+        # Make sure you read until you detect the command echo (avoid getting out of sync)
+        new_data = self.read_until_pattern(pattern=pattern, read_timeout=read_timeout)
+
+        # There can be echoed prompts that haven't been cleared before the cmd echo
+        # this can later mess up the trailing prompt pattern detection. Clear this out.
+        lines = new_data.split(cmd)
+        if len(lines) in [2, 3]:
+            # lines[-1] should realistically just be the null string
+            new_data = f"{cmd}{lines[-1]}"
+        else:
+            # cmd exists in the output multiple times? Just retain the original output
+            pass
+
+        return new_data
+
     def save_config(
         self, cmd: str = "", confirm: bool = False, confirm_response: str = ""
     ) -> str:

netmiko/mikrotik/mikrotik_ssh.py

@@ -222,8 +222,8 @@ def remote_space_available(self, search_pattern: str = "") -> int:
         sys_res = self.ssh_ctl_chan._send_command_timing_str(remote_cmd).splitlines()
         for res in sys_res:
             if "free-memory" in res:
-                spaceMib = res.strip().replace("free-memory: ", "").replace("MiB", "")
-                return int(float(spaceMib) * 1048576)
+                space_str = res.strip().replace("free-memory: ", "")
+                return self._format_to_bytes(space_str)
         raise ValueError("Unexpected output from remote_space_available")
 
     def remote_file_size(

netmiko/nokia/__init__.py

@@ -5,5 +5,12 @@
 )
 
 from netmiko.nokia.nokia_srl import NokiaSrlSSH
+from netmiko.nokia.nokia_isam import NokiaIsamSSH
 
-__all__ = ["NokiaSrosSSH", "NokiaSrosFileTransfer", "NokiaSrosTelnet", "NokiaSrlSSH"]
+__all__ = [
+    "NokiaSrosSSH",
+    "NokiaSrosFileTransfer",
+    "NokiaSrosTelnet",
+    "NokiaSrlSSH",
+    "NokiaIsamSSH",
+]

netmiko/nokia/nokia_isam.py

@@ -0,0 +1,73 @@
+import time
+import re
+from typing import Any
+
+from netmiko.base_connection import BaseConnection
+from netmiko.no_enable import NoEnable
+
+
+class NokiaIsamSSH(BaseConnection, NoEnable):
+
+    def session_preparation(self) -> None:
+        self._test_channel_read()
+        self.set_base_prompt()
+        commands = [
+            "environment inhibit-alarms",
+            "environment screen-length 0",
+        ]
+        for command in commands:
+            self.disable_paging(command=command, cmd_verify=True, pattern=r"#")
+        time.sleep(0.3 * self.global_delay_factor)
+        self.clear_buffer()
+
+    def set_base_prompt(self, *args: Any, **kwargs: Any) -> str:
+        """Remove the > when navigating into the different config level."""
+        cur_base_prompt = super().set_base_prompt(*args, **kwargs)
+        match = re.search(r"\*?(.*?)(>.*)*#", cur_base_prompt)
+        if match:
+            # strip off >... from base_prompt; strip off leading *
+            self.base_prompt: str = match.group(1)
+
+        return self.base_prompt
+
+    def cleanup(self, command: str = "logout") -> None:
+        """Gracefully exit the SSH session."""
+        try:
+            if self.check_config_mode():
+                self.exit_config_mode()
+        except Exception:
+            pass
+        # Always try to send final command
+        if self.session_log:
+            self.session_log.fin = True
+        self.write_channel(command + self.RETURN)
+
+    def check_config_mode(
+        self,
+        check_string: str = ">configure",
+        pattern: str = "#",
+        force_regex: bool = False,
+    ) -> bool:
+        """Use equivalent enable method."""
+        return super().check_config_mode(
+            check_string=check_string, pattern=pattern, force_regex=force_regex
+        )
+
+    def config_mode(
+        self, config_command: str = "configure", pattern: str = "", re_flags: int = 0
+    ) -> str:
+        return super().config_mode(
+            config_command=config_command, pattern=pattern, re_flags=re_flags
+        )
+
+    def exit_config_mode(self, exit_config: str = "exit", pattern: str = "") -> str:
+        return super().exit_config_mode(exit_config=exit_config)
+
+    def save_config(
+        self, cmd: str = "admin save", confirm: bool = False, confirm_response: str = ""
+    ) -> str:
+        return self._send_command_str(
+            command_string=cmd,
+            strip_prompt=False,
+            strip_command=False,
+        )

netmiko/perle/__init__.py

@@ -0,0 +1,3 @@
+from netmiko.perle.perle_ssh import PerleIolanSSH
+
+__all__ = ("PerleIolanSSH",)

netmiko/perle/perle_ssh.py

@@ -0,0 +1,94 @@
+import re
+from typing import Any, Dict, List, Optional, Union
+
+from netmiko.cisco_base_connection import CiscoBaseConnection
+from netmiko.no_config import NoConfig
+from netmiko.utilities import structured_data_converter
+
+
+class PerleIolanSSH(NoConfig, CiscoBaseConnection):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self.default_enter = kwargs.get("default_enter", "\r")
+
+    def session_preparation(self) -> None:
+        self._test_channel_read()
+        self.set_base_prompt(alt_prompt_terminator="$")
+
+    def enable(
+        self,
+        cmd: str = "admin",
+        pattern: str = "ssword:",
+        enable_pattern: Optional[str] = None,
+        check_state: bool = True,
+        re_flags: int = re.IGNORECASE,
+    ) -> str:
+        return super().enable(cmd, pattern, enable_pattern, check_state, re_flags)
+
+    def exit_enable_mode(self, *args: Any, **kwargs: Any) -> str:
+        # Perle does not have this concept
+        return ""
+
+    def save_config(
+        self, cmd: str = "save", confirm: bool = True, confirm_response: str = "y"
+    ) -> str:
+        return super().save_config(cmd, confirm, confirm_response)
+
+    def send_command_timing(
+        self,
+        command_string: str,
+        *args: Any,
+        **kwargs: Any,
+    ) -> Union[str, List[Any], Dict[str, Any]]:
+        real_command_string = command_string
+        real_strip_command = kwargs.get("strip_command", True)
+        real_strip_prompt = kwargs.get("strip_prompt", True)
+        command = command_string
+        more = r"< Hit any key >"
+        kwargs["strip_prompt"] = False
+        kwargs["strip_command"] = False
+
+        output = str(super().send_command_timing(command_string, *args, **kwargs))
+
+        command_string = " "
+        kwargs["normalize"] = False
+        kwargs["strip_command"] = True
+        while more in output:
+            output = re.sub(r"\n" + more, "", output)
+            output += str(
+                super().send_command_timing(
+                    command_string,
+                    *args,
+                    **kwargs,
+                )
+            )
+
+        output = self._sanitize_output(
+            output,
+            strip_command=real_strip_command,
+            command_string=command,
+            strip_prompt=real_strip_prompt,
+        )
+        return_data = structured_data_converter(
+            command=real_command_string,
+            raw_data=output,
+            platform=self.device_type,
+            use_textfsm=kwargs.get("use_textfsm", False),
+            use_ttp=kwargs.get("use_ttp", False),
+            use_genie=kwargs.get("use_genie", False),
+            textfsm_template=kwargs.get("textfsm_template", None),
+            ttp_template=kwargs.get("ttp_template", None),
+            raise_parsing_error=kwargs.get("raise_parsing_error", False),
+        )
+
+        return return_data
+
+    def strip_prompt(self, a_string: str) -> str:
+        # Delete repeated prompts
+        old_string = a_string
+        while (a_string := super().strip_prompt(a_string)) != old_string:
+            old_string = a_string
+        return a_string
+
+    def cleanup(self, command: str = "logout") -> None:
+        return super().cleanup(command)

netmiko/ssh_dispatcher.py

@@ -125,6 +125,7 @@
 from netmiko.mrv import MrvOptiswitchSSH
 from netmiko.netapp import NetAppcDotSSH
 from netmiko.nokia import (
+    NokiaIsamSSH,
     NokiaSrosSSH,
     NokiaSrosFileTransfer,
     NokiaSrosTelnet,
@@ -140,6 +141,7 @@
 from netmiko.paloalto import PaloAltoPanosSSH
 from netmiko.paloalto import PaloAltoPanosTelnet
 from netmiko.pluribus import PluribusSSH
+from netmiko.perle import PerleIolanSSH
 from netmiko.quanta import QuantaMeshSSH
 from netmiko.rad import RadETXSSH
 from netmiko.rad import RadETXTelnet
@@ -303,12 +305,14 @@
     "netapp_cdot": NetAppcDotSSH,
     "netgear_prosafe": NetgearProSafeSSH,
     "netscaler": NetscalerSSH,
+    "nokia_isam": NokiaIsamSSH,
     "nokia_sros": NokiaSrosSSH,
     "nokia_srl": NokiaSrlSSH,
     "oneaccess_oneos": OneaccessOneOSSSH,
     "ovs_linux": OvsLinuxSSH,
     "paloalto_panos": PaloAltoPanosSSH,
     "pluribus": PluribusSSH,
+    "perle_iolan": PerleIolanSSH,
     "quanta_mesh": QuantaMeshSSH,
     "rad_etx": RadETXSSH,
     "raisecom_roap": RaisecomRoapSSH,