Merge branch 'x3me_devices' of https://github.com/bsdemon/netmiko int…

…o bsdemon-x3me_devices2

netmiko/optilink/__init__.py

@@ -0,0 +1,11 @@
+from netmiko.optilink.op_golt_944 import OptilinkGOLT944Telnet
+from netmiko.optilink.op_eolt_11444 import OptilinkEOLT11444Telnet
+from netmiko.optilink.op_eolt_9702 import OptilinkEOLT9702Telnet
+from netmiko.optilink.op_golt_924 import OptilinkGOLT924Telnet
+
+__all__ = [
+    "OptilinkGOLT944Telnet",
+    "OptilinkEOLT11444Telnet",
+    "OptilinkEOLT9702Telnet",
+    "OptilinkGOLT924Telnet",
+]

netmiko/optilink/op_eolt_11444.py

@@ -0,0 +1,36 @@
+from netmiko.cisco_base_connection import CiscoBaseConnection
+
+
+class OptilinkEOLT11444Base(CiscoBaseConnection):
+    """
+    Optilink EOLT 11444
+    Optilink EOLT 11448
+    """
+
+    def session_preparation(self) -> None:
+        self._test_channel_read(pattern=r"[>#]")
+        self.set_base_prompt()
+        self.enable()
+        self.disable_paging()
+        self.clear_buffer()
+        self.exit_enable_mode()
+
+    def config_mode(
+        self,
+        config_command: str = "configure",
+        pattern: str = "",
+        re_flags: int = 0,
+    ) -> str:
+        """Enter into configuration mode."""
+        return super().config_mode(
+            config_command=config_command, pattern=pattern, re_flags=re_flags
+        )
+
+
+class OptilinkEOLT11444Telnet(OptilinkEOLT11444Base):
+    """
+    Optilink EOLT 11444 telnet driver
+    Optilink EOLT 11448 telnet driver
+    """
+
+    pass

netmiko/optilink/op_eolt_9702.py

@@ -0,0 +1,58 @@
+from netmiko.cisco_base_connection import CiscoBaseConnection
+
+
+class OptilinkEOLT9702Base(CiscoBaseConnection):
+    """
+    Optilink EOLT 97028P2AB
+    Optilink EOLT 97024P
+    """
+
+    def session_preparation(self) -> None:
+        self._test_channel_read(pattern=r"[>#]")
+        self.set_base_prompt()
+        self.enable()
+        self.config_mode()
+        self.send_command("vty output show-all")
+        self.exit_config_mode()
+        self.exit_enable_mode()
+
+    def config_mode(
+        self,
+        config_command: str = "config",
+        pattern: str = "",
+        re_flags: int = 0,
+    ) -> str:
+        """Enter into configuration mode."""
+        return super().config_mode(
+            config_command=config_command, pattern=pattern, re_flags=re_flags
+        )
+
+    def check_enable_mode(self, check_string: str = "#") -> bool:
+        """Check if in enable mode. Return a boolean."""
+        self.write_channel(self.RETURN)
+        output = self.read_until_prompt(read_entire_line=True)
+        return check_string in output
+
+    def exit_enable_mode(self, exit_command: str = "exit") -> str:
+        """Exit from enable mode."""
+        output = ""
+        if self.check_enable_mode():
+            self.write_channel(self.normalize_cmd(exit_command))
+            self.read_until_pattern(pattern=exit_command)
+            output += self.read_until_pattern(pattern=r"OP_OLT>")
+            if self.check_enable_mode():
+                raise ValueError("Failed to exit enable mode.")
+        return output
+
+    def exit_config_mode(self, exit_config: str = "exit", pattern: str = r"#.*") -> str:
+        """Exit from configuration mode."""
+        return super().exit_config_mode(exit_config=exit_config, pattern=pattern)
+
+
+class OptilinkEOLT9702Telnet(OptilinkEOLT9702Base):
+    """
+    Optilink EOLT 97028P2AB telnet driver
+    Optilink EOLT 97024P telnet driver
+    """
+
+    pass

netmiko/optilink/op_golt_924.py

@@ -0,0 +1,32 @@
+from netmiko.cisco_base_connection import CiscoBaseConnection
+
+
+class OptilinkGOLT924Base(CiscoBaseConnection):
+    """
+    Optilink GOLT 92408A
+    Optilink GOLT 92408A16A
+    """
+
+    def session_preparation(self) -> None:
+        self._test_channel_read(pattern=r"[>#]")
+        self.set_base_prompt()
+
+    def exit_enable_mode(self, exit_command: str = "exit") -> str:
+        """Exit from enable mode."""
+        output = ""
+        if self.check_enable_mode():
+            self.write_channel(self.normalize_cmd(exit_command))
+            self.read_until_pattern(pattern=exit_command)
+            output += self.read_until_pattern(pattern=r"gpon>")
+            if self.check_enable_mode():
+                raise ValueError("Failed to exit enable mode.")
+        return output
+
+
+class OptilinkGOLT924Telnet(OptilinkGOLT924Base):
+    """
+    Optilink GOLT 92416A telnet driver
+    Optilink GOLT 92408A telnet driver
+    """
+
+    pass

netmiko/optilink/op_golt_944.py

@@ -0,0 +1,151 @@
+# from netmiko.base_connection import BaseConnection
+import re
+from typing import Optional
+from netmiko.exceptions import NetmikoTimeoutException
+from netmiko.huawei.huawei import HuaweiBase, HuaweiTelnet
+
+
+class OptilinkGOLT944Base(HuaweiBase):
+    """Optilink GOLT 944"""
+
+    pass
+
+
+class OptilinkGOLT944Telnet(HuaweiTelnet):
+    """Optilink GOLT 944 telnet driver"""
+
+    def check_enable_mode(self, check_string: str = "<") -> bool:
+        """Check if in enable mode. Return a boolean."""
+        self.write_channel(self.RETURN)
+        output = self.read_until_prompt(read_entire_line=True)
+        return check_string in output
+
+    def check_config_mode(
+        self, check_string: str = "]", pattern: str = "", force_regex: bool = False
+    ) -> bool:
+        """Check if the device is in configuration mode or not."""
+        self.write_channel(self.RETURN)
+        if not pattern:
+            output = self.read_channel_timing(read_timeout=10.0)
+        else:
+            output = self.read_until_pattern(pattern=pattern)
+
+        if force_regex:
+            return bool(re.search(check_string, output))
+        else:
+            return check_string in output
+
+    def enable(
+        self,
+        cmd: str = "enable",
+        pattern: str = "ssword",
+        enable_pattern: Optional[str] = None,
+        check_state: bool = True,
+        re_flags: int = re.IGNORECASE,
+    ) -> str:
+        """Enter enable mode."""
+        output = ""
+        msg = (
+            "Failed to enter enable mode. Please ensure you pass "
+            "the 'secret' argument to ConnectHandler."
+        )
+
+        # Check if in enable mode already.
+        if check_state and self.check_enable_mode():
+            return output
+
+        # Send "enable" mode command
+        self.write_channel(self.normalize_cmd(cmd))
+        try:
+            # Read the command echo
+            if self.global_cmd_verify is not False:
+                output += self.read_until_pattern(pattern=re.escape(cmd.strip()))
+
+            # Search for trailing prompt or password pattern
+            output += self.read_until_prompt_or_pattern(
+                pattern=pattern, re_flags=re_flags
+            )
+
+            # Send the "secret" in response to password pattern
+            if re.search(pattern, output):
+                self.write_channel(self.normalize_cmd(self.secret))
+                output += self.read_until_prompt()
+
+            # Search for terminating pattern if defined
+            if enable_pattern and not re.search(enable_pattern, output):
+                output += self.read_until_pattern(pattern=enable_pattern)
+            elif not self.check_enable_mode():
+                raise ValueError(msg)
+
+        except NetmikoTimeoutException as e:
+            raise ValueError(msg) from e
+
+        return output
+
+    def config_mode(
+        self, config_command: str = "sys", pattern: str = "", re_flags: int = 0
+    ) -> str:
+        """Enter into configuration mode."""
+        output = ""
+        if not self.check_config_mode():
+            self.write_channel(self.normalize_cmd(config_command))
+            # Make sure you read until you detect the command echo (avoid getting out of sync)
+            if self.global_cmd_verify is not False:
+                output += self.read_until_pattern(
+                    pattern=re.escape(config_command.strip())
+                )
+            if pattern:
+                output += self.read_until_pattern(pattern=pattern, re_flags=re_flags)
+            else:
+                output += self.read_until_prompt(read_entire_line=True)
+            if not self.check_config_mode():
+                raise ValueError("Failed to enter configuration mode.")
+        return output
+
+    def exit_enable_mode(self, exit_command: str = "quit") -> str:
+        """Exit from enable mode."""
+        output = ""
+        if self.check_enable_mode():
+            self.write_channel(self.normalize_cmd(exit_command))
+            self.read_until_pattern(pattern=exit_command)
+            # output += self.read_until_pattern(pattern=r">")
+            output += self.read_until_prompt()
+            if self.check_enable_mode():
+                raise ValueError("Failed to exit enable mode.")
+        return output
+
+    def exit_config_mode(self, exit_config: str = "quit", pattern: str = "") -> str:
+        """Exit from configuration mode.
+
+        :param exit_config: Command to exit configuration mode
+        :type exit_config: str
+
+        :param pattern: Pattern to terminate reading of channel
+        :type pattern: str
+        """
+        output = ""
+        if self.check_config_mode():
+            self.write_channel(self.normalize_cmd(exit_config))
+            # Make sure you read until you detect the command echo (avoid getting out of sync)
+            if self.global_cmd_verify is not False:
+                output += self.read_until_pattern(
+                    pattern=re.escape(exit_config.strip())
+                )
+            if pattern:
+                output += self.read_until_pattern(pattern=pattern)
+            else:
+                output += self.read_until_prompt(read_entire_line=True)
+            if self.check_config_mode():
+                raise ValueError("Failed to exit configuration mode")
+        return output
+
+    def session_preparation(self) -> None:
+        self._test_channel_read(pattern=r"[>\]]")
+        self.set_base_prompt()
+        self.enable()
+        self.config_mode()
+        self.disable_paging(command="screen-rows per-page 0")
+        self.exit_config_mode()
+        self.exit_enable_mode()
+
+    pass

netmiko/ssh_dispatcher.py

@@ -112,6 +112,10 @@
 )
 from netmiko.netgear import NetgearProSafeSSH
 from netmiko.oneaccess import OneaccessOneOSTelnet, OneaccessOneOSSSH
+from netmiko.optilink import OptilinkEOLT9702Telnet
+from netmiko.optilink import OptilinkEOLT11444Telnet
+from netmiko.optilink import OptilinkGOLT924Telnet
+from netmiko.optilink import OptilinkGOLT944Telnet
 from netmiko.ovs import OvsLinuxSSH
 from netmiko.paloalto import PaloAltoPanosSSH
 from netmiko.paloalto import PaloAltoPanosTelnet
@@ -360,6 +364,10 @@
 CLASS_MAPPER["maipu_telnet"] = MaipuTelnet
 CLASS_MAPPER["nokia_sros_telnet"] = NokiaSrosTelnet
 CLASS_MAPPER["oneaccess_oneos_telnet"] = OneaccessOneOSTelnet
+CLASS_MAPPER["optilink_eolt9702_telnet"] = OptilinkEOLT9702Telnet
+CLASS_MAPPER["optilink_eolt11444_telnet"] = OptilinkEOLT11444Telnet
+CLASS_MAPPER["optilink_golt924_telnet"] = OptilinkGOLT924Telnet
+CLASS_MAPPER["optilink_golt944_telnet"] = OptilinkGOLT944Telnet
 CLASS_MAPPER["paloalto_panos_telnet"] = PaloAltoPanosTelnet
 CLASS_MAPPER["rad_etx_telnet"] = RadETXTelnet
 CLASS_MAPPER["raisecom_telnet"] = RaisecomRoapTelnet