|
| 1 | +import re |
| 2 | +from typing import Any, Dict, List, Optional, Union |
| 3 | + |
| 4 | +from netmiko.cisco_base_connection import CiscoBaseConnection |
| 5 | +from netmiko.no_config import NoConfig |
| 6 | +from netmiko.utilities import structured_data_converter |
| 7 | + |
| 8 | + |
| 9 | +class PerleIolanSSH(NoConfig, CiscoBaseConnection): |
| 10 | + def __init__(self, *args: Any, **kwargs: Any) -> None: |
| 11 | + super().__init__(*args, **kwargs) |
| 12 | + self.default_enter = kwargs.get("default_enter", "\r") |
| 13 | + |
| 14 | + def session_preparation(self) -> None: |
| 15 | + self._test_channel_read() |
| 16 | + self.set_base_prompt(pri_prompt_terminator="$") |
| 17 | + |
| 18 | + def enable( |
| 19 | + self, |
| 20 | + cmd: str = "admin", |
| 21 | + pattern: str = "ssword:", |
| 22 | + enable_pattern: Optional[str] = None, |
| 23 | + check_state: bool = True, |
| 24 | + re_flags: int = re.IGNORECASE, |
| 25 | + ) -> str: |
| 26 | + return super().enable(cmd, pattern, enable_pattern, check_state, re_flags) |
| 27 | + |
| 28 | + def exit_enable_mode(self, *args: Any, **kwargs: Any) -> str: |
| 29 | + # Perle does not have this concept |
| 30 | + return "" |
| 31 | + |
| 32 | + def save_config( |
| 33 | + self, cmd: str = "save", confirm: bool = True, confirm_response: str = "y" |
| 34 | + ) -> str: |
| 35 | + return super().save_config(cmd, confirm, confirm_response) |
| 36 | + |
| 37 | + def send_command_timing( |
| 38 | + self, |
| 39 | + *args: Any, |
| 40 | + **kwargs: Any, |
| 41 | + ) -> Union[str, List[Any], Dict[str, Any]]: |
| 42 | + real_command_string = kwargs["command_string"] |
| 43 | + real_strip_command = kwargs.get("strip_command", True) |
| 44 | + real_strip_prompt = kwargs.get("strip_prompt", True) |
| 45 | + command = kwargs["command_string"] + "\n" |
| 46 | + more = r"< Hit any key >" |
| 47 | + kwargs["strip_prompt"] = False |
| 48 | + kwargs["strip_command"] = False |
| 49 | + |
| 50 | + output = str(super().send_command_timing(*args, **kwargs)) |
| 51 | + |
| 52 | + kwargs["command_string"] = " " |
| 53 | + kwargs["normalize"] = False |
| 54 | + kwargs["strip_command"] = True |
| 55 | + while more in output: |
| 56 | + output = re.sub(r"\n" + more, "", output) |
| 57 | + output += str( |
| 58 | + super().send_command_timing( |
| 59 | + *args, |
| 60 | + **kwargs, |
| 61 | + ) |
| 62 | + ) |
| 63 | + |
| 64 | + output = self._sanitize_output( |
| 65 | + output, |
| 66 | + strip_command=real_strip_command, |
| 67 | + command_string=command, |
| 68 | + strip_prompt=real_strip_prompt, |
| 69 | + ) |
| 70 | + return_data = structured_data_converter( |
| 71 | + command=real_command_string, |
| 72 | + raw_data=output, |
| 73 | + platform=self.device_type, |
| 74 | + use_textfsm=kwargs.get("use_textfsm", False), |
| 75 | + use_ttp=kwargs.get("use_ttp", False), |
| 76 | + use_genie=kwargs.get("use_genie", False), |
| 77 | + textfsm_template=kwargs.get("textfsm_template", None), |
| 78 | + ttp_template=kwargs.get("ttp_template", None), |
| 79 | + raise_parsing_error=kwargs.get("raise_parsing_error", False), |
| 80 | + ) |
| 81 | + |
| 82 | + return return_data |
| 83 | + |
| 84 | + def strip_prompt(self, a_string: str) -> str: |
| 85 | + # Delete repeated prompts |
| 86 | + old_string = a_string |
| 87 | + while (a_string := super().strip_prompt(a_string)) != old_string: |
| 88 | + old_string = a_string |
| 89 | + return a_string |
| 90 | + |
| 91 | + def cleanup(self, command: str = "logout") -> None: |
| 92 | + return super().cleanup(command) |
0 commit comments