From 519032588a770fcae31167a86abfc86b635dd9f6 Mon Sep 17 00:00:00 2001 From: Josh VanDeraa Date: Sat, 8 Aug 2026 14:22:36 -0500 Subject: [PATCH 1/3] Add arista_eos_ssh driver for Arista EOS over SSH Adds EOSSSHDevice for environments where eAPI is not enabled. It subclasses EOSDevice and overrides only the transport surface (__init__, open, close, show, config, reboot, vlans) plus a native_ssh alias, so the fact properties, boot-option handling, install_os and the whole file-transfer family are inherited unchanged. Structured data comes from the CLI's '| json' pipe, which renders the same document eAPI returns; the pipe is a CLI feature and does not require 'management api http-commands'. The pipe is gated on '^show' because EOSDevice routes five EXEC/config commands through show(raw_text=False), and per-command read timeouts are resolved from the command text so show()'s signature stays identical to EOSDevice.show(). Validated against a DCS-7050TX-64-R running EOS 4.28.5M; the unit fixtures are that capture, sanitised. Public API parity with EOSDevice is enforced by test rather than by convention, and a further test feeds both drivers the same document and requires identical facts. --- changes/418.added | 1 + docs/user/lib_getting_started.md | 5 +- mkdocs.yml | 1 + pyntc/devices/__init__.py | 2 + pyntc/devices/eos_ssh_device.py | 430 ++++++++ tests/fixtures/.ntc.conf.sample | 8 +- tests/integration/conftest.py | 1 + tests/integration/test_eos_ssh_device.py | 229 ++++ tests/unit/conftest.py | 61 +- .../device_mocks/eos_ssh/README.md | 41 + .../test_devices/device_mocks/eos_ssh/dir | 18 + .../device_mocks/eos_ssh/show_boot | 6 + .../eos_ssh/show_boot-config_json | 18 + .../device_mocks/eos_ssh/show_hostname_json | 4 + .../eos_ssh/show_interfaces_status_json | 978 ++++++++++++++++++ .../device_mocks/eos_ssh/show_running-config | 48 + .../device_mocks/eos_ssh/show_startup-config | 49 + .../device_mocks/eos_ssh/show_version_json | 20 + .../device_mocks/eos_ssh/show_vlan_json | 35 + .../unit/test_devices/test_eos_ssh_device.py | 962 +++++++++++++++++ tests/unit/test_infra.py | 11 +- 21 files changed, 2922 insertions(+), 6 deletions(-) create mode 100644 changes/418.added create mode 100644 pyntc/devices/eos_ssh_device.py create mode 100644 tests/integration/test_eos_ssh_device.py create mode 100644 tests/unit/test_devices/device_mocks/eos_ssh/README.md create mode 100644 tests/unit/test_devices/device_mocks/eos_ssh/dir create mode 100644 tests/unit/test_devices/device_mocks/eos_ssh/show_boot create mode 100644 tests/unit/test_devices/device_mocks/eos_ssh/show_boot-config_json create mode 100644 tests/unit/test_devices/device_mocks/eos_ssh/show_hostname_json create mode 100644 tests/unit/test_devices/device_mocks/eos_ssh/show_interfaces_status_json create mode 100644 tests/unit/test_devices/device_mocks/eos_ssh/show_running-config create mode 100644 tests/unit/test_devices/device_mocks/eos_ssh/show_startup-config create mode 100644 tests/unit/test_devices/device_mocks/eos_ssh/show_version_json create mode 100644 tests/unit/test_devices/device_mocks/eos_ssh/show_vlan_json create mode 100644 tests/unit/test_devices/test_eos_ssh_device.py diff --git a/changes/418.added b/changes/418.added new file mode 100644 index 00000000..40e405fb --- /dev/null +++ b/changes/418.added @@ -0,0 +1 @@ +Added the `arista_eos_ssh` device type, an SSH-only Arista EOS driver for environments where eAPI is not enabled; it exposes the same API as `arista_eos_eapi` and obtains structured data via the CLI's `| json` pipe. diff --git a/docs/user/lib_getting_started.md b/docs/user/lib_getting_started.md index 8d45927d..a7f962bf 100644 --- a/docs/user/lib_getting_started.md +++ b/docs/user/lib_getting_started.md @@ -11,16 +11,19 @@ The first way is to use the `ntc_device` object. Just pass in all required param Like many libraries, we need to pass in the host/IP and credentials. Because this is a multi-vendor/API library, we also use the `device_type` parameter to identify which device we are building an instance of. -pyntc currently supports seven device types: +pyntc currently supports the following device types: - cisco_aireos_ssh - cisco_asa_ssh - cisco_ios_ssh - cisco_nxos_nxapi - arista_eos_eapi +- arista_eos_ssh - juniper_junos_netconf - f5_tmos_icontrol +Arista EOS is supported over two transports. `arista_eos_eapi` uses eAPI (JSON-RPC over HTTP/HTTPS) and requires `management api http-commands` to be enabled on the device. `arista_eos_ssh` uses SSH only, for environments where eAPI is not available; it exposes exactly the same methods and properties as the eAPI driver, so the two are interchangeable. + The example below shows how to build a device object when working with a Cisco IOS router. ```python diff --git a/mkdocs.yml b/mkdocs.yml index 9d9a9e76..2125dabc 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -162,6 +162,7 @@ nav: - pyntc.devices.asa_device: "code-reference/pyntc/devices/asa_device.md" - pyntc.devices.base_device: "code-reference/pyntc/devices/base_device.md" - pyntc.devices.eos_device: "code-reference/pyntc/devices/eos_device.md" + - pyntc.devices.eos_ssh_device: "code-reference/pyntc/devices/eos_ssh_device.md" - pyntc.devices.f5_device: "code-reference/pyntc/devices/f5_device.md" - pyntc.devices.ios_device: "code-reference/pyntc/devices/ios_device.md" - pyntc.devices.iosxewlc_device: "code-reference/pyntc/devices/iosxewlc_device.md" diff --git a/pyntc/devices/__init__.py b/pyntc/devices/__init__.py index 3756dd33..83bae9c1 100644 --- a/pyntc/devices/__init__.py +++ b/pyntc/devices/__init__.py @@ -3,6 +3,7 @@ from .aireos_device import AIREOSDevice from .asa_device import ASADevice from .eos_device import EOSDevice +from .eos_ssh_device import EOSSSHDevice from .f5_device import F5Device from .ios_device import IOSDevice from .iosxewlc_device import IOSXEWLCDevice @@ -13,6 +14,7 @@ supported_devices = { "cisco_asa_ssh": ASADevice, "arista_eos_eapi": EOSDevice, + "arista_eos_ssh": EOSSSHDevice, "f5_tmos_icontrol": F5Device, "cisco_ios_ssh": IOSDevice, "cisco_iosxr_ssh": IOSXRDevice, diff --git a/pyntc/devices/eos_ssh_device.py b/pyntc/devices/eos_ssh_device.py new file mode 100644 index 00000000..0f33030e --- /dev/null +++ b/pyntc/devices/eos_ssh_device.py @@ -0,0 +1,430 @@ +"""Module for using an Arista EOS device over SSH. + +This driver exists for environments where eAPI (``management api http-commands``) is not +available. It exposes the same public API as +:class:`~pyntc.devices.eos_device.EOSDevice`; only the transport differs. + +Structured output is obtained with the CLI's ``| json`` pipe, which renders the same +document eAPI returns -- the pipe is a pure CLI feature and does **not** require eAPI to be +enabled. Because the key names match, every fact property, the boot-option handling and the +whole file-transfer family are inherited from ``EOSDevice`` unchanged. +""" + +import json +import os +import re + +from netmiko import ConnectHandler + +from pyntc import log +from pyntc.devices.base_device import BaseDevice, fix_docs +from pyntc.devices.eos_device import DEFAULT_REBOOT_TIMEOUT, EOSDevice +from pyntc.errors import CommandError, CommandListError, FileTransferError, SocketClosedError + +DEFAULT_SSH_PORT = 22 + +# Only "show" commands may be piped to "| json". EOSDevice routes five EXEC/config commands +# through show(raw_text=False) whose return value it discards -- "copy running-config ...", +# "reload now", "configure replace ... force" and "install source ..." -- and +# "reload now | json" is not a valid command. +RE_JSON_ELIGIBLE = re.compile(r"^\s*show\b") + +# EOS reports CLI failures with a leading "% " token. "Invalid input" also appears for +# commands that have no JSON renderer, which _load_json turns into a CommandError rather +# than silently falling back to text parsing (a fallback would return a differently shaped +# document and produce wrong facts instead of an error). +RE_EOS_CLI_ERROR = re.compile(r"^%\s|^Invalid input|^Error:", re.MULTILINE) + +# Commands whose default Netmiko read timeout (100s) is too short. Resolving the timeout +# from the command text -- rather than adding a **netmiko_args parameter -- keeps show()'s +# signature byte-identical to EOSDevice.show(), so inherited callers such as +# ``set_boot_options``, ``save``, ``checkpoint`` and ``rollback`` work without overrides. +COMMAND_READ_TIMEOUTS = ( + (re.compile(r"^\s*install\s+source\b"), 1800), + (re.compile(r"^\s*copy\s+running-config\b"), 300), + (re.compile(r"^\s*configure\s+replace\b"), 300), + (re.compile(r"^\s*show\s+(running|startup)-config\b"), 120), +) +DEFAULT_READ_TIMEOUT = 100 + + +@fix_docs +class EOSSSHDevice(EOSDevice): + """Arista EOS Device Implementation over SSH.""" + + # pylint: disable=too-many-arguments, too-many-positional-arguments, super-init-not-called + def __init__(self, host, username, password, secret="", port=None, **kwargs): # nosec # noqa: D403 + """PyNTC Device implementation for Arista EOS over SSH. + + Args: + host (str): The address of the network device. + username (str): The username to authenticate with the device. + password (str): The password to authenticate with the device. + secret (str): The password to escalate privilege on the device. + port (int): The SSH port to connect on. Defaults to 22. Note this differs from + ``EOSDevice.port``, which is the eAPI port. + kwargs (dict): Additional arguments passed to Netmiko's ``ConnectHandler``. + """ + # Deliberately skips EOSDevice.__init__, which eagerly builds a pyeapi connection + # and takes eAPI-only arguments (transport/timeout). Going straight to BaseDevice + # keeps the shared state without the eAPI wiring. + BaseDevice.__init__( # pylint: disable=non-parent-init-called + self, host, username, password, device_type="arista_eos_ssh" + ) + self.native = None + self.secret = secret + self.port = int(port) if port else DEFAULT_SSH_PORT + self.netmiko_kwargs = kwargs + self._connected = False + self.open() + log.init(host=host) + + @property + def native_ssh(self): + """Alias for ``native`` so inherited Netmiko-backed code works unchanged. + + ``EOSDevice`` reaches for ``self.native_ssh`` in ``enable``, ``file_copy``, + ``check_file_exists``, ``get_remote_checksum`` and ``remote_file_copy``. Only + ``EOSDevice.open`` ever assigns it, and this class overrides ``open``, so exposing + it read-only is safe. + + Returns: + (netmiko.BaseConnection): The active Netmiko connection. + """ + return self.native + + @staticmethod + def _read_timeout_for(command): + """Resolve the Netmiko read timeout to use for ``command``. + + Args: + command (str): The command about to be sent. + + Returns: + (int): Timeout in seconds. + """ + for pattern, timeout in COMMAND_READ_TIMEOUTS: + if pattern.match(command): + return timeout + return DEFAULT_READ_TIMEOUT + + def _check_output_for_errors(self, command, output): + """Raise ``CommandError`` when the device reported a CLI error. + + Args: + command (str): The command that was sent. + output (str): The device response. + + Raises: + CommandError: When ``output`` reports an error. + """ + if RE_EOS_CLI_ERROR.search(output): + log.error("Host %s: Error in %s with response: %s", self.host, command, output) + raise CommandError(command, output) + + def _load_json(self, command, output): + """Parse ``| json`` output. + + Args: + command (str): The command that produced ``output``. + output (str): Raw device response. + + Returns: + (dict): The parsed document. + + Raises: + CommandError: When the output is not valid JSON, which on EOS means the command + has no JSON renderer. + """ + try: + return json.loads(output) + except ValueError: + log.error("Host %s: Command %s did not return JSON: %s", self.host, command, output) + raise CommandError(command, f"Command does not support JSON output: {output}") + + def _send_command(self, command, error_command=None, **netmiko_args): + """Send a single command and check the response for errors. + + Args: + command (str): The command to send on the wire. + error_command (str, optional): The command to name in a raised ``CommandError``. + Defaults to ``command``. ``show`` passes the caller's original command so + errors do not leak the ``| json`` suffix, matching the plain command name + that pyeapi reports on ``EOSDevice``. + netmiko_args (dict): Additional arguments for Netmiko's ``send_command``. + + Returns: + (str): The raw device response. + """ + netmiko_args.setdefault("read_timeout", self._read_timeout_for(command)) + response = self.native.send_command(command, **netmiko_args) + self._check_output_for_errors(error_command or command, response) + return response + + def open(self): + """Open, or re-validate, the Netmiko SSH connection to the device.""" + if self._connected: + try: + self.native.find_prompt() + except Exception: # pylint: disable=broad-except + self._connected = False + + if not self._connected: + self.native = ConnectHandler( + device_type="arista_eos", + host=self.host, + username=self.username, + password=self.password, + port=self.port, + secret=self.secret, + verbose=False, + **self.netmiko_kwargs, + ) + self._connected = True + + log.debug("Host %s: Connection to device was opened successfully.", self.host) + + def close(self): + """Disconnect from the device. + + Note this differs from ``EOSDevice.close``, which is a no-op because eAPI is + stateless. An SSH session holds a real socket that should be released. + """ + if self._connected: + self.native.disconnect() + self._connected = False + log.debug("Host %s: Connection closed.", self.host) + + def show(self, commands, raw_text=False): + """Send show command(s) to the device. + + Args: + commands (str, list): String with single command, or list with multiple commands. + raw_text (bool, optional): False to return structured data via the ``| json`` + pipe, True to return the raw CLI text. Defaults to False. + + Returns: + (dict): When ``commands`` is a str and ``raw_text`` is False. + (str): When ``commands`` is a str and ``raw_text`` is True. + (list): When ``commands`` is a list. + + Raises: + CommandError: When ``commands`` is a str and the device reports an error. + CommandListError: When ``commands`` is a list and one command reports an error. + """ + self.open() + self.enable() + + original_commands_is_str = isinstance(commands, str) + command_list = [commands] if original_commands_is_str else list(commands) + + responses = [] + entered_commands = [] + for command in command_list: + entered_commands.append(command) + as_json = not raw_text and bool(RE_JSON_ELIGIBLE.match(command)) + cli_command = f"{command} | json" if as_json else command + try: + output = self._send_command(cli_command, error_command=command) + except CommandError as err: + if original_commands_is_str: + raise + raise CommandListError(entered_commands, command, err.cli_error_msg) from err + + if raw_text: + responses.append(output) + elif as_json: + responses.append(self._load_json(command, output)) + else: + # Non-show command sent with raw_text=False (checkpoint, save, rollback, + # reboot, set_boot_options). Every inherited caller discards the result, + # so an empty dict preserves EOSDevice's contract. + responses.append({}) + + if original_commands_is_str: + return responses[0] + + log.debug("Host %s: Successfully executed command 'show' with responses %s.", self.host, responses) + return responses + + def config(self, commands): + """Send configuration commands to a device. + + Args: + commands (str, list): String with single command, or list with multiple commands. + + Raises: + CommandError: When ``commands`` is a str and the device reports an error. + CommandListError: When ``commands`` is a list and one command reports an error. + """ + self.open() + self.enable() + + original_commands_is_str = isinstance(commands, str) + command_list = [commands] if original_commands_is_str else list(commands) + + entered_commands = [] + try: + for command in command_list: + entered_commands.append(command) + output = self.native.send_config_set(command, exit_config_mode=False) + try: + self._check_output_for_errors(command, output) + except CommandError as err: + if original_commands_is_str: + raise + raise CommandListError(entered_commands, command, err.cli_error_msg) from err + finally: + # Never leave the session parked in config mode, even on failure. + self.native.exit_config_mode() + + log.info("Host %s: Device configured with commands %s.", self.host, commands) + + def reboot(self, wait_for_reload=False, timeout=DEFAULT_REBOOT_TIMEOUT, **kwargs): + """Reload the device. + + Unlike eAPI, the SSH session dies as the reload executes, so the command is sent + with ``send_command_timing`` and the resulting transport error is expected. + + Args: + wait_for_reload (bool): When True, block until the device's boot time advances + past the pre-reboot value. Defaults to False. + timeout (int): Max seconds to poll when ``wait_for_reload`` is True. + kwargs (dict): Additional keyword arguments, such as confirm. + + Raises: + RebootTimeoutError: When the device does not return within ``timeout``. + + Example: + >>> device = EOSSSHDevice(**connection_args) + >>> device.reboot() + >>> + """ + if kwargs.get("confirm"): + log.warning("Passing 'confirm' to reboot method is deprecated.") + + original_boot_time = self.boot_time if wait_for_reload else None + try: + self.native.send_command_timing("reload now") + except Exception as err: # pylint: disable=broad-except + log.debug("Host %s: Session dropped during reload, as expected (%s).", self.host, err) + + # The socket is gone regardless of how the command returned; force the next + # operation to reconnect rather than reuse a dead handle. + self._connected = False + log.info("Host %s: Device rebooted.", self.host) + + if wait_for_reload: + # Both arguments are numeric; naming them prevents a transposition from + # silently satisfying the "boot time advanced" check on the first poll. + self._wait_for_device_reboot(original_boot_time=original_boot_time, timeout=timeout) + + def file_copy_remote_exists(self, src, dest=None, file_system=None): + """Check whether ``src`` already exists on the device with a matching checksum. + + ``EOSDevice`` answers this through Netmiko's ``AristaFileTransfer``, which drops into + the switch's Linux shell (``bash`` then ``/bin/ls``). That requires shell privileges + the connecting account may not have. This override uses the CLI instead -- + ``dir /`` and ``verify /md5 `` -- matching how + ``IOSDevice`` already behaves. + + Args: + src (str): Path to the local file to check for. + dest (str, optional): Remote filename. Defaults to the basename of ``src``. + file_system (str, optional): Target filesystem. Auto-detected when omitted. + + Returns: + (bool): True when the remote file exists and its checksum matches ``src``. + """ + self.open() + self.enable() + if file_system is None: + file_system = self._get_file_system() + + dest = dest or os.path.basename(src) + local_checksum = self.get_local_checksum(src) + exists = self.verify_file(local_checksum, dest, file_system=file_system) + + log.debug("Host %s: File %s already on remote: %s.", self.host, src, exists) + return exists + + def file_copy(self, src, dest=None, file_system=None): + """Copy a local file to the device over SCP. + + Mirrors ``IOSDevice.file_copy``: existence and integrity are established with CLI + commands rather than Netmiko's shell-based helpers, so no ``bash`` access is needed. + ``AristaFileTransfer.enable_scp()`` raises ``NotImplementedError``, so unlike IOS + there is no SCP-enable step -- EOS serves SCP without one. + + Args: + src (str): Path to the local file to send. + dest (str, optional): Remote filename. Defaults to the basename of ``src``. + file_system (str, optional): Target filesystem. Auto-detected when omitted. + + Raises: + SocketClosedError: When the session drops mid-transfer and the file did not land. + FileTransferError: When the transfer fails, or the file cannot be verified + afterwards. + NotEnoughFreeSpaceError: When ``file_system`` has less room than ``src`` needs. + """ + self.open() + self.enable() + if file_system is None: + file_system = self._get_file_system() + + dest = dest or os.path.basename(src) + local_checksum = self.get_local_checksum(src) + log.debug("Host %s: Local checksum for file %s is %s.", self.host, src, local_checksum) + + if self.verify_file(local_checksum, dest, file_system=file_system): + log.info("Host %s: File %s already present and verified; skipping.", self.host, dest) + return + + self._check_free_space(os.path.getsize(src), file_system=file_system) + file_copy = self._file_copy_instance(src, dest, file_system=file_system) + + try: + file_copy.establish_scp_conn() + file_copy.transfer_file() + log.info("Host %s: File %s transferred successfully.", self.host, src) + except OSError as error: + # A dropped control channel does not necessarily mean a failed transfer; + # compare_md5() uses the CLI "verify" command, so it is safe without a shell. + if not file_copy.compare_md5(): + log.error("Host %s: Socket closed error %s", self.host, error) + raise SocketClosedError(message=error) from error + log.error("Host %s: OS error %s", self.host, error) + except: # noqa: E722 + log.error("Host %s: File transfer error %s", self.host, FileTransferError.default_message) + raise FileTransferError + finally: + file_copy.close_scp_chan() + + # Long transfers can outlive the control channel; make sure it is usable again. + self.open() + + if not self.verify_file(local_checksum, dest, file_system=file_system): + log.error( + "Host %s: Attempted file copy, but could not validate file existed after transfer %s", + self.host, + FileTransferError.default_message, + ) + raise FileTransferError + + @property + def vlans(self): + """Get list of VLANs on device. + + ``EOSDevice`` delegates to ``EOSVlans``, which is pyeapi-only + (``device.native.api("vlans")``). Over SSH the same data comes from + ``show vlan | json``, whose ``vlans`` key is a dict keyed by VLAN id. + + Returns: + (list): List of VLAN ids as strings. + """ + if self._vlans is None: + # sorted() over str keys, matching EOSVlans.get_list()'s lexicographic ordering. + self._vlans = sorted(self.show("show vlan")["vlans"].keys()) + + log.debug("Host %s: Vlans %s", self.host, self._vlans) + return self._vlans diff --git a/tests/fixtures/.ntc.conf.sample b/tests/fixtures/.ntc.conf.sample index e3ff40c0..fff8d330 100644 --- a/tests/fixtures/.ntc.conf.sample +++ b/tests/fixtures/.ntc.conf.sample @@ -9,7 +9,13 @@ username: user password: arista transport: http +[arista_eos_ssh:test_eos_ssh] +host: 192.168.43.4 +username: user +password: arista +port: 22 + [cisco_ios_ssh:test_ios] username: user password: pass -secret: cisco \ No newline at end of file +secret: cisco diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 6e4fb233..a423e157 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -19,6 +19,7 @@ # integration tests. _PLATFORM_HASH_ALGOS = { "test_eos_device": "sha512", + "test_eos_ssh_device": "sha512", "test_asa_device": "sha512", "test_jnpr_device": "sha256", "test_ios_device": "md5", diff --git a/tests/integration/test_eos_ssh_device.py b/tests/integration/test_eos_ssh_device.py new file mode 100644 index 00000000..39883330 --- /dev/null +++ b/tests/integration/test_eos_ssh_device.py @@ -0,0 +1,229 @@ +"""Integration tests for EOSSSHDevice. + +These tests connect to an actual Arista EOS device over SSH and are run manually. +They are NOT part of the CI unit test suite. + +This suite is the hardware validation for the ``arista_eos_ssh`` driver. It deliberately +covers two things the unit tests cannot: + +1. That ``show | json`` really does return the eAPI-shaped document the driver + relies on -- ``test_json_key_contract`` asserts every key the inherited fact properties + dereference. This is the design's load-bearing assumption. +2. That eAPI is genuinely not required -- nothing here enables or touches + ``management api http-commands``. + +Usage (from project root): + export EOS_SSH_HOST= + export EOS_SSH_USER= + export EOS_SSH_PASS= + export SCP_URL=scp://:@/ + export HTTP_URL=http://:@:8081/ + export FILE_CHECKSUM_512= + export FILE_SIZE= + export FILE_SIZE_UNIT=megabytes # optional; defaults to "bytes" + poetry run pytest tests/integration/test_eos_ssh_device.py -v + +Set only the protocol URL vars for the servers you have available; each protocol test +skips automatically if its URL is not set. + +Environment variables: + EOS_SSH_HOST - IP address or hostname of the lab EOS device + EOS_SSH_USER - SSH username + EOS_SSH_PASS - SSH password + EOS_SSH_SECRET - Enable secret (optional) + EOS_SSH_PORT - SSH port (optional; defaults to 22) + FTP_URL - FTP URL of the file to transfer + TFTP_URL - TFTP URL of the file to transfer + SCP_URL - SCP URL of the file to transfer + HTTP_URL - HTTP URL of the file to transfer + HTTPS_URL - HTTPS URL of the file to transfer + SFTP_URL - SFTP URL of the file to transfer + FILE_NAME - Destination filename on the device (default: basename of URL path) + FILE_CHECKSUM_512 - Expected sha512 checksum of the file + FILE_SIZE - Expected size of the file expressed in FILE_SIZE_UNIT units + FILE_SIZE_UNIT - One of "bytes", "megabytes", or "gigabytes" (default: "bytes") +""" + +import os + +import pytest + +from pyntc.devices import EOSSSHDevice + +from ._helpers import build_file_copy_model + +# Every key the inherited EOSDevice fact properties dereference, per command. If this test +# passes on real hardware, the "| json" output is eAPI-compatible and the driver's whole +# inheritance strategy is sound. +JSON_KEY_CONTRACT = { + "show version": ["bootupTimestamp", "modelName", "internalVersion", "serialNumber"], + "show hostname": ["hostname", "fqdn"], + "show boot-config": ["softwareImage"], + "show interfaces status": ["interfaceStatuses"], + "show vlan": ["vlans"], +} + +# Per-interface keys consumed by _interfaces_status_list via INTERFACES_KM. +INTERFACE_KEYS = ["bandwidth", "duplex", "linkStatus", "description"] + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def device(): + """Connect to the lab EOS device over SSH. Skips all tests if credentials are not set.""" + host = os.environ.get("EOS_SSH_HOST") + user = os.environ.get("EOS_SSH_USER") + password = os.environ.get("EOS_SSH_PASS") + + if not all([host, user, password]): + pytest.skip("EOS_SSH_HOST / EOS_SSH_USER / EOS_SSH_PASS environment variables not set") + + dev = EOSSSHDevice( + host, + user, + password, + secret=os.environ.get("EOS_SSH_SECRET", ""), + port=os.environ.get("EOS_SSH_PORT"), + ) + yield dev + dev.close() + + +# --------------------------------------------------------------------------- +# The load-bearing assumption +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("command,keys", sorted(JSON_KEY_CONTRACT.items())) +def test_json_key_contract(device, command, keys): + """``show ... | json`` must return the eAPI-shaped document the driver depends on.""" + result = device.show(command) + assert isinstance(result, dict), f"{command} | json did not return a JSON object" + for key in keys: + assert key in result, f"{command} | json is missing key '{key}'" + + +def test_interface_status_keys(device): + """Each interface entry must carry the keys ``_interfaces_status_list`` reshapes.""" + statuses = device.show("show interfaces status")["interfaceStatuses"] + assert statuses, "device reported no interfaces" + for name, interface in statuses.items(): + for key in INTERFACE_KEYS: + assert key in interface, f"interface {name} is missing key '{key}'" + + +def test_non_show_commands_are_not_piped_to_json(device): + """A non-show command must not gain a ``| json`` suffix, which would be invalid.""" + # "show clock" proves the pipe is applied; a bare command proves it is not. + assert isinstance(device.show("show clock"), dict) + + +# --------------------------------------------------------------------------- +# Facts and config retrieval +# --------------------------------------------------------------------------- + + +def test_device_connects(device): + """Verify the device is reachable and responds to show commands.""" + assert device.hostname + assert device.os_version + + +def test_facts(device): + """Every fact property must resolve without error.""" + assert isinstance(device.uptime, int) + assert isinstance(device.boot_time, float) + assert device.model + assert isinstance(device.serial_number, str) + assert isinstance(device.interfaces, list) + assert isinstance(device.vlans, list) + assert device.boot_options["sys"] + + +def test_running_config(device): + """The running config must come back as non-empty text.""" + assert "hostname" in device.running_config + + +def test_startup_config(device): + """The startup config must come back as non-empty text.""" + assert device.startup_config.strip() + + +def test_config_round_trip(device): + """Apply a harmless config change and confirm it lands in the running config.""" + marker = "pyntc integration test" + device.config(f"banner motd {marker}\nEOF") + assert marker in device.running_config + + +def test_show_raises_on_bad_command(device): + """A bogus show command must raise CommandError, not return garbage.""" + from pyntc.errors import CommandError + + with pytest.raises(CommandError): + device.show("show definitely-not-a-command") + + +# --------------------------------------------------------------------------- +# Filesystem and transfer +# --------------------------------------------------------------------------- + + +def test_file_system_detection(device): + """The default filesystem must parse out of ``dir`` output.""" + assert device._get_file_system().endswith(":") + + +def test_free_space(device): + """Free space must parse out of ``dir`` output as a positive integer.""" + assert device._get_free_space() > 0 + + +def test_remote_file_copy_scp(device): + """Transfer the file using SCP and verify it exists on the device.""" + model = build_file_copy_model("SCP_URL") + device.remote_file_copy(model) + assert device.check_file_exists(model.file_name) + + +def test_remote_file_copy_http(device): + """Transfer the file using HTTP and verify it exists on the device.""" + model = build_file_copy_model("HTTP_URL") + device.remote_file_copy(model) + assert device.check_file_exists(model.file_name) + + +def test_remote_file_copy_ftp(device): + """Transfer the file using FTP and verify it exists on the device.""" + model = build_file_copy_model("FTP_URL") + device.remote_file_copy(model) + assert device.check_file_exists(model.file_name) + + +def test_remote_file_copy_tftp(device): + """Transfer the file using TFTP and verify it exists on the device.""" + model = build_file_copy_model("TFTP_URL") + device.remote_file_copy(model) + assert device.check_file_exists(model.file_name) + + +def test_get_remote_checksum(device): + """If the transferred file exists, its checksum must come back non-empty.""" + model = build_file_copy_model("SCP_URL") + if not device.check_file_exists(model.file_name): + pytest.skip("File does not exist on device; run a remote_file_copy test first") + checksum = device.get_remote_checksum(model.file_name, hashing_algorithm="sha512") + assert checksum + + +def test_verify_file(device): + """verify_file must confirm the transferred file against its expected checksum.""" + model = build_file_copy_model("SCP_URL") + if not device.check_file_exists(model.file_name): + pytest.skip("File does not exist on device; run a remote_file_copy test first") + assert device.verify_file(model.checksum, model.file_name, hashing_algorithm="sha512") is True diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index fe9fbede..516871a2 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -3,7 +3,7 @@ import pytest -from pyntc.devices import AIREOSDevice, ASADevice, EOSDevice, IOSDevice, IOSXEWLCDevice +from pyntc.devices import AIREOSDevice, ASADevice, EOSDevice, EOSSSHDevice, IOSDevice, IOSXEWLCDevice def get_side_effects(mock_path, side_effects): @@ -59,6 +59,65 @@ def _mock(side_effects, existing_device=None, device=eos_device): return _mock +# EOS SSH fixtures + + +@pytest.fixture +def eos_ssh_device(): + with mock.patch("pyntc.devices.eos_ssh_device.ConnectHandler") as ch: + device = EOSSSHDevice("host", "user", "password") + device.native = ch + # Model the normal steady state: already privileged, not parked in config mode. + # Without this the inherited enable() would call exit_config_mode() on every + # show()/config(), polluting call-count assertions. + device.native.check_enable_mode.return_value = True + device.native.check_config_mode.return_value = False + yield device + + +@pytest.fixture +def eos_ssh_device_path(): + return "pyntc.devices.eos_ssh_device.EOSSSHDevice" + + +@pytest.fixture +def eos_ssh_mock_path(mock_path): + return f"{mock_path}/eos_ssh" + + +@pytest.fixture +def eos_ssh_send_command(eos_ssh_device, eos_ssh_mock_path): + def _mock(side_effects, existing_device=None, device=eos_ssh_device): + if existing_device is not None: + device = existing_device + device.native.send_command.side_effect = get_side_effects(eos_ssh_mock_path, side_effects) + return device + + return _mock + + +@pytest.fixture +def eos_ssh_send_command_timing(eos_ssh_device, eos_ssh_mock_path): + def _mock(side_effects, existing_device=None, device=eos_ssh_device): + if existing_device is not None: + device = existing_device + device.native.send_command_timing.side_effect = get_side_effects(eos_ssh_mock_path, side_effects) + return device + + return _mock + + +@pytest.fixture +def eos_ssh_config(eos_ssh_device, eos_ssh_mock_path): + def _mock(side_effects, existing_device=None, device=eos_ssh_device): + if existing_device is not None: + device = existing_device + device.native.send_config_set.side_effect = get_side_effects(eos_ssh_mock_path, side_effects) + return device + + return _mock + + @pytest.fixture def aireos_boot_image(): return "8.2.170.0" diff --git a/tests/unit/test_devices/device_mocks/eos_ssh/README.md b/tests/unit/test_devices/device_mocks/eos_ssh/README.md new file mode 100644 index 00000000..76cb5839 --- /dev/null +++ b/tests/unit/test_devices/device_mocks/eos_ssh/README.md @@ -0,0 +1,41 @@ +# `arista_eos_ssh` mock fixtures + +Golden CLI output for `EOSSSHDevice` unit tests. Loaded by `get_side_effects()` in `tests/unit/conftest.py` — any string in a side-effect list naming a file here is replaced by that file's contents. + +## Provenance + +Captured from a real **Arista DCS-7050TX-64-R running EOS 4.28.5M**, over SSH, via `show | json`. This capture is what confirmed the driver's central design assumption: the `| json` pipe returns the same document, with the same key names, that eAPI returns — without eAPI being involved. + +| Fixture | Source | +| --- | --- | +| `show_version_json` | real capture, `show version \| json` | +| `show_hostname_json` | real capture, `show hostname \| json` | +| `show_interfaces_status_json` | real capture, `show interfaces status \| json` (65 interfaces) | +| `show_boot-config_json` | real capture, `show boot-config \| json` | +| `show_vlan_json` | real capture, `show vlan \| json` | +| `dir` | real capture, `dir` | +| `show_boot` | real capture, `show boot` | +| `show_running-config` | **not** from hardware — the repo's sanitised vEOS config (see below) | +| `show_startup-config` | **not** from hardware — the repo's sanitised vEOS config (see below) | + +### Sanitisation + +Three values were replaced in `show_version_json`; everything else is verbatim: + +- `serialNumber` → `JPE00000000` +- `systemMacAddress` / `hwMacAddress` → `00:1c:73:00:00:01` + +And in `show_interfaces_status_json`, the `Ethernet1` description was replaced with `lab uplink` (it named a client). + +### Why the configs are not real captures + +`running_config` and `startup_config` are returned verbatim by the driver and never parsed, so a real capture would validate nothing — while writing device hostnames, SNMP communities and password hashes into a committed test tree. Those two files are the repo's existing sanitised vEOS config, kept only so the tests have non-empty text to work with. That is why they say `eos-spine1` while every other fixture says `nyc-eos-01`. + +The one genuine risk those commands carry is that a config line beginning with `% ` (inside a banner, say) would false-positive the driver's CLI error regex and make `running_config` raise on a healthy device. That was checked directly on hardware with `show running-config | include ^%` — no matches. `design-notes/capture_eos_ssh_fixtures.py` re-runs that scan and reports only a line count, never content. + +## Edge cases this capture pins down + +Both were absent from the older vEOS fixtures and would have gone untested: + +- **`softwareImage` carries a `flash:/` prefix** (`flash:/EOS-4.28.5M.swi`). `boot_options` strips it with `.replace("flash:/", "")`; the vEOS fixture had no prefix, so that line had never been exercised against realistic input. +- **`Management1` is routed and has no `vlanId`** inside `vlanInformation`. The interface key map must resolve that to `None` rather than raising. diff --git a/tests/unit/test_devices/device_mocks/eos_ssh/dir b/tests/unit/test_devices/device_mocks/eos_ssh/dir new file mode 100644 index 00000000..fc7b0c64 --- /dev/null +++ b/tests/unit/test_devices/device_mocks/eos_ssh/dir @@ -0,0 +1,18 @@ +Directory of flash:/ + + -rwx 1386 Aug 5 20:51 AsuFastPktTransmit.log + -rwx 764915173 Aug 5 20:49 EOS-4.28.5M.swi + -rwx 764834205 Aug 5 17:46 EOS-4.28.9M.swi + drwx 4096 Aug 3 21:32 Fossil + -rwx 852 Aug 5 20:51 SsuRestore.log + -rwx 852 Aug 5 20:51 SsuRestoreLegacy.log + -rwx 27 Aug 5 20:49 boot-config + drwx 4096 Aug 8 19:25 debug + drwx 4096 Aug 3 21:32 fastpkttx.backup + -rwx 94038 Apr 7 15:42 nautobot.png + drwx 4096 Aug 8 19:24 persist + drwx 4096 Aug 12 2024 schedule + -rwx 3610 Aug 5 20:48 startup-config + -rwx 0 Dec 2 2024 zerotouch-config + +3634421760 bytes total (1327603712 bytes free) diff --git a/tests/unit/test_devices/device_mocks/eos_ssh/show_boot b/tests/unit/test_devices/device_mocks/eos_ssh/show_boot new file mode 100644 index 00000000..ed29ea9b --- /dev/null +++ b/tests/unit/test_devices/device_mocks/eos_ssh/show_boot @@ -0,0 +1,6 @@ +Software image: flash:/EOS-4.28.5M.swi +Console speed: (not set) +Aboot password (encrypted): (not set) +Memory test iterations: (not set) +Checksum: 0a2c4f1390395f61a042f46c3fe19b86152fc2ca +Checksum algorithm: sha1 diff --git a/tests/unit/test_devices/device_mocks/eos_ssh/show_boot-config_json b/tests/unit/test_devices/device_mocks/eos_ssh/show_boot-config_json new file mode 100644 index 00000000..d92ff9c9 --- /dev/null +++ b/tests/unit/test_devices/device_mocks/eos_ssh/show_boot-config_json @@ -0,0 +1,18 @@ +{ + "upgradeCert": "", + "userCert": "", + "securebootSupported": false, + "tpmPassword": false, + "fileChecksum": "0a2c4f1390395f61a042f46c3fe19b86152fc2ca", + "aristaCertEnabled": false, + "spiUpdateEnabled": false, + "abootPassword": "(not set)", + "memTestIterations": 0, + "softwareImage": "flash:/EOS-4.28.5M.swi", + "aristaCert": "", + "fileChecksumAlg": "sha1", + "securebootEnabled": false, + "measuredbootEnabled": false, + "certsLoaded": false, + "spiFlashWriteProtected": false +} diff --git a/tests/unit/test_devices/device_mocks/eos_ssh/show_hostname_json b/tests/unit/test_devices/device_mocks/eos_ssh/show_hostname_json new file mode 100644 index 00000000..dec90d32 --- /dev/null +++ b/tests/unit/test_devices/device_mocks/eos_ssh/show_hostname_json @@ -0,0 +1,4 @@ +{ + "fqdn": "nyc-eos-01", + "hostname": "nyc-eos-01" +} diff --git a/tests/unit/test_devices/device_mocks/eos_ssh/show_interfaces_status_json b/tests/unit/test_devices/device_mocks/eos_ssh/show_interfaces_status_json new file mode 100644 index 00000000..21944bb5 --- /dev/null +++ b/tests/unit/test_devices/device_mocks/eos_ssh/show_interfaces_status_json @@ -0,0 +1,978 @@ +{ + "interfaceStatuses": { + "Ethernet1": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 10, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "lab uplink", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet2": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 10, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet3": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 10, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet4": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 10, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet5": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 10, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet6": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 11, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet7": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 11, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet8": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 11, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet9": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet10": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 12, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet11": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 12, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet12": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 12, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet13": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 12, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet14": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 12, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet15": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 12, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet16": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet17": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet18": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet19": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet20": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet21": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet22": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet23": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet24": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet25": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet26": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet27": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet28": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet29": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet30": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet31": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet32": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet33": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet34": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet35": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet36": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet37": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet38": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet39": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet40": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet41": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet42": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet43": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet44": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet45": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet46": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet47": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet48": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 0, + "interfaceType": "10GBASE-T", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexUnknown", + "autoNegotigateActive": true, + "linkStatus": "notconnect", + "lineProtocolStatus": "down" + }, + "Ethernet49/1": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 10000000000, + "interfaceType": "Not Present", + "description": "", + "autoNegotiateActive": false, + "duplex": "duplexFull", + "autoNegotigateActive": false, + "linkStatus": "notconnect", + "lineProtocolStatus": "notPresent" + }, + "Ethernet49/2": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 10000000000, + "interfaceType": "Not Present", + "description": "", + "autoNegotiateActive": false, + "duplex": "duplexFull", + "autoNegotigateActive": false, + "linkStatus": "notconnect", + "lineProtocolStatus": "notPresent" + }, + "Ethernet49/3": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 10000000000, + "interfaceType": "Not Present", + "description": "", + "autoNegotiateActive": false, + "duplex": "duplexFull", + "autoNegotigateActive": false, + "linkStatus": "notconnect", + "lineProtocolStatus": "notPresent" + }, + "Ethernet49/4": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 10000000000, + "interfaceType": "Not Present", + "description": "", + "autoNegotiateActive": false, + "duplex": "duplexFull", + "autoNegotigateActive": false, + "linkStatus": "notconnect", + "lineProtocolStatus": "notPresent" + }, + "Ethernet50/1": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 10000000000, + "interfaceType": "Not Present", + "description": "", + "autoNegotiateActive": false, + "duplex": "duplexFull", + "autoNegotigateActive": false, + "linkStatus": "notconnect", + "lineProtocolStatus": "notPresent" + }, + "Ethernet50/2": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 10000000000, + "interfaceType": "Not Present", + "description": "", + "autoNegotiateActive": false, + "duplex": "duplexFull", + "autoNegotigateActive": false, + "linkStatus": "notconnect", + "lineProtocolStatus": "notPresent" + }, + "Ethernet50/3": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 10000000000, + "interfaceType": "Not Present", + "description": "", + "autoNegotiateActive": false, + "duplex": "duplexFull", + "autoNegotigateActive": false, + "linkStatus": "notconnect", + "lineProtocolStatus": "notPresent" + }, + "Ethernet50/4": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 10000000000, + "interfaceType": "Not Present", + "description": "", + "autoNegotiateActive": false, + "duplex": "duplexFull", + "autoNegotigateActive": false, + "linkStatus": "notconnect", + "lineProtocolStatus": "notPresent" + }, + "Ethernet51/1": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 10000000000, + "interfaceType": "Not Present", + "description": "", + "autoNegotiateActive": false, + "duplex": "duplexFull", + "autoNegotigateActive": false, + "linkStatus": "notconnect", + "lineProtocolStatus": "notPresent" + }, + "Ethernet51/2": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 10000000000, + "interfaceType": "Not Present", + "description": "", + "autoNegotiateActive": false, + "duplex": "duplexFull", + "autoNegotigateActive": false, + "linkStatus": "notconnect", + "lineProtocolStatus": "notPresent" + }, + "Ethernet51/3": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 10000000000, + "interfaceType": "Not Present", + "description": "", + "autoNegotiateActive": false, + "duplex": "duplexFull", + "autoNegotigateActive": false, + "linkStatus": "notconnect", + "lineProtocolStatus": "notPresent" + }, + "Ethernet51/4": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 10000000000, + "interfaceType": "Not Present", + "description": "", + "autoNegotiateActive": false, + "duplex": "duplexFull", + "autoNegotigateActive": false, + "linkStatus": "notconnect", + "lineProtocolStatus": "notPresent" + }, + "Ethernet52/1": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 10000000000, + "interfaceType": "Not Present", + "description": "", + "autoNegotiateActive": false, + "duplex": "duplexFull", + "autoNegotigateActive": false, + "linkStatus": "notconnect", + "lineProtocolStatus": "notPresent" + }, + "Ethernet52/2": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 10000000000, + "interfaceType": "Not Present", + "description": "", + "autoNegotiateActive": false, + "duplex": "duplexFull", + "autoNegotigateActive": false, + "linkStatus": "notconnect", + "lineProtocolStatus": "notPresent" + }, + "Ethernet52/3": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 10000000000, + "interfaceType": "Not Present", + "description": "", + "autoNegotiateActive": false, + "duplex": "duplexFull", + "autoNegotigateActive": false, + "linkStatus": "notconnect", + "lineProtocolStatus": "notPresent" + }, + "Ethernet52/4": { + "vlanInformation": { + "interfaceMode": "bridged", + "vlanId": 1, + "interfaceForwardingModel": "bridged" + }, + "bandwidth": 10000000000, + "interfaceType": "Not Present", + "description": "", + "autoNegotiateActive": false, + "duplex": "duplexFull", + "autoNegotigateActive": false, + "linkStatus": "notconnect", + "lineProtocolStatus": "notPresent" + }, + "Management1": { + "vlanInformation": { + "interfaceMode": "routed", + "interfaceForwardingModel": "routed" + }, + "bandwidth": 100000000, + "interfaceType": "10/100/1000", + "description": "", + "autoNegotiateActive": true, + "duplex": "duplexFull", + "autoNegotigateActive": true, + "linkStatus": "connected", + "lineProtocolStatus": "up" + } + } +} diff --git a/tests/unit/test_devices/device_mocks/eos_ssh/show_running-config b/tests/unit/test_devices/device_mocks/eos_ssh/show_running-config new file mode 100644 index 00000000..f983a02b --- /dev/null +++ b/tests/unit/test_devices/device_mocks/eos_ssh/show_running-config @@ -0,0 +1,48 @@ +! Command: show running-config +! device: eos-spine1 (vEOS, EOS-4.14.7M) +! +! boot system flash:/new_image.swi +! +transceiver qsfp default-mode 4x10G +! +hostname eos-spine1 +ip domain-name ntc.com +! +snmp-server community public ro +! +spanning-tree mode mstp +! +no aaa root +! +username admin privilege 15 role network-admin secret 5 $1$7yUmRiH6$9F1Io4WMwAWSc2GMjeK3h/ +username ntc privilege 15 role network-admin secret 5 $1$yLXNmzh4$eltlOr6yIb8IpRGCjp8Bj/ +! +interface Ethernet1 +! +interface Ethernet2 +! +interface Ethernet3 +! +interface Ethernet4 +! +interface Ethernet5 +! +interface Ethernet6 +! +interface Ethernet7 +! +interface Ethernet8 +! +interface Management1 + ip address 10.0.0.11/24 +! +ip route 0.0.0.0/0 10.0.0.2 +! +ip routing +! +management api http-commands + protocol http + no shutdown +! +! +end diff --git a/tests/unit/test_devices/device_mocks/eos_ssh/show_startup-config b/tests/unit/test_devices/device_mocks/eos_ssh/show_startup-config new file mode 100644 index 00000000..bc65cb1d --- /dev/null +++ b/tests/unit/test_devices/device_mocks/eos_ssh/show_startup-config @@ -0,0 +1,49 @@ +! Command: show startup-config +! Startup-config last modified at Sat Jan 23 16:50:06 2016 by ntc +! device: eos-spine1 (vEOS, EOS-4.14.7M) +! +! boot system flash:EOS.swi +! +transceiver qsfp default-mode 4x10G +! +hostname eos-spine1 +ip domain-name ntc.com +! +snmp-server community public ro +! +spanning-tree mode mstp +! +no aaa root +! +username admin privilege 15 role network-admin secret 5 $1$7yUmRiH6$9F1Io4WMwAWSc2GMjeK3h/ +username ntc privilege 15 role network-admin secret 5 $1$yLXNmzh4$eltlOr6yIb8IpRGCjp8Bj/ +! +interface Ethernet1 +! +interface Ethernet2 +! +interface Ethernet3 +! +interface Ethernet4 +! +interface Ethernet5 +! +interface Ethernet6 +! +interface Ethernet7 +! +interface Ethernet8 +! +interface Management1 + ip address 10.0.0.11/24 +! +ip route 0.0.0.0/0 10.0.0.2 +! +ip routing +! +management api http-commands + protocol http + no shutdown +! +! +end diff --git a/tests/unit/test_devices/device_mocks/eos_ssh/show_version_json b/tests/unit/test_devices/device_mocks/eos_ssh/show_version_json new file mode 100644 index 00000000..a6a9abcf --- /dev/null +++ b/tests/unit/test_devices/device_mocks/eos_ssh/show_version_json @@ -0,0 +1,20 @@ +{ + "imageFormatVersion": "3.0", + "uptime": 254314.71, + "modelName": "DCS-7050TX-64-R", + "internalVersion": "4.28.5M-29792660.4285M", + "memTotal": 3982512, + "mfgName": "Arista", + "serialNumber": "JPE00000000", + "systemMacAddress": "00:1c:73:00:00:01", + "bootupTimestamp": 1785963023.376446, + "memFree": 2485644, + "version": "4.28.5M", + "configMacAddress": "00:00:00:00:00:00", + "isIntlVersion": false, + "imageOptimization": "Strata-4GB", + "internalBuildId": "d9aad8b6-4e46-4507-8815-ada0b879f38a", + "hardwareRevision": "01.01", + "hwMacAddress": "00:1c:73:00:00:01", + "architecture": "i686" +} diff --git a/tests/unit/test_devices/device_mocks/eos_ssh/show_vlan_json b/tests/unit/test_devices/device_mocks/eos_ssh/show_vlan_json new file mode 100644 index 00000000..5f564907 --- /dev/null +++ b/tests/unit/test_devices/device_mocks/eos_ssh/show_vlan_json @@ -0,0 +1,35 @@ +{ + "sourceDetail": "", + "vlans": { + "1": { + "status": "active", + "name": "default", + "interfaces": {}, + "dynamic": false + }, + "11": { + "status": "active", + "name": "HR", + "interfaces": {}, + "dynamic": false + }, + "12": { + "status": "active", + "name": "FIN", + "interfaces": {}, + "dynamic": false + }, + "10": { + "status": "active", + "name": "IT_DEP", + "interfaces": {}, + "dynamic": false + }, + "9": { + "status": "active", + "name": "AP", + "interfaces": {}, + "dynamic": false + } + } +} diff --git a/tests/unit/test_devices/test_eos_ssh_device.py b/tests/unit/test_devices/test_eos_ssh_device.py new file mode 100644 index 00000000..a54c744f --- /dev/null +++ b/tests/unit/test_devices/test_eos_ssh_device.py @@ -0,0 +1,962 @@ +"""Unit tests for the ``arista_eos_ssh`` driver. + +Fixtures are a real ``show ... | json`` capture from a DCS-7050TX-64-R running EOS 4.28.5M +(serial, MACs and one port description sanitised). Real hardware output covers two shapes +the older vEOS fixtures did not: ``softwareImage`` carrying a ``flash:/`` prefix, and a +routed ``Management1`` whose ``vlanInformation`` has no ``vlanId`` at all. + +Because the two drivers no longer share fixture data, +``test_facts_match_eapi_driver_for_identical_payload`` is what guards against drift: it +feeds the same document to both drivers and asserts every fact comes out identical. +""" + +import hashlib +import inspect +import json +import os +import time +from unittest import mock + +import pytest + +from pyntc import ntc_device +from pyntc.devices import EOSDevice, EOSSSHDevice +from pyntc.devices.base_device import RollbackError +from pyntc.devices.eos_device import DEFAULT_REBOOT_TIMEOUT +from pyntc.devices.eos_ssh_device import DEFAULT_READ_TIMEOUT +from pyntc.devices.eos_ssh_device import EOSSSHDevice as Driver +from pyntc.errors import ( + CommandError, + CommandListError, + FileTransferError, + NotEnoughFreeSpaceError, + OSInstallError, + SocketClosedError, +) +from pyntc.utils.models import FileCopyModel + +BOOT_TIMESTAMP = 1785963023.376446 +MODEL = "DCS-7050TX-64-R" +OS_VERSION = "4.28.5M-29792660.4285M" +HOSTNAME = "nyc-eos-01" +SERIAL_NUMBER = "JPE00000000" +BOOT_IMAGE = "EOS-4.28.5M.swi" +FREE_BYTES = 1327603712 +INTERFACE_COUNT = 65 + + +# --------------------------------------------------------------------------- +# Registration +# --------------------------------------------------------------------------- + + +def test_registered_device_type(): + with mock.patch.object(EOSSSHDevice, "open"): + device = ntc_device("arista_eos_ssh", "host", "user", "password") + assert isinstance(device, EOSSSHDevice) + assert device.device_type == "arista_eos_ssh" + + +def test_vendor(eos_ssh_device): + assert eos_ssh_device.vendor == "arista" + + +# --------------------------------------------------------------------------- +# Connection handling +# --------------------------------------------------------------------------- + + +def test_init_defaults(): + with mock.patch("pyntc.devices.eos_ssh_device.ConnectHandler"): + device = EOSSSHDevice("host", "user", "password") + assert device.port == 22 + assert device.secret == "" + assert device.device_type == "arista_eos_ssh" + + +def test_init_accepts_port_and_secret(): + with mock.patch("pyntc.devices.eos_ssh_device.ConnectHandler"): + device = EOSSSHDevice("host", "user", "password", secret="enable_me", port="2222") + assert device.port == 2222 + assert device.secret == "enable_me" + + +def test_init_does_not_build_an_eapi_connection(): + # EOSDevice.__init__ would call pyeapi.connect(); the SSH driver must not. + with mock.patch("pyntc.devices.eos_ssh_device.ConnectHandler"): + with mock.patch("pyntc.devices.eos_device.eos_connect") as mock_connect: + EOSSSHDevice("host", "user", "password") + mock_connect.assert_not_called() + + +def test_open_connects_with_arista_eos_driver(): + with mock.patch("pyntc.devices.eos_ssh_device.ConnectHandler") as ch: + EOSSSHDevice("host", "user", "password", port=2222, secret="s3cret") + _, kwargs = ch.call_args + assert kwargs["device_type"] == "arista_eos" + assert kwargs["host"] == "host" + assert kwargs["port"] == 2222 + assert kwargs["secret"] == "s3cret" + + +def test_open_passes_extra_netmiko_kwargs(): + with mock.patch("pyntc.devices.eos_ssh_device.ConnectHandler") as ch: + EOSSSHDevice("host", "user", "password", global_delay_factor=2) + assert ch.call_args[1]["global_delay_factor"] == 2 + + +def test_open_is_noop_when_already_connected(eos_ssh_device): + with mock.patch("pyntc.devices.eos_ssh_device.ConnectHandler") as ch: + eos_ssh_device.open() + ch.assert_not_called() + eos_ssh_device.native.find_prompt.assert_called() + + +def test_open_reconnects_when_session_is_dead(): + with mock.patch("pyntc.devices.eos_ssh_device.ConnectHandler") as ch: + device = EOSSSHDevice("host", "user", "password") + assert ch.call_count == 1 + device.native.find_prompt.side_effect = OSError("socket closed") + device.open() + assert ch.call_count == 2 + assert device._connected is True + + +def test_native_ssh_is_native(eos_ssh_device): + # Inherited file-transfer code reaches for native_ssh; it must be the Netmiko handler. + assert eos_ssh_device.native_ssh is eos_ssh_device.native + + +def test_close_disconnects(eos_ssh_device): + eos_ssh_device.close() + eos_ssh_device.native.disconnect.assert_called_once() + assert eos_ssh_device._connected is False + + +def test_close_is_idempotent(eos_ssh_device): + eos_ssh_device.close() + eos_ssh_device.close() + eos_ssh_device.native.disconnect.assert_called_once() + + +# --------------------------------------------------------------------------- +# show() +# --------------------------------------------------------------------------- + + +def test_show_single_command_returns_dict(eos_ssh_send_command): + device = eos_ssh_send_command(["show_version_json"]) + result = device.show("show version") + assert isinstance(result, dict) + assert result["modelName"] == MODEL + device.native.send_command.assert_called_with("show version | json", read_timeout=DEFAULT_READ_TIMEOUT) + + +def test_show_list_returns_list(eos_ssh_send_command): + device = eos_ssh_send_command(["show_version_json", "show_hostname_json"]) + results = device.show(["show version", "show hostname"]) + assert isinstance(results, list) + assert len(results) == 2 + assert results[0]["modelName"] == MODEL + assert results[1]["hostname"] == HOSTNAME + + +def test_show_raw_text_returns_str_without_json_pipe(eos_ssh_send_command): + device = eos_ssh_send_command(["dir"]) + result = device.show("dir", raw_text=True) + assert isinstance(result, str) + assert "bytes free" in result + device.native.send_command.assert_called_with("dir", read_timeout=DEFAULT_READ_TIMEOUT) + + +def test_show_raw_text_list(eos_ssh_send_command): + device = eos_ssh_send_command(["dir", "show_boot"]) + results = device.show(["dir", "show boot"], raw_text=True) + assert [isinstance(item, str) for item in results] == [True, True] + + +@pytest.mark.parametrize( + "command", + [ + "reload now", + "copy running-config startup-config", + "configure replace flash:cp force", + "install source flash:EOS.swi", + ], +) +def test_show_does_not_pipe_non_show_commands_to_json(eos_ssh_send_command, command): + # "reload now | json" is not a valid command. Non-show commands must go through bare, + # and every inherited caller discards the return value, so {} preserves the contract. + device = eos_ssh_send_command([""]) + result = device.show(command) + assert result == {} + assert device.native.send_command.call_args[0][0] == command + + +def test_show_raises_command_error(eos_ssh_send_command): + device = eos_ssh_send_command(["% Invalid input (at token 1: 'bogus')"]) + with pytest.raises(CommandError) as err: + device.show("show bogus") + # The caller's command, not the wire command -- errors must not leak the "| json" + # suffix, matching the plain command name pyeapi reports on EOSDevice. + assert err.value.command == "show bogus" + assert "Invalid input" in err.value.cli_error_msg + + +def test_show_list_raises_command_list_error(eos_ssh_send_command): + device = eos_ssh_send_command(["show_version_json", "% Invalid input"]) + with pytest.raises(CommandListError) as err: + device.show(["show version", "show bogus"]) + assert err.value.commands == ["show version", "show bogus"] + assert err.value.command == "show bogus" + + +def test_show_raises_when_output_is_not_json(eos_ssh_send_command): + # A command with no JSON renderer must raise, never silently fall back to text + # parsing -- a fallback returns a differently shaped document and yields wrong facts. + device = eos_ssh_send_command(["This command is not converted to JSON"]) + with pytest.raises(CommandError) as err: + device.show("show something-unconverted") + assert "does not support JSON output" in err.value.cli_error_msg + + +def test_show_reopens_connection(eos_ssh_send_command): + # Guards reboot polling: _wait_for_device_reboot survives only because show() re-opens. + device = eos_ssh_send_command(["show_version_json"]) + with mock.patch.object(Driver, "open") as mock_open: + device.show("show version") + mock_open.assert_called_once() + + +@pytest.mark.parametrize( + "command,expected_timeout", + [ + ("install source flash:EOS.swi", 1800), + ("copy running-config startup-config", 300), + ("configure replace flash:cp force", 300), + ("show running-config", 120), + ("show startup-config", 120), + ("show version", DEFAULT_READ_TIMEOUT), + ], +) +def test_show_resolves_read_timeout_per_command(eos_ssh_send_command, command, expected_timeout): + # Timeouts are derived from the command text so show()'s signature can stay identical + # to EOSDevice.show() while inherited long-running callers still work. + device = eos_ssh_send_command(['{"x": 1}']) + device.show(command, raw_text=True) + assert device.native.send_command.call_args[1]["read_timeout"] == expected_timeout + + +# --------------------------------------------------------------------------- +# config() +# --------------------------------------------------------------------------- + + +def test_config_single_command_returns_none(eos_ssh_config): + device = eos_ssh_config([""]) + assert device.config("interface Ethernet1") is None + device.native.send_config_set.assert_called_with("interface Ethernet1", exit_config_mode=False) + + +def test_config_list_returns_none(eos_ssh_config): + device = eos_ssh_config(["", ""]) + assert device.config(["interface Ethernet1", "no shutdown"]) is None + assert device.native.send_config_set.call_count == 2 + + +def test_config_exits_config_mode(eos_ssh_config): + device = eos_ssh_config([""]) + device.config("interface Ethernet1") + device.native.exit_config_mode.assert_called_once() + + +def test_config_raises_command_error(eos_ssh_config): + device = eos_ssh_config(["% Invalid input"]) + with pytest.raises(CommandError) as err: + device.config("bogus command") + assert err.value.command == "bogus command" + + +def test_config_list_raises_command_list_error(eos_ssh_config): + device = eos_ssh_config(["", "% Invalid input"]) + with pytest.raises(CommandListError) as err: + device.config(["interface Ethernet1", "bogus"]) + assert err.value.command == "bogus" + assert err.value.commands == ["interface Ethernet1", "bogus"] + + +def test_config_exits_config_mode_on_error(eos_ssh_config): + # A failed command must not leave the session parked in config mode. + device = eos_ssh_config(["% Invalid input"]) + with pytest.raises(CommandError): + device.config("bogus command") + device.native.exit_config_mode.assert_called_once() + + +# --------------------------------------------------------------------------- +# Fact properties -- expectations match test_eos_device.py exactly +# --------------------------------------------------------------------------- + + +def test_hostname(eos_ssh_send_command): + device = eos_ssh_send_command(["show_hostname_json"]) + assert device.hostname == HOSTNAME + + +def test_fqdn(eos_ssh_send_command): + device = eos_ssh_send_command(["show_hostname_json"]) + assert device.fqdn == HOSTNAME + + +def test_model(eos_ssh_send_command): + device = eos_ssh_send_command(["show_version_json"]) + assert device.model == MODEL + + +def test_os_version(eos_ssh_send_command): + device = eos_ssh_send_command(["show_version_json"]) + assert device.os_version == OS_VERSION + + +def test_serial_number(eos_ssh_send_command): + device = eos_ssh_send_command(["show_version_json"]) + assert device.serial_number == SERIAL_NUMBER + + +def test_boot_time(eos_ssh_send_command): + device = eos_ssh_send_command(["show_version_json"]) + boot_time = device.boot_time + assert isinstance(boot_time, float) + assert boot_time == BOOT_TIMESTAMP + + +def test_uptime(eos_ssh_send_command): + device = eos_ssh_send_command(["show_version_json"]) + uptime = device.uptime + assert isinstance(uptime, int) + assert uptime == pytest.approx(int(time.time() - BOOT_TIMESTAMP), abs=2) + + +def test_uptime_string(eos_ssh_send_command): + device = eos_ssh_send_command(["show_version_json"]) + with mock.patch.object(Driver, "_uptime_to_string", return_value="02:00:03:38"): + assert device.uptime_string == "02:00:03:38" + + +def test_interfaces(eos_ssh_send_command): + device = eos_ssh_send_command(["show_interfaces_status_json"]) + interfaces = device.interfaces + assert len(interfaces) == INTERFACE_COUNT + # Sorted lexicographically, matching EOSDevice: "Ethernet10" precedes "Ethernet2". + assert interfaces == sorted(interfaces) + assert interfaces[:3] == ["Ethernet1", "Ethernet10", "Ethernet11"] + assert interfaces[-1] == "Management1" + assert "Ethernet49/1" in interfaces + + +def test_routed_interface_without_vlan_id(eos_ssh_send_command): + # Management1 is routed: its vlanInformation carries no vlanId. The key map must + # resolve that to None rather than raising. + device = eos_ssh_send_command(["show_interfaces_status_json"]) + management = [i for i in device._interfaces_status_list() if i["interface"] == "Management1"][0] + assert management["vlan"] is None + assert management["state"] == "connected" + + +def test_vlans(eos_ssh_send_command): + device = eos_ssh_send_command(["show_vlan_json"]) + # Lexicographic ordering, matching EOSVlans.get_list() -- so "9" sorts last. + assert device.vlans == ["1", "10", "11", "12", "9"] + + +def test_vlans_uses_show_vlan_not_pyeapi_api(eos_ssh_send_command): + # EOSVlans reaches for device.native.api("vlans"), which does not exist over SSH. + device = eos_ssh_send_command(["show_vlan_json"]) + device.vlans # noqa: B018 + assert device.native.send_command.call_args[0][0] == "show vlan | json" + device.native.api.assert_not_called() + + +def test_boot_options_strips_flash_prefix(eos_ssh_send_command): + # Real hardware returns softwareImage as "flash:/EOS-4.28.5M.swi"; the vEOS fixture had + # no prefix, so boot_options' .replace("flash:/", "") was previously untested. + device = eos_ssh_send_command(["show_boot-config_json"]) + assert device.boot_options == {"sys": BOOT_IMAGE} + + +def test_running_config(eos_ssh_send_command): + device = eos_ssh_send_command(["show_running-config"]) + running_config = device.running_config + assert isinstance(running_config, str) + assert "hostname eos-spine1" in running_config + device.native.send_command.assert_called_with("show running-config", read_timeout=120) + + +def test_startup_config(eos_ssh_send_command): + device = eos_ssh_send_command(["show_startup-config"]) + assert "hostname eos-spine1" in device.startup_config + + +def test_facts_are_cached(eos_ssh_send_command): + device = eos_ssh_send_command(["show_hostname_json"]) + assert device.hostname == HOSTNAME + assert device.hostname == HOSTNAME + # A single device round-trip: the second read comes from the cache. + assert device.native.send_command.call_count == 1 + + +def test_backup_running_config(eos_ssh_send_command, tmp_path): + # Inherited backup_running_config reads running_config twice (once to write, once to + # log) and running_config is not cached, so two round-trips are expected. + device = eos_ssh_send_command(["show_running-config", "show_running-config"]) + target = tmp_path / "backup.cfg" + device.backup_running_config(str(target)) + assert "hostname eos-spine1" in target.read_text() + + +# --------------------------------------------------------------------------- +# Filesystem helpers (inherited, exercised over SSH) +# --------------------------------------------------------------------------- + + +def test_get_file_system(eos_ssh_send_command): + device = eos_ssh_send_command(["dir"]) + assert device._get_file_system() == "flash:" + + +def test_get_free_space(eos_ssh_send_command): + device = eos_ssh_send_command(["dir"]) + assert device._get_free_space() == FREE_BYTES + + +def test_get_free_space_raises_when_unparseable(eos_ssh_send_command): + device = eos_ssh_send_command(["nothing useful here"]) + with pytest.raises(CommandError): + device._get_free_space() + + +def test_check_free_space_raises_when_insufficient(eos_ssh_send_command): + from pyntc.errors import NotEnoughFreeSpaceError + + device = eos_ssh_send_command(["dir"]) + with pytest.raises(NotEnoughFreeSpaceError): + device._check_free_space(99_999_999_999, file_system="flash:") + + +def test_image_booted(eos_ssh_send_command): + device = eos_ssh_send_command(["show_boot", "show_boot"]) + assert device._image_booted(BOOT_IMAGE) is True + # The other image present on flash is not the booted one. + assert device._image_booted("EOS-4.28.9M.swi") is False + + +# --------------------------------------------------------------------------- +# Inherited file operations (these already used Netmiko on the eAPI driver) +# --------------------------------------------------------------------------- + + +def test_check_file_exists_true(eos_ssh_send_command): + device = eos_ssh_send_command(["dir", "Directory of flash:/EOS.swi\n\n-rwx 1234 EOS.swi\n"]) + assert device.check_file_exists("EOS.swi") is True + + +def test_check_file_exists_false(eos_ssh_send_command): + device = eos_ssh_send_command(["dir", "% Error listing directory"]) + assert device.check_file_exists("missing.swi") is False + + +def test_check_file_exists_raises_on_unknown_output(eos_ssh_send_command): + device = eos_ssh_send_command(["dir", "something unexpected"]) + with pytest.raises(CommandError): + device.check_file_exists("EOS.swi") + + +def test_get_remote_checksum(eos_ssh_send_command): + device = eos_ssh_send_command(["dir", "verify /sha512 (flash:EOS.swi) = abc123"]) + assert device.get_remote_checksum("EOS.swi", hashing_algorithm="sha512") == "abc123" + + +def test_get_remote_checksum_rejects_unsupported_algorithm(eos_ssh_device): + with pytest.raises(ValueError, match="Unsupported hashing algorithm"): + eos_ssh_device.get_remote_checksum("EOS.swi", hashing_algorithm="blake3") + + +def test_verify_file_matches(eos_ssh_send_command): + device = eos_ssh_send_command( + ["dir", "Directory of flash:/EOS.swi\n", "dir", "verify /md5 (flash:EOS.swi) = ABC123"] + ) + assert device.verify_file("abc123", "EOS.swi") is True + + +def test_verify_file_missing_file(eos_ssh_send_command): + device = eos_ssh_send_command(["dir", "No such file"]) + assert device.verify_file("abc123", "EOS.swi") is False + + +# --------------------------------------------------------------------------- +# Pushing code onto the box: file_copy (local -> device, over SCP) +# --------------------------------------------------------------------------- + + +def test_file_copy_instance_uses_the_netmiko_session(eos_ssh_device): + # The whole reason native_ssh is aliased: inherited FileTransfer code must receive the + # SSH driver's own Netmiko handler, not a separate session. + with mock.patch("pyntc.devices.eos_device.FileTransfer") as file_transfer: + eos_ssh_device._file_copy_instance("/local/EOS.swi", "EOS.swi", file_system="flash:") + args, kwargs = file_transfer.call_args + assert args[0] is eos_ssh_device.native + # "flash:" is the CLI name; SCP addresses the same filesystem by its Linux path. + assert kwargs["file_system"] == "/mnt/flash" + + +@pytest.fixture +def local_image(tmp_path): + """A small local file plus its md5, standing in for an image to upload.""" + source = tmp_path / "EOS.swi" + source.write_bytes(b"x" * 32) + return source, hashlib.md5(b"x" * 32).hexdigest() # noqa: S324 + + +def _present(checksum): + """Side effects for a verify_file() that finds a matching remote file.""" + return ["Directory of flash:/EOS.swi\n", f"verify /md5 (flash:EOS.swi) = {checksum}"] + + +ABSENT = ["No such file"] + + +def test_file_copy_skips_transfer_when_already_present(eos_ssh_send_command, local_image): + source, checksum = local_image + device = eos_ssh_send_command(["dir", *_present(checksum)]) + with mock.patch("pyntc.devices.eos_device.FileTransfer") as file_transfer: + device.file_copy(str(source)) + file_transfer.return_value.transfer_file.assert_not_called() + + +def test_file_copy_transfers_when_missing(eos_ssh_send_command, local_image): + source, checksum = local_image + device = eos_ssh_send_command(["dir", *ABSENT, "dir", *_present(checksum)]) + with mock.patch("pyntc.devices.eos_device.FileTransfer") as file_transfer: + device.file_copy(str(source)) + file_transfer.return_value.establish_scp_conn.assert_called_once() + file_transfer.return_value.transfer_file.assert_called_once() + file_transfer.return_value.close_scp_chan.assert_called_once() + # Arista's FileTransfer raises NotImplementedError here, so it must never be called. + file_transfer.return_value.enable_scp.assert_not_called() + + +def test_file_copy_never_enters_the_shell(eos_ssh_send_command, local_image): + # The reason this override exists: the connecting account may have no bash access. + source, checksum = local_image + device = eos_ssh_send_command(["dir", *ABSENT, "dir", *_present(checksum)]) + with mock.patch("pyntc.devices.eos_device.FileTransfer"): + device.file_copy(str(source)) + commands = [call[0][0] for call in device.native.send_command.call_args_list] + assert not any(command.strip() == "bash" or command.startswith("/bin/") for command in commands) + + +def test_file_copy_verifies_over_the_cli(eos_ssh_send_command, local_image): + source, checksum = local_image + device = eos_ssh_send_command(["dir", *ABSENT, "dir", *_present(checksum)]) + with mock.patch("pyntc.devices.eos_device.FileTransfer"): + device.file_copy(str(source)) + commands = [call[0][0] for call in device.native.send_command.call_args_list] + assert "dir flash:/EOS.swi" in commands + assert "verify /md5 flash:EOS.swi" in commands + + +def test_file_copy_raises_when_transfer_fails(eos_ssh_send_command, local_image): + source, _ = local_image + device = eos_ssh_send_command(["dir", *ABSENT, "dir"]) + with mock.patch("pyntc.devices.eos_device.FileTransfer") as file_transfer: + file_transfer.return_value.transfer_file.side_effect = RuntimeError("scp blew up") + with pytest.raises(FileTransferError): + device.file_copy(str(source)) + # The SCP channel must be closed even when the transfer blows up. + file_transfer.return_value.close_scp_chan.assert_called_once() + + +def test_file_copy_raises_socket_closed_when_session_drops_and_file_missing(eos_ssh_send_command, local_image): + source, _ = local_image + device = eos_ssh_send_command(["dir", *ABSENT, "dir"]) + with mock.patch("pyntc.devices.eos_device.FileTransfer") as file_transfer: + file_transfer.return_value.transfer_file.side_effect = OSError("socket closed") + file_transfer.return_value.compare_md5.return_value = False + with pytest.raises(SocketClosedError): + device.file_copy(str(source)) + + +def test_file_copy_tolerates_dropped_session_when_file_landed(eos_ssh_send_command, local_image): + # A dropped control channel is survivable if the file actually made it. + source, checksum = local_image + device = eos_ssh_send_command(["dir", *ABSENT, "dir", *_present(checksum)]) + with mock.patch("pyntc.devices.eos_device.FileTransfer") as file_transfer: + file_transfer.return_value.transfer_file.side_effect = OSError("socket closed") + file_transfer.return_value.compare_md5.return_value = True + device.file_copy(str(source)) + + +def test_file_copy_raises_when_file_absent_after_transfer(eos_ssh_send_command, local_image): + source, _ = local_image + device = eos_ssh_send_command(["dir", *ABSENT, "dir", *ABSENT]) + with mock.patch("pyntc.devices.eos_device.FileTransfer"): + with pytest.raises(FileTransferError): + device.file_copy(str(source)) + + +def test_file_copy_raises_when_not_enough_free_space(eos_ssh_send_command, local_image): + source, _ = local_image + device = eos_ssh_send_command(["dir", *ABSENT, "dir"]) + with mock.patch("pyntc.devices.eos_device.FileTransfer") as file_transfer: + with mock.patch("os.path.getsize", return_value=FREE_BYTES + 1): + with pytest.raises(NotEnoughFreeSpaceError): + device.file_copy(str(source)) + file_transfer.return_value.transfer_file.assert_not_called() + + +def test_file_copy_remote_exists_true(eos_ssh_send_command, local_image): + source, checksum = local_image + device = eos_ssh_send_command(["dir", *_present(checksum)]) + assert device.file_copy_remote_exists(str(source)) is True + + +def test_file_copy_remote_exists_false_when_checksum_differs(eos_ssh_send_command, local_image): + source, _ = local_image + device = eos_ssh_send_command(["dir", *_present("deadbeef")]) + assert device.file_copy_remote_exists(str(source)) is False + + +def test_file_copy_accepts_explicit_file_system_and_dest(eos_ssh_send_command, local_image): + # With both supplied there is no "dir" filesystem probe -- verification goes first. + source, checksum = local_image + device = eos_ssh_send_command( + ["Directory of flash:/boot.swi\n", f"verify /md5 (flash:boot.swi) = {checksum}"], + ) + with mock.patch("pyntc.devices.eos_device.FileTransfer") as file_transfer: + device.file_copy(str(source), dest="boot.swi", file_system="flash:") + file_transfer.return_value.transfer_file.assert_not_called() + assert device.native.send_command.call_args_list[0][0][0] == "dir flash:/boot.swi" + + +def test_file_copy_remote_exists_accepts_explicit_file_system(eos_ssh_send_command, local_image): + source, checksum = local_image + device = eos_ssh_send_command(_present(checksum)) + assert device.file_copy_remote_exists(str(source), file_system="flash:") is True + + +def test_file_copy_remote_exists_never_enters_the_shell(eos_ssh_send_command, local_image): + source, checksum = local_image + device = eos_ssh_send_command(["dir", *_present(checksum)]) + device.file_copy_remote_exists(str(source)) + commands = [call[0][0] for call in device.native.send_command.call_args_list] + assert not any(command.strip() == "bash" or command.startswith("/bin/") for command in commands) + + +# --------------------------------------------------------------------------- +# Pulling code onto the box: remote_file_copy (device fetches from a server) +# --------------------------------------------------------------------------- + + +def _model(url="http://192.0.2.5/EOS.swi", checksum="abc123", **kwargs): + return FileCopyModel(download_url=url, checksum=checksum, file_name="EOS.swi", **kwargs) + + +def test_remote_file_copy_issues_copy_command_and_verifies(eos_ssh_send_command): + device = eos_ssh_send_command( + [ + "dir", # _get_file_system + "", # the copy command itself + "Directory of flash:/EOS.swi\n", # verify_file -> check_file_exists + "verify /md5 (flash:EOS.swi) = abc123", # verify_file -> get_remote_checksum + ] + ) + device.remote_file_copy(_model()) + commands = [call[0][0] for call in device.native.send_command.call_args_list] + assert "copy http://192.0.2.5/EOS.swi flash:" in commands + + +def test_remote_file_copy_embeds_credentials_for_http(eos_ssh_send_command): + device = eos_ssh_send_command(["dir", "", "Directory of flash:/EOS.swi\n", "verify /md5 (flash:EOS.swi) = abc123"]) + device.remote_file_copy(_model(url="http://user:token@192.0.2.5/EOS.swi")) + commands = [call[0][0] for call in device.native.send_command.call_args_list] + assert "copy http://user:token@192.0.2.5/EOS.swi flash:" in commands + + +def test_remote_file_copy_prompts_for_scp_password(eos_ssh_send_command, eos_ssh_send_command_timing): + # SCP cannot carry the password in the URL, so the driver answers the prompt interactively. + device = eos_ssh_send_command(["dir", "Directory of flash:/EOS.swi\n", "verify /md5 (flash:EOS.swi) = abc123"]) + eos_ssh_send_command_timing(["Password:", ""], existing_device=device) + device.remote_file_copy(_model(url="scp://user:token@192.0.2.5/EOS.swi")) + timing_commands = [call[0][0] for call in device.native.send_command_timing.call_args_list] + assert timing_commands[0] == "copy scp://user@192.0.2.5/EOS.swi flash:" + assert timing_commands[1] == "token" # the password, sent only after the prompt appears + + +def test_remote_file_copy_rejects_non_model(eos_ssh_device): + with pytest.raises(TypeError): + eos_ssh_device.remote_file_copy("http://192.0.2.5/EOS.swi") + + +def test_remote_file_copy_rejects_unsupported_scheme(eos_ssh_device): + with pytest.raises(ValueError, match="Unsupported scheme"): + eos_ssh_device.remote_file_copy(_model(url="rsync://192.0.2.5/EOS.swi")) + + +def test_remote_file_copy_rejects_query_string(eos_ssh_device): + # The EOS CLI cannot handle "?" in a copy URL. + with pytest.raises(ValueError, match="query strings"): + eos_ssh_device.remote_file_copy(_model(url="https://192.0.2.5/EOS.swi?token=x")) + + +def test_remote_file_copy_checks_free_space_first(eos_ssh_send_command): + device = eos_ssh_send_command(["dir", "dir"]) + with pytest.raises(NotEnoughFreeSpaceError): + device.remote_file_copy(_model(file_size=10, file_size_unit="gigabytes")) + # Nothing was transferred. + commands = [call[0][0] for call in device.native.send_command.call_args_list] + assert not any(command.startswith("copy ") for command in commands) + + +def test_remote_file_copy_raises_on_error_output(eos_ssh_send_command): + device = eos_ssh_send_command(["dir", "Error: connection refused"]) + with pytest.raises(FileTransferError): + device.remote_file_copy(_model()) + + +def test_remote_file_copy_raises_when_checksum_mismatches(eos_ssh_send_command): + device = eos_ssh_send_command( + ["dir", "", "Directory of flash:/EOS.swi\n", "verify /md5 (flash:EOS.swi) = deadbeef"] + ) + with pytest.raises(FileTransferError): + device.remote_file_copy(_model(checksum="abc123")) + + +# --------------------------------------------------------------------------- +# Upgrading: install_os +# --------------------------------------------------------------------------- + +NEW_IMAGE = "EOS-4.28.9M.swi" +NEW_IMAGE_BOOTED = "Software image: flash:/EOS-4.28.9M.swi\n" + + +def _set_boot_options_effects(): + """Side effects consumed by set_boot_options: fs probe, dir listing, install, readback.""" + return ["dir", "dir", "", '{"softwareImage": "flash:/EOS-4.28.9M.swi"}'] + + +def test_install_os_returns_false_when_image_already_booted(eos_ssh_send_command): + device = eos_ssh_send_command(["show_boot"]) + assert device.install_os(BOOT_IMAGE) is False + + +def test_install_os_sets_boot_options_then_reboots(eos_ssh_send_command): + device = eos_ssh_send_command(["show_boot", *_set_boot_options_effects(), NEW_IMAGE_BOOTED]) + with mock.patch.object(Driver, "reboot") as mock_reboot: + assert device.install_os(NEW_IMAGE) is True + mock_reboot.assert_called_once_with(wait_for_reload=True, timeout=DEFAULT_REBOOT_TIMEOUT) + commands = [call[0][0] for call in device.native.send_command.call_args_list] + assert f"install source flash:{NEW_IMAGE}" in commands + + +def test_install_os_honours_custom_timeout(eos_ssh_send_command): + device = eos_ssh_send_command(["show_boot", *_set_boot_options_effects(), NEW_IMAGE_BOOTED]) + with mock.patch.object(Driver, "reboot") as mock_reboot: + device.install_os(NEW_IMAGE, timeout=120) + mock_reboot.assert_called_once_with(wait_for_reload=True, timeout=120) + + +def test_install_os_without_reboot_does_not_reboot(eos_ssh_send_command): + device = eos_ssh_send_command(["show_boot", *_set_boot_options_effects()]) + with mock.patch.object(Driver, "reboot") as mock_reboot: + assert device.install_os(NEW_IMAGE, reboot=False) is True + mock_reboot.assert_not_called() + + +def test_install_os_raises_when_image_not_booted_after_reboot(eos_ssh_send_command): + # Device comes back still running the old image. The final side effect feeds + # self.hostname, which OSInstallError reads when building its message. + device = eos_ssh_send_command(["show_boot", *_set_boot_options_effects(), "show_boot", "show_hostname_json"]) + with mock.patch.object(Driver, "reboot"): + with pytest.raises(OSInstallError): + device.install_os(NEW_IMAGE) + + +# --------------------------------------------------------------------------- +# reboot / rollback / save +# --------------------------------------------------------------------------- + + +def test_reboot_sends_reload_now(eos_ssh_device): + eos_ssh_device.reboot() + eos_ssh_device.native.send_command_timing.assert_called_with("reload now") + + +def test_reboot_marks_session_disconnected(eos_ssh_device): + eos_ssh_device.reboot() + assert eos_ssh_device._connected is False + + +def test_reboot_tolerates_dropped_session(eos_ssh_device): + # The session dies mid-command by design; that must not surface as an error. + eos_ssh_device.native.send_command_timing.side_effect = OSError("Socket is closed") + eos_ssh_device.reboot() + assert eos_ssh_device._connected is False + + +def test_reboot_without_wait_does_not_poll(eos_ssh_device): + with mock.patch.object(Driver, "_wait_for_device_reboot") as mock_wait: + eos_ssh_device.reboot() + mock_wait.assert_not_called() + + +def test_reboot_wait_for_reload_polls_with_original_boot_time(eos_ssh_send_command): + device = eos_ssh_send_command(["show_version_json"]) + with mock.patch.object(Driver, "_wait_for_device_reboot") as mock_wait: + device.reboot(wait_for_reload=True, timeout=42) + mock_wait.assert_called_once_with(original_boot_time=BOOT_TIMESTAMP, timeout=42) + + +def test_reboot_warns_on_deprecated_confirm(eos_ssh_device, caplog): + eos_ssh_device.reboot(confirm=True) + assert "deprecated" in caplog.text + + +def test_vlans_are_cached(eos_ssh_send_command): + device = eos_ssh_send_command(["show_vlan_json"]) + assert device.vlans == ["1", "10", "11", "12", "9"] + assert device.vlans == ["1", "10", "11", "12", "9"] + assert device.native.send_command.call_count == 1 + + +def test_rollback(eos_ssh_send_command): + device = eos_ssh_send_command([""]) + device.rollback("good_checkpoint") + assert device.native.send_command.call_args[0][0] == "configure replace good_checkpoint force" + + +def test_rollback_raises_on_failure(eos_ssh_send_command): + device = eos_ssh_send_command(["% Invalid input"]) + with pytest.raises(RollbackError): + device.rollback("bad_checkpoint") + + +def test_save(eos_ssh_send_command): + device = eos_ssh_send_command([""]) + assert device.save() is True + assert device.native.send_command.call_args[0][0] == "copy running-config startup-config" + + +def test_checkpoint(eos_ssh_send_command): + device = eos_ssh_send_command([""]) + device.checkpoint("good_checkpoint") + assert device.native.send_command.call_args[0][0] == "copy running-config good_checkpoint" + + +def test_set_boot_options(eos_ssh_send_command): + # Side effects: _get_file_system, dir , install source, then boot_options readback. + device = eos_ssh_send_command( + ["dir", "dir", "", '{"softwareImage": "flash:/EOS-4.28.9M.swi"}'], + ) + device.set_boot_options("EOS-4.28.9M.swi") + calls = [call[0][0] for call in device.native.send_command.call_args_list] + assert "install source flash:EOS-4.28.9M.swi" in calls + + +def test_set_boot_options_uses_long_read_timeout(eos_ssh_send_command): + device = eos_ssh_send_command( + ["dir", "dir", "", '{"softwareImage": "flash:/EOS-4.28.9M.swi"}'], + ) + device.set_boot_options("EOS-4.28.9M.swi") + install_call = [c for c in device.native.send_command.call_args_list if "install source" in c[0][0]][0] + assert install_call[1]["read_timeout"] == 1800 + + +def test_set_boot_options_missing_image(eos_ssh_send_command): + from pyntc.errors import NTCFileNotFoundError + + # Third side effect feeds self.hostname, which NTCFileNotFoundError reads. + device = eos_ssh_send_command(["dir", "dir", "show_hostname_json"]) + with pytest.raises(NTCFileNotFoundError): + device.set_boot_options("not-on-the-box.swi") + + +def test_set_boot_options_raises_when_readback_mismatches(eos_ssh_send_command): + device = eos_ssh_send_command(["dir", "dir", "", '{"softwareImage": "flash:/EOS-4.28.5M.swi"}']) + with pytest.raises(CommandError): + device.set_boot_options("EOS-4.28.9M.swi") + + +def test_install_mode_remains_unimplemented(eos_ssh_device): + # EOSDevice does not implement install_mode; parity means neither does this driver. + with pytest.raises(NotImplementedError): + eos_ssh_device.install_mode # noqa: B018 + + +# --------------------------------------------------------------------------- +# API parity with EOSDevice +# --------------------------------------------------------------------------- + +# EOSDevice assigns native_ssh as an *instance* attribute inside open(); the SSH driver +# exposes it as a class-level property so inherited code resolves it. That is the only +# permitted addition to the public surface. +KNOWN_ADDITIONS = {"native_ssh"} + + +def _fixture(name): + path = os.path.join(os.path.dirname(__file__), "device_mocks", "eos_ssh", name) + with open(path) as handle: + return json.load(handle) + + +# Facts derived purely from show output. "vlans" is excluded: EOSDevice sources it from +# pyeapi's native.api("vlans"), which has no SSH equivalent by design. +SHARED_FACTS = ["boot_time", "hostname", "fqdn", "model", "os_version", "serial_number", "interfaces", "boot_options"] + + +@pytest.mark.parametrize("fact", SHARED_FACTS) +def test_facts_match_eapi_driver_for_identical_payload(fact): + """Both drivers must derive identical facts from identical device output. + + The two drivers no longer share fixture files, so this is the anti-drift guard: feed + the same documents to each and require the same answer. + """ + payloads = { + "show version": _fixture("show_version_json"), + "show hostname": _fixture("show_hostname_json"), + "show interfaces status": _fixture("show_interfaces_status_json"), + "show boot-config": _fixture("show_boot-config_json"), + } + + def fake_show(command, raw_text=False): + return payloads[command] + + with mock.patch("pyntc.devices.eos_ssh_device.ConnectHandler"): + ssh_device = EOSSSHDevice("host", "user", "password") + with mock.patch("pyeapi.client.Node", autospec=True): + with mock.patch("pyntc.devices.eos_device.eos_connect"): + eapi_device = EOSDevice("host", "user", "password") + + with mock.patch.object(ssh_device, "show", side_effect=fake_show): + with mock.patch.object(eapi_device, "show", side_effect=fake_show): + assert getattr(ssh_device, fact) == getattr(eapi_device, fact) + + +def _public_api(cls): + return {name for name in dir(cls) if not name.startswith("_")} + + +def test_public_api_matches_eos_device(): + assert _public_api(EOSSSHDevice) - KNOWN_ADDITIONS == _public_api(EOSDevice) + + +def test_no_eos_device_member_is_missing(): + assert _public_api(EOSDevice) - _public_api(EOSSSHDevice) == set() + + +@pytest.mark.parametrize("name", sorted(_public_api(EOSDevice))) +def test_member_parity(name): + eapi_attr = inspect.getattr_static(EOSDevice, name) + ssh_attr = inspect.getattr_static(EOSSSHDevice, name) + assert isinstance(ssh_attr, property) == isinstance(eapi_attr, property), f"{name} kind differs" + if callable(eapi_attr) and not isinstance(eapi_attr, property): + assert inspect.signature(ssh_attr) == inspect.signature(eapi_attr), f"{name} signature differs" diff --git a/tests/unit/test_infra.py b/tests/unit/test_infra.py index 0be3ff1a..c69490d1 100644 --- a/tests/unit/test_infra.py +++ b/tests/unit/test_infra.py @@ -5,7 +5,7 @@ import pytest from pyntc import ntc_device, ntc_device_by_name -from pyntc.devices import EOSDevice, IOSDevice, NXOSDevice, supported_devices +from pyntc.devices import EOSDevice, EOSSSHDevice, IOSDevice, NXOSDevice, supported_devices from pyntc.errors import ConfFileNotFoundError, UnsupportedDeviceError BAD_DEVICE_TYPE = "238nzsvkn3981" @@ -18,12 +18,13 @@ @mock.patch("pyntc.devices.ios_device.IOSDevice.open") @mock.patch("pyntc.devices.iosxr_device.IOSXRDevice.open") @mock.patch("pyntc.devices.nxos_device.NXOSDevice.open") +@mock.patch("pyntc.devices.eos_ssh_device.EOSSSHDevice.open") @mock.patch("pyntc.devices.jnpr_device.JunosNativeSW") @mock.patch("pyntc.devices.jnpr_device.JunosNativeDevice.open") @mock.patch("pyntc.devices.jnpr_device.JunosNativeDevice.timeout") @pytest.mark.parametrize("device_type,expected", supported_devices.items(), ids=list(supported_devices)) def test_device_creation( - j_timeout, j_open, j_nsw, nx_open, xr_open, i_open, a_open, f_mr, air_open, device_type, expected + j_timeout, j_open, j_nsw, eos_ssh_open, nx_open, xr_open, i_open, a_open, f_mr, air_open, device_type, expected ): # Skip f5 on python >3.11 if sys.version_info >= (3, 12) and device_type == "f5_tmos_icontrol": @@ -39,8 +40,9 @@ def test_unsupported_device(): @mock.patch("pyntc.devices.ios_device.IOSDevice.open") @mock.patch("pyntc.devices.nxos_device.NXOSDevice.open") +@mock.patch("pyntc.devices.eos_ssh_device.EOSSSHDevice.open") @mock.patch("pyntc.devices.jnpr_device.JunosDevice.open") -def test_device_by_name(j_open, nx_open, i_open): +def test_device_by_name(j_open, eos_ssh_open, nx_open, i_open): config_filepath = os.path.join(FIXTURES_DIR, ".ntc.conf.sample") nxos_device = ntc_device_by_name("test_nxos", filename=config_filepath) @@ -49,6 +51,9 @@ def test_device_by_name(j_open, nx_open, i_open): eos_device = ntc_device_by_name("test_eos", filename=config_filepath) assert isinstance(eos_device, EOSDevice) + eos_ssh_device = ntc_device_by_name("test_eos_ssh", filename=config_filepath) + assert isinstance(eos_ssh_device, EOSSSHDevice) + ios_device = ntc_device_by_name("test_ios", filename=config_filepath) assert isinstance(ios_device, IOSDevice) From 41ce6a01900ce21caecf53f852068be58ffce86b Mon Sep 17 00:00:00 2001 From: Jeff Kala Date: Tue, 11 Aug 2026 08:50:21 -0600 Subject: [PATCH 2/3] fix a few potential bugs, switch timeouts on file calls to match nxos --- pyntc/devices/eos_ssh_device.py | 11 ++++++----- tests/unit/test_devices/test_eos_ssh_device.py | 14 ++++++++++++-- 2 files changed, 18 insertions(+), 7 deletions(-) diff --git a/pyntc/devices/eos_ssh_device.py b/pyntc/devices/eos_ssh_device.py index 0f33030e..4aff97ae 100644 --- a/pyntc/devices/eos_ssh_device.py +++ b/pyntc/devices/eos_ssh_device.py @@ -40,7 +40,7 @@ # signature byte-identical to EOSDevice.show(), so inherited callers such as # ``set_boot_options``, ``save``, ``checkpoint`` and ``rollback`` work without overrides. COMMAND_READ_TIMEOUTS = ( - (re.compile(r"^\s*install\s+source\b"), 1800), + (re.compile(r"^\s*install\s+source\b"), 3600), (re.compile(r"^\s*copy\s+running-config\b"), 300), (re.compile(r"^\s*configure\s+replace\b"), 300), (re.compile(r"^\s*show\s+(running|startup)-config\b"), 120), @@ -204,7 +204,8 @@ def show(self, commands, raw_text=False): pipe, True to return the raw CLI text. Defaults to False. Returns: - (dict): When ``commands`` is a str and ``raw_text`` is False. + (dict): When ``commands`` is a str and ``raw_text`` is False. Non-show commands + cannot be piped to ``| json``; they run as plain text and return an empty dict. (str): When ``commands`` is a str and ``raw_text`` is True. (list): When ``commands`` is a list. @@ -226,15 +227,15 @@ def show(self, commands, raw_text=False): cli_command = f"{command} | json" if as_json else command try: output = self._send_command(cli_command, error_command=command) + if as_json: + output = self._load_json(command, output) except CommandError as err: if original_commands_is_str: raise raise CommandListError(entered_commands, command, err.cli_error_msg) from err - if raw_text: + if raw_text or as_json: responses.append(output) - elif as_json: - responses.append(self._load_json(command, output)) else: # Non-show command sent with raw_text=False (checkpoint, save, rollback, # reboot, set_boot_options). Every inherited caller discards the result, diff --git a/tests/unit/test_devices/test_eos_ssh_device.py b/tests/unit/test_devices/test_eos_ssh_device.py index a54c744f..38def50a 100644 --- a/tests/unit/test_devices/test_eos_ssh_device.py +++ b/tests/unit/test_devices/test_eos_ssh_device.py @@ -220,6 +220,16 @@ def test_show_raises_when_output_is_not_json(eos_ssh_send_command): assert "does not support JSON output" in err.value.cli_error_msg +def test_show_list_raises_command_list_error_when_output_is_not_json(eos_ssh_send_command): + # The list contract must hold for JSON parse failures too, not only device-reported + # errors: a non-JSON response mid-list raises CommandListError, never bare CommandError. + device = eos_ssh_send_command(["show_version_json", "This command is not converted to JSON"]) + with pytest.raises(CommandListError) as err: + device.show(["show version", "show something-unconverted"]) + assert err.value.commands == ["show version", "show something-unconverted"] + assert err.value.command == "show something-unconverted" + + def test_show_reopens_connection(eos_ssh_send_command): # Guards reboot polling: _wait_for_device_reboot survives only because show() re-opens. device = eos_ssh_send_command(["show_version_json"]) @@ -231,7 +241,7 @@ def test_show_reopens_connection(eos_ssh_send_command): @pytest.mark.parametrize( "command,expected_timeout", [ - ("install source flash:EOS.swi", 1800), + ("install source flash:EOS.swi", 3600), ("copy running-config startup-config", 300), ("configure replace flash:cp force", 300), ("show running-config", 120), @@ -868,7 +878,7 @@ def test_set_boot_options_uses_long_read_timeout(eos_ssh_send_command): ) device.set_boot_options("EOS-4.28.9M.swi") install_call = [c for c in device.native.send_command.call_args_list if "install source" in c[0][0]][0] - assert install_call[1]["read_timeout"] == 1800 + assert install_call[1]["read_timeout"] == 3600 def test_set_boot_options_missing_image(eos_ssh_send_command): From c283d1a9570ea19b6d7e4c32900a5ffc9478ec62 Mon Sep 17 00:00:00 2001 From: Jeff Kala Date: Tue, 11 Aug 2026 09:07:32 -0600 Subject: [PATCH 3/3] fixes from testing against actual CML eos device via ssh --- pyntc/devices/eos_ssh_device.py | 5 ++++- tests/integration/test_eos_ssh_device.py | 11 ++++++++--- tests/unit/test_devices/test_eos_ssh_device.py | 12 +++++++++++- 3 files changed, 23 insertions(+), 5 deletions(-) diff --git a/pyntc/devices/eos_ssh_device.py b/pyntc/devices/eos_ssh_device.py index 4aff97ae..5cbe100e 100644 --- a/pyntc/devices/eos_ssh_device.py +++ b/pyntc/devices/eos_ssh_device.py @@ -268,7 +268,10 @@ def config(self, commands): try: for command in command_list: entered_commands.append(command) - output = self.native.send_config_set(command, exit_config_mode=False) + # Multi-line commands (e.g. "banner motd\n...\nEOF") drop the CLI into an + # input mode whose echo Netmiko's cmd_verify cannot match; verification must + # be disabled for them or send_config_set raises ReadTimeout. + output = self.native.send_config_set(command, exit_config_mode=False, cmd_verify="\n" not in command) try: self._check_output_for_errors(command, output) except CommandError as err: diff --git a/tests/integration/test_eos_ssh_device.py b/tests/integration/test_eos_ssh_device.py index 39883330..87168d7f 100644 --- a/tests/integration/test_eos_ssh_device.py +++ b/tests/integration/test_eos_ssh_device.py @@ -155,10 +155,15 @@ def test_startup_config(device): def test_config_round_trip(device): - """Apply a harmless config change and confirm it lands in the running config.""" + """Apply a harmless config change, confirm it lands in the running config, then remove it.""" marker = "pyntc integration test" - device.config(f"banner motd {marker}\nEOF") - assert marker in device.running_config + # Multi-line on purpose: banners exercise the cmd_verify=False path in config(). + device.config(f"banner motd\n{marker}\nEOF") + try: + assert marker in device.running_config + finally: + device.config("no banner motd") + assert marker not in device.running_config def test_show_raises_on_bad_command(device): diff --git a/tests/unit/test_devices/test_eos_ssh_device.py b/tests/unit/test_devices/test_eos_ssh_device.py index 38def50a..868cccf6 100644 --- a/tests/unit/test_devices/test_eos_ssh_device.py +++ b/tests/unit/test_devices/test_eos_ssh_device.py @@ -265,7 +265,17 @@ def test_show_resolves_read_timeout_per_command(eos_ssh_send_command, command, e def test_config_single_command_returns_none(eos_ssh_config): device = eos_ssh_config([""]) assert device.config("interface Ethernet1") is None - device.native.send_config_set.assert_called_with("interface Ethernet1", exit_config_mode=False) + device.native.send_config_set.assert_called_with("interface Ethernet1", exit_config_mode=False, cmd_verify=True) + + +def test_config_disables_cmd_verify_for_multiline_commands(eos_ssh_config): + # Multi-line input modes (banner motd ... EOF) echo in a way cmd_verify cannot match; + # verified against real hardware (vEOS): with cmd_verify netmiko raises ReadTimeout. + device = eos_ssh_config([""]) + device.config("banner motd\npyntc\nEOF") + device.native.send_config_set.assert_called_with( + "banner motd\npyntc\nEOF", exit_config_mode=False, cmd_verify=False + ) def test_config_list_returns_none(eos_ssh_config):