diff --git a/.gitignore b/.gitignore index c6ef50b6fc5864c9206debfea0e32033f9487ff4..b6e90d4d4385fdae6d12e79ba3d524f943165c53 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ *.pyc *.swp + tests/old_format/DEVICE_CREDS.py tests/test_prep.sh tests/testx.txt @@ -8,6 +9,11 @@ tests/etc/test_devices.yml tests/etc/commands.yml tests/etc/responses.yml tests/etc/test_devices_exc.yml + +tests_new/etc/commands.yml +tests_new/etc/responses.yml +tests_new/etc/test_devices.yml + examples/SECRET_DEVICE_CREDS.py ./build ./build/* @@ -42,3 +48,8 @@ docs/build/doctrees/ tests/cisco3-out.txt tests/test.log +tests_new/test_out/*.out +tests_new/test_out/*.stderr +tests_new/cisco-xrv.txt +tests_new/*.out +tests/*.out diff --git a/EXAMPLES.md b/EXAMPLES.md index cf995ef87686f02f9e0fd80676a1d6faa92baf48..92c62fd821653715959651cbe77b8d2e2cbce5cc 100644 --- a/EXAMPLES.md +++ b/EXAMPLES.md @@ -404,6 +404,50 @@ Password: <br /> +## Using TTP + +```py +cisco1 = { + "device_type": "cisco_ios", + "host": "cisco1.lasthop.io", + "username": "pyclass", + "password": getpass(), +} + +# write template to file +ttp_raw_template = """ +interface {{ interface }} + description {{ description }} +""" + +with open("show_run_interfaces.ttp", "w") as writer: + writer.write(ttp_raw_template) + +command = "show run | s interfaces" +with ConnectHandler(**cisco1) as net_connect: + # Use TTP to retrieve structured data + output = net_connect.send_command( + command, use_ttp=True, ttp_template="show_run_interfaces.ttp" + ) + +print() +pprint(output) +print() +``` + + +#### Output from the above execution: + +``` + [[[{'description': 'Router-id-loopback', + 'interface': 'Loopback0'}, + {'description': 'CPE_Acces_Vlan', + 'interface': 'Vlan778'}]]] +``` + +<br /> + + ## Using Genie ```py diff --git a/PLATFORMS.md b/PLATFORMS.md index a0c1c8e9b50678bc3bc8b6291cfa88cbefe7f7da..ff3790b70402c73698806fd9036b39a232f3afa1 100644 --- a/PLATFORMS.md +++ b/PLATFORMS.md @@ -48,6 +48,7 @@ - Pluribus - Ruckus ICX/FastIron - Ruijie Networks +- TPLink JetStream - Ubiquiti EdgeSwitch - Vyatta VyOS - Yamaha diff --git a/docs/netmiko/index.html b/docs/netmiko/index.html index ca51da7948f0c4a83ba878d849b46041e4683d2f..21bc651159bd1444075256c288a05c109580c2c2 100644 --- a/docs/netmiko/index.html +++ b/docs/netmiko/index.html @@ -1900,6 +1900,8 @@ Device settings: {self.device_type} {self.host}:{self.port} normalize=True, use_textfsm=False, textfsm_template=None, + use_ttp=False, + ttp_template=None, use_genie=False, cmd_verify=False, cmd_echo=None, @@ -1933,6 +1935,13 @@ Device settings: {self.device_type} {self.host}:{self.port} path, relative path, or name of file in current directory. (default: None). :type textfsm_template: str + :param use_ttp: Process command output through TTP template (default: False). + :type use_ttp: bool + + :param ttp_template: Name of template to parse output with; can be fully qualified + path, relative path, or name of file in current directory. (default: None). + :type ttp_template: str + :param use_genie: Process command output through PyATS/Genie parser (default: False). :type use_genie: bool @@ -2095,6 +2104,13 @@ Device settings: {self.device_type} {self.host}:{self.port} :param textfsm_template: Name of template to parse output with; can be fully qualified path, relative path, or name of file in current directory. (default: None). + :param use_ttp: Process command output through TTP template (default: False). + :type use_ttp: bool + + :param ttp_template: Name of template to parse output with; can be fully qualified + path, relative path, or name of file in current directory. (default: None). + :type ttp_template: str + :param use_genie: Process command output through PyATS/Genie parser (default: False). :type normalize: bool diff --git a/examples/send_command_ttp.py b/examples/send_command_ttp.py new file mode 100644 index 0000000000000000000000000000000000000000..249f6b55c724dad84d343d0b04cbbb5ef44429ae --- /dev/null +++ b/examples/send_command_ttp.py @@ -0,0 +1,31 @@ +#!/usr/bin/env python +from netmiko import ConnectHandler +from getpass import getpass +from pprint import pprint + +cisco1 = { + "device_type": "cisco_ios", + "host": "cisco1.lasthop.io", + "username": "pyclass", + "password": getpass(), +} + +# write template to file +ttp_raw_template = """ +interface {{ interface }} + description {{ description }} +""" + +with open("show_run_interfaces.ttp", "w") as writer: + writer.write(ttp_raw_template) + +command = "show run" +with ConnectHandler(**cisco1) as net_connect: + # Use TTP to retrieve structured data + output = net_connect.send_command( + command, use_ttp=True, ttp_template="show_run_interfaces.ttp" + ) + +print() +pprint(output) +print() diff --git a/netmiko/__init__.py b/netmiko/__init__.py index b3befa0e20c09ac7a11b141911f34708baafa6e1..dbef5b5664e8ebf50c8904502b17cfcc5f7b3778 100644 --- a/netmiko/__init__.py +++ b/netmiko/__init__.py @@ -23,7 +23,7 @@ from netmiko.scp_functions import file_transfer, progress_bar # Alternate naming Netmiko = ConnectHandler -__version__ = "3.3.0" +__version__ = "3.3.2" __all__ = ( "ConnectHandler", "ssh_dispatcher", diff --git a/netmiko/arista/arista.py b/netmiko/arista/arista.py index ecba8f5cae6e450ddd247268812215a86962e1a0..8378858acd3f4c973948fb6db2b48131b150b856 100644 --- a/netmiko/arista/arista.py +++ b/netmiko/arista/arista.py @@ -1,18 +1,21 @@ -import time +import re from netmiko.cisco_base_connection import CiscoSSHConnection from netmiko.cisco_base_connection import CiscoFileTransfer class AristaBase(CiscoSSHConnection): + def __init__(self, *args, **kwargs): + kwargs.setdefault("fast_cli", True) + kwargs.setdefault("_legacy_mode", False) + return super().__init__(*args, **kwargs) + def session_preparation(self): """Prepare the session after the connection has been established.""" - self._test_channel_read(pattern=r"[>#]") + cmd = "terminal width 511" + # Arista will echo immediately and then when the device really responds (like NX-OS) + self.set_terminal_width(command=cmd, pattern=r"Width set to") + self.disable_paging(cmd_verify=False, pattern=r"Pagination disabled") self.set_base_prompt() - self.set_terminal_width(command="terminal width 511", pattern="terminal") - self.disable_paging() - # Clear the read buffer - time.sleep(0.3 * self.global_delay_factor) - self.clear_buffer() def check_config_mode(self, check_string=")#", pattern=""): """ @@ -33,6 +36,20 @@ class AristaBase(CiscoSSHConnection): output = output.replace("(s2)", "") return check_string in output + def config_mode(self, config_command="configure terminal", pattern="", re_flags=0): + """Force arista to read pattern all the way to prompt on the next line.""" + + if not re_flags: + re_flags = re.DOTALL + check_string = re.escape(")#") + + if not pattern: + pattern = re.escape(self.base_prompt[:16]) + pattern = f"{pattern}.*{check_string}" + return super().config_mode( + config_command=config_command, pattern=pattern, re_flags=re_flags + ) + def _enter_shell(self): """Enter the Bourne Shell.""" return self.send_command("bash", expect_string=r"[\$#]") diff --git a/netmiko/aruba/aruba_ssh.py b/netmiko/aruba/aruba_ssh.py index 6651e7be7fb7cbd5e572a282736f1e7970f38c2f..8ecdc6998b3edc692385689090720759cc62059e 100644 --- a/netmiko/aruba/aruba_ssh.py +++ b/netmiko/aruba/aruba_ssh.py @@ -1,4 +1,9 @@ -"""Aruba OS support""" +""" +Aruba OS support. + +For use with Aruba OS Controllers. + +""" import time import re from netmiko.cisco_base_connection import CiscoSSHConnection diff --git a/netmiko/base_connection.py b/netmiko/base_connection.py index c278f6900caae74e8ddba068e626d323bb8eeae8..189558065f7e969c4cbe1a1708aee85ebecd3a9c 100644 --- a/netmiko/base_connection.py +++ b/netmiko/base_connection.py @@ -17,6 +17,7 @@ from threading import Lock import paramiko import serial +from tenacity import retry, stop_after_attempt, wait_exponential from netmiko import log from netmiko.netmiko_globals import MAX_BUFFER, BACKSPACE_CHAR @@ -29,6 +30,7 @@ from netmiko.utilities import ( check_serial_port, get_structured_data, get_structured_data_genie, + get_structured_data_ttp, select_cmd_verify, ) from netmiko.utilities import m_exec_time # noqa @@ -980,6 +982,7 @@ Device settings: {self.device_type} {self.host}:{self.port} print("Interactive SSH session established") return "" + # @m_exec_time def _test_channel_read(self, count=40, pattern=""): """Try to read the channel (generally post login) verify you receive data back. @@ -1011,6 +1014,7 @@ Device settings: {self.device_type} {self.host}:{self.port} break else: self.write_channel(self.RETURN) + main_delay = _increment_delay(main_delay) time.sleep(main_delay) i += 1 @@ -1045,7 +1049,7 @@ Device settings: {self.device_type} {self.host}:{self.port} :type delay_factor: int """ if self.fast_cli: - if delay_factor <= self.global_delay_factor: + if delay_factor and delay_factor <= self.global_delay_factor: return delay_factor else: return self.global_delay_factor @@ -1114,6 +1118,10 @@ Device settings: {self.device_type} {self.host}:{self.port} output = self.read_until_prompt() return output + # Retry by sleeping .33 and then double sleep until 5 attempts (.33, .66, 1.32, etc) + @retry( + wait=wait_exponential(multiplier=0.33, min=0, max=5), stop=stop_after_attempt(5) + ) def set_base_prompt( self, pri_prompt_terminator="#", alt_prompt_terminator=">", delay_factor=1 ): @@ -1208,6 +1216,8 @@ Device settings: {self.device_type} {self.host}:{self.port} normalize=True, use_textfsm=False, textfsm_template=None, + use_ttp=False, + ttp_template=None, use_genie=False, cmd_verify=False, cmd_echo=None, @@ -1241,6 +1251,13 @@ Device settings: {self.device_type} {self.host}:{self.port} path, relative path, or name of file in current directory. (default: None). :type textfsm_template: str + :param use_ttp: Process command output through TTP template (default: False). + :type use_ttp: bool + + :param ttp_template: Name of template to parse output with; can be fully qualified + path, relative path, or name of file in current directory. (default: None). + :type ttp_template: str + :param use_genie: Process command output through PyATS/Genie parser (default: False). :type use_genie: bool @@ -1250,13 +1267,19 @@ Device settings: {self.device_type} {self.host}:{self.port} :param cmd_echo: Deprecated (use cmd_verify instead) :type cmd_echo: bool """ - # For compatibility remove cmd_echo in Netmiko 4.x.x + + # For compatibility; remove cmd_echo in Netmiko 4.x.x if cmd_echo is not None: cmd_verify = cmd_echo output = "" + delay_factor = self.select_delay_factor(delay_factor) - self.clear_buffer() + # Cleanup in future versions of Netmiko + if delay_factor < 1: + if not self._legacy_mode and self.fast_cli: + delay_factor = 1 + if normalize: command_string = self.normalize_cmd(command_string) @@ -1291,7 +1314,7 @@ Device settings: {self.device_type} {self.host}:{self.port} strip_prompt=strip_prompt, ) - # If both TextFSM and Genie are set, try TextFSM then Genie + # If both TextFSM, TTP and Genie are set, try TextFSM then TTP then Genie if use_textfsm: structured_output = get_structured_data( output, @@ -1302,6 +1325,11 @@ Device settings: {self.device_type} {self.host}:{self.port} # If we have structured data; return it. if not isinstance(structured_output, str): return structured_output + if use_ttp: + structured_output = get_structured_data_ttp(output, template=ttp_template) + # If we have structured data; return it. + if not isinstance(structured_output, str): + return structured_output if use_genie: structured_output = get_structured_data_genie( output, platform=self.device_type, command=command_string.strip() @@ -1366,6 +1394,8 @@ Device settings: {self.device_type} {self.host}:{self.port} normalize=True, use_textfsm=False, textfsm_template=None, + use_ttp=False, + ttp_template=None, use_genie=False, cmd_verify=True, ): @@ -1403,6 +1433,13 @@ Device settings: {self.device_type} {self.host}:{self.port} :param textfsm_template: Name of template to parse output with; can be fully qualified path, relative path, or name of file in current directory. (default: None). + :param use_ttp: Process command output through TTP template (default: False). + :type use_ttp: bool + + :param ttp_template: Name of template to parse output with; can be fully qualified + path, relative path, or name of file in current directory. (default: None). + :type ttp_template: str + :param use_genie: Process command output through PyATS/Genie parser (default: False). :type normalize: bool @@ -1484,7 +1521,7 @@ Device settings: {self.device_type} {self.host}:{self.port} new_data = self.read_channel() else: # nobreak raise IOError( - "Search pattern never detected in send_command_expect: {}".format( + "Search pattern never detected in send_command: {}".format( search_pattern ) ) @@ -1496,7 +1533,7 @@ Device settings: {self.device_type} {self.host}:{self.port} strip_prompt=strip_prompt, ) - # If both TextFSM and Genie are set, try TextFSM then Genie + # If both TextFSM, TTP and Genie are set, try TextFSM then TTP then Genie if use_textfsm: structured_output = get_structured_data( output, @@ -1507,6 +1544,11 @@ Device settings: {self.device_type} {self.host}:{self.port} # If we have structured data; return it. if not isinstance(structured_output, str): return structured_output + if use_ttp: + structured_output = get_structured_data_ttp(output, template=ttp_template) + # If we have structured data; return it. + if not isinstance(structured_output, str): + return structured_output if use_genie: structured_output = get_structured_data_genie( output, platform=self.device_type, command=command_string.strip() @@ -1663,7 +1705,7 @@ Device settings: {self.device_type} {self.host}:{self.port} output = self.read_until_pattern(pattern=pattern) return check_string in output - def config_mode(self, config_command="", pattern=""): + def config_mode(self, config_command="", pattern="", re_flags=0): """Enter into config_mode. :param config_command: Configuration command to send to the device @@ -1671,6 +1713,9 @@ Device settings: {self.device_type} {self.host}:{self.port} :param pattern: Pattern to terminate reading of channel :type pattern: str + + :param re_flags: Regular expression flags + :type re_flags: RegexFlag """ output = "" if not self.check_config_mode(): @@ -1680,8 +1725,8 @@ Device settings: {self.device_type} {self.host}:{self.port} output += self.read_until_pattern( pattern=re.escape(config_command.strip()) ) - if not re.search(pattern, output, flags=re.M): - output += self.read_until_pattern(pattern=pattern) + if not re.search(pattern, output, flags=re_flags): + output += self.read_until_pattern(pattern=pattern, re_flags=re_flags) if not self.check_config_mode(): raise ValueError("Failed to enter configuration mode.") return output diff --git a/netmiko/cisco/__init__.py b/netmiko/cisco/__init__.py index ac3868e550efa37e2decbbc571f6add9c03f9959..66744c2379fe8b623ad345546937c96a286599c2 100644 --- a/netmiko/cisco/__init__.py +++ b/netmiko/cisco/__init__.py @@ -7,6 +7,7 @@ from netmiko.cisco.cisco_ios import ( from netmiko.cisco.cisco_ios import CiscoIosFileTransfer from netmiko.cisco.cisco_ios import InLineTransfer from netmiko.cisco.cisco_asa_ssh import CiscoAsaSSH, CiscoAsaFileTransfer +from netmiko.cisco.cisco_ftd_ssh import CiscoFtdSSH from netmiko.cisco.cisco_nxos_ssh import CiscoNxosSSH, CiscoNxosFileTransfer from netmiko.cisco.cisco_xr import CiscoXrSSH, CiscoXrTelnet, CiscoXrFileTransfer from netmiko.cisco.cisco_wlc_ssh import CiscoWlcSSH @@ -17,6 +18,7 @@ __all__ = [ "CiscoIosSSH", "CiscoIosTelnet", "CiscoAsaSSH", + "CiscoFtdSSH", "CiscoNxosSSH", "CiscoXrSSH", "CiscoXrTelnet", diff --git a/netmiko/cisco/cisco_ftd_ssh.py b/netmiko/cisco/cisco_ftd_ssh.py new file mode 100644 index 0000000000000000000000000000000000000000..3d2199c45e70680ba2a0abb76c305d245458a29b --- /dev/null +++ b/netmiko/cisco/cisco_ftd_ssh.py @@ -0,0 +1,27 @@ +"""Subclass specific to Cisco FTD.""" +from netmiko.cisco_base_connection import CiscoSSHConnection + + +class CiscoFtdSSH(CiscoSSHConnection): + """Subclass specific to Cisco FTD.""" + + def session_preparation(self): + """Prepare the session after the connection has been established.""" + self._test_channel_read() + self.set_base_prompt() + + def send_config_set(self, *args, **kwargs): + """Canot change config on FTD via ssh""" + raise NotImplementedError + + def enable(self, *args, **kwargs): + """No enable mode on firepower ssh""" + return "" + + def config_mode(self, *args, **kwargs): + """No config mode on firepower ssh""" + return "" + + def check_config_mode(self, *args, **kwargs): + """No config mode on firepower ssh""" + return False diff --git a/netmiko/cisco/cisco_nxos_ssh.py b/netmiko/cisco/cisco_nxos_ssh.py index 8df4e5a4b644693a658e86690a04da37fce38e8b..88721bbb7985ad2b01d8630b920a0fcf78914d7a 100644 --- a/netmiko/cisco/cisco_nxos_ssh.py +++ b/netmiko/cisco/cisco_nxos_ssh.py @@ -1,21 +1,26 @@ import re -import time import os from netmiko.cisco_base_connection import CiscoSSHConnection from netmiko.cisco_base_connection import CiscoFileTransfer class CiscoNxosSSH(CiscoSSHConnection): + def __init__(self, *args, **kwargs): + # Cisco NX-OS defaults to fast_cli=True and legacy_mode=False + kwargs.setdefault("fast_cli", True) + kwargs.setdefault("_legacy_mode", False) + return super().__init__(*args, **kwargs) + def session_preparation(self): """Prepare the session after the connection has been established.""" - self._test_channel_read(pattern=r"[>#]") self.ansi_escape_codes = True - self.set_base_prompt() - self.set_terminal_width(command="terminal width 511", pattern="terminal") + # NX-OS has an issue where it echoes the command even though it hasn't returned the prompt + self._test_channel_read(pattern=r"[>#]") + self.set_terminal_width( + command="terminal width 511", pattern=r"terminal width 511" + ) self.disable_paging() - # Clear the read buffer - time.sleep(0.3 * self.global_delay_factor) - self.clear_buffer() + self.set_base_prompt() def normalize_linefeeds(self, a_string): """Convert '\r\n' or '\r\r\n' to '\n, and remove extra '\r's in the text.""" diff --git a/netmiko/cisco/cisco_xr.py b/netmiko/cisco/cisco_xr.py index de2644add22d32a41eae3d40bcca5849bcee330c..8978aa842b0813dc3028e365bcc47d1942746171 100644 --- a/netmiko/cisco/cisco_xr.py +++ b/netmiko/cisco/cisco_xr.py @@ -1,22 +1,27 @@ -import time import re from netmiko.cisco_base_connection import CiscoBaseConnection, CiscoFileTransfer class CiscoXrBase(CiscoBaseConnection): + def __init__(self, *args, **kwargs): + # Cisco NX-OS defaults to fast_cli=True and legacy_mode=False + kwargs.setdefault("fast_cli", True) + kwargs.setdefault("_legacy_mode", False) + return super().__init__(*args, **kwargs) + def establish_connection(self): """Establish SSH connection to the network device""" super().establish_connection(width=511, height=511) def session_preparation(self): """Prepare the session after the connection has been established.""" - self._test_channel_read() - self.set_base_prompt() - self.set_terminal_width(command="terminal width 511", pattern="terminal") + # IOS-XR has an issue where it echoes the command even though it hasn't returned the prompt + self._test_channel_read(pattern=r"[>#]") + cmd = "terminal width 511" + self.set_terminal_width(command=cmd, pattern=cmd) self.disable_paging() - # Clear the read buffer - time.sleep(0.3 * self.global_delay_factor) - self.clear_buffer() + self._test_channel_read(pattern=r"[>#]") + self.set_base_prompt() def send_config_set(self, config_commands=None, exit_config_mode=False, **kwargs): """IOS-XR requires you not exit from configuration mode.""" @@ -121,17 +126,21 @@ class CiscoXrBase(CiscoBaseConnection): output = output.replace("(admin)", "") return check_string in output - def exit_config_mode(self, exit_config="end"): + def exit_config_mode(self, exit_config="end", pattern=""): """Exit configuration mode.""" output = "" if self.check_config_mode(): - output = self.send_command_timing( - exit_config, strip_prompt=False, strip_command=False - ) - if "Uncommitted changes found" in output: - output += self.send_command_timing( - "no", strip_prompt=False, strip_command=False + 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 "Uncommitted changes found" in output: + self.write_channel(self.normalize_cmd("no\n")) + output += self.read_until_pattern(pattern=r"[>#]") + if not re.search(pattern, output, flags=re.M): + output += self.read_until_pattern(pattern=pattern) if self.check_config_mode(): raise ValueError("Failed to exit configuration mode") return output diff --git a/netmiko/cisco_base_connection.py b/netmiko/cisco_base_connection.py index 112a6764e6e039120cb02a940f0047775c5a0be9..25a4a59e4a168128ca7002337850143a0e220d2c 100644 --- a/netmiko/cisco_base_connection.py +++ b/netmiko/cisco_base_connection.py @@ -29,7 +29,7 @@ class CiscoBaseConnection(BaseConnection): """ return super().check_config_mode(check_string=check_string, pattern=pattern) - def config_mode(self, config_command="configure terminal", pattern=""): + def config_mode(self, config_command="configure terminal", pattern="", re_flags=0): """ Enter into configuration mode on remote device. @@ -37,7 +37,9 @@ class CiscoBaseConnection(BaseConnection): """ if not pattern: pattern = re.escape(self.base_prompt[:16]) - return super().config_mode(config_command=config_command, pattern=pattern) + return super().config_mode( + config_command=config_command, pattern=pattern, re_flags=re_flags + ) def exit_config_mode(self, exit_config="end", pattern="#"): """Exit from configuration mode.""" diff --git a/netmiko/fortinet/fortinet_ssh.py b/netmiko/fortinet/fortinet_ssh.py index 1e42d22eb59ab1598aefd6a40168aef9dbd9c7bc..7131af75753289d875d486e061bc53cf032be3db 100644 --- a/netmiko/fortinet/fortinet_ssh.py +++ b/netmiko/fortinet/fortinet_ssh.py @@ -46,7 +46,7 @@ class FortinetSSH(CiscoSSHConnection): self.vdoms = False self._output_mode = "more" - if "Virtual domain configuration: enable" in output: + if re.search(r"Virtual domain configuration: (multiple|enable)", output): self.vdoms = True vdom_additional_command = "config global" output = self.send_command_timing(vdom_additional_command, delay_factor=2) diff --git a/netmiko/huawei/huawei.py b/netmiko/huawei/huawei.py index adc8b143885ffc3064d21c7128a63f2e92473f57..12b9435c1e49d9db9555b92348eb1b2a7022097e 100644 --- a/netmiko/huawei/huawei.py +++ b/netmiko/huawei/huawei.py @@ -28,9 +28,6 @@ class HuaweiBase(CiscoBaseConnection): pattern = rf" {code_cursor_left}" output = re.sub(pattern, "", output) - log.debug("Stripping ANSI escape codes") - log.debug(f"new_output = {output}") - log.debug(f"repr = {repr(output)}") return super().strip_ansi_escape_codes(output) def config_mode(self, config_command="system-view"): diff --git a/netmiko/paloalto/paloalto_panos.py b/netmiko/paloalto/paloalto_panos.py index 60f5371cec584dfc10e33e128e55abe89cdd9af9..4676b1b5bb126f28ed1417558e57453c3f9da721 100644 --- a/netmiko/paloalto/paloalto_panos.py +++ b/netmiko/paloalto/paloalto_panos.py @@ -51,6 +51,7 @@ class PaloAltoPanosBase(BaseConnection): def commit( self, + comment=None, force=False, partial=False, device_and_network=False, @@ -87,6 +88,8 @@ class PaloAltoPanosBase(BaseConnection): # Select proper command string based on arguments provided command_string = "commit" commit_marker = "configuration committed successfully" + if comment: + command_string += f' description "{comment}"' if force: command_string += " force" if partial: diff --git a/netmiko/scp_functions.py b/netmiko/scp_functions.py index 37f49712bb420955d4b3ff2216b46b44efe35afb..6c30dea11f3454c4cd496927da0fafef187b18ad 100644 --- a/netmiko/scp_functions.py +++ b/netmiko/scp_functions.py @@ -10,7 +10,8 @@ from netmiko import FileTransfer, InLineTransfer def progress_bar(filename, size, sent, peername=None): max_width = 50 - filename = filename.decode() + if isinstance(filename, bytes): + filename = filename.decode() clear_screen = chr(27) + "[2J" terminating_char = "|" diff --git a/netmiko/ssh_dispatcher.py b/netmiko/ssh_dispatcher.py index 908f06a9c461ece9226845fd44fc8bc76aefc52f..64bfe6a7c489f1864decbafd10dd2d753df9afd1 100755 --- a/netmiko/ssh_dispatcher.py +++ b/netmiko/ssh_dispatcher.py @@ -13,6 +13,7 @@ from netmiko.centec import CentecOSSSH, CentecOSTelnet from netmiko.checkpoint import CheckPointGaiaSSH from netmiko.ciena import CienaSaosSSH, CienaSaosTelnet, CienaSaosFileTransfer from netmiko.cisco import CiscoAsaSSH, CiscoAsaFileTransfer +from netmiko.cisco import CiscoFtdSSH from netmiko.cisco import ( CiscoIosSSH, CiscoIosFileTransfer, @@ -84,6 +85,8 @@ from netmiko.sixwind import SixwindOSSSH from netmiko.sophos import SophosSfosSSH from netmiko.terminal_server import TerminalServerSSH from netmiko.terminal_server import TerminalServerTelnet +from netmiko.tplink import TPLinkJetStreamSSH, TPLinkJetStreamTelnet +from netmiko.ubiquiti import UbiquitiEdgeRouterSSH from netmiko.ubiquiti import UbiquitiEdgeSSH from netmiko.ubiquiti import UbiquitiUnifiSwitchSSH from netmiko.vyos import VyOSSSH @@ -106,6 +109,8 @@ CLASS_MAPPER_BASE = { "apresia_aeos": ApresiaAeosSSH, "arista_eos": AristaSSH, "aruba_os": ArubaSSH, + "aruba_osswitch": HPProcurveSSH, + "aruba_procurve": HPProcurveSSH, "avaya_ers": ExtremeErsSSH, "avaya_vsp": ExtremeVspSSH, "broadcom_icos": BroadcomIcosSSH, @@ -119,6 +124,7 @@ CLASS_MAPPER_BASE = { "centec_os": CentecOSSSH, "ciena_saos": CienaSaosSSH, "cisco_asa": CiscoAsaSSH, + "cisco_ftd": CiscoFtdSSH, "cisco_ios": CiscoIosSSH, "cisco_nxos": CiscoNxosSSH, "cisco_s300": CiscoS300SSH, @@ -190,7 +196,9 @@ CLASS_MAPPER_BASE = { "ruijie_os": RuijieOSSSH, "sixwind_os": SixwindOSSSH, "sophos_sfos": SophosSfosSSH, + "tplink_jetstream": TPLinkJetStreamSSH, "ubiquiti_edge": UbiquitiEdgeSSH, + "ubiquiti_edgerouter": UbiquitiEdgeRouterSSH, "ubiquiti_edgeswitch": UbiquitiEdgeSSH, "ubiquiti_unifiswitch": UbiquitiUnifiSwitchSSH, "vyatta_vyos": VyOSSSH, @@ -232,6 +240,7 @@ FILE_TRANSFER_MAP = new_mapper # Add telnet drivers CLASS_MAPPER["apresia_aeos_telnet"] = ApresiaAeosTelnet CLASS_MAPPER["arista_eos_telnet"] = AristaTelnet +CLASS_MAPPER["aruba_procurve_telnet"] = HPProcurveTelnet CLASS_MAPPER["brocade_fastiron_telnet"] = RuckusFastironTelnet CLASS_MAPPER["brocade_netiron_telnet"] = ExtremeNetironTelnet CLASS_MAPPER["calix_b6_telnet"] = CalixB6Telnet @@ -259,6 +268,7 @@ CLASS_MAPPER["rad_etx_telnet"] = RadETXTelnet CLASS_MAPPER["raisecom_telnet"] = RaisecomRoapTelnet CLASS_MAPPER["ruckus_fastiron_telnet"] = RuckusFastironTelnet CLASS_MAPPER["ruijie_os_telnet"] = RuijieOSTelnet +CLASS_MAPPER["tplink_jetstream_telnet"] = TPLinkJetStreamTelnet CLASS_MAPPER["yamaha_telnet"] = YamahaTelnet CLASS_MAPPER["zte_zxros_telnet"] = ZteZxrosTelnet diff --git a/netmiko/tplink/__init__.py b/netmiko/tplink/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..4e53b9a9942f245ef7872b318a03e4205795725f --- /dev/null +++ b/netmiko/tplink/__init__.py @@ -0,0 +1,3 @@ +from netmiko.tplink.tplink_jetstream import TPLinkJetStreamSSH, TPLinkJetStreamTelnet + +__all__ = ["TPLinkJetStreamSSH", "TPLinkJetStreamTelnet"] diff --git a/netmiko/tplink/tplink_jetstream.py b/netmiko/tplink/tplink_jetstream.py new file mode 100644 index 0000000000000000000000000000000000000000..45a61a1f614adb76e822a2709a9d19b05edda8c8 --- /dev/null +++ b/netmiko/tplink/tplink_jetstream.py @@ -0,0 +1,172 @@ +import re +import time + +from cryptography import utils as crypto_utils +from cryptography.hazmat.primitives.asymmetric import dsa + +from netmiko import log +from netmiko.cisco_base_connection import CiscoSSHConnection +from netmiko.ssh_exception import NetmikoTimeoutException + + +class TPLinkJetStreamBase(CiscoSSHConnection): + def __init__(self, **kwargs): + # TP-Link doesn't have a way to set terminal width which breaks cmd_verify + if kwargs.get("global_cmd_verify") is None: + kwargs["global_cmd_verify"] = False + # TP-Link uses "\r\n" as default_enter for SSH and Telnet + if kwargs.get("default_enter") is None: + kwargs["default_enter"] = "\r\n" + return super().__init__(**kwargs) + + def session_preparation(self): + """ + Prepare the session after the connection has been established. + """ + delay_factor = self.select_delay_factor(delay_factor=0) + time.sleep(0.3 * delay_factor) + self.clear_buffer() + self._test_channel_read(pattern=r"[>#]") + self.set_base_prompt() + self.enable() + self.disable_paging() + # Clear the read buffer + time.sleep(0.3 * self.global_delay_factor) + self.clear_buffer() + + def enable(self, cmd="", pattern="ssword", re_flags=re.IGNORECASE): + """ + TPLink JetStream requires you to first execute "enable" and then execute "enable-admin". + This is necessary as "configure" is generally only available at "enable-admin" level + + If the user does not have the Admin role, he will need to execute enable-admin to really + enable all functions. + """ + + # If end-user passes in "cmd" execute that using normal process. + if cmd: + return super().enable(cmd=cmd, pattern=pattern, re_flags=re_flags) + + output = "" + msg = ( + "Failed to enter enable mode. Please ensure you pass " + "the 'secret' argument to ConnectHandler." + ) + + cmds = ["enable", "enable-admin"] + if not self.check_enable_mode(): + for cmd in cmds: + self.write_channel(self.normalize_cmd(cmd)) + try: + output += self.read_until_prompt_or_pattern( + pattern=pattern, re_flags=re_flags + ) + self.write_channel(self.normalize_cmd(self.secret)) + output += self.read_until_prompt() + except NetmikoTimeoutException: + raise ValueError(msg) + if not self.check_enable_mode(): + raise ValueError(msg) + return output + + def config_mode(self, config_command="configure"): + """Enter configuration mode.""" + return super().config_mode(config_command=config_command) + + def exit_config_mode(self, exit_config="exit", pattern=r"#"): + """ + Exit config mode. + + Like the Mellanox equipment, the TP-Link Jetstream does not + support a single command to completely exit the configuration mode. + + Consequently, need to keep checking and sending "exit". + """ + output = "" + check_count = 12 + while check_count >= 0: + if self.check_config_mode(): + self.write_channel(self.normalize_cmd(exit_config)) + output += self.read_until_pattern(pattern=pattern) + else: + break + check_count -= 1 + + if self.check_config_mode(): + raise ValueError("Failed to exit configuration mode") + log.debug(f"exit_config_mode: {output}") + + return output + + def check_config_mode(self, check_string="(config", pattern=r"#"): + """Check whether device is in configuration mode. Return a boolean.""" + return super().check_config_mode(check_string=check_string, pattern=pattern) + + def set_base_prompt( + self, pri_prompt_terminator=">", alt_prompt_terminator="#", delay_factor=1 + ): + """ + Sets self.base_prompt + + Used as delimiter for stripping of trailing prompt in output. + + Should be set to something that is general and applies in multiple + contexts. For TP-Link this will be the router prompt with > or # + stripped off. + + This will be set on logging in, but not when entering system-view + """ + return super().set_base_prompt( + pri_prompt_terminator=pri_prompt_terminator, + alt_prompt_terminator=alt_prompt_terminator, + delay_factor=delay_factor, + ) + + +class TPLinkJetStreamSSH(TPLinkJetStreamBase): + def _override_check_dsa_parameters(parameters): + """ + Override check_dsa_parameters from cryptography's dsa.py + + Without this the error below occurs: + + ValueError: p must be exactly 1024, 2048, or 3072 bits long + + Allows for shorter or longer parameters.p to be returned + from the server's host key. This is a HORRIBLE hack and a + security risk, please remove if possible! + + By now, with firmware: + + 2.0.5 Build 20200109 Rel.36203(s) + + It's still not possible to remove this hack. + """ + if crypto_utils.bit_length(parameters.q) not in [160, 256]: + raise ValueError("q must be exactly 160 or 256 bits long") + + if not (1 < parameters.g < parameters.p): + raise ValueError("g, p don't satisfy 1 < g < p.") + + dsa._check_dsa_parameters = _override_check_dsa_parameters + + +class TPLinkJetStreamTelnet(TPLinkJetStreamBase): + def telnet_login( + self, + pri_prompt_terminator="#", + alt_prompt_terminator=">", + username_pattern=r"User:", + pwd_pattern=r"Password:", + delay_factor=1, + max_loops=60, + ): + """Telnet login: can be username/password or just password.""" + super().telnet_login( + pri_prompt_terminator=pri_prompt_terminator, + alt_prompt_terminator=alt_prompt_terminator, + username_pattern=username_pattern, + pwd_pattern=pwd_pattern, + delay_factor=delay_factor, + max_loops=max_loops, + ) diff --git a/netmiko/ubiquiti/__init__.py b/netmiko/ubiquiti/__init__.py index 214673edf75f9a7f8cba2dc85edb3609a1b02418..f650a7a640b2eb66c8b5c90853b3769f7034029c 100644 --- a/netmiko/ubiquiti/__init__.py +++ b/netmiko/ubiquiti/__init__.py @@ -1,4 +1,10 @@ from netmiko.ubiquiti.edge_ssh import UbiquitiEdgeSSH +from netmiko.ubiquiti.edgerouter_ssh import UbiquitiEdgeRouterSSH from netmiko.ubiquiti.unifiswitch_ssh import UbiquitiUnifiSwitchSSH -__all__ = ["UbiquitiEdgeSSH", "UnifiSwitchSSH", "UbiquitiUnifiSwitchSSH"] +__all__ = [ + "UbiquitiEdgeRouterSSH", + "UbiquitiEdgeSSH", + "UnifiSwitchSSH", + "UbiquitiUnifiSwitchSSH", +] diff --git a/netmiko/ubiquiti/edgerouter_ssh.py b/netmiko/ubiquiti/edgerouter_ssh.py new file mode 100644 index 0000000000000000000000000000000000000000..dbf88c6bfa5195c0dd1b28f2968d3fa249c5c43e --- /dev/null +++ b/netmiko/ubiquiti/edgerouter_ssh.py @@ -0,0 +1,25 @@ +import time +from netmiko.vyos.vyos_ssh import VyOSSSH + + +class UbiquitiEdgeRouterSSH(VyOSSSH): + """Implement methods for interacting with EdgeOS EdgeRouter network devices.""" + + def session_preparation(self): + """Prepare the session after the connection has been established.""" + self._test_channel_read() + self.set_base_prompt() + self.set_terminal_width(command="terminal width 512") + self.disable_paging(command="terminal length 0") + # Clear the read buffer + time.sleep(0.3 * self.global_delay_factor) + self.clear_buffer() + + def save_config(self, cmd="save", confirm=False, confirm_response=""): + """Saves Config.""" + if confirm is True: + raise ValueError("EdgeRouter does not support save_config confirmation.") + output = self.send_command(command_string=cmd) + if "Done" not in output: + raise ValueError(f"Save failed with following errors:\n\n{output}") + return output diff --git a/netmiko/utilities.py b/netmiko/utilities.py index 5a5d4f0f4882647732ca7897a5000a8c88ea11d2..9acfd448c9494dfd1f0087397ec1ecb370eb8ec2 100644 --- a/netmiko/utilities.py +++ b/netmiko/utilities.py @@ -9,6 +9,14 @@ from datetime import datetime from netmiko._textfsm import _clitable as clitable from netmiko._textfsm._clitable import CliTableError +try: + from ttp import ttp + + TTP_INSTALLED = True + +except ImportError: + TTP_INSTALLED = False + try: from genie.conf.base import Device from genie.libs.parser.utils import get_parser @@ -18,6 +26,12 @@ try: except ImportError: GENIE_INSTALLED = False +# If we are on python < 3.7, we need to force the import of importlib.resources backport +try: + from importlib.resources import path as importresources_path +except ModuleNotFoundError: + from importlib_resources import path as importresources_path + try: import serial.tools.list_ports @@ -25,7 +39,6 @@ try: except ImportError: PYSERIAL_INSTALLED = False - # Dictionary mapping 'show run' for vendors with different command SHOW_RUN_MAPPER = { "juniper": "show configuration", @@ -109,8 +122,8 @@ def find_cfg_file(file_name=None): if files: return files[0] raise IOError( - ".netmiko.yml file not found in NETMIKO_TOOLS environment variable directory, current " - "directory, or home directory." + ".netmiko.yml file not found in NETMIKO_TOOLS environment variable directory," + " current directory, or home directory." ) @@ -224,25 +237,59 @@ def check_serial_port(name): raise ValueError(msg) -def get_template_dir(): - """Find and return the ntc-templates/templates dir.""" - try: - template_dir = os.path.expanduser(os.environ["NET_TEXTFSM"]) +def get_template_dir(_skip_ntc_package=False): + """ + Find and return the directory containing the TextFSM index file. + + Order of preference is: + 1) Find directory in `NET_TEXTFSM` Environment Variable. + 2) Check for pip installed `ntc-templates` location in this environment. + 3) ~/ntc-templates/templates. + + If `index` file is not found in any of these locations, raise ValueError + + :return: directory containing the TextFSM index file + + """ + + msg = """ +Directory containing TextFSM index file not found. + +Please set the NET_TEXTFSM environment variable to point at the directory containing your TextFSM +index file. + +Alternatively, `pip install ntc-templates` (if using ntc-templates). + +""" + + # Try NET_TEXTFSM environment variable + template_dir = os.environ.get("NET_TEXTFSM") + if template_dir is not None: + template_dir = os.path.expanduser(template_dir) index = os.path.join(template_dir, "index") if not os.path.isfile(index): # Assume only base ./ntc-templates specified template_dir = os.path.join(template_dir, "templates") - except KeyError: - # Construct path ~/ntc-templates/templates - home_dir = os.path.expanduser("~") - template_dir = os.path.join(home_dir, "ntc-templates", "templates") + + else: + # Try 'pip installed' ntc-templates + try: + with importresources_path( + package="ntc_templates", resource="templates" + ) as posix_path: + # Example: /opt/venv/netmiko/lib/python3.8/site-packages/ntc_templates/templates + template_dir = str(posix_path) + # This is for Netmiko automated testing + if _skip_ntc_package: + raise ModuleNotFoundError() + + except ModuleNotFoundError: + # Finally check in ~/ntc-templates/templates + home_dir = os.path.expanduser("~") + template_dir = os.path.join(home_dir, "ntc-templates", "templates") index = os.path.join(template_dir, "index") if not os.path.isdir(template_dir) or not os.path.isfile(index): - msg = """ -Valid ntc-templates not found, please install https://github.com/networktocode/ntc-templates -and then set the NET_TEXTFSM environment variable to point to the ./ntc-templates/templates -directory.""" raise ValueError(msg) return os.path.abspath(template_dir) @@ -305,6 +352,25 @@ def get_structured_data(raw_output, platform=None, command=None, template=None): ) +def get_structured_data_ttp(raw_output, template=None): + """ + Convert raw CLI output to structured data using TTP template. + + You can use a straight TextFSM file i.e. specify "template" + """ + if not TTP_INSTALLED: + msg = "\nTTP is not installed. Please PIP install ttp:\n" "pip install ttp\n" + raise ValueError(msg) + + try: + if template: + ttp_parser = ttp(data=raw_output, template=template) + ttp_parser.parse(one=True) + return ttp_parser.result(format="raw") + except Exception: + return raw_output + + def get_structured_data_genie(raw_output, platform, command): if not sys.version_info >= (3, 4): raise ValueError("Genie requires Python >= 3.4") @@ -345,7 +411,7 @@ def get_structured_data_genie(raw_output, platform, command): device.custom["abstraction"]["order"] = ["os"] device.cli = AttrDict({"execute": None}) try: - # Test of whether their is a parser for the given command (will return Exception if fails) + # Test whether there is a parser for given command (return Exception if fails) get_parser(command, device) parsed_output = device.parse(command, output=raw_output) return parsed_output diff --git a/requirements-dev.txt b/requirements-dev.txt index 862bdef84d634ec8f00863ef4aa11e864444f801..85ea63a32579226d3fbdf04ad0bb9422749362fa 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -6,3 +6,4 @@ twine==1.13.0 pysnmp==4.4.12 pdoc3==0.6.3 -r requirements.txt +-r requirements-ttp.txt diff --git a/requirements-genie.txt b/requirements-genie.txt index e1290255e894a03cc5cb20b68a2a527e77c74adb..eb69647334bb74db69b9ea5b97746606b07f307e 100644 --- a/requirements-genie.txt +++ b/requirements-genie.txt @@ -1,2 +1,2 @@ -pyats==20.2 -genie==20.2 +pyats==20.9 +genie==20.9 diff --git a/requirements-ttp.txt b/requirements-ttp.txt new file mode 100644 index 0000000000000000000000000000000000000000..806325661d1e628f94a8c07e30a1ec8c4ba92398 --- /dev/null +++ b/requirements-ttp.txt @@ -0,0 +1 @@ +ttp>=0.4.0 diff --git a/setup.py b/setup.py index 59c9fb7baebeaa02de6e23524047ccaecece36e1..ae68afddd8b95ed1add828c2e1dc7ac3854a2d42 100644 --- a/setup.py +++ b/setup.py @@ -47,10 +47,12 @@ setup( packages=find_packages(exclude=("test*",)), install_requires=[ "setuptools>=38.4.0", - "paramiko>=2.4.3", + "paramiko>=2.6.0", "scp>=0.13.2", + "tenacity", + "ntc-templates", "pyserial", - "textfsm", + "importlib_resources ; python_version<'3.7'", ], - extras_require={"test": ["pyyaml==5.1.2", "pytest>=5.1.2"]}, + extras_require={"test": ["pyyaml>=5.1.2", "pytest>=5.1.2"]}, ) diff --git a/tests/conftest.py b/tests/conftest.py index 5c8912f5836083054f035afa0b1819d98a05da79..943eb26417376572e736f4d2adafebe1c3313e0a 100755 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -164,10 +164,10 @@ def delete_file_nxos(ssh_conn, dest_file_system, dest_file): full_file_name = "{}{}".format(dest_file_system, dest_file) cmd = "delete {}".format(full_file_name) - output = ssh_conn.send_command_timing(cmd) + output = ssh_conn.send_command(cmd, expect_string=r"Do you want to delete") if "yes/no/abort" in output and dest_file in output: - output += ssh_conn.send_command_timing( - "y", strip_command=False, strip_prompt=False + output += ssh_conn.send_command( + "y", expect_string=r"#", strip_command=False, strip_prompt=False ) return output else: diff --git a/tests/etc/commands.yml.example b/tests/etc/commands.yml.example index 56592afd03549a036bc1120aeab35ead6d48f942..0149b10421bf7ddb3690b7517428319f2ec13f41 100644 --- a/tests/etc/commands.yml.example +++ b/tests/etc/commands.yml.example @@ -442,3 +442,29 @@ huawei_smartax: config: - acl 2456 config_verification: "display current-configuration | include acl 2456" + +tplink_jetstream: + version: "show system-info" + basic: "show interface vlan 1" + extended_output: "show running-config all" + config: + - "no clipaging" + config_verification: "show running-config | include clipaging" + +tplink_jetstream_telnet: + version: "show system-info" + basic: "show interface vlan 1" + extended_output: "show running-config all" + config: + - "no clipaging" + config_verification: "show running-config | include clipaging" + +ubiquiti_edgerouter: + version: "show version" + basic: "show interfaces switch switch0" + extended_output: "show configuration all" + config: + - "set system coredump enabled true" + - "set system coredump enabled false" + support_commit: True + config_verification: "show system coredump enabled" diff --git a/tests/etc/responses.yml.example b/tests/etc/responses.yml.example index 47519dcc8e0862f84760b649a8fa572c70abf9c9..e001f330e28c0138a1a62e02f34516fcefe998df 100644 --- a/tests/etc/responses.yml.example +++ b/tests/etc/responses.yml.example @@ -290,3 +290,31 @@ huawei_smartax: interface_ip: 192.0.2.1 version_banner: "VERSION :" multiple_line_output: "" + +tplink_jetstream: + base_prompt: "T1500G-10PS" + router_prompt: "T1500G-10PS>" + enable_prompt: "T1500G-10PS#" + interface_ip: 192.168.0.1 + version_banner: "Software Version" + multiple_line_output: "interface vlan 1" + cmd_response_final: "no clipaging" + +tplink_jetstream_telnet: + base_prompt: "T1500G-10PS" + router_prompt: "T1500G-10PS>" + enable_prompt: "T1500G-10PS#" + interface_ip: 192.168.0.1 + version_banner: "Software Version" + multiple_line_output: "interface vlan 1" + cmd_response_final: "no clipaging" + +ubiquiti_edgerouter: + base_prompt: "ubnt@edgerouter" + router_prompt: "ubnt@edgerouter:~$" + enable_prompt: "ubnt@edgerouter:~$" + interface_ip: "192.168.1.1" + version_banner: "Ubiquiti Networks, Inc" + multiple_line_output: "switch switch0 {" + cmd_response_init: "enabled true" + cmd_response_final: "enabled false" diff --git a/tests/etc/test_devices.yml.example b/tests/etc/test_devices.yml.example index 97c4722b44415d834770c3f3081c604884cfeefa..ab79e71e6f1ec441a26b806f62864b02baae1a6e 100644 --- a/tests/etc/test_devices.yml.example +++ b/tests/etc/test_devices.yml.example @@ -221,3 +221,21 @@ huawei_smartax: ip: 192.0.2.1 username: TEST password: TEST + +tplink_jetstream: + device_type: tplink_jetstream + ip: 192.168.0.1 + username: admin + password: 'admin' + +tplink_jetstream_telnet: + device_type: tplink_jetstream_telnet + ip: 192.168.0.1 + username: admin + password: 'admin' + +ubiquiti_edgerouter: + device_type: ubiquiti_edgerouter + ip: 192.168.1.1 + username: ubnt + password: ubnt diff --git a/tests/performance/test_devices.yml b/tests/performance/test_devices.yml deleted file mode 100644 index 3f0f0d05d76d6722ce251fcbcb70851db4c1bba0..0000000000000000000000000000000000000000 --- a/tests/performance/test_devices.yml +++ /dev/null @@ -1,15 +0,0 @@ ---- -cisco1: - device_type: cisco_ios - host: cisco1.lasthop.io - username: pyclass - -cisco3: - device_type: cisco_xe - host: cisco3.lasthop.io - username: pyclass - -cisco5: - device_type: cisco_xe - host: cisco5.lasthop.io - username: pyclass diff --git a/tests/test_ios/remove_delay_cisco881.sh b/tests/test_ios/remove_delay_cisco881.sh deleted file mode 100755 index 41ae9fe9ab861ec0d68dabf4a540bff2ffdc47d8..0000000000000000000000000000000000000000 --- a/tests/test_ios/remove_delay_cisco881.sh +++ /dev/null @@ -1,4 +0,0 @@ -#!/bin/bash - -# Remove delay for all; must be root or sudo to execute -sudo -s /sbin/tc qdisc del dev eth0 root diff --git a/tests/test_ios/test_ios.sh b/tests/test_ios/test_ios.sh deleted file mode 100755 index 9e3a0360d617c068b01a64b4920d5543758f8b18..0000000000000000000000000000000000000000 --- a/tests/test_ios/test_ios.sh +++ /dev/null @@ -1,42 +0,0 @@ -#!/bin/sh - -RETURN_CODE=0 - -echo "Starting tests...good luck:" \ -&& echo "Cisco IOS" \ -&& cd .. \ -&& py.test -s -x -v test_netmiko_show.py --test_device cisco881 \ -&& py.test -s -x -v test_netmiko_config.py --test_device cisco881 \ -&& py.test -s -x -v test_netmiko_config_acl.py --test_device cisco881 \ -&& py.test -s -x -v test_netmiko_tcl.py --test_device cisco881 \ -&& py.test -s -x -v test_netmiko_scp.py --test_device cisco881 \ -|| RETURN_CODE=1 - -exit $RETURN_CODE - -#&& py.test -v test_netmiko_tcl.py --test_device cisco881_key \ -#&& py.test -v test_netmiko_show.py --test_device cisco881_key \ -#&& py.test -v test_netmiko_config.py --test_device cisco881_key \ -#&& py.test -v test_netmiko_config_acl.py --test_device cisco881_key \ - -#&& py.test -v test_netmiko_session_log.py --test_device cisco881_slog \ - -#&& py.test -v test_netmiko_tcl.py --test_device cisco881_fast \ -#&& py.test -v test_netmiko_show.py --test_device cisco881_fast \ -#&& py.test -v test_netmiko_config.py --test_device cisco881_fast \ - -#&& py.test -v test_netmiko_show.py --test_device cisco881_ssh_config \ -#&& py.test -v test_netmiko_config.py --test_device cisco881_ssh_config \ -#&& py.test -v test_netmiko_config_acl.py --test_device cisco881_ssh_config \ - -#&& py.test -v test_netmiko_show.py --test_device cisco881_ssh_proxyjump \ -#&& py.test -v test_netmiko_config.py --test_device cisco881_ssh_proxyjump \ -#&& py.test -v test_netmiko_config_acl.py --test_device cisco881_ssh_proxyjump \ - -#&& py.test -v test_netmiko_show.py --test_device cisco881_telnet \ -#&& py.test -v test_netmiko_config.py --test_device cisco881_telnet \ -#&& py.test -v test_netmiko_config_acl.py --test_device cisco881_telnet \ -#&& py.test -s -v test_netmiko_autodetect.py --test_device cisco881 \ - -## && py.test -v test_netmiko_scp.py --test_device cisco881_key \ -## && py.test -v test_netmiko_scp.py --test_device cisco881_fast \ diff --git a/tests/test_netmiko_show.py b/tests/test_netmiko_show.py index 15dc876447da78549392b577698f455dfcee0dc3..3fa2ebcac855d3cfd418f472d9f9e8775c8c43d8 100755 --- a/tests/test_netmiko_show.py +++ b/tests/test_netmiko_show.py @@ -16,6 +16,7 @@ test_disconnect: cleanly disconnect the SSH session """ import pytest import time +import os from datetime import datetime from netmiko.utilities import select_cmd_verify @@ -28,9 +29,14 @@ def bogus_func(obj, *args, **kwargs): def test_disable_paging(net_connect, commands, expected_responses): """Verify paging is disabled by looking for string after when paging would normally occur.""" + # FIX: these really shouldn't be necessary. if net_connect.device_type == "arista_eos": # Arista logging buffer gets enormous net_connect.send_command("clear logging") + elif net_connect.device_type == "arista_eos": + # NX-OS logging buffer gets enormous (NX-OS fails when testing very high-latency + + # packet loss) + net_connect.send_command("clear logging logfile") multiple_line_output = net_connect.send_command(commands["extended_output"]) assert expected_responses["multiple_line_output"] in multiple_line_output @@ -185,6 +191,40 @@ def test_send_command_textfsm(net_connect, commands, expected_responses): assert isinstance(show_ip_alt, list) +def test_send_command_ttp(net_connect): + """ + Verify a command can be sent down the channel + successfully using send_command method. + """ + + base_platform = net_connect.device_type + if base_platform.count("_") >= 2: + # Strip off the _ssh, _telnet, _serial + base_platform = base_platform.split("_")[:-1] + base_platform = "_".join(base_platform) + if base_platform not in ["cisco_ios"]: + assert pytest.skip("TTP template not existing for this platform") + else: + time.sleep(1) + net_connect.clear_buffer() + + # write a simple template to file + ttp_raw_template = """ + interface {{ interface }} + description {{ description }} + """ + ttp_temp_filename = "show_run_interfaces.ttp" + with open(ttp_temp_filename, "w") as writer: + writer.write(ttp_raw_template) + + command = "show run" + show_ip_alt = net_connect.send_command( + command, use_ttp=True, ttp_template=ttp_temp_filename + ) + os.remove(ttp_temp_filename) + assert isinstance(show_ip_alt, list) + + def test_send_command_genie(net_connect, commands, expected_responses): """Verify a command can be sent down the channel successfully using send_command method.""" diff --git a/tests/test_nxos.sh b/tests/test_nxos.sh deleted file mode 100755 index 0dea482783d4f6df2cb6862f9838ac61972c8141..0000000000000000000000000000000000000000 --- a/tests/test_nxos.sh +++ /dev/null @@ -1,12 +0,0 @@ -#!/bin/sh - -RETURN_CODE=0 - -echo "Starting tests...good luck:" \ -&& echo "Cisco NXOS" \ -&& py.test -v test_netmiko_scp.py --test_device nxos1 \ -&& py.test -v test_netmiko_show.py --test_device nxos1 \ -&& py.test -v test_netmiko_config.py --test_device nxos1 \ -|| RETURN_CODE=1 - -exit $RETURN_CODE diff --git a/tests/test_tplink_jetstream.sh b/tests/test_tplink_jetstream.sh new file mode 100755 index 0000000000000000000000000000000000000000..9c338c0f3aab8722754af27ece8469263258640c --- /dev/null +++ b/tests/test_tplink_jetstream.sh @@ -0,0 +1,11 @@ +#!/bin/sh + +RETURN_CODE=0 + +# Exit on the first test failure and set RETURN_CODE = 1 +echo "Starting tests...good luck:" \ +&& py.test -v test_netmiko_show.py --test_device tplink_jetstream \ +&& py.test -v test_netmiko_config.py --test_device tplink_jetstream \ +|| RETURN_CODE=1 + +exit $RETURN_CODE diff --git a/tests/test_tplink_jetstream_telnet.sh b/tests/test_tplink_jetstream_telnet.sh new file mode 100755 index 0000000000000000000000000000000000000000..2d5007b8d5f7b8e34773b0fb9747ec9bf5a75161 --- /dev/null +++ b/tests/test_tplink_jetstream_telnet.sh @@ -0,0 +1,11 @@ +#!/bin/sh + +RETURN_CODE=0 + +# Exit on the first test failure and set RETURN_CODE = 1 +echo "Starting tests...good luck:" \ +&& py.test -v test_netmiko_show.py --test_device tplink_jetstream_telnet \ +&& py.test -v test_netmiko_config.py --test_device tplink_jetstream_telnet \ +|| RETURN_CODE=1 + +exit $RETURN_CODE diff --git a/tests/test_ubiquiti_edgerouter.sh b/tests/test_ubiquiti_edgerouter.sh new file mode 100755 index 0000000000000000000000000000000000000000..1ce213173b820372d37faff083ecea0eaa96f26a --- /dev/null +++ b/tests/test_ubiquiti_edgerouter.sh @@ -0,0 +1,11 @@ +#!/bin/sh + +RETURN_CODE=0 + +# Exit on the first test failure and set RETURN_CODE = 1 +echo "Starting tests...good luck:" \ +&& py.test -v test_netmiko_show.py --test_device ubiquiti_edgerouter \ +&& py.test -v test_netmiko_config.py --test_device ubiquiti_edgerouter \ +|| RETURN_CODE=1 + +exit $RETURN_CODE diff --git a/tests/unit/test_utilities.py b/tests/unit/test_utilities.py index 29d984b22e6c0656d4cd771f1d95bc865f6b54df..4fa8fceb372d8408e6f2c5f862687c53017e3cca 100755 --- a/tests/unit/test_utilities.py +++ b/tests/unit/test_utilities.py @@ -1,7 +1,9 @@ #!/usr/bin/env python import os +import sys from os.path import dirname, join, relpath +import pytest from netmiko import utilities from netmiko._textfsm import _clitable as clitable @@ -196,6 +198,43 @@ def test_textfsm_w_index(): assert result == [{"model": "4500"}] +def test_ntc_templates_discovery(): + """ + Verify Netmiko uses proper ntc-templates: + + Order of preference is: + 1. Find directory in `NET_TEXTFSM` Environment Variable. + 2. Check for pip installed `ntc-templates` location in this environment. + 3. ~/ntc-templates/templates. + + If `index` file is not found in any of these locations, raise ValueError + """ + + # Check environment variable first + os.environ["NET_TEXTFSM"] = RELATIVE_RESOURCE_FOLDER + ntc_path = utilities.get_template_dir() + assert ntc_path == RESOURCE_FOLDER + + # Next should be PIP installed ntc-tempaltes + del os.environ["NET_TEXTFSM"] + ntc_path = utilities.get_template_dir() + for py_path in sys.path: + if "site-packages" in py_path: + packages_dir = py_path + break + assert ntc_path == f"{packages_dir}/ntc_templates/templates" + + # Next should use local index file in ~ + home_dir = os.path.expanduser("~") + # Will not work for CI-CD without pain so just test locally + if "kbyers" in home_dir: + ntc_path = utilities.get_template_dir(_skip_ntc_package=True) + assert ntc_path == f"{home_dir}/ntc-templates/templates" + else: + with pytest.raises(ValueError): + ntc_path = utilities.get_template_dir(_skip_ntc_package=True) + + def test_textfsm_index_relative_path(): """Test relative path for textfsm ntc directory""" os.environ["NET_TEXTFSM"] = RELATIVE_RESOURCE_FOLDER diff --git a/tests_new/cisco_ios_commands.txt b/tests_new/cisco_ios_commands.txt new file mode 100644 index 0000000000000000000000000000000000000000..a5ee9c047743140cd6e30e0eca9e077491e2bf5a --- /dev/null +++ b/tests_new/cisco_ios_commands.txt @@ -0,0 +1,3 @@ +logging buffered 9990 +logging buffered 8880 +no logging console diff --git a/tests_new/conftest.py b/tests_new/conftest.py new file mode 100755 index 0000000000000000000000000000000000000000..19ee55afbb0a417829c744d414ba5ba75069ad05 --- /dev/null +++ b/tests_new/conftest.py @@ -0,0 +1,539 @@ +#!/usr/bin/env python +"""py.test fixtures to be used in netmiko test suite.""" +from os import path +import os + +import pytest + +from netmiko import ConnectHandler, FileTransfer, InLineTransfer, SSHDetect +from test_utils import parse_yaml + + +PWD = path.dirname(path.realpath(__file__)) + + +def pytest_addoption(parser): + """Add test_device option to py.test invocations.""" + parser.addoption( + "--test_device", + action="store", + dest="test_device", + type=str, + help="Specify the platform type to test on", + ) + + +@pytest.fixture(scope="module") +def net_connect(request): + """ + Create the SSH connection to the remote device + + Return the netmiko connection object + """ + device_under_test = request.config.getoption("test_device") + test_devices = parse_yaml(PWD + "/etc/test_devices.yml") + device = test_devices[device_under_test] + device["verbose"] = False + conn = ConnectHandler(**device) + return conn + + +@pytest.fixture(scope="module") +def net_connect_cmd_verify(request): + """ + Create the SSH connection to the remote device + + Return the netmiko connection object + + Set global_cmd_verify = False + """ + device_under_test = request.config.getoption("test_device") + test_devices = parse_yaml(PWD + "/etc/test_devices.yml") + device = test_devices[device_under_test] + device["verbose"] = False + device["fast_cli"] = False + device["global_cmd_verify"] = False + conn = ConnectHandler(**device) + return conn + + +@pytest.fixture(scope="function") +def net_connect_newconn(request): + """ + Create the SSH connection to the remote device + + Return the netmiko connection object. + Force a new connection for each test. + """ + device_under_test = request.config.getoption("test_device") + test_devices = parse_yaml(PWD + "/etc/test_devices.yml") + device = test_devices[device_under_test] + device["verbose"] = False + conn = ConnectHandler(**device) + return conn + + +@pytest.fixture() +def net_connect_cm(request): + """ + Create the SSH connection to the remote device using a context manager + retrieve the find_prompt() data and close the connection. + """ + device_under_test = request.config.getoption("test_device") + test_devices = parse_yaml(PWD + "/etc/test_devices.yml") + device = test_devices[device_under_test] + device["verbose"] = False + my_prompt = "" + with ConnectHandler(**device) as conn: + my_prompt = conn.find_prompt() + return my_prompt + + +@pytest.fixture(scope="module") +def net_connect_slog_wr(request): + """ + Create the SSH connection to the remote device. Modify session_log init arguments. + + Return the netmiko connection object. + """ + device_under_test = request.config.getoption("test_device") + test_devices = parse_yaml(PWD + "/etc/test_devices.yml") + device = test_devices[device_under_test] + device["verbose"] = False + # Overwrite default session_log location + device["session_log"] = "SLOG/cisco881_slog_wr.log" + device["session_log_record_writes"] = True + conn = ConnectHandler(**device) + return conn + + +@pytest.fixture(scope="module") +def device_slog(request): + """ + Create the SSH connection to the remote device. Modify session_log init arguments. + + Return the netmiko device (not connected) + """ + device_under_test = request.config.getoption("test_device") + test_devices = parse_yaml(PWD + "/etc/test_devices.yml") + device = test_devices[device_under_test] + device["verbose"] = False + device["session_log_file_mode"] = "append" + return device + + +@pytest.fixture(scope="module") +def expected_responses(request): + """ + Parse the responses.yml file to get a responses dictionary + """ + device_under_test = request.config.getoption("test_device") + responses = parse_yaml(PWD + "/etc/responses.yml") + return responses[device_under_test] + + +@pytest.fixture(scope="module") +def commands(request): + """ + Parse the commands.yml file to get a commands dictionary + """ + device_under_test = request.config.getoption("test_device") + test_devices = parse_yaml(PWD + "/etc/test_devices.yml") + device = test_devices[device_under_test] + test_platform = device["device_type"] + + commands_yml = parse_yaml(PWD + "/etc/commands.yml") + + # Nokia SR-OS driver is overloaded with both classical-CLI and MD-CLI + # Swap out the commands to be the MD-CLI commands + if device_under_test == "sros1_md": + test_platform = "nokia_sros_md" + + return commands_yml[test_platform] + + +def delete_file_nxos(ssh_conn, dest_file_system, dest_file): + """ + nxos1# delete bootflash:test2.txt + Do you want to delete "/test2.txt" ? (yes/no/abort) [y] y + """ + if not dest_file_system: + raise ValueError("Invalid file system specified") + if not dest_file: + raise ValueError("Invalid dest file specified") + + full_file_name = "{}{}".format(dest_file_system, dest_file) + + cmd = "delete {}".format(full_file_name) + output = ssh_conn.send_command(cmd, expect_string=r"Do you want to delete") + if "yes/no/abort" in output and dest_file in output: + output += ssh_conn.send_command( + "y", expect_string=r"#", strip_command=False, strip_prompt=False + ) + return output + else: + output += ssh_conn.send_command_timing("abort") + raise ValueError("An error happened deleting file on Cisco NX-OS") + + +def delete_file_xr(ssh_conn, dest_file_system, dest_file): + """ + Delete a remote file for a Cisco IOS-XR device: + + delete disk0:/test9.txt + Mon Aug 31 17:56:15.008 UTC + Delete disk0:/test9.txt[confirm] + """ + if not dest_file_system: + raise ValueError("Invalid file system specified") + if not dest_file: + raise ValueError("Invalid dest file specified") + + full_file_name = f"{dest_file_system}/{dest_file}" + + cmd = f"delete {full_file_name}" + output = ssh_conn.send_command(cmd, expect_string=r"Delete.*confirm") + if "Delete" in output and dest_file in output: + output += ssh_conn.send_command("\n", expect_string=r"#") + return output + + raise ValueError("An error happened deleting file on Cisco IOS-XR") + + +def delete_file_ios(ssh_conn, dest_file_system, dest_file): + """ + Delete a remote file for a Cisco IOS device: + + cisco1#del flash:/useless_file.cfg + Delete filename [useless_file.cfg]? + Delete flash:/useless_file.cfg? [confirm]y + +delete disk0:/test9.txt +Mon Aug 31 17:56:15.008 UTC +Delete disk0:/test9.txt[confirm] + """ + if not dest_file_system: + raise ValueError("Invalid file system specified") + if not dest_file: + raise ValueError("Invalid dest file specified") + + full_file_name = f"{dest_file_system}/{dest_file}" + + cmd = f"delete {full_file_name}" + output = ssh_conn.send_command(cmd, expect_string=r"Delete filename") + if "Delete" in output and dest_file in output: + output += ssh_conn.send_command("\n", expect_string=r"confirm") + output += ssh_conn.send_command("y", expect_string=r"#") + return output + + raise ValueError("An error happened deleting file on Cisco IOS") + + +def delete_file_dellos10(ssh_conn, dest_file_system, dest_file): + """Delete a remote file for a Dell OS10 device.""" + if not dest_file: + raise ValueError("Invalid dest file specified") + + cmd = "delete home://{}".format(dest_file) + output = ssh_conn.send_command_timing(cmd) + if "Proceed to delete" in output: + output = ssh_conn.send_command_timing("yes") + return output + + raise ValueError("An error happened deleting file on Dell OS10") + + +def delete_file_generic(ssh_conn, dest_file_system, dest_file): + """Delete a remote file for a Junos device.""" + full_file_name = "{}/{}".format(dest_file_system, dest_file) + cmd = "rm {}".format(full_file_name) + output = ssh_conn._enter_shell() + output += ssh_conn.send_command_timing(cmd, strip_command=False, strip_prompt=False) + output += ssh_conn._return_cli() + return output + + +def delete_file_ciena_saos(ssh_conn, dest_file_system, dest_file): + """Delete a remote file for a ciena device.""" + full_file_name = "{}/{}".format(dest_file_system, dest_file) + cmd = "file rm {}".format(full_file_name) + output = ssh_conn.send_command_timing(cmd, strip_command=False, strip_prompt=False) + return output + + +def delete_file_nokia_sros(ssh_conn, dest_file_system, dest_file): + """Delete a remote file for a Nokia SR OS device.""" + full_file_name = "{}/{}".format(dest_file_system, dest_file) + cmd = "file delete {} force".format(full_file_name) + cmd_prefix = "" + if "@" in ssh_conn.base_prompt: + cmd_prefix = "//" + ssh_conn.send_command(cmd_prefix + "environment no more") + output = ssh_conn.send_command_timing( + cmd_prefix + cmd, strip_command=False, strip_prompt=False + ) + return output + + +@pytest.fixture(scope="module") +def scp_fixture(request): + """ + Create an FileTransfer object. + + Return a tuple (ssh_conn, scp_handle) + """ + platform_args = get_platform_args() + + # Create the files + with open("test9.txt", "w") as f: + # Not important what it is in the file + f.write("no logging console\n") + + with open("test2_src.txt", "w") as f: + # Not important what it is in the file + f.write("no logging console\n") + f.write("logging buffered 10000\n") + + device_under_test = request.config.getoption("test_device") + test_devices = parse_yaml(PWD + "/etc/test_devices.yml") + device = test_devices[device_under_test] + device["verbose"] = False + ssh_conn = ConnectHandler(**device) + + platform = device["device_type"] + dest_file_system = platform_args[platform]["file_system"] + if "ciena_saos" in platform and ssh_conn.username: + dest_file_system = f"/tmp/users/{ssh_conn.username}" + + source_file = "test9.txt" + dest_file = "test9.txt" + local_file = f"test_{platform}/testx.txt" + direction = "put" + + scp_transfer = FileTransfer( + ssh_conn, + source_file=source_file, + dest_file=dest_file, + file_system=dest_file_system, + direction=direction, + ) + scp_transfer.establish_scp_conn() + + # Make sure SCP is enabled + if platform_args[platform]["enable_scp"]: + scp_transfer.enable_scp() + + # Delete the test transfer files + if scp_transfer.check_file_exists(): + func = platform_args[platform]["delete_file"] + func(ssh_conn, dest_file_system, dest_file) + if os.path.exists(local_file): + os.remove(local_file) + return (ssh_conn, scp_transfer) + + +@pytest.fixture(scope="module") +def scp_fixture_get(request): + """ + Create an FileTransfer object (direction=get) + + Return a tuple (ssh_conn, scp_handle) + """ + platform_args = get_platform_args() + device_under_test = request.config.getoption("test_device") + test_devices = parse_yaml(PWD + "/etc/test_devices.yml") + device = test_devices[device_under_test] + device["verbose"] = False + ssh_conn = ConnectHandler(**device) + + platform = device["device_type"] + dest_file_system = platform_args[platform]["file_system"] + source_file = "test9.txt" + local_file = f"test_{platform}/testx.txt" + dest_file = local_file + direction = "get" + + scp_transfer = FileTransfer( + ssh_conn, + source_file=source_file, + dest_file=dest_file, + file_system=dest_file_system, + direction=direction, + ) + scp_transfer.establish_scp_conn() + + # Make sure SCP is enabled + if platform_args[platform]["enable_scp"]: + scp_transfer.enable_scp() + + # Delete the test transfer files + if os.path.exists(local_file): + os.remove(local_file) + return (ssh_conn, scp_transfer) + + +@pytest.fixture(scope="module") +def tcl_fixture(request): + """ + Create an InLineTransfer object. + + Return a tuple (ssh_conn, tcl_handle) + """ + # Create the files + with open("test9.txt", "w") as f: + # Not important what it is in the file + f.write("no logging console\n") + + device_under_test = request.config.getoption("test_device") + test_devices = parse_yaml(PWD + "/etc/test_devices.yml") + device = test_devices[device_under_test] + device["verbose"] = False + platform = device["device_type"] + ssh_conn = ConnectHandler(**device) + + dest_file_system = "flash:" + source_file = "test9.txt" + dest_file = "test9.txt" + local_file = f"test_{platform}/testx.txt" + direction = "put" + + tcl_transfer = InLineTransfer( + ssh_conn, + source_file=source_file, + dest_file=dest_file, + file_system=dest_file_system, + direction=direction, + ) + + # Delete the test transfer files + if tcl_transfer.check_file_exists(): + delete_file_ios(ssh_conn, dest_file_system, dest_file) + if os.path.exists(local_file): + os.remove(local_file) + + return (ssh_conn, tcl_transfer) + + +@pytest.fixture(scope="module") +def ssh_autodetect(request): + """Create an SSH autodetect object. + + return (ssh_conn, real_device_type) + """ + device_under_test = request.config.getoption("test_device") + test_devices = parse_yaml(PWD + "/etc/test_devices.yml") + device = test_devices[device_under_test] + device["verbose"] = False + my_device_type = device.pop("device_type") + device["device_type"] = "autodetect" + conn = SSHDetect(**device) + return (conn, my_device_type) + + +@pytest.fixture(scope="module") +def scp_file_transfer(request): + """ + Testing file_transfer + + Return the netmiko connection object + """ + platform_args = get_platform_args() + + # Create the files + with open("test9.txt", "w") as f: + # Not important what it is in the file + f.write("no logging console\n") + + with open("test2_src.txt", "w") as f: + # Not important what it is in the file + f.write("no logging console\n") + f.write("logging buffered 10000\n") + + device_under_test = request.config.getoption("test_device") + test_devices = parse_yaml(PWD + "/etc/test_devices.yml") + device = test_devices[device_under_test] + device["verbose"] = False + ssh_conn = ConnectHandler(**device) + + platform = device["device_type"] + file_system = platform_args[platform]["file_system"] + source_file = "test9.txt" + dest_file = "test9.txt" + local_file = f"test_{platform}/testx.txt" + alt_file = "test2.txt" + direction = "put" + + scp_transfer = FileTransfer( + ssh_conn, + source_file=source_file, + dest_file=dest_file, + file_system=file_system, + direction=direction, + ) + scp_transfer.establish_scp_conn() + + # Delete the test transfer files + if scp_transfer.check_file_exists(): + func = platform_args[platform]["delete_file"] + func(ssh_conn, file_system, dest_file) + if os.path.exists(local_file): + os.remove(local_file) + if os.path.exists(alt_file): + os.remove(alt_file) + + return (ssh_conn, file_system) + + +def get_platform_args(): + return { + "cisco_ios": { + "file_system": "flash:", + "enable_scp": True, + "delete_file": delete_file_ios, + }, + "cisco_xe": { + "file_system": "flash:", + "enable_scp": True, + "delete_file": delete_file_ios, + }, + "juniper_junos": { + "file_system": "/var/tmp", + "enable_scp": False, + "delete_file": delete_file_generic, + }, + "arista_eos": { + "file_system": "/mnt/flash", + "enable_scp": False, + "delete_file": delete_file_generic, + }, + "cisco_nxos": { + "file_system": "bootflash:", + "enable_scp": False, + "delete_file": delete_file_nxos, + }, + "cisco_xr": { + "file_system": "disk0:", + "enable_scp": False, + "delete_file": delete_file_xr, + }, + "linux": { + "file_system": "/var/tmp", + "enable_scp": False, + "delete_file": delete_file_generic, + }, + "dellos10": { + "file_system": "/home/admin", + "enable_scp": False, + "delete_file": delete_file_dellos10, + }, + "ciena_saos": { + "file_system": "/tmp/users/ciena", + "enable_scp": False, + "delete_file": delete_file_ciena_saos, + }, + "nokia_sros": { + "file_system": "cf3:", + "enable_scp": False, + "delete_file": delete_file_nokia_sros, + }, + } diff --git a/tests_new/network_utilities.py b/tests_new/network_utilities.py new file mode 100644 index 0000000000000000000000000000000000000000..9d7b4bfe58467c2b826d72b4bc1f13d13a900619 --- /dev/null +++ b/tests_new/network_utilities.py @@ -0,0 +1,125 @@ +from ipaddress import ip_address + + +def generate_acl( + acl_name="netmiko_test_large_acl", + entries=100, + base_cmd="ip access-list extended", + base_addr="192.168.0.0", +): + cmd = f"{base_cmd} {acl_name}" + acl = [cmd] + for i in range(1, entries + 1): + addr = ip_address(base_addr) + cmd = f"permit ip host {addr + i} any" + acl.append(cmd) + return acl + + +def generate_ios_acl( + acl_name="netmiko_test_large_acl", + entries=100, + base_cmd="ip access-list extended", + base_addr="192.168.0.0", +): + return generate_acl( + acl_name=acl_name, entries=entries, base_cmd=base_cmd, base_addr=base_addr + ) + + +def generate_arista_eos_acl( + acl_name="netmiko_test_large_acl", + entries=100, + base_cmd="ip access-list", + base_addr="192.168.0.0", +): + return generate_acl( + acl_name=acl_name, entries=entries, base_cmd=base_cmd, base_addr=base_addr + ) + + +generate_cisco_ios_acl = generate_ios_acl +generate_cisco_xe_acl = generate_ios_acl + + +def generate_nxos_acl( + acl_name="netmiko_test_large_acl", + entries=100, + base_cmd="ip access-list", + base_addr="192.168.0.0", +): + return generate_acl( + acl_name=acl_name, entries=entries, base_cmd=base_cmd, base_addr=base_addr + ) + + +generate_cisco_nxos_acl = generate_nxos_acl + + +def generate_cisco_xr_acl( + acl_name="netmiko_test_large_acl", + entries=100, + base_cmd="ipv4 access-list", + base_addr="192.168.0.0", +): + + # Add base ACL command + cmd = f"{base_cmd} {acl_name}" + acl = [cmd] + for i in range(1, entries + 1): + addr = ip_address(base_addr) + cmd = f"permit ipv4 host {addr + i} any" + acl.append(cmd) + return acl + + +if __name__ == "__main__": + # Test code + acl = generate_ios_acl(entries=10) + ios_ref_acl = [ + "ip access-list extended netmiko_test_large_acl", + "permit ip host 192.168.0.1 any", + "permit ip host 192.168.0.2 any", + "permit ip host 192.168.0.3 any", + "permit ip host 192.168.0.4 any", + "permit ip host 192.168.0.5 any", + "permit ip host 192.168.0.6 any", + "permit ip host 192.168.0.7 any", + "permit ip host 192.168.0.8 any", + "permit ip host 192.168.0.9 any", + "permit ip host 192.168.0.10 any", + ] + + assert acl == ios_ref_acl + + acl = generate_nxos_acl(entries=10) + nxos_ref_acl = [ + "ip access-list netmiko_test_large_acl", + "permit ip host 192.168.0.1 any", + "permit ip host 192.168.0.2 any", + "permit ip host 192.168.0.3 any", + "permit ip host 192.168.0.4 any", + "permit ip host 192.168.0.5 any", + "permit ip host 192.168.0.6 any", + "permit ip host 192.168.0.7 any", + "permit ip host 192.168.0.8 any", + "permit ip host 192.168.0.9 any", + "permit ip host 192.168.0.10 any", + ] + assert acl == nxos_ref_acl + + acl = generate_cisco_xr_acl(entries=10) + xr_ref_acl = [ + "ipv4 access-list netmiko_test_large_acl", + "permit ipv4 host 192.168.0.1 any", + "permit ipv4 host 192.168.0.2 any", + "permit ipv4 host 192.168.0.3 any", + "permit ipv4 host 192.168.0.4 any", + "permit ipv4 host 192.168.0.5 any", + "permit ipv4 host 192.168.0.6 any", + "permit ipv4 host 192.168.0.7 any", + "permit ipv4 host 192.168.0.8 any", + "permit ipv4 host 192.168.0.9 any", + "permit ipv4 host 192.168.0.10 any", + ] + assert acl == xr_ref_acl diff --git a/tests_new/performance/arista1.out b/tests_new/performance/arista1.out new file mode 100644 index 0000000000000000000000000000000000000000..55555798140bc74f8649215773ae333c913e3a2e --- /dev/null +++ b/tests_new/performance/arista1.out @@ -0,0 +1,23 @@ +Last login: Wed Sep 23 17:35:27 2020 from 13.57.229.17 + +terminal width 511 +terminal width 511 +arista1#terminal width 511 +Width set to 511 columns. +arista1#terminal length 0 +Pagination disabled. +arista1# +arista1# +arista1# +arista1# +arista1# +arista1# +arista1#configure terminal +arista1(config)# +arista1(config)#no ip access-list netmiko_test_large_acl +! Hardware not present. ACL(s) not programmed in the hardware. +arista1(config)# +arista1(config)#end +arista1# +arista1# +arista1#exit diff --git a/tests/performance/netmiko_performance.csv b/tests_new/performance/netmiko_performance.csv similarity index 52% rename from tests/performance/netmiko_performance.csv rename to tests_new/performance/netmiko_performance.csv index 9ea8a738c828c82a77dd30d8b9aaf94a25c5a356..1812cb906843d9d7871c73b6f5a550c0b419c6f4 100644 --- a/tests/performance/netmiko_performance.csv +++ b/tests_new/performance/netmiko_performance.csv @@ -17,3 +17,19 @@ date,netmiko_version,device_name,connect,send_command_simple,send_config_simple, 2020-8-28 10:5:44,3.3.0_dev,cisco1,0:00:01.028332,0:00:01.059909,0:00:01.094017,0:00:03.527510 2020-8-28 10:5:55,3.3.0_dev,cisco3,0:00:01.071055,0:00:01.155505,0:00:01.022633,0:00:07.542166 2020-8-28 10:6:11,3.3.0_dev,cisco5,0:00:01.715099,0:00:01.778859,0:00:01.521635,0:00:11.627172 +2020-8-28 15:18:35,3.3.0_dev,cisco1,0:00:01.041867,0:00:01.099966,0:00:01.102853,0:00:03.506012 +2020-8-28 15:18:48,3.3.0_dev,cisco3,0:00:01.082106,0:00:01.162183,0:00:01.218904,0:00:08.922089 +2020-8-28 15:19:4,3.3.0_dev,cisco5,0:00:01.876103,0:00:01.442482,0:00:02.132264,0:00:11.050844 +2020-8-28 15:33:54,3.2.1_dev,nxos1,0:00:08.107724,0:00:08.487870,0:00:07.868704,0:00:17.714860 +2020-9-2 10:59:26,3.3.1_dev,cisco1,0:00:01.166879,0:00:01.138073,0:00:01.129265,0:00:03.751203 +2020-9-2 10:59:37,3.3.1_dev,cisco3,0:00:00.906176,0:00:00.981905,0:00:01.228977,0:00:07.456019 +2020-9-2 10:59:55,3.3.1_dev,cisco5,0:00:01.714811,0:00:01.851125,0:00:02.344312,0:00:12.557857 +2020-9-2 11:0:15,3.3.1_dev,nxos1,0:00:02.540320,0:00:02.657437,0:00:03.181819,0:00:08.347049 +2020-9-3 13:55:44,3.3.1_dev,cisco_xr_azure,0:00:06.205826,0:00:06.818236,0:00:10.011052,0:00:22.667603 +2020-9-4 8:54:56,3.3.2_dev,cisco1,0:00:01.068040,0:00:01.142141,0:00:01.151673,0:00:03.524082 +2020-9-4 8:55:9,3.3.2_dev,cisco3,0:00:01.105694,0:00:01.172320,0:00:01.280839,0:00:07.601024 +2020-9-4 8:55:27,3.3.2_dev,cisco5,0:00:01.794554,0:00:01.546943,0:00:01.895251,0:00:11.378513 +2020-9-4 8:55:44,3.3.2_dev,nxos1,0:00:02.504163,0:00:02.354322,0:00:02.715918,0:00:06.401413 +2020-9-4 8:56:12,3.3.2_dev,cisco_xr_azure,0:00:03.118182,0:00:03.106151,0:00:04.676095,0:00:12.314846 +2020-9-23 15:33:32,3.3.0,arista1,0:00:08.096496,0:00:08.734566,0:00:19.013243,0:00:39.441913 +2020-9-23 17:35:49,3.3.2_dev2,arista1,0:00:02.365521,0:00:02.964974,0:00:04.584866,0:00:19.276065 diff --git a/tests_new/performance/test_devices.yml b/tests_new/performance/test_devices.yml new file mode 100644 index 0000000000000000000000000000000000000000..1e98d82d4a73c727883caf854de9417b9afb3554 --- /dev/null +++ b/tests_new/performance/test_devices.yml @@ -0,0 +1,35 @@ +--- +cisco1: + device_type: cisco_ios + host: cisco1.lasthop.io + username: pyclass + +cisco3: + device_type: cisco_xe + host: cisco3.lasthop.io + username: pyclass + +cisco5: + device_type: cisco_xe + host: cisco5.lasthop.io + username: pyclass + +nxos1: + device_type: cisco_nxos + host: nxos1.lasthop.io + username: pyclass + # conn_timeout: 20 + # session_log: nxos1.out + +cisco_xr_azure: + device_type: cisco_xr + host: iosxr3.lasthop.io + username: pyclass + secret: '' + session_log: xr.out + +arista1: + device_type: arista_eos + host: arista1.lasthop.io + username: pyclass + session_log: arista1.out diff --git a/tests/performance/test_netmiko.py b/tests_new/performance/test_netmiko.py similarity index 66% rename from tests/performance/test_netmiko.py rename to tests_new/performance/test_netmiko.py index ab207e02e0711c3a7f533cb814e4f4cb683d8d1b..6139084c3c6539281ccd7f2ae8c0d58b5b423706 100644 --- a/tests/performance/test_netmiko.py +++ b/tests_new/performance/test_netmiko.py @@ -1,17 +1,26 @@ -from netmiko import ConnectHandler, __version__ import os -from ipaddress import ip_address +from os import path import yaml import functools from datetime import datetime import csv -# import logging -# logging.basicConfig(filename="test.log", level=logging.DEBUG) -# logger = logging.getLogger("netmiko") +from netmiko import ConnectHandler, __version__ +from test_utils import parse_yaml + +import network_utilities PRINT_DEBUG = False +PWD = path.dirname(path.realpath(__file__)) + + +def commands(platform): + """Parse the commands.yml file to get a commands dictionary.""" + test_platform = platform + commands_yml = parse_yaml(PWD + "/../etc/commands.yml") + return commands_yml[test_platform] + def generate_csv_timestamp(): """yyyy-MM-dd HH:mm:ss""" @@ -74,33 +83,49 @@ def connect(device): @f_exec_time def send_command_simple(device): with ConnectHandler(**device) as conn: - output = conn.send_command("show ip int brief") + platform = device["device_type"] + cmd = commands(platform)["basic"] + output = conn.send_command(cmd) PRINT_DEBUG and print(output) @f_exec_time def send_config_simple(device): with ConnectHandler(**device) as conn: - output = conn.send_config_set("logging buffered 20000") + platform = device["device_type"] + cmd = commands(platform)["config"][0] + output = conn.send_config_set(cmd) PRINT_DEBUG and print(output) @f_exec_time def send_config_large_acl(device): + + # Results will be marginally distorted by generating the ACL here. + device_type = device["device_type"] + func_name = f"generate_{device_type}_acl" + func = getattr(network_utilities, func_name) + with ConnectHandler(**device) as conn: - # Results will be marginally distorted by generating the ACL here. - cfg = generate_ios_acl(entries=100) + cfg = func(entries=100) output = conn.send_config_set(cfg) PRINT_DEBUG and print(output) -def generate_ios_acl(entries=100): - base_cmd = "ip access-list extended netmiko_test_large_acl" - acl = [base_cmd] - for i in range(1, entries + 1): - cmd = f"permit ip host {ip_address('192.168.0.0') + i} any" - acl.append(cmd) - return acl +@f_exec_time +def cleanup(device): + + # Results will be marginally distorted by generating the ACL here. + platform = device["device_type"] + base_acl_cmd = commands(platform)["config_long_acl"]["base_cmd"] + remove_acl_cmd = f"no {base_acl_cmd}" + cleanup_generic(device, remove_acl_cmd) + + +def cleanup_generic(device, command): + with ConnectHandler(**device) as conn: + output = conn.send_config_set(command) + PRINT_DEBUG and print(output) def main(): @@ -109,6 +134,8 @@ def main(): devices = read_devices() print("\n\n") for dev_name, dev_dict in devices.items(): + if dev_name != "arista1": + continue print("-" * 80) print(f"Device name: {dev_name}") print("-" * 12) @@ -121,12 +148,14 @@ def main(): "send_command_simple", "send_config_simple", "send_config_large_acl", + "cleanup", ] results = {} for op in operations: func = globals()[op] time_delta, result = func(dev_dict) - results[op] = time_delta + if op != "cleanup": + results[op] = time_delta print("-" * 80) print() diff --git a/tests_new/run_live_tests.sh b/tests_new/run_live_tests.sh new file mode 100755 index 0000000000000000000000000000000000000000..d6398418a0676ec5edce648c49166f6013ddc762 --- /dev/null +++ b/tests_new/run_live_tests.sh @@ -0,0 +1,19 @@ +cd test_cisco_ios/ +./test_ios.sh > ../test_out/test_cisco_ios.out 2> ../test_out/test_cisco_ios.stderr & + +cd .. +cd test_cisco_nxos/ +./test_cisco_nxos.sh > ../test_out/test_cisco_nxos.out 2> ../test_out/test_cisco_nxos.stderr & + +cd .. +cd test_cisco_xe +./test_cisco_xe.sh > ../test_out/test_cisco_xe.out 2> ../test_out/test_cisco_xe.stderr & + +cd .. +cd test_cisco_xr +./test_iosxr.sh > ../test_out/test_cisco_xr_xrv.out 2> ../test_out/test_cisco_xr_xrv.stderr & +./test_iosxr_azure.sh > ../test_out/test_cisco_xr_azure.out 2> ../test_out/test_cisco_xr_azure.stderr & + +cd .. +cd test_arista_eos +./test_arista_eos.sh > ../test_out/test_arista_eos.out 2> ../test_out/test_arista_eos.stderr & diff --git a/tests_new/show_run_interfaces.ttp b/tests_new/show_run_interfaces.ttp new file mode 100644 index 0000000000000000000000000000000000000000..d35d7376be6323d0afc0098a1d2c17f7aa6e969f --- /dev/null +++ b/tests_new/show_run_interfaces.ttp @@ -0,0 +1,3 @@ + + description {{ description }} + \ No newline at end of file diff --git a/tests_new/test2_src.txt b/tests_new/test2_src.txt new file mode 100644 index 0000000000000000000000000000000000000000..d2ddd3cb8d4e9e72990b0d3b7d23e172b0c8c032 --- /dev/null +++ b/tests_new/test2_src.txt @@ -0,0 +1,2 @@ +no logging console +logging buffered 10000 diff --git a/tests_new/test9.txt b/tests_new/test9.txt new file mode 100644 index 0000000000000000000000000000000000000000..acc8e3dd51d016a5c392edf6a28189215066c982 --- /dev/null +++ b/tests_new/test9.txt @@ -0,0 +1 @@ +no logging console diff --git a/tests_new/test_arista_eos/test9.txt b/tests_new/test_arista_eos/test9.txt new file mode 100644 index 0000000000000000000000000000000000000000..acc8e3dd51d016a5c392edf6a28189215066c982 --- /dev/null +++ b/tests_new/test_arista_eos/test9.txt @@ -0,0 +1 @@ +no logging console diff --git a/tests_new/test_arista_eos/test_arista_eos.sh b/tests_new/test_arista_eos/test_arista_eos.sh new file mode 100755 index 0000000000000000000000000000000000000000..95b85b731e2d5bf3a9d99741337ec6b3f60a1560 --- /dev/null +++ b/tests_new/test_arista_eos/test_arista_eos.sh @@ -0,0 +1,15 @@ +#!/bin/sh + +RETURN_CODE=0 + +echo "Starting tests...good luck:" \ +&& echo "Arista EOS" \ +&& cd .. \ +&& py.test -s -x -v test_netmiko_show.py --test_device arista_sw \ +&& py.test -s -x -v test_netmiko_config.py --test_device arista_sw \ +&& py.test -s -x -v test_netmiko_config_acl.py --test_device arista_sw \ +&& py.test -s -x -v test_netmiko_scp.py --test_device arista_sw \ +&& py.test -s -x -v test_netmiko_autodetect.py --test_device arista_sw \ +|| RETURN_CODE=1 + +exit $RETURN_CODE diff --git a/tests_new/test_arista_eos/testx.txt b/tests_new/test_arista_eos/testx.txt new file mode 100755 index 0000000000000000000000000000000000000000..acc8e3dd51d016a5c392edf6a28189215066c982 --- /dev/null +++ b/tests_new/test_arista_eos/testx.txt @@ -0,0 +1 @@ +no logging console diff --git a/tests/test_ios/add_delay_cisco881.sh b/tests_new/test_cisco_ios/add_delay_cisco881.sh similarity index 100% rename from tests/test_ios/add_delay_cisco881.sh rename to tests_new/test_cisco_ios/add_delay_cisco881.sh diff --git a/tests_new/test_cisco_ios/test9.txt b/tests_new/test_cisco_ios/test9.txt new file mode 100644 index 0000000000000000000000000000000000000000..acc8e3dd51d016a5c392edf6a28189215066c982 --- /dev/null +++ b/tests_new/test_cisco_ios/test9.txt @@ -0,0 +1 @@ +no logging console diff --git a/tests_new/test_cisco_ios/test_ios.sh b/tests_new/test_cisco_ios/test_ios.sh new file mode 100755 index 0000000000000000000000000000000000000000..fca55379eda5b729f7fbc731579f0198fc0e818a --- /dev/null +++ b/tests_new/test_cisco_ios/test_ios.sh @@ -0,0 +1,20 @@ +#!/bin/sh + +RETURN_CODE=0 + +echo "Starting tests...good luck:" \ +&& echo "Cisco IOS" \ +&& cd .. \ +&& py.test -s -x -v test_netmiko_show.py --test_device cisco1 \ +&& py.test -s -x -v test_netmiko_cmd_verify.py --test_device cisco1 \ +&& py.test -s -x -v test_netmiko_config.py --test_device cisco1 \ +&& py.test -s -x -v test_netmiko_config_acl.py --test_device cisco1 \ +&& py.test -s -x -v test_netmiko_tcl.py --test_device cisco1 \ +&& py.test -s -x -v test_netmiko_scp.py --test_device cisco1 \ +&& py.test -s -x -v test_netmiko_autodetect.py --test_device cisco1 \ +|| RETURN_CODE=1 + +exit $RETURN_CODE + +# Note, it is possible the test_netmiko_cmd_verify tests fail as you are +# attempting to do things with cmd_verify/global_cmd_verify set to False. diff --git a/tests_new/test_cisco_ios/testx.txt b/tests_new/test_cisco_ios/testx.txt new file mode 100644 index 0000000000000000000000000000000000000000..acc8e3dd51d016a5c392edf6a28189215066c982 --- /dev/null +++ b/tests_new/test_cisco_ios/testx.txt @@ -0,0 +1 @@ +no logging console diff --git a/tests_new/test_cisco_nxos/add_delay_cisco_nxos.sh b/tests_new/test_cisco_nxos/add_delay_cisco_nxos.sh new file mode 100755 index 0000000000000000000000000000000000000000..83d4d891a6f7b9c69ecf00605b69573019197efb --- /dev/null +++ b/tests_new/test_cisco_nxos/add_delay_cisco_nxos.sh @@ -0,0 +1,9 @@ +#!/bin/bash + +# Add delay for cisco3; must be root or sudo to execute +sudo -s /sbin/tc qdisc del dev eth0 root +sudo -s /sbin/tc qdisc add dev eth0 root handle 1: prio +sudo -s /sbin/tc qdisc add dev eth0 parent 1:3 handle 30: tbf rate 20kbit buffer 1600 limit 3000 +sudo -s /sbin/tc qdisc add dev eth0 parent 30:1 handle 31: netem delay 1000ms 10ms distribution normal loss 10% +sudo -s /sbin/tc filter add dev eth0 protocol ip parent 1:0 prio 3 u32 match ip dst 52.151.9.205/32 flowid 1:3 + diff --git a/tests_new/test_cisco_nxos/test9.txt b/tests_new/test_cisco_nxos/test9.txt new file mode 100644 index 0000000000000000000000000000000000000000..acc8e3dd51d016a5c392edf6a28189215066c982 --- /dev/null +++ b/tests_new/test_cisco_nxos/test9.txt @@ -0,0 +1 @@ +no logging console diff --git a/tests_new/test_cisco_nxos/test_cisco_nxos.sh b/tests_new/test_cisco_nxos/test_cisco_nxos.sh new file mode 100755 index 0000000000000000000000000000000000000000..c06de0657cdbce0f4fc702471139833caae30e43 --- /dev/null +++ b/tests_new/test_cisco_nxos/test_cisco_nxos.sh @@ -0,0 +1,14 @@ +#!/bin/sh + +RETURN_CODE=0 + +echo "Starting tests...good luck:" \ +&& echo "Cisco NXOS" \ +&& cd .. \ +&& py.test -v -s -x test_netmiko_show.py --test_device nxos1 \ +&& py.test -v -s -x test_netmiko_config.py --test_device nxos1 \ +&& py.test -v -s -x test_netmiko_config_acl.py --test_device nxos1 \ +&& py.test -v -s -x test_netmiko_scp.py --test_device nxos1 \ +|| RETURN_CODE=1 + +exit $RETURN_CODE diff --git a/tests_new/test_cisco_nxos/testx.txt b/tests_new/test_cisco_nxos/testx.txt new file mode 100644 index 0000000000000000000000000000000000000000..acc8e3dd51d016a5c392edf6a28189215066c982 --- /dev/null +++ b/tests_new/test_cisco_nxos/testx.txt @@ -0,0 +1 @@ +no logging console diff --git a/tests_new/test_cisco_xe/add_delay_cisco_xe.sh b/tests_new/test_cisco_xe/add_delay_cisco_xe.sh new file mode 100755 index 0000000000000000000000000000000000000000..5eef74d7523621d3978421514e6f76c114f9ec99 --- /dev/null +++ b/tests_new/test_cisco_xe/add_delay_cisco_xe.sh @@ -0,0 +1,9 @@ +#!/bin/bash + +# Add delay for cisco3; must be root or sudo to execute +sudo -s /sbin/tc qdisc del dev eth0 root +sudo -s /sbin/tc qdisc add dev eth0 root handle 1: prio +sudo -s /sbin/tc qdisc add dev eth0 parent 1:3 handle 30: tbf rate 20kbit buffer 1600 limit 3000 +sudo -s /sbin/tc qdisc add dev eth0 parent 30:1 handle 31: netem delay 1000ms 10ms distribution normal loss 10% +sudo -s /sbin/tc filter add dev eth0 protocol ip parent 1:0 prio 3 u32 match ip dst 184.105.247.89/32 flowid 1:3 + diff --git a/tests_new/test_cisco_xe/test2_src.txt b/tests_new/test_cisco_xe/test2_src.txt new file mode 100644 index 0000000000000000000000000000000000000000..d2ddd3cb8d4e9e72990b0d3b7d23e172b0c8c032 --- /dev/null +++ b/tests_new/test_cisco_xe/test2_src.txt @@ -0,0 +1,2 @@ +no logging console +logging buffered 10000 diff --git a/tests_new/test_cisco_xe/test9.txt b/tests_new/test_cisco_xe/test9.txt new file mode 100644 index 0000000000000000000000000000000000000000..acc8e3dd51d016a5c392edf6a28189215066c982 --- /dev/null +++ b/tests_new/test_cisco_xe/test9.txt @@ -0,0 +1 @@ +no logging console diff --git a/tests/test_ios/test_iosxe.sh b/tests_new/test_cisco_xe/test_cisco_xe.sh similarity index 97% rename from tests/test_ios/test_iosxe.sh rename to tests_new/test_cisco_xe/test_cisco_xe.sh index c02647d24e643bbf24c04ef5b6597e6dab06d5a3..71fa6163087b20215b65571fe3ca3e2d4c26bdf6 100755 --- a/tests/test_ios/test_iosxe.sh +++ b/tests_new/test_cisco_xe/test_cisco_xe.sh @@ -3,6 +3,7 @@ RETURN_CODE=0 echo "Starting tests...good luck:" \ +&& cd .. \ && echo "Cisco IOS-XE" \ && py.test -v test_netmiko_show.py --test_device cisco3 \ && py.test -v test_netmiko_config.py --test_device cisco3 \ diff --git a/tests_new/test_cisco_xe/testx.txt b/tests_new/test_cisco_xe/testx.txt new file mode 100644 index 0000000000000000000000000000000000000000..acc8e3dd51d016a5c392edf6a28189215066c982 --- /dev/null +++ b/tests_new/test_cisco_xe/testx.txt @@ -0,0 +1 @@ +no logging console diff --git a/tests_new/test_cisco_xr/add_delay_iosxr_azure.sh b/tests_new/test_cisco_xr/add_delay_iosxr_azure.sh new file mode 100755 index 0000000000000000000000000000000000000000..e8441f8807a4c3c8fb7ee886385c0d6b4612ed39 --- /dev/null +++ b/tests_new/test_cisco_xr/add_delay_iosxr_azure.sh @@ -0,0 +1,7 @@ +#!/bin/bash +sudo -s /sbin/tc qdisc del dev eth0 root +sudo -s /sbin/tc qdisc add dev eth0 root handle 1: prio +sudo -s /sbin/tc qdisc add dev eth0 parent 1:3 handle 30: tbf rate 20kbit buffer 1600 limit 3000 +sudo -s /sbin/tc qdisc add dev eth0 parent 30:1 handle 31: netem delay 1000ms 10ms distribution normal loss 10% +sudo -s /sbin/tc filter add dev eth0 protocol ip parent 1:0 prio 3 u32 match ip dst 40.121.206.54/32 flowid 1:3 + diff --git a/tests_new/test_cisco_xr/test9.txt b/tests_new/test_cisco_xr/test9.txt new file mode 100644 index 0000000000000000000000000000000000000000..acc8e3dd51d016a5c392edf6a28189215066c982 --- /dev/null +++ b/tests_new/test_cisco_xr/test9.txt @@ -0,0 +1 @@ +no logging console diff --git a/tests_new/test_cisco_xr/test_iosxr.sh b/tests_new/test_cisco_xr/test_iosxr.sh new file mode 100755 index 0000000000000000000000000000000000000000..632cfb2015d83a2489c6aec56366522010588803 --- /dev/null +++ b/tests_new/test_cisco_xr/test_iosxr.sh @@ -0,0 +1,16 @@ +#!/bin/sh + +RETURN_CODE=0 + +echo "Starting tests...good luck:" \ +&& echo "Cisco IOS-XR (xrv)" \ +&& cd .. \ +&& py.test -s -x -v test_netmiko_show.py --test_device cisco_xrv \ +&& py.test -s -x -v test_netmiko_config.py --test_device cisco_xrv \ +&& py.test -v -s -x test_netmiko_config_acl.py --test_device cisco_xrv \ +&& py.test -s -x -v test_netmiko_commit.py --test_device cisco_xrv \ +&& py.test -s -x -v test_netmiko_scp.py --test_device cisco_xrv \ +&& py.test -s -x -v test_netmiko_autodetect.py --test_device cisco_xrv \ +|| RETURN_CODE=1 + +exit $RETURN_CODE diff --git a/tests_new/test_cisco_xr/test_iosxr_azure.sh b/tests_new/test_cisco_xr/test_iosxr_azure.sh new file mode 100755 index 0000000000000000000000000000000000000000..c5d2d5981e97bcb07672f0c0538431fc92ac1551 --- /dev/null +++ b/tests_new/test_cisco_xr/test_iosxr_azure.sh @@ -0,0 +1,18 @@ +#!/bin/sh + +RETURN_CODE=0 + +echo "Starting tests...good luck:" \ +&& cd .. \ +&& echo "Cisco IOS-XR (Azure)" \ +&& py.test -s -x -v test_netmiko_show.py --test_device cisco_xr_azure \ +&& py.test -s -x -v test_netmiko_config.py --test_device cisco_xr_azure \ +&& py.test -s -x -v test_netmiko_commit.py --test_device cisco_xr_azure \ +&& py.test -s -x -v test_netmiko_autodetect.py --test_device cisco_xr_azure \ +\ +|| RETURN_CODE=1 + +exit $RETURN_CODE + +# FIX - some issue with SCP to the IOS-XR in Azure? +#&& py.test -s -x -v test_netmiko_scp.py --test_device cisco_xr_azure \ diff --git a/tests_new/test_cisco_xr/testx.txt b/tests_new/test_cisco_xr/testx.txt new file mode 100644 index 0000000000000000000000000000000000000000..acc8e3dd51d016a5c392edf6a28189215066c982 --- /dev/null +++ b/tests_new/test_cisco_xr/testx.txt @@ -0,0 +1 @@ +no logging console diff --git a/tests_new/test_commit.sh b/tests_new/test_commit.sh new file mode 100644 index 0000000000000000000000000000000000000000..13fc96fa757190b3fb13e9433488602718cd81f4 --- /dev/null +++ b/tests_new/test_commit.sh @@ -0,0 +1,7 @@ +for i in 1 2 3 4 5 6 7 8 +do + py.test -s -x -v test_netmiko_commit.py::test_confirm_delay --test_device cisco_xrv + if [ $? -ne 0 ]; then + break + fi +done diff --git a/tests_new/test_netmiko_autodetect.py b/tests_new/test_netmiko_autodetect.py new file mode 100755 index 0000000000000000000000000000000000000000..7051ee14a08df798ee1c1f7b92550d895df6f2d2 --- /dev/null +++ b/tests_new/test_netmiko_autodetect.py @@ -0,0 +1,7 @@ +#!/usr/bin/env python +def test_ssh_connect(ssh_autodetect): + """Verify the connection was established successfully.""" + net_conn, real_device_type = ssh_autodetect + device_type = net_conn.autodetect() + print(device_type) + assert device_type == real_device_type diff --git a/tests_new/test_netmiko_cmd_verify.py b/tests_new/test_netmiko_cmd_verify.py new file mode 100644 index 0000000000000000000000000000000000000000..a7360b934e2c4f92720eeb7f6bbb488686c604b4 --- /dev/null +++ b/tests_new/test_netmiko_cmd_verify.py @@ -0,0 +1,55 @@ +import pytest +from netmiko.utilities import select_cmd_verify + + +@select_cmd_verify +def bogus_func(obj, *args, **kwargs): + """Function that just returns the arguments modified by the decorator.""" + return (obj, args, kwargs) + + +def test_cmd_verify_decorator(net_connect_cmd_verify): + obj = net_connect_cmd_verify + # Global False should have precedence + assert obj.global_cmd_verify is False + (obj, args, kwargs) = bogus_func(net_connect_cmd_verify, cmd_verify=True) + assert kwargs["cmd_verify"] is False + (obj, args, kwargs) = bogus_func(net_connect_cmd_verify, cmd_verify=False) + assert kwargs["cmd_verify"] is False + + # Global True should have precedence + obj.global_cmd_verify = True + assert obj.global_cmd_verify is True + (obj, args, kwargs) = bogus_func(net_connect_cmd_verify, cmd_verify=True) + assert kwargs["cmd_verify"] is True + (obj, args, kwargs) = bogus_func(net_connect_cmd_verify, cmd_verify=False) + assert kwargs["cmd_verify"] is True + + # None should track the local argument + obj.global_cmd_verify = None + assert obj.global_cmd_verify is None + (obj, args, kwargs) = bogus_func(net_connect_cmd_verify, cmd_verify=True) + assert kwargs["cmd_verify"] is True + (obj, args, kwargs) = bogus_func(net_connect_cmd_verify, cmd_verify=False) + assert kwargs["cmd_verify"] is False + + # Set it back to proper False value (so later tests aren't messed up). + obj.global_cmd_verify = False + + +def test_send_command_global_cmd_verify( + net_connect_cmd_verify, commands, expected_responses +): + """ + Verify a command can be sent down the channel successfully using send_command method. + + Disable cmd_verify globally. + """ + net_connect = net_connect_cmd_verify + if net_connect.fast_cli is True: + assert pytest.skip() + net_connect.clear_buffer() + # cmd_verify should be disabled globally at this point + assert net_connect.global_cmd_verify is False + show_ip_alt = net_connect.send_command(commands["basic"]) + assert expected_responses["interface_ip"] in show_ip_alt diff --git a/tests_new/test_netmiko_commit.py b/tests_new/test_netmiko_commit.py new file mode 100755 index 0000000000000000000000000000000000000000..4985f79e74755061048eed2f019eacbc2b9b443b --- /dev/null +++ b/tests_new/test_netmiko_commit.py @@ -0,0 +1,413 @@ +#!/usr/bin/env python +""" +test_ssh_connect: verify ssh connectivity +test_config_mode: verify enter config mode +test_commit_base: test std .commit() +test_commit_confirm: test commit with confirm +test_confirm_delay: test commit-confirm with non-std delay +test_no_confirm: test commit-confirm with no confirm +test_commit_check: test commit check +test_commit_comment: test commit with comment +test_commit_andquit: test commit andquit +test_exit_config_mode: verify exit config mode +test_disconnect: cleanly disconnect the SSH session +""" + +import time +import re +import random +import string +import pytest + + +def gen_random(N=6): + return "".join( + [ + random.choice( + string.ascii_lowercase + string.ascii_uppercase + string.digits + ) + for x in range(N) + ] + ) + + +def retrieve_commands(commands): + """ + Retrieve context needed for a set of commit actions + """ + + config_commands = commands["config"] + support_commit = commands.get("support_commit") + config_verify = commands["config_verification"] + + return (config_commands, support_commit, config_verify) + + +def setup_initial_state(net_connect, commands, expected_responses): + """ + Setup initial configuration prior to change so that config change can be verified + """ + + # Setup initial state + config_commands, support_commit, config_verify = retrieve_commands(commands) + setup_base_config(net_connect, config_commands[0:1]) + + cmd_response = expected_responses.get("cmd_response_init", config_commands[0]) + initial_state = config_change_verify(net_connect, config_verify, cmd_response) + assert initial_state is True + + return config_commands, support_commit, config_verify + + +def setup_base_config(net_connect, config_changes): + """ + Send set of config commands and commit + """ + net_connect.send_config_set(config_changes) + net_connect.commit() + + +def config_change_verify(net_connect, verify_cmd, cmd_response): + """ + Send verify_cmd down channel, verify cmd_response in output + """ + config_commands_output = net_connect.send_command_expect(verify_cmd) + if cmd_response in config_commands_output: + return True + else: + return False + + +def test_ssh_connect(net_connect, commands, expected_responses): + """ + Verify the connection was established successfully + """ + show_version = net_connect.send_command_expect(commands["version"]) + assert expected_responses["version_banner"] in show_version + + +def test_config_mode(net_connect, commands, expected_responses): + """ + Test enter config mode + """ + net_connect.config_mode() + assert net_connect.check_config_mode() is True + + +def test_commit_base(net_connect, commands, expected_responses): + """ + Test .commit() with no options + """ + # Setup the initial config state + config_commands, support_commit, config_verify = setup_initial_state( + net_connect, commands, expected_responses + ) + + # Perform commit test + net_connect.send_config_set(config_commands) + net_connect.commit() + + cmd_response = expected_responses.get("cmd_response_final", config_commands[-1]) + final_state = config_change_verify(net_connect, config_verify, cmd_response) + assert final_state is True + + +def test_commit_confirm(net_connect, commands, expected_responses): + """ + Test confirm with confirm + """ + # Setup the initial config state + config_commands, support_commit, config_verify = setup_initial_state( + net_connect, commands, expected_responses + ) + + if net_connect.device_type in ["nokia_sros"]: + assert pytest.skip() + + # Perform commit-confirm test + net_connect.send_config_set(config_commands) + if net_connect.device_type == "cisco_xr": + net_connect.commit(confirm=True, confirm_delay=60) + else: + net_connect.commit(confirm=True) + + cmd_response = expected_responses.get("cmd_response_final", config_commands[-1]) + final_state = config_change_verify(net_connect, config_verify, cmd_response) + assert final_state is True + + # Perform confirm + net_connect.commit() + + +def test_confirm_delay(net_connect, commands, expected_responses): + """ + Test commit with confirm and non-default delay + """ + # Setup the initial config state + config_commands, support_commit, config_verify = setup_initial_state( + net_connect, commands, expected_responses + ) + + if net_connect.device_type in ["nokia_sros"]: + assert pytest.skip() + + # Perform commit-confirm test + net_connect.send_config_set(config_commands) + if net_connect.device_type == "cisco_xr": + net_connect.commit(confirm=True, confirm_delay=60) + else: + net_connect.commit(confirm=True, confirm_delay=5) + + cmd_response = expected_responses.get("cmd_response_final", config_commands[-1]) + final_state = config_change_verify(net_connect, config_verify, cmd_response) + assert final_state is True + + # Perform confirm + net_connect.commit() + + +def test_no_confirm(net_connect, commands, expected_responses): + """ + Perform commit-confirm, but don't confirm (verify rollback) + """ + # Setup the initial config state + config_commands, support_commit, config_verify = setup_initial_state( + net_connect, commands, expected_responses + ) + + if net_connect.device_type in ["nokia_sros"]: + assert pytest.skip() + + # Perform commit-confirm test + net_connect.send_config_set(config_commands) + if net_connect.device_type == "cisco_xr": + net_connect.commit(confirm=True, confirm_delay=30) + else: + net_connect.commit(confirm=True, confirm_delay=1) + + cmd_response = expected_responses.get("cmd_response_final", config_commands[-1]) + final_state = config_change_verify(net_connect, config_verify, cmd_response) + assert final_state is True + + time.sleep(130) + + # Verify rolled back to initial state + cmd_response = expected_responses.get("cmd_response_init", config_commands[0]) + init_state = config_change_verify(net_connect, config_verify, cmd_response) + assert init_state is True + + +def test_clear_msg(net_connect, commands, expected_responses): + """ + IOS-XR generates the following message upon a failed commit + + One or more commits have occurred from other + configuration sessions since this session started + or since the last commit was made from this session. + You can use the 'show configuration commit changes' + command to browse the changes. + Do you wish to proceed with this commit anyway? [no]: yes + + Clear it + """ + # Setup the initial config state + config_commands, support_commit, config_verify = retrieve_commands(commands) + + if net_connect.device_type == "cisco_xr": + output = net_connect.send_config_set(config_commands) + output += net_connect.send_command_expect( + "commit", expect_string=r"Do you wish to" + ) + output += net_connect.send_command_expect("yes", auto_find_prompt=False) + assert True + + +def test_commit_check(net_connect, commands, expected_responses): + """ + Test commit check + """ + # IOS-XR does not support commit check + if net_connect.device_type in ["cisco_xr", "nokia_sros"]: + assert pytest.skip() + else: + + # Setup the initial config state + config_commands, support_commit, config_verify = setup_initial_state( + net_connect, commands, expected_responses + ) + + # Perform commit-confirm test + net_connect.send_config_set(config_commands) + net_connect.commit(check=True) + + # Verify at initial state + cmd_response = expected_responses.get("cmd_response_init", config_commands[0]) + init_state = config_change_verify(net_connect, config_verify, cmd_response) + assert init_state is True + + rollback = commands.get("rollback") + net_connect.send_config_set([rollback]) + + +def test_commit_comment(net_connect, commands, expected_responses): + """ + Test commit with comment + """ + # Setup the initial config state + config_commands, support_commit, config_verify = setup_initial_state( + net_connect, commands, expected_responses + ) + + if net_connect.device_type in ["nokia_sros"]: + assert pytest.skip() + + # Perform commit with comment + net_connect.send_config_set(config_commands) + net_connect.commit(comment="Unit test on commit with comment") + + # Verify change was committed + cmd_response = expected_responses.get("cmd_response_final", config_commands[-1]) + final_state = config_change_verify(net_connect, config_verify, cmd_response) + assert final_state is True + + # Verify commit comment + commit_verification = commands.get("commit_verification") + tmp_output = net_connect.send_command(commit_verification) + if net_connect.device_type == "cisco_xr": + commit_comment = tmp_output + else: + commit_comment = tmp_output.strip().split("\n")[1] + assert expected_responses.get("commit_comment") in commit_comment.strip() + + +def test_commit_andquit(net_connect, commands, expected_responses): + """ + Test commit and immediately quit configure mode + """ + + # IOS-XR does not support commit and quit + if net_connect.device_type in ["cisco_xr", "nokia_sros"]: + assert pytest.skip() + else: + + # Setup the initial config state + config_commands, support_commit, config_verify = setup_initial_state( + net_connect, commands, expected_responses + ) + + # Execute change and commit + net_connect.send_config_set(config_commands) + net_connect.commit(and_quit=True) + + # Verify change was committed + cmd_response = expected_responses.get("cmd_response_final", config_commands[-1]) + config_verify = "show configuration | match archive" + final_state = config_change_verify(net_connect, config_verify, cmd_response) + assert final_state is True + + +def test_commit_label(net_connect, commands, expected_responses): + """Test commit label for IOS-XR.""" + + if net_connect.device_type != "cisco_xr": + assert True is True + else: + # Setup the initial config state + config_commands, support_commit, config_verify = setup_initial_state( + net_connect, commands, expected_responses + ) + + # Execute change and commit + net_connect.send_config_set(config_commands) + label = "test_lbl_" + gen_random() + net_connect.commit(label=label) + + # Verify commit label + commit_verification = commands.get("commit_verification") + tmp_output = net_connect.send_command(commit_verification) + match = re.search(r"Label: (.*)", tmp_output) + response_label = match.group(1) + response_label = response_label.strip() + assert label == response_label + + +def test_commit_label_comment(net_connect, commands, expected_responses): + """ + Test commit label for IOS-XR with comment + """ + # IOS-XR only test + if net_connect.device_type != "cisco_xr": + assert pytest.skip() + else: + # Setup the initial config state + config_commands, support_commit, config_verify = setup_initial_state( + net_connect, commands, expected_responses + ) + + # Execute change and commit + net_connect.send_config_set(config_commands) + label = "test_lbl_" + gen_random() + comment = "Test with comment and label" + net_connect.commit(label=label, comment=comment) + + # Verify commit label + commit_verification = commands.get("commit_verification") + tmp_output = net_connect.send_command(commit_verification) + match = re.search(r"Label: (.*)", tmp_output) + response_label = match.group(1) + response_label = response_label.strip() + assert label == response_label + match = re.search(r"Comment: (.*)", tmp_output) + response_comment = match.group(1) + response_comment = response_comment.strip() + response_comment = response_comment.strip('"') + assert comment == response_comment + + +def test_commit_label_confirm(net_connect, commands, expected_responses): + """ + Test commit label for IOS-XR with confirm + """ + # IOS-XR only test + if net_connect.device_type != "cisco_xr": + assert pytest.skip() + else: + # Setup the initial config state + config_commands, support_commit, config_verify = setup_initial_state( + net_connect, commands, expected_responses + ) + + # Execute change and commit + net_connect.send_config_set(config_commands) + label = "test_lbl_" + gen_random() + net_connect.commit(label=label, confirm=True, confirm_delay=120) + + cmd_response = expected_responses.get("cmd_response_final", config_commands[-1]) + final_state = config_change_verify(net_connect, config_verify, cmd_response) + assert final_state is True + + # Verify commit label + commit_verification = commands.get("commit_verification") + tmp_output = net_connect.send_command(commit_verification) + match = re.search(r"Label: (.*)", tmp_output) + response_label = match.group(1) + response_label = response_label.strip() + assert label == response_label + + net_connect.commit() + + +def test_exit_config_mode(net_connect, commands, expected_responses): + """ + Test exit config mode + """ + net_connect.exit_config_mode() + time.sleep(1) + assert net_connect.check_config_mode() is False + + +def test_disconnect(net_connect, commands, expected_responses): + """ + Terminate the SSH session + """ + net_connect.disconnect() diff --git a/tests_new/test_netmiko_config.py b/tests_new/test_netmiko_config.py new file mode 100755 index 0000000000000000000000000000000000000000..89eab786064ef5453e0de1cbd32aaedb5f63a956 --- /dev/null +++ b/tests_new/test_netmiko_config.py @@ -0,0 +1,118 @@ +#!/usr/bin/env python +import pytest + + +def test_ssh_connect(net_connect, commands, expected_responses): + """ + Verify the connection was established successfully + """ + show_version = net_connect.send_command(commands["version"]) + assert expected_responses["version_banner"] in show_version + + +def test_enable_mode(net_connect, commands, expected_responses): + """ + Test entering enable mode + + Catch exception for devices that don't support enable + """ + try: + net_connect.enable() + enable_prompt = net_connect.find_prompt() + assert enable_prompt == expected_responses["enable_prompt"] + except AttributeError: + assert True + + +def test_config_mode(net_connect, commands, expected_responses): + """ + Test enter config mode + """ + # Behavior for devices with no config mode is to return null string + if net_connect.config_mode() != "": + assert net_connect.check_config_mode() is True + else: + assert True + + +def test_exit_config_mode(net_connect, commands, expected_responses): + """ + Test exit config mode + """ + net_connect.exit_config_mode() + assert net_connect.check_config_mode() is False + + +def test_config_set(net_connect, commands, expected_responses): + """Test sending configuration commands.""" + config_commands = commands["config"] + support_commit = commands.get("support_commit") + config_verify = commands["config_verification"] + + # Set to initial value and testing sending command as a string + net_connect.send_config_set(config_commands[0]) + if support_commit: + net_connect.commit() + cmd_response = expected_responses.get("cmd_response_init") + config_commands_output = net_connect.send_command(config_verify) + if cmd_response: + assert cmd_response in config_commands_output + else: + assert config_commands[0] in config_commands_output + net_connect.send_config_set(config_commands) + if support_commit: + net_connect.commit() + cmd_response = expected_responses.get("cmd_response_final") + config_commands_output = net_connect.send_command_expect(config_verify) + if cmd_response: + assert cmd_response in config_commands_output + else: + assert config_commands[-1] in config_commands_output + + +def test_config_set_longcommand(net_connect, commands, expected_responses): + """Test sending configuration commands using long commands""" + config_commands = commands.get("config_long_command") + config_verify = commands["config_verification"] # noqa + if not config_commands: + assert True + return + output = net_connect.send_config_set(config_commands) # noqa + assert True + + +def test_config_hostname(net_connect, commands, expected_responses): + hostname = "test-netmiko1" + command = f"hostname {hostname}" + if "arista" in net_connect.device_type: + current_hostname = net_connect.find_prompt()[:-1] + net_connect.send_config_set(command) + new_hostname = net_connect.find_prompt() + assert hostname in new_hostname + # Reset prompt back to original value + net_connect.set_base_prompt() + net_connect.send_config_set(f"hostname {current_hostname}") + + +def test_config_from_file(net_connect, commands, expected_responses): + """ + Test sending configuration commands from a file + """ + config_file = commands.get("config_file") + config_verify = commands["config_verification"] + if config_file is not None: + net_connect.send_config_from_file(config_file) + config_commands_output = net_connect.send_command_expect(config_verify) + assert expected_responses["file_check_cmd"] in config_commands_output + else: + assert pytest.skip() + + if "nokia_sros" in net_connect.device_type: + net_connect.save_config() + + +def test_disconnect(net_connect, commands, expected_responses): + """ + Terminate the SSH session + """ + net_connect.disconnect() diff --git a/tests_new/test_netmiko_config_acl.py b/tests_new/test_netmiko_config_acl.py new file mode 100755 index 0000000000000000000000000000000000000000..c2d3b27cca67028b6b001c4aabe01579a23e4337 --- /dev/null +++ b/tests_new/test_netmiko_config_acl.py @@ -0,0 +1,62 @@ +#!/usr/bin/env python +import re +import pytest +from network_utilities import generate_ios_acl, generate_nxos_acl +from network_utilities import generate_cisco_xr_acl # noqa +from network_utilities import generate_arista_eos_acl # noqa + + +def remove_acl(net_connect, cmd, commit=False): + """Ensure ACL is removed.""" + net_connect.send_config_set(f"no {cmd}") + if commit: + net_connect.commit() + net_connect.exit_config_mode() + + +def test_large_acl(net_connect, commands, expected_responses, acl_entries=100): + """Test configuring a large ACL.""" + if commands.get("config_long_acl"): + acl_config = commands.get("config_long_acl") + base_cmd = acl_config["base_cmd"] + verify_cmd = acl_config["verify_cmd"] + offset = acl_config["offset"] + else: + pytest.skip("Platform not supported for ACL test") + + support_commit = commands.get("support_commit") + remove_acl(net_connect, cmd=base_cmd, commit=support_commit) + + # Generate the ACL + platform = net_connect.device_type + if "cisco_ios" in net_connect.device_type or "cisco_xe" in net_connect.device_type: + cfg_lines = generate_ios_acl() + elif "cisco_nxos" in net_connect.device_type: + cfg_lines = generate_nxos_acl() + else: + func_name = f"generate_{platform}_acl" + acl_func = globals()[func_name] + cfg_lines = acl_func() + + # Send ACL to remote devices + result = net_connect.send_config_set(cfg_lines) + if support_commit: + net_connect.commit() + net_connect.exit_config_mode() + + # send_config_set should return same num lines + offset lines for entering/exiting cfg-mode + # NX-OS is will have more than one newline (per line) + result_list = re.split(r"\n+", result) + assert len(result_list) == len(cfg_lines) + offset + + # Check that length of lines in show of the acl matches lines configured + verify = net_connect.send_command(verify_cmd) + verify_list = re.split(r"\n+", verify.strip()) + + # IOS-XR potentially has a timestamp on the show command + if "UTC" in verify_list[0]: + verify_list.pop(0) + assert len(verify_list) == len(cfg_lines) + + remove_acl(net_connect, cmd=base_cmd, commit=support_commit) + net_connect.disconnect() diff --git a/tests_new/test_netmiko_scp.py b/tests_new/test_netmiko_scp.py new file mode 100755 index 0000000000000000000000000000000000000000..d703704d5a936369e9c267262fda60a34b48ec29 --- /dev/null +++ b/tests_new/test_netmiko_scp.py @@ -0,0 +1,269 @@ +#!/usr/bin/env python +import pytest +from netmiko import file_transfer + + +def test_scp_put(scp_fixture): + ssh_conn, scp_transfer = scp_fixture + if scp_transfer.check_file_exists(): + assert False + else: + scp_transfer.put_file() + assert scp_transfer.check_file_exists() is True + + +def test_remote_space_available(scp_fixture, expected_responses): + ssh_conn, scp_transfer = scp_fixture + remote_space = scp_transfer.remote_space_available() + assert remote_space >= expected_responses["scp_remote_space"] + + +def test_local_space_available(scp_fixture): + ssh_conn, scp_transfer = scp_fixture + local_space = scp_transfer.local_space_available() + assert local_space >= 1_000_000_000 + + +def test_verify_space_available_put(scp_fixture): + ssh_conn, scp_transfer = scp_fixture + assert scp_transfer.verify_space_available() is True + # intentional make there not be enough space available + scp_transfer.file_size = 100_000_000_000 + assert scp_transfer.verify_space_available() is False + + +def test_remote_file_size(scp_fixture): + ssh_conn, scp_transfer = scp_fixture + if not scp_transfer.check_file_exists(): + scp_transfer.put_file() + remote_file_size = scp_transfer.remote_file_size() + assert remote_file_size == 19 + + +def test_md5_methods(scp_fixture): + ssh_conn, scp_transfer = scp_fixture + if "nokia_sros" in ssh_conn.device_type: + pytest.skip("MD5 not supported on this platform") + md5_value = "d8df36973ff832b564ad84642d07a261" + + remote_md5 = scp_transfer.remote_md5() + assert remote_md5 == md5_value + assert scp_transfer.compare_md5() is True + + +def test_disconnect(scp_fixture): + """Terminate the SSH session.""" + ssh_conn, scp_transfer = scp_fixture + ssh_conn.disconnect() + + +def test_verify_space_available_get(scp_fixture_get): + ssh_conn, scp_transfer = scp_fixture_get + assert scp_transfer.verify_space_available() is True + # intentional make there not be enough space available + scp_transfer.file_size = 100_000_000_000_000_000 + assert scp_transfer.verify_space_available() is False + + +def test_scp_get(scp_fixture_get): + ssh_conn, scp_transfer = scp_fixture_get + + if scp_transfer.check_file_exists(): + # File should not already exist + assert False + else: + scp_transfer.get_file() + if scp_transfer.check_file_exists(): + assert True + else: + assert False + + +def test_md5_methods_get(scp_fixture_get): + ssh_conn, scp_transfer = scp_fixture_get + platform = ssh_conn.device_type + if "nokia_sros" in platform: + pytest.skip("MD5 not supported on this platform") + md5_value = "d8df36973ff832b564ad84642d07a261" + local_md5 = scp_transfer.file_md5(f"test_{platform}/test9.txt") + assert local_md5 == md5_value + assert scp_transfer.compare_md5() is True + + +def test_disconnect_get(scp_fixture_get): + """Terminate the SSH session.""" + ssh_conn, scp_transfer = scp_fixture_get + ssh_conn.disconnect() + + +def test_file_transfer(scp_file_transfer): + """Test Netmiko file_transfer function.""" + ssh_conn, file_system = scp_file_transfer + platform = ssh_conn.device_type + source_file = f"test_{platform}/test9.txt" + dest_file = "test9.txt" + direction = "put" + + transfer_dict = file_transfer( + ssh_conn, + source_file=source_file, + dest_file=dest_file, + file_system=file_system, + direction=direction, + overwrite_file=True, + ) + + # No file on device at the beginning + assert ( + transfer_dict["file_exists"] + and transfer_dict["file_transferred"] + and transfer_dict["file_verified"] + ) + + # File exists on device at this point + transfer_dict = file_transfer( + ssh_conn, + source_file=source_file, + dest_file=dest_file, + file_system=file_system, + direction=direction, + overwrite_file=True, + ) + assert ( + transfer_dict["file_exists"] + and not transfer_dict["file_transferred"] + and transfer_dict["file_verified"] + ) + + # Don't allow a file overwrite (switch the source file, but same dest file name) + source_file = f"test_{platform}/test2_src.txt" + with pytest.raises(Exception): + transfer_dict = file_transfer( + ssh_conn, + source_file=source_file, + dest_file=dest_file, + file_system=file_system, + direction=direction, + overwrite_file=False, + ) + + # Don't allow MD5 and file overwrite not allowed + source_file = f"test_{platform}/test9.txt" + with pytest.raises(Exception): + transfer_dict = file_transfer( + ssh_conn, + source_file=source_file, + dest_file=dest_file, + disable_md5=True, + file_system=file_system, + direction=direction, + overwrite_file=False, + ) + + # Don't allow MD5 (this will force a re-transfer) + transfer_dict = file_transfer( + ssh_conn, + source_file=source_file, + dest_file=dest_file, + disable_md5=True, + file_system=file_system, + direction=direction, + overwrite_file=True, + ) + assert ( + transfer_dict["file_exists"] + and transfer_dict["file_transferred"] + and not transfer_dict["file_verified"] + ) + + # Transfer 'test2.txt' in preparation for get operations + source_file = "test2_src.txt" + dest_file = "test2.txt" + transfer_dict = file_transfer( + ssh_conn, + source_file=source_file, + dest_file=dest_file, + file_system=file_system, + direction=direction, + overwrite_file=True, + ) + assert transfer_dict["file_exists"] + + # GET Operations + direction = "get" + source_file = "test9.txt" + dest_file = f"test_{platform}/testx.txt" + transfer_dict = file_transfer( + ssh_conn, + source_file=source_file, + dest_file=dest_file, + disable_md5=False, + file_system=file_system, + direction=direction, + overwrite_file=True, + ) + # File get should occur here + assert ( + transfer_dict["file_exists"] + and transfer_dict["file_transferred"] + and transfer_dict["file_verified"] + ) + + # File should exist now + transfer_dict = file_transfer( + ssh_conn, + source_file=source_file, + dest_file=dest_file, + disable_md5=False, + file_system=file_system, + direction=direction, + overwrite_file=True, + ) + assert ( + transfer_dict["file_exists"] + and not transfer_dict["file_transferred"] + and transfer_dict["file_verified"] + ) + + # Don't allow a file overwrite (switch the file, but same dest file name) + source_file = f"test_{platform}/test2.txt" + dest_file = f"test_{platform}/testx.txt" + with pytest.raises(Exception): + transfer_dict = file_transfer( + ssh_conn, + source_file=source_file, + dest_file=dest_file, + file_system=file_system, + direction=direction, + overwrite_file=False, + ) + + # Don't allow MD5 and file overwrite not allowed + source_file = "test9.txt" + dest_file = f"test_{platform}/testx.txt" + with pytest.raises(Exception): + transfer_dict = file_transfer( + ssh_conn, + source_file=source_file, + dest_file=dest_file, + disable_md5=True, + file_system=file_system, + direction=direction, + overwrite_file=False, + ) + + # Don't allow MD5 (this will force a re-transfer) + transfer_dict = file_transfer( + ssh_conn, + source_file=source_file, + dest_file=dest_file, + disable_md5=True, + file_system=file_system, + direction=direction, + overwrite_file=True, + ) + assert ( + transfer_dict["file_exists"] + and transfer_dict["file_transferred"] + and not transfer_dict["file_verified"] + ) diff --git a/tests_new/test_netmiko_show.py b/tests_new/test_netmiko_show.py new file mode 100755 index 0000000000000000000000000000000000000000..3ae21b8a1657ff46f66a4f7ec5aa08b2ef66f7e5 --- /dev/null +++ b/tests_new/test_netmiko_show.py @@ -0,0 +1,280 @@ +#!/usr/bin/env python +""" +setup_module: setup variables for later use. + +test_disable_paging: disable paging +test_ssh_connect: verify ssh connectivity +test_send_command: send a command +test_send_command_timing: send a command +test_base_prompt: test the base prompt +test_strip_prompt: test removing the prompt +test_strip_command: test stripping extraneous info after sending a command +test_normalize_linefeeds: ensure \n is the only line termination character in output +test_clear_buffer: clear text buffer +test_enable_mode: verify enter enable mode +test_disconnect: cleanly disconnect the SSH session +""" +import pytest +import time +from datetime import datetime + + +def test_disable_paging(net_connect, commands, expected_responses): + """Verify paging is disabled by looking for string after when paging would normally occur.""" + # FIX: these really shouldn't be necessary. + if net_connect.device_type == "arista_eos": + # Arista logging buffer gets enormous + net_connect.send_command("clear logging") + elif net_connect.device_type == "arista_eos": + # NX-OS logging buffer gets enormous (NX-OS fails when testing very high-latency + + # packet loss) + net_connect.send_command("clear logging logfile") + multiple_line_output = net_connect.send_command(commands["extended_output"]) + assert expected_responses["multiple_line_output"] in multiple_line_output + + +def test_terminal_width(net_connect, commands, expected_responses): + """Verify long commands work properly.""" + wide_command = commands.get("wide_command") + if wide_command: + net_connect.send_command(wide_command) + assert True + + +def test_ssh_connect(net_connect, commands, expected_responses): + """Verify the connection was established successfully.""" + show_version = net_connect.send_command(commands["version"]) + assert expected_responses["version_banner"] in show_version + + +def test_ssh_connect_cm(net_connect_cm, commands, expected_responses): + """Test the context manager.""" + prompt_str = net_connect_cm + assert expected_responses["base_prompt"] in prompt_str + + +def test_send_command_timing(net_connect, commands, expected_responses): + """Verify a command can be sent down the channel successfully.""" + time.sleep(1) + net_connect.clear_buffer() + # Force verification of command echo + show_ip = net_connect.send_command_timing(commands["basic"], cmd_verify=True) + assert expected_responses["interface_ip"] in show_ip + + +def test_send_command_timing_no_cmd_verify(net_connect, commands, expected_responses): + # Skip devices that are performance optimized (i.e. cmd_verify is required there) + if net_connect.fast_cli is True: + assert pytest.skip() + time.sleep(1) + net_connect.clear_buffer() + # cmd_verify=False is the default + show_ip = net_connect.send_command_timing(commands["basic"], cmd_verify=False) + assert expected_responses["interface_ip"] in show_ip + + +def test_send_command(net_connect, commands, expected_responses): + """Verify a command can be sent down the channel successfully using send_command method.""" + net_connect.clear_buffer() + show_ip_alt = net_connect.send_command(commands["basic"]) + assert expected_responses["interface_ip"] in show_ip_alt + + +def test_send_command_no_cmd_verify(net_connect, commands, expected_responses): + # Skip devices that are performance optimized (i.e. cmd_verify is required there) + if net_connect.fast_cli is True: + assert pytest.skip() + net_connect.clear_buffer() + show_ip_alt = net_connect.send_command(commands["basic"], cmd_verify=False) + assert expected_responses["interface_ip"] in show_ip_alt + + +def test_complete_on_space_disabled(net_connect, commands, expected_responses): + """Verify complete on space is disabled.""" + # If complete on space is enabled will get re-written to "show configuration groups" + if net_connect.device_type in ["juniper_junos", "nokia_sros"]: + if ( + net_connect.device_type == "nokia_sros" + and "@" not in net_connect.base_prompt + ): + # Only MD-CLI supports disable of command complete on space + assert pytest.skip() + cmd = commands.get("abbreviated_cmd") + cmd_full = commands.get("abbreviated_cmd_full") + net_connect.write_channel(f"{cmd}\n") + output = net_connect.read_until_prompt() + assert cmd_full not in output + else: + assert pytest.skip() + + +def test_send_command_textfsm(net_connect, commands, expected_responses): + """Verify a command can be sent down the channel successfully using send_command method.""" + + base_platform = net_connect.device_type + if base_platform.count("_") >= 2: + # Strip off the _ssh, _telnet, _serial + base_platform = base_platform.split("_")[:-1] + base_platform = "_".join(base_platform) + if base_platform not in [ + "cisco_ios", + # "cisco_xe", FIX: re-enable after a translation is made for ntc-templates + "cisco_xr", + "cisco_nxos", + "arista_eos", + "cisco_asa", + "juniper_junos", + "hp_procurve", + ]: + assert pytest.skip("TextFSM/ntc-templates not supported on this platform") + else: + time.sleep(1) + net_connect.clear_buffer() + fallback_cmd = commands.get("basic") + command = commands.get("basic_textfsm", fallback_cmd) + show_ip_alt = net_connect.send_command(command, use_textfsm=True) + assert isinstance(show_ip_alt, list) + + +def test_send_command_ttp(net_connect): + """ + Verify a command can be sent down the channel + successfully using send_command method. + """ + + base_platform = net_connect.device_type + if base_platform.count("_") >= 2: + # Strip off the _ssh, _telnet, _serial + base_platform = base_platform.split("_")[:-1] + base_platform = "_".join(base_platform) + if base_platform not in ["cisco_ios"]: + assert pytest.skip("TTP template not existing for this platform") + else: + time.sleep(1) + net_connect.clear_buffer() + + # write a simple template to file + ttp_raw_template = """ + description {{ description }} + """ + with open("show_run_interfaces.ttp", "w") as writer: + writer.write(ttp_raw_template) + + command = "show run | s interfaces" + show_ip_alt = net_connect.send_command( + command, use_ttp=True, ttp_template="show_run_interfaces.ttp" + ) + assert isinstance(show_ip_alt, list) + + +def test_send_command_genie(net_connect, commands, expected_responses): + """Verify a command can be sent down the channel successfully using send_command method.""" + + base_platform = net_connect.device_type + if base_platform.count("_") >= 2: + # Strip off the _ssh, _telnet, _serial + base_platform = base_platform.split("_")[:-1] + base_platform = "_".join(base_platform) + if base_platform not in [ + "cisco_ios", + "cisco_xe", + "cisco_xr", + "cisco_nxos", + "cisco_asa", + ]: + assert pytest.skip("Genie not supported on this platform") + else: + time.sleep(1) + net_connect.clear_buffer() + fallback_cmd = commands.get("basic") + command = commands.get("basic_textfsm", fallback_cmd) + show_ip_alt = net_connect.send_command(command, use_genie=True) + assert isinstance(show_ip_alt, dict) + + +def test_base_prompt(net_connect, commands, expected_responses): + """Verify the router prompt is detected correctly.""" + assert net_connect.base_prompt == expected_responses["base_prompt"] + + +def test_strip_prompt(net_connect, commands, expected_responses): + """Ensure the router prompt is not in the command output.""" + + if expected_responses["base_prompt"] == "": + return + show_ip = net_connect.send_command_timing(commands["basic"]) + show_ip_alt = net_connect.send_command(commands["basic"]) + assert expected_responses["base_prompt"] not in show_ip + assert expected_responses["base_prompt"] not in show_ip_alt + + +def test_strip_command(net_connect, commands, expected_responses): + """Ensure that the command that was executed does not show up in the command output.""" + show_ip = net_connect.send_command_timing(commands["basic"]) + show_ip_alt = net_connect.send_command(commands["basic"]) + + # dlink_ds has an echo of the command in the command output + if "dlink_ds" in net_connect.device_type: + show_ip = "\n".join(show_ip.split("\n")[2:]) + show_ip_alt = "\n".join(show_ip_alt.split("\n")[2:]) + assert commands["basic"] not in show_ip + assert commands["basic"] not in show_ip_alt + + +def test_normalize_linefeeds(net_connect, commands, expected_responses): + """Ensure no '\r\n' sequences.""" + show_version = net_connect.send_command_timing(commands["version"]) + show_version_alt = net_connect.send_command(commands["version"]) + assert "\r\n" not in show_version + assert "\r\n" not in show_version_alt + + +def test_clear_buffer(net_connect, commands, expected_responses): + """Test that clearing the buffer works.""" + # Manually send a command down the channel so that data needs read. + net_connect.write_channel(commands["basic"] + "\n") + time.sleep(4) + net_connect.clear_buffer() + + # Should not be anything there on the second pass + clear_buffer_check = net_connect.clear_buffer() + assert clear_buffer_check is None + + +def test_enable_mode(net_connect, commands, expected_responses): + """ + Test entering enable mode + + Catch exception for devices that don't support enable + """ + try: + net_connect.enable() + enable_prompt = net_connect.find_prompt() + assert enable_prompt == expected_responses["enable_prompt"] + except AttributeError: + assert True + + +def test_disconnect(net_connect, commands, expected_responses): + """Terminate the SSH session.""" + start_time = datetime.now() + net_connect.disconnect() + end_time = datetime.now() + time_delta = end_time - start_time + assert net_connect.remote_conn is None + assert time_delta.total_seconds() < 8 + + +def test_disconnect_no_enable(net_connect_newconn, commands, expected_responses): + """Terminate the SSH session from privilege level1""" + net_connect = net_connect_newconn + if "cisco_ios" in net_connect.device_type: + net_connect.send_command_timing("disable") + start_time = datetime.now() + net_connect.disconnect() + end_time = datetime.now() + time_delta = end_time - start_time + assert net_connect.remote_conn is None + assert time_delta.total_seconds() < 5 + else: + assert True diff --git a/tests_new/test_netmiko_tcl.py b/tests_new/test_netmiko_tcl.py new file mode 100755 index 0000000000000000000000000000000000000000..ccb08e961153c0991fa9857a36dc6ccd6ed7ecaf --- /dev/null +++ b/tests_new/test_netmiko_tcl.py @@ -0,0 +1,45 @@ +#!/usr/bin/env python +def test_tcl_put(tcl_fixture): + ssh_conn, transfer = tcl_fixture + if transfer.check_file_exists(): + assert False + else: + transfer._enter_tcl_mode() + transfer.put_file() + transfer._exit_tcl_mode() + assert transfer.check_file_exists() is True + + +def test_remote_space_available(tcl_fixture): + ssh_conn, transfer = tcl_fixture + remote_space = transfer.remote_space_available() + assert remote_space >= 30000000 + + +def test_verify_space_available_put(tcl_fixture): + ssh_conn, transfer = tcl_fixture + assert transfer.verify_space_available() is True + # intentional make there not be enough space available + transfer.file_size = 1000000000 + assert transfer.verify_space_available() is False + + +def test_remote_file_size(tcl_fixture): + ssh_conn, transfer = tcl_fixture + remote_file_size = transfer.remote_file_size() + assert remote_file_size == 20 + + +def test_md5_methods(tcl_fixture): + ssh_conn, transfer = tcl_fixture + md5_value = "4313f1adae86a21117441b0a95d482a7" + + remote_md5 = transfer.remote_md5() + assert remote_md5 == md5_value + assert transfer.compare_md5() is True + + +def test_disconnect(tcl_fixture): + """Terminate the SSH session.""" + ssh_conn, transfer = tcl_fixture + ssh_conn.disconnect() diff --git a/tests_new/test_utils.py b/tests_new/test_utils.py new file mode 100644 index 0000000000000000000000000000000000000000..719a96cf2ec7b43ac03a936060466a53f7c5545b --- /dev/null +++ b/tests_new/test_utils.py @@ -0,0 +1,23 @@ +#!/usr/bin/env python +""" +Implement common functions for tests +""" +import io +import sys + + +def parse_yaml(yaml_file): + """ + Parses a yaml file, returning its contents as a dict. + """ + + try: + import yaml + except ImportError: + sys.exit("Unable to import yaml module.") + + try: + with io.open(yaml_file, encoding="utf-8") as fname: + return yaml.safe_load(fname) + except IOError: + sys.exit("Unable to open YAML file: {0}".format(yaml_file))