From 966c1172884268773f0f551eb0528909f356e9ea Mon Sep 17 00:00:00 2001 From: James Williams Date: Mon, 15 Jun 2026 23:33:48 -0500 Subject: [PATCH 1/7] initial commit of xr driver --- .gitignore | 3 + docs/user/lib_overview.md | 34 + mkdocs.yml | 1 + pyntc/devices/__init__.py | 2 + pyntc/devices/iosxr_device.py | 713 +++++++++++++++++++ tests/unit/test_devices/test_iosxr_device.py | 479 +++++++++++++ tests/unit/test_infra.py | 5 +- 7 files changed, 1236 insertions(+), 1 deletion(-) create mode 100644 pyntc/devices/iosxr_device.py create mode 100644 tests/unit/test_devices/test_iosxr_device.py diff --git a/.gitignore b/.gitignore index 8141a6a9..f8b76915 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,6 @@ +# Local IOS-XR lab harness (not part of the package) +/test-iosxr.py + # Ansible Retry Files *.retry diff --git a/docs/user/lib_overview.md b/docs/user/lib_overview.md index 30f29070..b899bdac 100644 --- a/docs/user/lib_overview.md +++ b/docs/user/lib_overview.md @@ -15,7 +15,41 @@ It's main purpose is to simplify the execution of common tasks including: - Cisco AireOS - uses netmiko (SSH) - Cisco ASA - uses netmiko (SSH) - Cisco IOS platforms - uses netmiko (SSH) +- Cisco IOS-XR (eXR / 64-bit) - uses netmiko (SSH) - Cisco NX-OS - uses pynxos (NX-API) - Arista EOS - uses pyeapi (eAPI) - Juniper Junos - uses PyEz (NETCONF) - F5 Networks - uses f5-sdk (ReST) + +!!! note "IOS-XR upgrades: install the base ISO plus matching feature RPMs" + The `cisco_iosxr_ssh` driver (`IOSXRDevice`) performs eXR OS upgrades using the + asynchronous native install workflow (`install add` → poll → `install activate` → + poll → reload → `install commit` → verify). + + On eXR the base ISO **cannot be activated on its own** when optional feature + packages (IS-IS, OSPF, MPLS, multicast, etc.) are active: `install activate` + aborts demanding the matching-version RPMs be activated in the same operation. + Pass those RPMs via the `additional_files` argument to `install_os` so the base + ISO and the RPMs are added and activated together: + + ```python + device.install_os( + "ncs5k-mini-x-7.11.2.iso", + additional_files=[ + "ncs5k-isis-2.1.0.0-r7112.x86_64.rpm", + "ncs5k-ospf-2.1.0.0-r7112.x86_64.rpm", + # ... the remaining matching-version feature RPMs + ], + ) + ``` + + All files (ISO + RPMs) must already be staged on `harddisk:` (use + `remote_file_copy` for each). Omitting the RPMs on a device that runs feature + packages will cause `install activate` to abort. + +!!! warning "Nautobot OS Upgrades passes a single image today" + The Nautobot OS Upgrades `InstallOsJob` currently passes a single image to + `install_os` and does not yet supply `additional_files`. Driving an eXR upgrade + through that app will therefore attempt an ISO-only activation and abort on any + device with feature packages. Multi-file support in OS Upgrades is tracked as + separate follow-up work. diff --git a/mkdocs.yml b/mkdocs.yml index de0ef850..9fec2fcd 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -163,6 +163,7 @@ nav: - 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" + - pyntc.devices.iosxr_device: "code-reference/pyntc/devices/iosxr_device.md" - pyntc.devices.jnpr_device: "code-reference/pyntc/devices/jnpr_device.md" - pyntc.devices.nxos_device: "code-reference/pyntc/devices/nxos_device.md" - pyntc.devices.system_features: "code-reference/pyntc/devices/system_features/__init__.md" diff --git a/pyntc/devices/__init__.py b/pyntc/devices/__init__.py index a9dcead4..3756dd33 100644 --- a/pyntc/devices/__init__.py +++ b/pyntc/devices/__init__.py @@ -6,6 +6,7 @@ from .f5_device import F5Device from .ios_device import IOSDevice from .iosxewlc_device import IOSXEWLCDevice +from .iosxr_device import IOSXRDevice from .jnpr_device import JunosDevice from .nxos_device import NXOSDevice @@ -14,6 +15,7 @@ "arista_eos_eapi": EOSDevice, "f5_tmos_icontrol": F5Device, "cisco_ios_ssh": IOSDevice, + "cisco_iosxr_ssh": IOSXRDevice, "juniper_junos_netconf": JunosDevice, "cisco_nxos_nxapi": NXOSDevice, "cisco_aireos_ssh": AIREOSDevice, diff --git a/pyntc/devices/iosxr_device.py b/pyntc/devices/iosxr_device.py new file mode 100644 index 00000000..5f0ef35d --- /dev/null +++ b/pyntc/devices/iosxr_device.py @@ -0,0 +1,713 @@ +"""Module for using a Cisco IOS-XR (eXR / 64-bit) device over SSH. + +This driver targets 64-bit IOS-XR (eXR) platforms (initial target: NCS5000 / +NCS-5011) and implements the asynchronous, package-based OS upgrade workflow: + + install add -> poll for completion -> install activate -> reload -> install commit -> verify + +Phase 1 installs the base ISO only. The ``additional_files`` argument to +``install_os`` is accepted for forward compatibility but is currently inert: +when supplied it logs a warning and the files are ignored, and only the base +ISO is installed. Feature RPMs are therefore not installed and must be handled +manually until full-bundle support lands. +""" + +import re +import time + +from netmiko import ConnectHandler +from netmiko.exceptions import ReadTimeout + +from pyntc import log +from pyntc.devices.base_device import BaseDevice, fix_docs +from pyntc.errors import CommandError, CommandListError, FileTransferError, OSInstallError, RebootTimeoutError +from pyntc.utils.models import FileCopyModel + +DEFAULT_FILE_SYSTEM = "harddisk:" +# Parse the operation id from an "install add" response, e.g. "Install operation 17 started". +RE_INSTALL_OP = re.compile(r"[Ii]nstall operation (\d+)") +# Parse the active boot package and its version from "show install active", e.g. "ncs5k-xr-7.11.2". +RE_XR_BOOT_IMAGE = re.compile(r"(?P\S*xr-(?P\d+\.\d+\.\d+\w*))") +# Parse the running version from "show version", e.g. "Version 7.11.2". +RE_XR_VERSION = re.compile(r"Version\s+(\d+\.\d+\.\d+\w*)") +# Success / error markers emitted by the IOS-XR "copy" command (eXR uses +# "Successfully copied ... Bytes" / "Copy operation success"). +RE_COPY_SUCCESS = re.compile( + r"Successfully copied|Copy operation success|bytes copied|copied in|\[OK\]|Download Complete|transfer successful", + re.IGNORECASE, +) +RE_COPY_ERROR = re.compile( + r"%Error|Error opening|Invalid input|Failed|Aborted|denied|No such file|Connection refused|timed out|could not", + re.IGNORECASE, +) + + +@fix_docs +class IOSXRDevice(BaseDevice): + """Cisco IOS-XR (eXR / 64-bit) Device Implementation.""" + + vendor = "cisco" + + # pylint: disable=too-many-arguments, too-many-positional-arguments + def __init__(self, host, username, password, secret="", port=None, **kwargs): # noqa: D403 # nosec + """PyNTC Device implementation for Cisco IOS-XR (eXR). + + 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 port to use to establish the connection. Defaults to 22. + kwargs (dict): Additional arguments to pass to the Netmiko ConnectHandler. + """ + super().__init__(host, username, password, device_type="cisco_iosxr_ssh") + + self.native = None + self.secret = secret + self.port = int(port) if port else 22 + self.read_timeout_override = kwargs.get("read_timeout_override") + self._connected = False + self.open() + log.init(host=host) + + def _send_command(self, command, expect_string=None, **kwargs): + command_args = {"command_string": command} + if expect_string is not None: + command_args["expect_string"] = expect_string + command_args.update(kwargs) + + response = self.native.send_command(**command_args) + + if "% " in response or "Error:" in response: + log.error("Host %s: Error in %s with response: %s", self.host, command, response) + raise CommandError(command, response) + + log.info("Host %s: Command %s was executed successfully.", self.host, command) + return response + + def _uptime_components(self, uptime_full_string): + match_weeks = re.search(r"(\d+) weeks?", uptime_full_string) + match_days = re.search(r"(\d+) days?", uptime_full_string) + match_hours = re.search(r"(\d+) hours?", uptime_full_string) + match_minutes = re.search(r"(\d+) minutes?", uptime_full_string) + + weeks = int(match_weeks.group(1)) if match_weeks else 0 + days = int(match_days.group(1)) if match_days else 0 + hours = int(match_hours.group(1)) if match_hours else 0 + minutes = int(match_minutes.group(1)) if match_minutes else 0 + + return weeks, days, hours, minutes + + def _uptime_to_seconds(self, uptime_full_string): + weeks, days, hours, minutes = self._uptime_components(uptime_full_string) + seconds = weeks * 7 * 24 * 60 * 60 + seconds += days * 24 * 60 * 60 + seconds += hours * 60 * 60 + seconds += minutes * 60 + return seconds + + def _uptime_to_string(self, uptime_full_string): + weeks, days, hours, minutes = self._uptime_components(uptime_full_string) + days = days + weeks * 7 + return f"{days:02d}:{hours:02d}:{minutes:02d}:00" + + def _install_add(self, source, packages): + """Stage packages into the install repository. + + ``install add`` returns immediately and continues asynchronously in the + background, printing an operation id that subsequent steps reference. + + Args: + source (str): The on-device source path, e.g. ``harddisk:/``. + packages (list): Package filenames to add (the base ISO for Phase 1). + + Returns: + (int): The install operation id parsed from the device response. + + Raises: + OSInstallError: When the device response does not contain an operation id. + """ + command = f"install add source {source} {' '.join(packages)}" + response = self.native.send_command(command, read_timeout=120) + + match = RE_INSTALL_OP.search(response) + if match is None: + log.error("Host %s: Unable to parse install operation id from response: %s", self.host, response) + raise OSInstallError(hostname=self.host, desired_boot=" ".join(packages)) + + op_id = int(match.group(1)) + log.info("Host %s: install add started operation %s.", self.host, op_id) + return op_id + + def _wait_for_install_op(self, op_id, timeout=3600, interval=30): + """Poll ``show install log `` until the operation reaches a terminal state. + + Args: + op_id (int): The install operation id to track. + timeout (int): Maximum seconds to wait for a terminal state. Defaults to 3600. + interval (int): Seconds to wait between polls. Defaults to 30. + + Raises: + OSInstallError: When the operation aborts/fails or the timeout is exceeded. + """ + success = re.compile(rf"operation\s+{op_id}\b.*(completed successfully|succeeded)", re.IGNORECASE | re.DOTALL) + failure = re.compile(rf"operation\s+{op_id}\b.*(aborted|failed)", re.IGNORECASE | re.DOTALL) + + start = time.time() + while time.time() - start < timeout: + output = self.native.send_command(f"show install log {op_id}", read_timeout=120) + if failure.search(output): + log.error("Host %s: install operation %s aborted/failed.", self.host, op_id) + raise OSInstallError(hostname=self.host, desired_boot=f"operation {op_id}") + if success.search(output): + log.info("Host %s: install operation %s completed successfully.", self.host, op_id) + return + time.sleep(interval) + + log.error("Host %s: install operation %s timed out after %s seconds.", self.host, op_id, timeout) + raise OSInstallError(hostname=self.host, desired_boot=f"operation {op_id}") + + def _install_activate(self, op_id, poll_interval=60, timeout=3600): + """Activate a staged install operation and track it to completion. + + The activation is issued with ``noprompt`` (so eXR does not wait on the interactive + reload confirmation) and runs **asynchronously** so the SSH session stays free to poll + the install status. A synchronous activate would hold the session and, when the reload + tore the device down, trap the read on a half-open socket until ``read_timeout``. + + The activation creates its own operation id (distinct from the ``install add`` id). + This method polls ``show install request`` once per ``poll_interval`` — logging each + poll — until that operation finishes successfully, then returns so the caller can wait + for the reload. + + Args: + op_id (int): The staged ``install add`` operation id to activate. + poll_interval (int): Seconds between status polls. Defaults to 60. + timeout (int): Maximum seconds to wait for the activation to finish. Defaults to 3600. + + Raises: + OSInstallError: When the activation operation aborts/fails or does not finish in time. + """ + command = f"install activate id {op_id} noprompt" + log.info("Host %s: issuing activation: %s", self.host, command) + try: + self.native.send_command_timing(command, read_timeout=180) + except Exception as exc: # noqa: BLE001 # pylint: disable=broad-exception-caught + log.info("Host %s: activation issue dropped the session (reload already underway): %s", self.host, exc) + return + + start = time.time() + while time.time() - start < timeout: + time.sleep(poll_interval) + try: + request = self.native.send_command("show install request", read_timeout=120) + except Exception as exc: # noqa: BLE001 # pylint: disable=broad-exception-caught + log.info("Host %s: session dropped while polling activation (reload underway): %s", self.host, exc) + return + op_ids = [int(match) for match in re.findall(r"Operation Id\s*:\s*(\d+)", request)] + newest = max(op_ids) if op_ids else op_id + log.info("Host %s: polled activation status (latest operation %s).", self.host, newest) + if newest > op_id and re.search(r"install activate", request, re.IGNORECASE): + if re.search(r"State\s*:\s*Failure|aborted", request, re.IGNORECASE): + log.error("Host %s: activation operation %s failed: %s", self.host, newest, request) + raise OSInstallError(hostname=self.host, desired_boot=f"operation {newest}") + if re.search(r"State\s*:\s*Success", request, re.IGNORECASE): + log.info("Host %s: activation operation %s finished successfully.", self.host, newest) + return + + log.error("Host %s: activation of operation %s did not finish within %ss.", self.host, op_id, timeout) + raise OSInstallError(hostname=self.host, desired_boot=f"operation {op_id}") + + def _install_commit(self): + """Persist the activated software so it survives the reload.""" + self.native.send_command("install commit") + log.info("Host %s: install commit issued.", self.host) + + def _wait_for_device_reboot(self, timeout=3600, interval=60): + """Wait for the activation reload: watch the session drop, then poll until it returns. + + After a successful activation the device reloads. This probes the device once per + ``interval`` — logging each poll — and requires a *drop-then-recover* transition: it + first waits for the session to drop (reload in progress), then keeps polling on fresh + connections until one succeeds. Requiring the drop avoids mistaking the brief still-up + window before the reload for a completed reboot, and using fresh short-lived + connections avoids the half-open-socket hang a long-lived read would hit. + + Args: + timeout (int): Maximum seconds to wait for the device to return. Defaults to 3600. + interval (int): Seconds between probes. Defaults to 60. + + Raises: + RebootTimeoutError: When the device does not return within ``timeout``. + """ + # Drop any existing session so each probe is a fresh connection. + try: + self.close() + except Exception as close_exc: # noqa: BLE001 # pylint: disable=broad-exception-caught + log.debug("Host %s: pre-reboot disconnect raised %s (ignored).", self.host, close_exc) + self.native = None + self._connected = False + + start = time.time() + seen_down = False + while time.time() - start < timeout: + try: + self.open() + self.show("show version") + if seen_down: + log.info("Host %s: device is back up after reload.", self.host) + return + log.info("Host %s: device still reachable; waiting for the reload to drop the session...", self.host) + except Exception as exc: # noqa: BLE001 # pylint: disable=broad-exception-caught + self.native = None + self._connected = False + if not seen_down: + log.info("Host %s: device disconnected; reload in progress (%s).", self.host, exc) + seen_down = True + else: + log.info("Host %s: device still down (%s); polling again in %ss.", self.host, exc, interval) + time.sleep(interval) + + log.error("Host %s: device did not return within %ss while rebooting.", self.host, timeout) + raise RebootTimeoutError(hostname=self.host, wait_time=timeout) + + @property + def boot_options(self): + """Get the current boot image and version from ``show install active``. + + Returns: + (dict): ``{"sys": , "version": }``; + both values are ``None`` when the output cannot be parsed. + """ + show_install_active = self.show("show install active") + match = RE_XR_BOOT_IMAGE.search(show_install_active) + if match: + boot_options = {"sys": match.group("sys"), "version": match.group("version")} + else: + boot_options = {"sys": None, "version": None} + + log.debug("Host %s: boot options %s.", self.host, boot_options) + return boot_options + + def close(self): + """Disconnect from the device.""" + if self.connected: + self.native.disconnect() + self._connected = False + log.debug("Host %s: Connection closed.", self.host) + + @property + def connected(self): # noqa: D401 + """Get the connection status of the device. + + Returns: + (bool): True if the device is connected, else False. + """ + return self._connected + + @connected.setter + def connected(self, value): + self._connected = value + + def enable(self): + """No-op for IOS-XR. + + IOS-XR EXEC mode is already privileged, so there is no enable step + analogous to IOS. Provided for API compatibility. + """ + log.debug("Host %s: enable() is a no-op on IOS-XR.", self.host) + + def check_file_exists(self, filename, file_system=None): + """Check whether a file exists on the device filesystem. + + Args: + filename (str): The filename to look for. + file_system (str, optional): Filesystem to inspect. Defaults to ``harddisk:``. + + Returns: + (bool): True if the file is present, False otherwise. + """ + if file_system is None: + file_system = DEFAULT_FILE_SYSTEM + + result = self.native.send_command(f"dir {file_system}/{filename}", read_timeout=30) + if re.search(r"No such file|No files matched|not found|Path does not exist|Error", result, re.IGNORECASE): + log.debug("Host %s: File %s does not exist on %s.", self.host, filename, file_system) + return False + if re.search(re.escape(filename), result): + log.debug("Host %s: File %s exists on %s.", self.host, filename, file_system) + return True + + log.debug("Host %s: File %s not found in 'dir' output on %s.", self.host, filename, file_system) + return False + + def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kwargs): + """Copy a file from a remote URL onto the device filesystem. + + Pulls the file specified by ``src`` from a remote server (FTP/TFTP/SCP/HTTP/HTTPS) + using the IOS-XR ``copy`` command and saves it to ``file_system`` (default + ``harddisk:``). The transfer is verified by confirming the file exists after + copy. **Checksum verification is not performed** on IOS-XR in this release; the + ``checksum`` on ``src`` is not validated. + + Args: + src (FileCopyModel): The source specification (URL, credentials, timeout). + dest (str, optional): Destination filename. Defaults to ``src.file_name``. + file_system (str, optional): Target filesystem. Defaults to ``harddisk:``. + kwargs (dict): Additional keyword arguments (unused). + + Raises: + TypeError: When ``src`` is not a ``FileCopyModel``. + FileTransferError: When the transfer fails or the file is absent afterward. + """ + if not isinstance(src, FileCopyModel): + raise TypeError("src must be an instance of FileCopyModel") + + if file_system is None: + file_system = DEFAULT_FILE_SYSTEM + if dest is None: + dest = src.file_name + + if self.check_file_exists(dest, file_system=file_system): + log.info("Host %s: File %s already present on %s; skipping copy.", self.host, dest, file_system) + return + + self._pre_transfer_space_check(src, file_system) + current_prompt = self.native.find_prompt() + + # Prompts IOS-XR may emit during an interactive copy. + prompt_answers = { + r"Destination filename": "", + r"Host name or IP address": "", + r"Source username|Username": src.username or "", + r"Password": src.token or "", + r"yes/no|\[confirm\]|Are you sure": "", + } + keys = list(prompt_answers.keys()) + [re.escape(current_prompt)] + expect_regex = f"({'|'.join(keys)})" + + command = f"copy {src.clean_url} {file_system}/{dest}" + if src.vrf and src.scheme not in {"http", "https"}: + command = f"{command} vrf {src.vrf}" + + # Bypass _send_command: a copy may emit benign "%" lines that are not failures. + output = self.native.send_command(command, expect_string=expect_regex, read_timeout=src.timeout) + + # Walk any interactive prompts. Netmiko strips the trailing prompt from the output, + # so the post-copy existence check (below) is the authoritative success signal; this + # loop only answers prompts and surfaces explicit error markers early. + for _ in range(10): + if RE_COPY_SUCCESS.search(output): + log.info("Host %s: File %s transfer reported success.", self.host, dest) + break + if RE_COPY_ERROR.search(output): + log.error("Host %s: File transfer error for %s: %s", self.host, dest, output) + raise FileTransferError + for prompt, answer in prompt_answers.items(): + if re.search(prompt, output, re.IGNORECASE): + is_password = prompt == r"Password" + output = self.native.send_command( + answer, expect_string=expect_regex, read_timeout=src.timeout, cmd_verify=not is_password + ) + break + else: + # No recognised prompt and no explicit marker; defer to the existence check. + break + + if not self.check_file_exists(dest, file_system=file_system): + log.error("Host %s: File %s not found after transfer.", self.host, dest) + raise FileTransferError + + log.info("Host %s: File %s copied to %s and verified present.", self.host, dest, file_system) + + @property + def hostname(self): + """Get the hostname of the device. + + Returns: + (str): The device hostname derived from the CLI prompt. + """ + if self._hostname is None: + prompt = self.native.find_prompt() + self._hostname = re.sub(r"^RP/\S+/CPU\d+:", "", prompt).strip().rstrip("#>") + return self._hostname + + def _image_booted(self, image_name, image_pattern=r"(\d+\.\d+\.\d+\w*)", **vendor_specifics): + image_match = re.search(image_pattern, image_name) + if image_match is None: + log.info("Host %s: Unable to parse a version from image %s.", self.host, image_name) + return False + image_version = image_match.group(1) + + booted_version = self.boot_options.get("version") + if booted_version is None: + version_data = self.show("show version") + version_match = RE_XR_VERSION.search(version_data) + booted_version = version_match.group(1) if version_match else None + + booted = booted_version == image_version + if booted: + log.info("Host %s: Image %s booted successfully.", self.host, image_name) + else: + log.info("Host %s: Image %s not booted (running %s).", self.host, image_name, booted_version) + return booted + + @property + def install_mode(self): + """Indicate whether the device is operating in install mode. + + eXR is always install-mode (there is no legacy boot-from-image state), + so this always returns ``True``. Provided for ``BaseDevice`` parity. + + Returns: + (bool): Always ``True``. + """ + return True + + def _get_free_space(self, file_system=None): + """Return free bytes on ``file_system`` as reported by ``dir`` output. + + Args: + file_system (str, optional): Target filesystem. Defaults to ``harddisk:``. + + Returns: + (int): Free bytes available on ``file_system``. + + Raises: + CommandError: When the free space cannot be parsed from ``dir`` output. + """ + if file_system is None: + file_system = DEFAULT_FILE_SYSTEM + + raw_data = self.show(f"dir {file_system}") + # eXR reports the trailer in kbytes (e.g. "9948012 kbytes total (9396256 kbytes free)"); + # other contexts may use plain bytes. Capture the unit and normalise to bytes. + match = re.search(r"\((\d+)\s+(k|m|g)?bytes\s+free", raw_data, re.IGNORECASE) + if match is None: + log.error("Host %s: could not parse free space from 'dir %s'.", self.host, file_system) + raise CommandError(command=f"dir {file_system}", message="Unable to parse free space from dir output.") + + multipliers = {"": 1, "k": 1024, "m": 1024**2, "g": 1024**3} + unit = (match.group(2) or "").lower() + free_bytes = int(match.group(1)) * multipliers[unit] + log.debug("Host %s: %s bytes free on %s.", self.host, free_bytes, file_system) + return free_bytes + + def install_os(self, image_name, reboot=True, additional_files=None, **vendor_specifics): + """Install the base IOS-XR ISO and verify the device boots into it. + + Orchestrates the eXR install workflow over the native primitives: + ``install add`` (ISO + any feature RPMs) -> poll for completion -> + ``install activate`` -> poll the activation operation -> wait for reboot -> + ``install commit`` -> verify. + + On eXR the base ISO cannot be activated on its own when feature packages + (IS-IS, OSPF, MPLS, etc.) are active: ``install activate`` aborts demanding the + matching-version RPMs be activated in the same operation. Pass those RPMs via + ``additional_files`` so the whole set is added and activated together. + + Args: + image_name (str): The base ISO filename already staged on ``harddisk:``. + reboot (bool): Must be ``True``; activation reloads the device automatically. + additional_files (list, optional): Feature RPM filenames (already staged on + ``harddisk:``) to add and activate alongside the base ISO. Required on any + device that runs optional feature packages. + vendor_specifics (dict, optional): Supports ``timeout`` (default 3600) for the + install-operation and reboot waits. + + Returns: + (bool): True if the install ran and succeeded, False if the device was + already running ``image_name``. + + Raises: + ValueError: When ``reboot`` is False (eXR always reloads during activation). + OSInstallError: When activation aborts or the device does not boot into + ``image_name`` after install. + """ + timeout = vendor_specifics.get("timeout", 3600) + + if self._image_booted(image_name): + log.info("Host %s: OS image %s already booted; nothing to install.", self.host, image_name) + return False + + if not reboot: + raise ValueError( + "IOS-XR devices reload automatically during 'install activate'; " + "the reboot argument cannot be set to False." + ) + + packages = [image_name, *(additional_files or [])] + add_id = self._install_add(f"{DEFAULT_FILE_SYSTEM}/", packages) + self._wait_for_install_op(add_id, timeout=timeout) + self._install_activate(add_id, timeout=timeout) + self._wait_for_device_reboot(timeout=timeout) + self._install_commit() + + if not self._image_booted(image_name): + log.error("Host %s: OS install error for image %s.", self.host, image_name) + raise OSInstallError(hostname=self.host, desired_boot=image_name) + + log.info("Host %s: OS image %s installed successfully.", self.host, image_name) + return True + + def open(self): + """Open a connection to the network device.""" + if self.connected: + try: + self.native.find_prompt() + except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught + self._connected = False + + if not self.connected: + self.native = ConnectHandler( + device_type="cisco_xr", + ip=self.host, + username=self.username, + password=self.password, + port=self.port, + read_timeout_override=self.read_timeout_override, + secret=self.secret, + # Keepalives let the status polls notice a dropped session promptly. + keepalive=30, + verbose=False, + ) + self._connected = True + + log.debug("Host %s: Connection to controller was opened successfully.", self.host) + + @property + def os_version(self): + """Get the running OS version from ``show version``. + + Returns: + (str): The version string (e.g. ``7.11.2``), or ``None`` if unparsable. + """ + if self._os_version is None: + version_data = self.show("show version") + match = RE_XR_VERSION.search(version_data) + self._os_version = match.group(1) if match else None + + log.debug("Host %s: OS version %s.", self.host, self._os_version) + return self._os_version + + def reboot(self, wait_for_reload=False, **kwargs): + """Reboot the device. + + Args: + wait_for_reload (bool): Whether to also run ``_wait_for_device_reboot``. Defaults to False. + kwargs (dict): Additional arguments to pass to Netmiko. + """ + if kwargs.get("confirm"): + log.warning("Passing 'confirm' to reboot method is deprecated.") + + try: + self.native.send_command_timing("reload") + # IOS-XR prompts "Proceed with reload?" — confirm with a newline. + try: + self.native.send_command_timing("\n", read_timeout=10) + except ReadTimeout as expected_exception: + log.info("Host %s: Device rebooted.", self.host) + log.info("Hit expected exception during reload: %s", expected_exception.__class__) + if wait_for_reload: + time.sleep(10) + self._wait_for_device_reboot() + except Exception as err: # noqa: BLE001 # pylint: disable=broad-exception-caught + log.error(err) + log.error(err.__class__) + + def set_boot_options(self, image_name, **vendor_specifics): + """Not supported on IOS-XR. + + eXR has no separate set-boot step: boot selection is performed atomically + by ``install_os`` via ``install activate`` + ``install commit``. + + Raises: + NotImplementedError: Always. + """ + raise NotImplementedError( + "IOS-XR has no separate set-boot step; boot selection is performed by install_os " + "via 'install activate' + 'install commit'." + ) + + def config(self, command, **netmiko_args): + """Not implemented for IOS-XR. + + Configuration management is out of scope for this OS-upgrade driver in the current + release; only the upgrade workflow is supported. + + Raises: + NotImplementedError: Always. + """ + raise NotImplementedError("config() is not implemented for the IOS-XR driver in this release.") + + def save(self, filename=None): + """Not supported on IOS-XR. + + eXR software state is committed atomically by ``install_os`` (via ``install commit``); + there is no standalone save step. + + Raises: + NotImplementedError: Always. + """ + raise NotImplementedError( + "IOS-XR has no standalone save; software is committed by install_os via 'install commit'." + ) + + def show(self, command, expect_string=None, **netmiko_args): + """Run a command on the device. + + IOS-XR EXEC mode is already privileged, so no enable step is performed. + + Args: + command (str|list): Command(s) to run. + expect_string (str, optional): Expected prompt string. Defaults to None. + netmiko_args (dict): Additional arguments passed to Netmiko's send_command. + + Returns: + (str|list): Command output; a list when ``command`` is a list. + + Raises: + CommandListError: When ``command`` is a list and one of the commands fails. + """ + if isinstance(command, list): + responses = [] + entered_commands = [] + for command_instance in command: + entered_commands.append(command_instance) + try: + responses.append(self._send_command(command_instance)) + except CommandError as e: + raise CommandListError(entered_commands, command_instance, e.cli_error_msg) + return responses + return self._send_command(command, expect_string=expect_string, **netmiko_args) + + @property + def uptime(self): + """Get uptime from the device. + + Returns: + (int): Uptime in seconds. + """ + if self._uptime is None: + version_data = self.show("show version") + match = re.search(r"uptime is (.+)", version_data) + uptime_full_string = match.group(1) if match else "" + self._uptime = self._uptime_to_seconds(uptime_full_string) + + log.debug("Host %s: Uptime %s.", self.host, self._uptime) + return self._uptime + + @property + def uptime_string(self): + """Get uptime in ``dd:hh:mm:ss`` format. + + Returns: + (str): Uptime of the device. + """ + if self._uptime_string is None: + version_data = self.show("show version") + match = re.search(r"uptime is (.+)", version_data) + uptime_full_string = match.group(1) if match else "" + self._uptime_string = self._uptime_to_string(uptime_full_string) + + return self._uptime_string diff --git a/tests/unit/test_devices/test_iosxr_device.py b/tests/unit/test_devices/test_iosxr_device.py new file mode 100644 index 00000000..4e7c62eb --- /dev/null +++ b/tests/unit/test_devices/test_iosxr_device.py @@ -0,0 +1,479 @@ +import unittest + +import mock + +from pyntc.devices import IOSXRDevice, supported_devices +from pyntc.devices import iosxr_device as iosxr_module +from pyntc.errors import FileTransferError +from pyntc.utils.models import FileCopyModel + +ISO = "ncs5k-mini-x-7.11.2.iso" +ACTIVE_VERSION = "7.11.2" +ISO_URL = "http://10.1.100.220/IOS-XR/7.11.2/ncs5k-mini-x-7.11.2.iso" +PROMPT = "RP/0/RP0/CPU0:ncs#" + +DIR_FILE_PRESENT = ( + "Mon Jun 15 12:00:00.000 UTC\n\n" + "Directory of harddisk:\n" + "15 -rw- 1500000000 Jun 15 12:00 ncs5k-mini-x-7.11.2.iso\n" +) +DIR_FILE_ABSENT = "Mon Jun 15 12:00:00.000 UTC\n%Error: dir: '/harddisk:/ncs5k-mini-x-7.11.2.iso': No such file\n" +COPY_SUCCESS = ( + "Mon Jun 15 12:00:00.000 UTC\n" + "Destination filename [/harddisk:/ncs5k-mini-x-7.11.2.iso]?\n" + "Accessing http://10.1.100.220/IOS-XR/7.11.2/ncs5k-mini-x-7.11.2.iso\n" + "1500000000 bytes copied in 42 secs (35714285 bytes/sec)\n" + "RP/0/RP0/CPU0:ncs#" +) +COPY_ERROR = ( + "Mon Jun 15 12:00:00.000 UTC\n" + "%Error opening http://10.1.100.220/IOS-XR/7.11.2/ncs5k-mini-x-7.11.2.iso: Connection refused\n" +) +# Real eXR (NCS5011) copy output: netmiko strips the trailing prompt, and the +# success markers are "Successfully copied ... Bytes" / "Copy operation success". +COPY_SUCCESS_EXR = ( + "\nAccessing http://10.1.100.220/IOS-XR/7.11.2/ncs5k-mini-x-7.11.2.iso\n" + + ("!" * 60) + + "\nSuccessfully copied 1432756224 Bytes\n\n\nCopy operation success\n" +) + +SHOW_INSTALL_ACTIVE_SINGLE = ( + "Node 0/RP0/CPU0 [RP]\n" + " Boot Partition: xr_lv0\n" + " Active Packages: 1\n" + " ncs5k-xr-7.11.2 version=7.11.2 [Boot image]\n" +) + +SHOW_INSTALL_ACTIVE_MULTI = ( + "Node 0/RP0/CPU0 [RP]\n" + " Active Packages: 8\n" + " ncs5k-xr-7.11.2 version=7.11.2 [Boot image]\n" + " ncs5k-isis-7.11.2\n" + " ncs5k-ospf-7.11.2\n" + " ncs5k-mpls-7.11.2\n" + " ncs5k-mpls-te-rsvp-7.11.2\n" + " ncs5k-mcast-7.11.2\n" + " ncs5k-m2m-7.11.2\n" + " ncs5k-mgbl-7.11.2\n" +) + +INSTALL_ADD_RESPONSE = ( + "Mon Jun 15 12:00:00.000 UTC\n" + "Install operation 17 started by admin:\n" + " install add source harddisk:/ ncs5k-mini-x-7.11.2.iso\n" + "This operation will continue asynchronously.\n" + "Install add operation 17 will continue in the background.\n" +) + +INSTALL_ADD_NO_OP_ID = "Mon Jun 15 12:00:00.000 UTC\n% Unexpected output without an operation id\n" + +SHOW_INSTALL_LOG_INPROGRESS = ( + "Install operation 17: 'install add source harddisk:/ ...' started\nAction 17 in progress\n" +) + +SHOW_INSTALL_LOG_SUCCESS = ( + "Install operation 17: 'install add source harddisk:/ ...' started\nInstall operation 17 completed successfully\n" +) + +SHOW_INSTALL_LOG_ABORT = ( + "Install operation 17: 'install add source harddisk:/ ...' started\nInstall operation 17 aborted\n" +) + +SHOW_VERSION = ( + "Cisco IOS XR Software, Version 7.11.2\n" + "Copyright (c) 2013-2024 by Cisco Systems, Inc.\n\n" + "cisco NCS-5011 () processor\n" + "System uptime is 1 week, 2 days, 3 hours, 4 minutes\n" +) + +DIR_HARDDISK = ( + "Mon Jun 15 12:00:00.000 UTC\n\n" + "Directory of harddisk:\n" + "15 -rw- 1500000000 Jun 15 12:00 ncs5k-mini-x-7.11.2.iso\n\n" + "3000000000 bytes total (2000000000 bytes free)\n" +) + +DIR_HARDDISK_LOW = "Mon Jun 15 12:00:00.000 UTC\n\nDirectory of harddisk:\n3000000000 bytes total (1000 bytes free)\n" + +# Real eXR (NCS5011) trailer reports kbytes, not bytes. +DIR_HARDDISK_KBYTES = ( + "Mon Jun 15 23:43:01.321 UTC\n\n" + "Directory of harddisk:\n" + " 13 drwxr-xr-x. 2 4096 Jun 15 20:22 .tmp\n" + " 12 -rw-r--r--. 1 382788 Jun 15 23:35 nvgen_bkup.log\n\n" + "9948012 kbytes total (9396256 kbytes free)\n" +) + +# 'show install request' states: add op (activation not finished), activate succeeded, activate failed. +SHOW_INSTALL_REQUEST_ADD = ( + "Tue Jun 16 02:15:50.891 UTC\n" + "No install operation in progress\n\n" + "Last operation performed:\n" + "Operation Id : 17\nRequest : Install add\nState : Success\n" +) +SHOW_INSTALL_REQUEST_ACTIVATE_SUCCESS = ( + "Tue Jun 16 02:20:00.000 UTC\n" + "No install operation in progress\n\n" + "Last operation performed:\n" + "Operation Id : 18\nRequest : Install activate\nState : Success\n" +) +SHOW_INSTALL_REQUEST_ACTIVATE_FAILURE = ( + "Tue Jun 16 02:20:00.000 UTC\n" + "No install operation in progress\n\n" + "Last operation performed:\n" + "Operation Id : 18\nRequest : Install activate\nState : Failure\n" +) + + +def _fake_clock(values): + """Return a time.time() stand-in that yields ``values`` then a large constant. + + The standard ``logging`` module also calls ``time.time()``, so a finite + ``side_effect`` list raises StopIteration unpredictably. This helper returns + a huge value once exhausted, keeping the polling-loop timeout logic + deterministic regardless of interleaved log calls. + """ + seq = list(values) + + def _inner(*args, **kwargs): + return seq.pop(0) if seq else 1e12 + + return _inner + + +class TestIOSXRDevice(unittest.TestCase): + @mock.patch.object(IOSXRDevice, "open") + @mock.patch.object(IOSXRDevice, "close") + def setUp(self, mock_close, mock_open): # pylint: disable=arguments-differ + self.device = IOSXRDevice("host", "user", "pass") + self.device.native = mock.MagicMock() + + def tearDown(self): + if self.device.native is not None: + self.device.native.reset_mock() + + # --- basics / registration --- + + def test_port(self): + self.assertEqual(self.device.port, 22) + + def test_device_type(self): + self.assertEqual(self.device.device_type, "cisco_iosxr_ssh") + + def test_registration(self): + self.assertIs(supported_devices["cisco_iosxr_ssh"], IOSXRDevice) + + # --- facts --- + + def test_os_version(self): + self.device.native.send_command.return_value = SHOW_VERSION + self.assertEqual(self.device.os_version, ACTIVE_VERSION) + + def test_uptime_parses_weeks(self): + self.device.native.send_command.return_value = SHOW_VERSION + # 1 week + 2 days + 3 hours + 4 minutes + expected = (7 * 86400) + (2 * 86400) + (3 * 3600) + (4 * 60) + self.assertEqual(self.device.uptime, expected) + + def test_uptime_string_folds_weeks_into_days(self): + self.device.native.send_command.return_value = SHOW_VERSION + # 1 week + 2 days -> 9 days, 3 hours, 4 minutes -> dd:hh:mm:ss + self.assertEqual(self.device.uptime_string, "09:03:04:00") + + def test_hostname_strips_rp_prefix(self): + self.device.native.find_prompt.return_value = "RP/0/RP0/CPU0:NCS5011-LAB#" + self.assertEqual(self.device.hostname, "NCS5011-LAB") + + # --- show / config / save --- + + def test_show_list_returns_list(self): + self.device.native.send_command.side_effect = ["out-a", "out-b"] + self.assertEqual(self.device.show(["show foo", "show bar"]), ["out-a", "out-b"]) + + def test_show_raises_command_error_on_error_response(self): + self.device.native.send_command.return_value = "% Invalid input detected" + with self.assertRaises(iosxr_module.CommandError): + self.device.show("show bogus") + + def test_show_list_raises_command_list_error(self): + self.device.native.send_command.side_effect = ["% Invalid input detected", "ok"] + with self.assertRaises(iosxr_module.CommandListError): + self.device.show(["show bad", "show good"]) + + def test_config_not_implemented(self): + with self.assertRaises(NotImplementedError): + self.device.config("hostname FOO") + + def test_save_not_implemented(self): + with self.assertRaises(NotImplementedError): + self.device.save() + + # --- boot_options / install_mode / set_boot_options --- + + def test_boot_options(self): + self.device.native.send_command.return_value = SHOW_INSTALL_ACTIVE_SINGLE + self.assertEqual(self.device.boot_options, {"sys": "ncs5k-xr-7.11.2", "version": "7.11.2"}) + + def test_boot_options_multi_package(self): + self.device.native.send_command.return_value = SHOW_INSTALL_ACTIVE_MULTI + self.assertEqual(self.device.boot_options, {"sys": "ncs5k-xr-7.11.2", "version": "7.11.2"}) + + def test_boot_options_none_when_unmatched(self): + self.device.native.send_command.return_value = "No active packages found" + self.assertEqual(self.device.boot_options, {"sys": None, "version": None}) + + def test_install_mode_always_true(self): + self.assertTrue(self.device.install_mode) + + def test_set_boot_options_not_implemented(self): + with self.assertRaises(NotImplementedError): + self.device.set_boot_options(ISO) + + # --- _image_booted --- + + def test_image_booted_true(self): + self.device.native.send_command.return_value = SHOW_INSTALL_ACTIVE_SINGLE + self.assertTrue(self.device._image_booted(ISO)) + + def test_image_booted_false(self): + self.device.native.send_command.return_value = SHOW_INSTALL_ACTIVE_SINGLE + self.assertFalse(self.device._image_booted("ncs5k-mini-x-7.10.1.iso")) + + # --- _get_free_space --- + + def test_get_free_space(self): + self.device.native.send_command.return_value = DIR_HARDDISK + self.assertEqual(self.device._get_free_space(), 2000000000) + + def test_get_free_space_default_file_system_is_harddisk(self): + self.device.native.send_command.return_value = DIR_HARDDISK + self.device._get_free_space() + self.device.native.send_command.assert_any_call(command_string="dir harddisk:") + + def test_get_free_space_kbytes_units(self): + self.device.native.send_command.return_value = DIR_HARDDISK_KBYTES + self.assertEqual(self.device._get_free_space(), 9396256 * 1024) + + def test_get_free_space_unparsable_raises(self): + self.device.native.send_command.return_value = "garbage output" + with self.assertRaises(iosxr_module.CommandError): + self.device._get_free_space() + + # --- async install primitives --- + + def test_install_add_parses_op_id(self): + self.device.native.send_command.return_value = INSTALL_ADD_RESPONSE + self.assertEqual(self.device._install_add("harddisk:/", [ISO]), 17) + + def test_install_add_no_op_id_raises(self): + self.device.native.send_command.return_value = INSTALL_ADD_NO_OP_ID + with self.assertRaises(iosxr_module.OSInstallError): + self.device._install_add("harddisk:/", [ISO]) + + @mock.patch("pyntc.devices.iosxr_device.time.sleep") + def test_wait_for_install_op_success(self, mock_sleep): + self.device.native.send_command.side_effect = [ + SHOW_INSTALL_LOG_INPROGRESS, + SHOW_INSTALL_LOG_INPROGRESS, + SHOW_INSTALL_LOG_SUCCESS, + ] + self.device._wait_for_install_op(17) + self.assertEqual(self.device.native.send_command.call_count, 3) + + @mock.patch("pyntc.devices.iosxr_device.time.sleep") + def test_wait_for_install_op_abort_raises(self, mock_sleep): + self.device.native.send_command.return_value = SHOW_INSTALL_LOG_ABORT + with self.assertRaises(iosxr_module.OSInstallError): + self.device._wait_for_install_op(17) + + @mock.patch("pyntc.devices.iosxr_device.time.sleep") + @mock.patch("pyntc.devices.iosxr_device.time.time", side_effect=_fake_clock([0, 0])) + def test_wait_for_install_op_timeout_raises(self, mock_time, mock_sleep): + self.device.native.send_command.return_value = SHOW_INSTALL_LOG_INPROGRESS + with self.assertRaises(iosxr_module.OSInstallError): + self.device._wait_for_install_op(17, timeout=3600) + + @mock.patch("pyntc.devices.iosxr_device.time.sleep") + def test_install_activate_issues_async_and_polls_until_finished(self, mock_sleep): + self.device.native.send_command_timing.return_value = "Install operation 18 started by ntc" + self.device.native.send_command.return_value = SHOW_INSTALL_REQUEST_ACTIVATE_SUCCESS + self.device._install_activate(17) + self.device.native.send_command_timing.assert_any_call("install activate id 17 noprompt", read_timeout=180) + + @mock.patch("pyntc.devices.iosxr_device.time.sleep") + def test_install_activate_raises_on_failure(self, mock_sleep): + self.device.native.send_command_timing.return_value = "Install operation 18 started by ntc" + self.device.native.send_command.return_value = SHOW_INSTALL_REQUEST_ACTIVATE_FAILURE + with self.assertRaises(iosxr_module.OSInstallError): + self.device._install_activate(17) + + @mock.patch("pyntc.devices.iosxr_device.time.sleep") + def test_install_activate_tolerates_session_drop(self, mock_sleep): + # The reload drops the session while polling: that is the success signal, not an error. + self.device.native.send_command_timing.return_value = "Install operation 18 started by ntc" + self.device.native.send_command.side_effect = OSError("socket closed") + self.device._install_activate(17) # must not raise + + @mock.patch("pyntc.devices.iosxr_device.time.sleep") + @mock.patch("pyntc.devices.iosxr_device.time.time", side_effect=_fake_clock([0, 0])) + def test_install_activate_timeout_raises(self, mock_time, mock_sleep): + # Activation never finishes (status keeps showing only the add op) -> timeout -> raise. + self.device.native.send_command_timing.return_value = "Install operation 18 started by ntc" + self.device.native.send_command.return_value = SHOW_INSTALL_REQUEST_ADD + with self.assertRaises(iosxr_module.OSInstallError): + self.device._install_activate(17, timeout=3600) + + def test_install_commit(self): + self.device.native.send_command.return_value = "Install operation 18 completed successfully" + self.device._install_commit() + self.device.native.send_command.assert_any_call("install commit") + + # --- reboot / _wait_for_device_reboot --- + + def test_reboot(self): + self.device.reboot() + self.device.native.send_command_timing.assert_any_call("reload") + + @mock.patch("pyntc.devices.iosxr_device.time.sleep") + @mock.patch.object(IOSXRDevice, "show") + @mock.patch.object(IOSXRDevice, "open", side_effect=[None, OSError("down"), None]) + @mock.patch.object(IOSXRDevice, "close") + def test_wait_for_device_reboot(self, mock_close, mock_open, mock_show, mock_sleep): + # Reachable -> disconnect (reload) -> back up: requires the drop-then-recover transition. + self.device._wait_for_device_reboot(timeout=600) + self.assertEqual(mock_open.call_count, 3) + + @mock.patch("pyntc.devices.iosxr_device.time.sleep") + @mock.patch("pyntc.devices.iosxr_device.time.time", side_effect=_fake_clock([0, 0])) + @mock.patch.object(IOSXRDevice, "open", side_effect=OSError("down")) + @mock.patch.object(IOSXRDevice, "close") + def test_wait_for_device_reboot_timeout(self, mock_close, mock_open, mock_time, mock_sleep): + with self.assertRaises(iosxr_module.RebootTimeoutError): + self.device._wait_for_device_reboot(timeout=3600) + + # --- install_os orchestration --- + + @mock.patch.object(IOSXRDevice, "_install_commit") + @mock.patch.object(IOSXRDevice, "_wait_for_device_reboot") + @mock.patch.object(IOSXRDevice, "_install_activate", return_value=18) + @mock.patch.object(IOSXRDevice, "_wait_for_install_op") + @mock.patch.object(IOSXRDevice, "_install_add", return_value=17) + @mock.patch.object(IOSXRDevice, "uptime", new_callable=mock.PropertyMock, return_value=1000) + @mock.patch.object(IOSXRDevice, "_image_booted", side_effect=[False, True]) + def test_install_os( + self, mock_booted, mock_uptime, mock_add, mock_wait_op, mock_activate, mock_wait_reboot, mock_commit + ): + result = self.device.install_os(ISO) + self.assertTrue(result) + mock_add.assert_called_once_with("harddisk:/", [ISO]) + mock_wait_op.assert_called_once_with(17, timeout=3600) # add op only + mock_activate.assert_called_once_with(17, timeout=3600) + mock_wait_reboot.assert_called_once_with(timeout=3600) + mock_commit.assert_called_once() + + @mock.patch.object(IOSXRDevice, "_install_commit") + @mock.patch.object(IOSXRDevice, "_wait_for_device_reboot") + @mock.patch.object(IOSXRDevice, "_install_activate", return_value=18) + @mock.patch.object(IOSXRDevice, "_wait_for_install_op") + @mock.patch.object(IOSXRDevice, "_install_add", return_value=17) + @mock.patch.object(IOSXRDevice, "uptime", new_callable=mock.PropertyMock, return_value=1000) + @mock.patch.object(IOSXRDevice, "_image_booted", side_effect=[False, True]) + def test_install_os_with_additional_files( + self, mock_booted, mock_uptime, mock_add, mock_wait_op, mock_activate, mock_wait_reboot, mock_commit + ): + rpms = ["ncs5k-mpls-7.11.2.rpm", "ncs5k-ospf-7.11.2.rpm"] + result = self.device.install_os(ISO, additional_files=rpms) + self.assertTrue(result) + # The base ISO and the feature RPMs are added together as a single set. + mock_add.assert_called_once_with("harddisk:/", [ISO, *rpms]) + + @mock.patch.object(IOSXRDevice, "_install_add") + @mock.patch.object(IOSXRDevice, "_image_booted", return_value=True) + def test_install_os_already_installed(self, mock_booted, mock_add): + result = self.device.install_os(ISO) + self.assertFalse(result) + mock_add.assert_not_called() + + @mock.patch.object(IOSXRDevice, "_image_booted", return_value=False) + def test_install_os_reboot_false_raises(self, mock_booted): + with self.assertRaises(ValueError): + self.device.install_os(ISO, reboot=False) + + @mock.patch.object(IOSXRDevice, "_install_commit") + @mock.patch.object(IOSXRDevice, "_wait_for_device_reboot") + @mock.patch.object(IOSXRDevice, "_install_activate", return_value=18) + @mock.patch.object(IOSXRDevice, "_wait_for_install_op") + @mock.patch.object(IOSXRDevice, "_install_add", return_value=17) + @mock.patch.object(IOSXRDevice, "uptime", new_callable=mock.PropertyMock, return_value=1000) + @mock.patch.object(IOSXRDevice, "_image_booted", side_effect=[False, False]) + def test_install_os_verify_failure_raises( + self, mock_booted, mock_uptime, mock_add, mock_wait_op, mock_activate, mock_wait_reboot, mock_commit + ): + with self.assertRaises(iosxr_module.OSInstallError): + self.device.install_os(ISO) + + # --- check_file_exists --- + + def test_check_file_exists_true(self): + self.device.native.send_command.return_value = DIR_FILE_PRESENT + self.assertTrue(self.device.check_file_exists(ISO)) + + def test_check_file_exists_false(self): + self.device.native.send_command.return_value = DIR_FILE_ABSENT + self.assertFalse(self.device.check_file_exists(ISO)) + + # --- remote_file_copy --- + + def test_remote_file_copy_requires_model(self): + with self.assertRaises(TypeError): + self.device.remote_file_copy(ISO_URL) + + @mock.patch.object(IOSXRDevice, "check_file_exists", side_effect=[False, True]) + def test_remote_file_copy_success(self, mock_exists): + self.device.native.find_prompt.return_value = PROMPT + self.device.native.send_command.return_value = COPY_SUCCESS + src = FileCopyModel(download_url=ISO_URL, checksum="", file_name=ISO) + + self.device.remote_file_copy(src) + + copy_calls = [ + call + for call in self.device.native.send_command.call_args_list + if call.args and call.args[0].startswith("copy ") + ] + self.assertTrue(copy_calls) + self.assertEqual(copy_calls[0].args[0], f"copy {ISO_URL} harddisk:/{ISO}") + + @mock.patch.object(IOSXRDevice, "check_file_exists", return_value=True) + def test_remote_file_copy_idempotent_when_present(self, mock_exists): + self.device.native.find_prompt.return_value = PROMPT + src = FileCopyModel(download_url=ISO_URL, checksum="", file_name=ISO) + + self.device.remote_file_copy(src) + + copy_calls = [ + call + for call in self.device.native.send_command.call_args_list + if call.args and call.args[0].startswith("copy ") + ] + self.assertEqual(copy_calls, []) + + @mock.patch.object(IOSXRDevice, "check_file_exists", side_effect=[False, True]) + def test_remote_file_copy_success_exr_output(self, mock_exists): + # Real eXR success output (no trailing prompt, "Successfully copied"/"Copy operation success"). + self.device.native.find_prompt.return_value = PROMPT + self.device.native.send_command.return_value = COPY_SUCCESS_EXR + src = FileCopyModel(download_url=ISO_URL, checksum="", file_name=ISO) + + self.device.remote_file_copy(src) # must not raise + + self.assertEqual(mock_exists.call_count, 2) # idempotency check + post-copy verify + + @mock.patch.object(IOSXRDevice, "check_file_exists", side_effect=[False]) + def test_remote_file_copy_error_raises(self, mock_exists): + self.device.native.find_prompt.return_value = PROMPT + self.device.native.send_command.return_value = COPY_ERROR + src = FileCopyModel(download_url=ISO_URL, checksum="", file_name=ISO) + + with self.assertRaises(FileTransferError): + self.device.remote_file_copy(src) diff --git a/tests/unit/test_infra.py b/tests/unit/test_infra.py index 881bd12a..97e9ca79 100644 --- a/tests/unit/test_infra.py +++ b/tests/unit/test_infra.py @@ -15,11 +15,14 @@ @mock.patch("pyntc.devices.f5_device.ManagementRoot") @mock.patch("pyntc.devices.asa_device.ASADevice.open") @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.jnpr_device.JunosNativeSW") @mock.patch("pyntc.devices.jnpr_device.JunosNativeDevice.open") @mock.patch("pyntc.devices.jnpr_device.JunosNativeDevice.timeout") -def test_device_creation(j_timeout, j_open, j_nsw, nx_open, i_open, a_open, f_mr, air_open, device_type, expected): +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 +): device = ntc_device(device_type, "host", "user", "pass") assert isinstance(device, expected) From e3c46fb8a7d17b1d72b2d72053dcd87d4265fd03 Mon Sep 17 00:00:00 2001 From: James Williams Date: Tue, 16 Jun 2026 00:01:53 -0500 Subject: [PATCH 2/7] updates to iosxr driver --- pyntc/devices/iosxr_device.py | 29 +++++------ tests/unit/test_devices/test_iosxr_device.py | 53 +++++++++----------- 2 files changed, 39 insertions(+), 43 deletions(-) diff --git a/pyntc/devices/iosxr_device.py b/pyntc/devices/iosxr_device.py index 5f0ef35d..9a26538b 100644 --- a/pyntc/devices/iosxr_device.py +++ b/pyntc/devices/iosxr_device.py @@ -175,10 +175,12 @@ def _install_activate(self, op_id, poll_interval=60, timeout=3600): the install status. A synchronous activate would hold the session and, when the reload tore the device down, trap the read on a half-open socket until ``read_timeout``. - The activation creates its own operation id (distinct from the ``install add`` id). - This method polls ``show install request`` once per ``poll_interval`` — logging each - poll — until that operation finishes successfully, then returns so the caller can wait - for the reload. + For an ISO upgrade the activation always ends in a reload, so "success" manifests as + ``show install request`` reporting *completed, pending reload* (or the SSH session + dropping as the reload starts) — not a committed ``State : Success`` (the device + reloads before that appears). This method polls ``show install request`` once per + ``poll_interval`` — logging each poll — and returns when it sees the pending-reload + marker or the session drops, and raises if the operation reports an abort/error. Args: op_id (int): The staged ``install add`` operation id to activate. @@ -204,16 +206,15 @@ def _install_activate(self, op_id, poll_interval=60, timeout=3600): except Exception as exc: # noqa: BLE001 # pylint: disable=broad-exception-caught log.info("Host %s: session dropped while polling activation (reload underway): %s", self.host, exc) return - op_ids = [int(match) for match in re.findall(r"Operation Id\s*:\s*(\d+)", request)] - newest = max(op_ids) if op_ids else op_id - log.info("Host %s: polled activation status (latest operation %s).", self.host, newest) - if newest > op_id and re.search(r"install activate", request, re.IGNORECASE): - if re.search(r"State\s*:\s*Failure|aborted", request, re.IGNORECASE): - log.error("Host %s: activation operation %s failed: %s", self.host, newest, request) - raise OSInstallError(hostname=self.host, desired_boot=f"operation {newest}") - if re.search(r"State\s*:\s*Success", request, re.IGNORECASE): - log.info("Host %s: activation operation %s finished successfully.", self.host, newest) - return + log.info("Host %s: polled activation status.", self.host) + if re.search(r"abort|Error[:!]", request, re.IGNORECASE): + log.error("Host %s: activation of operation %s failed: %s", self.host, op_id, request) + raise OSInstallError(hostname=self.host, desired_boot=f"operation {op_id}") + if re.search( + r"completed, pending reload|finished successfully|completed successfully", request, re.IGNORECASE + ): + log.info("Host %s: activation completed; reload imminent.", self.host) + return log.error("Host %s: activation of operation %s did not finish within %ss.", self.host, op_id, timeout) raise OSInstallError(hostname=self.host, desired_boot=f"operation {op_id}") diff --git a/tests/unit/test_devices/test_iosxr_device.py b/tests/unit/test_devices/test_iosxr_device.py index 4e7c62eb..0f0dcc0d 100644 --- a/tests/unit/test_devices/test_iosxr_device.py +++ b/tests/unit/test_devices/test_iosxr_device.py @@ -104,24 +104,19 @@ "9948012 kbytes total (9396256 kbytes free)\n" ) -# 'show install request' states: add op (activation not finished), activate succeeded, activate failed. -SHOW_INSTALL_REQUEST_ADD = ( - "Tue Jun 16 02:15:50.891 UTC\n" - "No install operation in progress\n\n" - "Last operation performed:\n" - "Operation Id : 17\nRequest : Install add\nState : Success\n" -) -SHOW_INSTALL_REQUEST_ACTIVATE_SUCCESS = ( - "Tue Jun 16 02:20:00.000 UTC\n" - "No install operation in progress\n\n" - "Last operation performed:\n" - "Operation Id : 18\nRequest : Install activate\nState : Success\n" +# 'show install request' states observed on real eXR during an activation. +SHOW_INSTALL_REQUEST_IN_PROGRESS = ( + "Tue Jun 16 03:31:04.338 UTC\n" + "User ntc, Op Id 26\ninstall activate\nncs5k-mini-x-7.11.2\n" + "install operation 26 is in progress\n" + "Install prepare operation 26 is in progress\n" + "0/RP0 In Progress Partition preparation in progress\n" ) +SHOW_INSTALL_REQUEST_PENDING_RELOAD = "Tue Jun 16 03:35:12.838 UTC\nInstall operation completed, pending reload\n" SHOW_INSTALL_REQUEST_ACTIVATE_FAILURE = ( - "Tue Jun 16 02:20:00.000 UTC\n" - "No install operation in progress\n\n" - "Last operation performed:\n" - "Operation Id : 18\nRequest : Install activate\nState : Failure\n" + "Tue Jun 16 00:29:55.000 UTC\n" + "Error: An exception is hit while executing the install operation.\n" + "Install operation 26 aborted\n" ) @@ -294,34 +289,34 @@ def test_wait_for_install_op_timeout_raises(self, mock_time, mock_sleep): self.device._wait_for_install_op(17, timeout=3600) @mock.patch("pyntc.devices.iosxr_device.time.sleep") - def test_install_activate_issues_async_and_polls_until_finished(self, mock_sleep): - self.device.native.send_command_timing.return_value = "Install operation 18 started by ntc" - self.device.native.send_command.return_value = SHOW_INSTALL_REQUEST_ACTIVATE_SUCCESS - self.device._install_activate(17) - self.device.native.send_command_timing.assert_any_call("install activate id 17 noprompt", read_timeout=180) + def test_install_activate_issues_async_and_returns_on_pending_reload(self, mock_sleep): + self.device.native.send_command_timing.return_value = "Install operation 26 started by ntc" + self.device.native.send_command.return_value = SHOW_INSTALL_REQUEST_PENDING_RELOAD + self.device._install_activate(25) + self.device.native.send_command_timing.assert_any_call("install activate id 25 noprompt", read_timeout=180) @mock.patch("pyntc.devices.iosxr_device.time.sleep") def test_install_activate_raises_on_failure(self, mock_sleep): - self.device.native.send_command_timing.return_value = "Install operation 18 started by ntc" + self.device.native.send_command_timing.return_value = "Install operation 26 started by ntc" self.device.native.send_command.return_value = SHOW_INSTALL_REQUEST_ACTIVATE_FAILURE with self.assertRaises(iosxr_module.OSInstallError): - self.device._install_activate(17) + self.device._install_activate(25) @mock.patch("pyntc.devices.iosxr_device.time.sleep") def test_install_activate_tolerates_session_drop(self, mock_sleep): # The reload drops the session while polling: that is the success signal, not an error. - self.device.native.send_command_timing.return_value = "Install operation 18 started by ntc" + self.device.native.send_command_timing.return_value = "Install operation 26 started by ntc" self.device.native.send_command.side_effect = OSError("socket closed") - self.device._install_activate(17) # must not raise + self.device._install_activate(25) # must not raise @mock.patch("pyntc.devices.iosxr_device.time.sleep") @mock.patch("pyntc.devices.iosxr_device.time.time", side_effect=_fake_clock([0, 0])) def test_install_activate_timeout_raises(self, mock_time, mock_sleep): - # Activation never finishes (status keeps showing only the add op) -> timeout -> raise. - self.device.native.send_command_timing.return_value = "Install operation 18 started by ntc" - self.device.native.send_command.return_value = SHOW_INSTALL_REQUEST_ADD + # Activation never reaches pending-reload (status stays in progress) -> timeout -> raise. + self.device.native.send_command_timing.return_value = "Install operation 26 started by ntc" + self.device.native.send_command.return_value = SHOW_INSTALL_REQUEST_IN_PROGRESS with self.assertRaises(iosxr_module.OSInstallError): - self.device._install_activate(17, timeout=3600) + self.device._install_activate(25, timeout=3600) def test_install_commit(self): self.device.native.send_command.return_value = "Install operation 18 completed successfully" From d44b5513afdbc0d6d6e98b014c2f47f66c9edfef Mon Sep 17 00:00:00 2001 From: James Williams Date: Tue, 16 Jun 2026 13:57:01 -0500 Subject: [PATCH 3/7] update the xr driver to only work with golden iso images --- .gitignore | 3 - changes/398.added | 1 + docs/user/lib_overview.md | 64 ++++++++----- pyntc/devices/iosxr_device.py | 99 +++++++++++++------- tests/unit/test_devices/test_iosxr_device.py | 63 +++++++------ 5 files changed, 139 insertions(+), 91 deletions(-) create mode 100644 changes/398.added diff --git a/.gitignore b/.gitignore index f8b76915..8141a6a9 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,3 @@ -# Local IOS-XR lab harness (not part of the package) -/test-iosxr.py - # Ansible Retry Files *.retry diff --git a/changes/398.added b/changes/398.added new file mode 100644 index 00000000..e01c13de --- /dev/null +++ b/changes/398.added @@ -0,0 +1 @@ +Added native Cisco IOS-XR (eXR / 64-bit) support via the `cisco_iosxr_ssh` driver (`IOSXRDevice`), including `remote_file_copy` (FTP/TFTP/SCP/HTTP/HTTPS) and OS upgrades from a single golden ISO (built with Cisco's gisobuild tool) through the asynchronous `install add` / `install activate` / `install commit` workflow. diff --git a/docs/user/lib_overview.md b/docs/user/lib_overview.md index b899bdac..6a7770e4 100644 --- a/docs/user/lib_overview.md +++ b/docs/user/lib_overview.md @@ -21,35 +21,53 @@ It's main purpose is to simplify the execution of common tasks including: - Juniper Junos - uses PyEz (NETCONF) - F5 Networks - uses f5-sdk (ReST) -!!! note "IOS-XR upgrades: install the base ISO plus matching feature RPMs" +!!! note "IOS-XR upgrades require a golden ISO" The `cisco_iosxr_ssh` driver (`IOSXRDevice`) performs eXR OS upgrades using the asynchronous native install workflow (`install add` → poll → `install activate` → poll → reload → `install commit` → verify). - On eXR the base ISO **cannot be activated on its own** when optional feature - packages (IS-IS, OSPF, MPLS, multicast, etc.) are active: `install activate` - aborts demanding the matching-version RPMs be activated in the same operation. - Pass those RPMs via the `additional_files` argument to `install_os` so the base - ISO and the RPMs are added and activated together: + The driver upgrades from a single **golden ISO**. On eXR a bare base ISO + **cannot be activated on its own** when optional feature packages (IS-IS, OSPF, + MPLS, multicast, etc.) are active: `install activate` aborts demanding the + matching-version RPMs be activated in the same operation. A golden ISO solves + this by bundling the base XR image and the matching feature RPMs into one file, + so it activates cleanly without supplying separate packages: ```python - device.install_os( - "ncs5k-mini-x-7.11.2.iso", - additional_files=[ - "ncs5k-isis-2.1.0.0-r7112.x86_64.rpm", - "ncs5k-ospf-2.1.0.0-r7112.x86_64.rpm", - # ... the remaining matching-version feature RPMs - ], - ) + device.install_os("ncs5k-golden-x-7.11.2-NTC7112.iso") ``` - All files (ISO + RPMs) must already be staged on `harddisk:` (use - `remote_file_copy` for each). Omitting the RPMs on a device that runs feature - packages will cause `install activate` to abort. + The golden ISO must already be staged on `harddisk:` (use `remote_file_copy`). + Installing a base ISO plus separate feature RPMs is **not** supported by this + driver. -!!! warning "Nautobot OS Upgrades passes a single image today" - The Nautobot OS Upgrades `InstallOsJob` currently passes a single image to - `install_os` and does not yet supply `additional_files`. Driving an eXR upgrade - through that app will therefore attempt an ISO-only activation and abort on any - device with feature packages. Multi-file support in OS Upgrades is tracked as - separate follow-up work. +!!! tip "Building a golden ISO with gisobuild" + Build a golden ISO with Cisco's [gisobuild](https://github.com/ios-xr/gisobuild) + tool. Provide the platform's base (`mini`) ISO, a repository of the matching + feature RPMs you want bundled, a label (which becomes part of the resulting + filename), and an output directory: + + ```bash + ntc@linux-server:~/xr/gisobuild$ mkdir -p /home/ntc/xr/7.11.1/giso_out + + ./src/gisobuild.py \ + --iso /home/ntc/xr/7.11.1/ncs5k-mini-x-7.11.1.iso \ + --repo /home/ntc/xr/7.11.1/ \ + --pkglist ncs5k-isis-1.0.0.0-r7111.x86_64.rpm \ + ncs5k-ospf-1.0.0.0-r7111.x86_64.rpm \ + ncs5k-mpls-1.0.0.0-r7111.x86_64.rpm \ + ncs5k-mpls-te-rsvp-1.0.0.0-r7111.x86_64.rpm \ + ncs5k-mcast-1.0.0.0-r7111.x86_64.rpm \ + ncs5k-mgbl-1.0.0.0-r7111.x86_64.rpm \ + ncs5k-m2m-1.0.0.0-r7111.x86_64.rpm \ + --label NTC711 \ + --out-directory /home/ntc/xr/7.11.1/giso_out \ + --create-checksum \ + --clean \ + --docker + ``` + + On success gisobuild writes the golden ISO (e.g. + `ncs5k-golden-x-7.11.1-NTC711.iso`) into the output directory. Publish that file + to the server `remote_file_copy` pulls from, then pass its filename to + `install_os`. diff --git a/pyntc/devices/iosxr_device.py b/pyntc/devices/iosxr_device.py index 9a26538b..3b5784a5 100644 --- a/pyntc/devices/iosxr_device.py +++ b/pyntc/devices/iosxr_device.py @@ -1,15 +1,16 @@ """Module for using a Cisco IOS-XR (eXR / 64-bit) device over SSH. This driver targets 64-bit IOS-XR (eXR) platforms (initial target: NCS5000 / -NCS-5011) and implements the asynchronous, package-based OS upgrade workflow: +NCS-5011) and implements the asynchronous OS upgrade workflow: install add -> poll for completion -> install activate -> reload -> install commit -> verify -Phase 1 installs the base ISO only. The ``additional_files`` argument to -``install_os`` is accepted for forward compatibility but is currently inert: -when supplied it logs a warning and the files are ignored, and only the base -ISO is installed. Feature RPMs are therefore not installed and must be handled -manually until full-bundle support lands. +The driver upgrades from a single **golden ISO** image. A golden ISO bundles the +base XR image together with the matching-version feature RPMs (IS-IS, OSPF, MPLS, +multicast, etc.) into one file, so it can be added and activated on its own — eXR +will not abort the activation demanding separate feature RPMs. Build a golden ISO +with Cisco's gisobuild tool (https://github.com/ios-xr/gisobuild). Installing a +bare base ISO plus separate feature RPMs is not supported by this driver. """ import re @@ -111,15 +112,15 @@ def _uptime_to_string(self, uptime_full_string): days = days + weeks * 7 return f"{days:02d}:{hours:02d}:{minutes:02d}:00" - def _install_add(self, source, packages): - """Stage packages into the install repository. + def _install_add(self, source, image_name): + """Stage the golden ISO into the install repository. ``install add`` returns immediately and continues asynchronously in the background, printing an operation id that subsequent steps reference. Args: source (str): The on-device source path, e.g. ``harddisk:/``. - packages (list): Package filenames to add (the base ISO for Phase 1). + image_name (str): The golden ISO filename to add. Returns: (int): The install operation id parsed from the device response. @@ -127,13 +128,13 @@ def _install_add(self, source, packages): Raises: OSInstallError: When the device response does not contain an operation id. """ - command = f"install add source {source} {' '.join(packages)}" + command = f"install add source {source} {image_name}" response = self.native.send_command(command, read_timeout=120) match = RE_INSTALL_OP.search(response) if match is None: log.error("Host %s: Unable to parse install operation id from response: %s", self.host, response) - raise OSInstallError(hostname=self.host, desired_boot=" ".join(packages)) + raise OSInstallError(hostname=self.host, desired_boot=image_name) op_id = int(match.group(1)) log.info("Host %s: install add started operation %s.", self.host, op_id) @@ -219,10 +220,45 @@ def _install_activate(self, op_id, poll_interval=60, timeout=3600): log.error("Host %s: activation of operation %s did not finish within %ss.", self.host, op_id, timeout) raise OSInstallError(hostname=self.host, desired_boot=f"operation {op_id}") - def _install_commit(self): - """Persist the activated software so it survives the reload.""" - self.native.send_command("install commit") - log.info("Host %s: install commit issued.", self.host) + def _install_commit(self, retries=3, retry_delay=30, read_timeout=120): + """Persist the activated software so it survives future reloads. + + Issued immediately after the activation reload, where the install manager can be slow + to respond, so the command uses a generous ``read_timeout`` and is retried a few times + on failure. Re-issuing ``install commit`` when there is nothing left to commit is a + harmless no-op, so retrying is safe even if a prior attempt actually committed but the + prompt was slow to return. + + Args: + retries (int): Number of attempts before giving up. Defaults to 3. + retry_delay (int): Seconds to wait between attempts. Defaults to 30. + read_timeout (int): Per-attempt Netmiko read timeout. Defaults to 120. + + Raises: + OSInstallError: When the commit does not complete cleanly after ``retries`` attempts. + """ + for attempt in range(1, retries + 1): + try: + self.native.send_command("install commit", read_timeout=read_timeout) + log.info("Host %s: install commit issued.", self.host) + return + except Exception as exc: # noqa: BLE001 # pylint: disable=broad-exception-caught + log.warning( + "Host %s: install commit attempt %s/%s did not return cleanly (%s).", + self.host, + attempt, + retries, + exc, + ) + if attempt < retries: + time.sleep(retry_delay) + try: + self.open() + except Exception as open_exc: # noqa: BLE001 # pylint: disable=broad-exception-caught + log.debug("Host %s: reopen before commit retry failed (%s).", self.host, open_exc) + + log.error("Host %s: install commit did not complete after %s attempts.", self.host, retries) + raise OSInstallError(hostname=self.host, desired_boot="install commit") def _wait_for_device_reboot(self, timeout=3600, interval=60): """Wait for the activation reload: watch the session drop, then poll until it returns. @@ -282,10 +318,8 @@ def boot_options(self): """ show_install_active = self.show("show install active") match = RE_XR_BOOT_IMAGE.search(show_install_active) - if match: - boot_options = {"sys": match.group("sys"), "version": match.group("version")} - else: - boot_options = {"sys": None, "version": None} + # The regex's named groups are exactly "sys" and "version". + boot_options = match.groupdict() if match else {"sys": None, "version": None} log.debug("Host %s: boot options %s.", self.host, boot_options) return boot_options @@ -494,25 +528,23 @@ def _get_free_space(self, file_system=None): log.debug("Host %s: %s bytes free on %s.", self.host, free_bytes, file_system) return free_bytes - def install_os(self, image_name, reboot=True, additional_files=None, **vendor_specifics): - """Install the base IOS-XR ISO and verify the device boots into it. + def install_os(self, image_name, reboot=True, **vendor_specifics): + """Install a golden IOS-XR ISO and verify the device boots into it. Orchestrates the eXR install workflow over the native primitives: - ``install add`` (ISO + any feature RPMs) -> poll for completion -> - ``install activate`` -> poll the activation operation -> wait for reboot -> - ``install commit`` -> verify. + ``install add`` (golden ISO) -> poll for completion -> ``install activate`` -> + poll the activation operation -> wait for reboot -> ``install commit`` -> verify. - On eXR the base ISO cannot be activated on its own when feature packages - (IS-IS, OSPF, MPLS, etc.) are active: ``install activate`` aborts demanding the - matching-version RPMs be activated in the same operation. Pass those RPMs via - ``additional_files`` so the whole set is added and activated together. + ``image_name`` must be a **golden ISO** that already bundles the base XR image + and the matching-version feature RPMs (IS-IS, OSPF, MPLS, multicast, etc.). A + bare base ISO cannot be activated on its own when feature packages are active — + eXR aborts the activation demanding the matching RPMs — so build a golden ISO + with Cisco's gisobuild tool (https://github.com/ios-xr/gisobuild) and stage that + single file. Installing a base ISO plus separate feature RPMs is not supported. Args: - image_name (str): The base ISO filename already staged on ``harddisk:``. + image_name (str): The golden ISO filename already staged on ``harddisk:``. reboot (bool): Must be ``True``; activation reloads the device automatically. - additional_files (list, optional): Feature RPM filenames (already staged on - ``harddisk:``) to add and activate alongside the base ISO. Required on any - device that runs optional feature packages. vendor_specifics (dict, optional): Supports ``timeout`` (default 3600) for the install-operation and reboot waits. @@ -537,8 +569,7 @@ def install_os(self, image_name, reboot=True, additional_files=None, **vendor_sp "the reboot argument cannot be set to False." ) - packages = [image_name, *(additional_files or [])] - add_id = self._install_add(f"{DEFAULT_FILE_SYSTEM}/", packages) + add_id = self._install_add(f"{DEFAULT_FILE_SYSTEM}/", image_name) self._wait_for_install_op(add_id, timeout=timeout) self._install_activate(add_id, timeout=timeout) self._wait_for_device_reboot(timeout=timeout) diff --git a/tests/unit/test_devices/test_iosxr_device.py b/tests/unit/test_devices/test_iosxr_device.py index 0f0dcc0d..6a953549 100644 --- a/tests/unit/test_devices/test_iosxr_device.py +++ b/tests/unit/test_devices/test_iosxr_device.py @@ -7,32 +7,34 @@ from pyntc.errors import FileTransferError from pyntc.utils.models import FileCopyModel -ISO = "ncs5k-mini-x-7.11.2.iso" +ISO = "ncs5k-golden-x-7.11.2-NTC7112.iso" ACTIVE_VERSION = "7.11.2" -ISO_URL = "http://10.1.100.220/IOS-XR/7.11.2/ncs5k-mini-x-7.11.2.iso" +ISO_URL = "http://10.1.100.220/IOS-XR/7.11.2/ncs5k-golden-x-7.11.2-NTC7112.iso" PROMPT = "RP/0/RP0/CPU0:ncs#" DIR_FILE_PRESENT = ( "Mon Jun 15 12:00:00.000 UTC\n\n" "Directory of harddisk:\n" - "15 -rw- 1500000000 Jun 15 12:00 ncs5k-mini-x-7.11.2.iso\n" + "15 -rw- 1500000000 Jun 15 12:00 ncs5k-golden-x-7.11.2-NTC7112.iso\n" +) +DIR_FILE_ABSENT = ( + "Mon Jun 15 12:00:00.000 UTC\n%Error: dir: '/harddisk:/ncs5k-golden-x-7.11.2-NTC7112.iso': No such file\n" ) -DIR_FILE_ABSENT = "Mon Jun 15 12:00:00.000 UTC\n%Error: dir: '/harddisk:/ncs5k-mini-x-7.11.2.iso': No such file\n" COPY_SUCCESS = ( "Mon Jun 15 12:00:00.000 UTC\n" - "Destination filename [/harddisk:/ncs5k-mini-x-7.11.2.iso]?\n" - "Accessing http://10.1.100.220/IOS-XR/7.11.2/ncs5k-mini-x-7.11.2.iso\n" + "Destination filename [/harddisk:/ncs5k-golden-x-7.11.2-NTC7112.iso]?\n" + "Accessing http://10.1.100.220/IOS-XR/7.11.2/ncs5k-golden-x-7.11.2-NTC7112.iso\n" "1500000000 bytes copied in 42 secs (35714285 bytes/sec)\n" "RP/0/RP0/CPU0:ncs#" ) COPY_ERROR = ( "Mon Jun 15 12:00:00.000 UTC\n" - "%Error opening http://10.1.100.220/IOS-XR/7.11.2/ncs5k-mini-x-7.11.2.iso: Connection refused\n" + "%Error opening http://10.1.100.220/IOS-XR/7.11.2/ncs5k-golden-x-7.11.2-NTC7112.iso: Connection refused\n" ) # Real eXR (NCS5011) copy output: netmiko strips the trailing prompt, and the # success markers are "Successfully copied ... Bytes" / "Copy operation success". COPY_SUCCESS_EXR = ( - "\nAccessing http://10.1.100.220/IOS-XR/7.11.2/ncs5k-mini-x-7.11.2.iso\n" + "\nAccessing http://10.1.100.220/IOS-XR/7.11.2/ncs5k-golden-x-7.11.2-NTC7112.iso\n" + ("!" * 60) + "\nSuccessfully copied 1432756224 Bytes\n\n\nCopy operation success\n" ) @@ -60,7 +62,7 @@ INSTALL_ADD_RESPONSE = ( "Mon Jun 15 12:00:00.000 UTC\n" "Install operation 17 started by admin:\n" - " install add source harddisk:/ ncs5k-mini-x-7.11.2.iso\n" + " install add source harddisk:/ ncs5k-golden-x-7.11.2-NTC7112.iso\n" "This operation will continue asynchronously.\n" "Install add operation 17 will continue in the background.\n" ) @@ -89,7 +91,7 @@ DIR_HARDDISK = ( "Mon Jun 15 12:00:00.000 UTC\n\n" "Directory of harddisk:\n" - "15 -rw- 1500000000 Jun 15 12:00 ncs5k-mini-x-7.11.2.iso\n\n" + "15 -rw- 1500000000 Jun 15 12:00 ncs5k-golden-x-7.11.2-NTC7112.iso\n\n" "3000000000 bytes total (2000000000 bytes free)\n" ) @@ -107,7 +109,7 @@ # 'show install request' states observed on real eXR during an activation. SHOW_INSTALL_REQUEST_IN_PROGRESS = ( "Tue Jun 16 03:31:04.338 UTC\n" - "User ntc, Op Id 26\ninstall activate\nncs5k-mini-x-7.11.2\n" + "User ntc, Op Id 26\ninstall activate\nncs5k-golden-x-7.11.2-NTC7112\n" "install operation 26 is in progress\n" "Install prepare operation 26 is in progress\n" "0/RP0 In Progress Partition preparation in progress\n" @@ -258,12 +260,12 @@ def test_get_free_space_unparsable_raises(self): def test_install_add_parses_op_id(self): self.device.native.send_command.return_value = INSTALL_ADD_RESPONSE - self.assertEqual(self.device._install_add("harddisk:/", [ISO]), 17) + self.assertEqual(self.device._install_add("harddisk:/", ISO), 17) def test_install_add_no_op_id_raises(self): self.device.native.send_command.return_value = INSTALL_ADD_NO_OP_ID with self.assertRaises(iosxr_module.OSInstallError): - self.device._install_add("harddisk:/", [ISO]) + self.device._install_add("harddisk:/", ISO) @mock.patch("pyntc.devices.iosxr_device.time.sleep") def test_wait_for_install_op_success(self, mock_sleep): @@ -321,7 +323,22 @@ def test_install_activate_timeout_raises(self, mock_time, mock_sleep): def test_install_commit(self): self.device.native.send_command.return_value = "Install operation 18 completed successfully" self.device._install_commit() - self.device.native.send_command.assert_any_call("install commit") + self.device.native.send_command.assert_any_call("install commit", read_timeout=120) + + @mock.patch("pyntc.devices.iosxr_device.time.sleep") + @mock.patch.object(IOSXRDevice, "open") + def test_install_commit_retries_then_succeeds(self, mock_open, mock_sleep): + # The install manager can be slow right after the reload; the first attempt fails. + self.device.native.send_command.side_effect = [Exception("read timeout"), "ok"] + self.device._install_commit() + self.assertEqual(self.device.native.send_command.call_count, 2) + + @mock.patch("pyntc.devices.iosxr_device.time.sleep") + @mock.patch.object(IOSXRDevice, "open") + def test_install_commit_raises_after_exhausting_retries(self, mock_open, mock_sleep): + self.device.native.send_command.side_effect = Exception("read timeout") + with self.assertRaises(iosxr_module.OSInstallError): + self.device._install_commit(retries=2) # --- reboot / _wait_for_device_reboot --- @@ -360,28 +377,12 @@ def test_install_os( ): result = self.device.install_os(ISO) self.assertTrue(result) - mock_add.assert_called_once_with("harddisk:/", [ISO]) + mock_add.assert_called_once_with("harddisk:/", ISO) mock_wait_op.assert_called_once_with(17, timeout=3600) # add op only mock_activate.assert_called_once_with(17, timeout=3600) mock_wait_reboot.assert_called_once_with(timeout=3600) mock_commit.assert_called_once() - @mock.patch.object(IOSXRDevice, "_install_commit") - @mock.patch.object(IOSXRDevice, "_wait_for_device_reboot") - @mock.patch.object(IOSXRDevice, "_install_activate", return_value=18) - @mock.patch.object(IOSXRDevice, "_wait_for_install_op") - @mock.patch.object(IOSXRDevice, "_install_add", return_value=17) - @mock.patch.object(IOSXRDevice, "uptime", new_callable=mock.PropertyMock, return_value=1000) - @mock.patch.object(IOSXRDevice, "_image_booted", side_effect=[False, True]) - def test_install_os_with_additional_files( - self, mock_booted, mock_uptime, mock_add, mock_wait_op, mock_activate, mock_wait_reboot, mock_commit - ): - rpms = ["ncs5k-mpls-7.11.2.rpm", "ncs5k-ospf-7.11.2.rpm"] - result = self.device.install_os(ISO, additional_files=rpms) - self.assertTrue(result) - # The base ISO and the feature RPMs are added together as a single set. - mock_add.assert_called_once_with("harddisk:/", [ISO, *rpms]) - @mock.patch.object(IOSXRDevice, "_install_add") @mock.patch.object(IOSXRDevice, "_image_booted", return_value=True) def test_install_os_already_installed(self, mock_booted, mock_add): From dbf3751098dcc6d9894c4b510f1413c4f24dc064 Mon Sep 17 00:00:00 2001 From: James Williams Date: Tue, 16 Jun 2026 17:50:55 -0500 Subject: [PATCH 4/7] updates per testing with os upgrades --- pyntc/devices/iosxr_device.py | 96 ++++++++++++++++---- tests/unit/test_devices/test_iosxr_device.py | 41 +++++++++ 2 files changed, 121 insertions(+), 16 deletions(-) diff --git a/pyntc/devices/iosxr_device.py b/pyntc/devices/iosxr_device.py index 3b5784a5..943e8ffc 100644 --- a/pyntc/devices/iosxr_device.py +++ b/pyntc/devices/iosxr_device.py @@ -17,7 +17,7 @@ import time from netmiko import ConnectHandler -from netmiko.exceptions import ReadTimeout +from netmiko.exceptions import AuthenticationException, ReadTimeout, SSHException from pyntc import log from pyntc.devices.base_device import BaseDevice, fix_docs @@ -25,6 +25,15 @@ from pyntc.utils.models import FileCopyModel DEFAULT_FILE_SYSTEM = "harddisk:" +# A freshly reloaded eXR node — or one hit by the upgrade workflow's rapid, short-lived +# sessions — refuses new SSH connections once it exceeds its `ssh server rate-limit`, +# closing the socket before the version exchange (the client sees "Error reading SSH +# protocol banner" or a connection timeout). These failures are transient: waiting a few +# seconds lets the per-minute rate-limit window drain, so connections are retried with a +# backoff. Override per device via the `ssh_connect_attempts` / `ssh_connect_retry_delay` +# kwargs. +DEFAULT_SSH_CONNECT_ATTEMPTS = 5 +DEFAULT_SSH_CONNECT_RETRY_DELAY = 15 # Parse the operation id from an "install add" response, e.g. "Install operation 17 started". RE_INSTALL_OP = re.compile(r"[Ii]nstall operation (\d+)") # Parse the active boot package and its version from "show install active", e.g. "ncs5k-xr-7.11.2". @@ -67,6 +76,8 @@ def __init__(self, host, username, password, secret="", port=None, **kwargs): # self.secret = secret self.port = int(port) if port else 22 self.read_timeout_override = kwargs.get("read_timeout_override") + self._connect_attempts = int(kwargs.get("ssh_connect_attempts", DEFAULT_SSH_CONNECT_ATTEMPTS)) + self._connect_retry_delay = int(kwargs.get("ssh_connect_retry_delay", DEFAULT_SSH_CONNECT_RETRY_DELAY)) self._connected = False self.open() log.init(host=host) @@ -289,7 +300,9 @@ def _wait_for_device_reboot(self, timeout=3600, interval=60): seen_down = False while time.time() - start < timeout: try: - self.open() + # This loop is itself the retry mechanism, so each probe fails fast + # (retry=False) rather than paying the connect backoff on every poll. + self.open(retry=False) self.show("show version") if seen_down: log.info("Host %s: device is back up after reload.", self.host) @@ -582,8 +595,70 @@ def install_os(self, image_name, reboot=True, **vendor_specifics): log.info("Host %s: OS image %s installed successfully.", self.host, image_name) return True - def open(self): - """Open a connection to the network device.""" + def _connect(self, attempts): + """Establish a Netmiko connection, retrying transient SSH failures with backoff. + + eXR refuses new SSH sessions once its ``ssh server rate-limit`` is exceeded — and + the upgrade workflow opens many short-lived sessions in quick succession, so a fresh + connection (e.g. the post-reboot verification) can be rejected: the device closes the + socket before the version exchange, surfacing as ``Error reading SSH protocol banner`` + or a connection timeout. These are transient, so connect is retried with a backoff + long enough to let the per-minute window drain. Authentication failures are not + transient and are re-raised immediately. + + Args: + attempts (int): Maximum number of connection attempts. + + Returns: + ConnectHandler: A live Netmiko connection. + + Raises: + AuthenticationException: On a genuine auth failure (never retried). + SSHException: When every attempt fails to connect (last error re-raised). + """ + last_exc = None + for attempt in range(1, attempts + 1): + try: + return ConnectHandler( + device_type="cisco_xr", + ip=self.host, + username=self.username, + password=self.password, + port=self.port, + read_timeout_override=self.read_timeout_override, + secret=self.secret, + # Keepalives let the status polls notice a dropped session promptly. + keepalive=30, + verbose=False, + ) + except AuthenticationException: + # Bad credentials are not transient — fail fast. + raise + except (SSHException, OSError, EOFError) as exc: + # SSHException covers Netmiko's banner/timeout wrappers and raw paramiko + # banner errors; OSError/EOFError cover the rate-limited socket close. + last_exc = exc + if attempt < attempts: + log.info( + "Host %s: SSH connect attempt %s/%s failed (%s); retrying in %ss.", + self.host, + attempt, + attempts, + exc, + self._connect_retry_delay, + ) + time.sleep(self._connect_retry_delay) + log.error("Host %s: SSH connect failed after %s attempts.", self.host, attempts) + raise last_exc + + def open(self, retry=True): + """Open a connection to the network device. + + Args: + retry (bool): Retry transient SSH failures (rate-limit / banner) with backoff. + Defaults to True. Callers that run their own polling loop (e.g. + ``_wait_for_device_reboot``) pass False so each probe fails fast. + """ if self.connected: try: self.native.find_prompt() @@ -591,18 +666,7 @@ def open(self): self._connected = False if not self.connected: - self.native = ConnectHandler( - device_type="cisco_xr", - ip=self.host, - username=self.username, - password=self.password, - port=self.port, - read_timeout_override=self.read_timeout_override, - secret=self.secret, - # Keepalives let the status polls notice a dropped session promptly. - keepalive=30, - verbose=False, - ) + self.native = self._connect(self._connect_attempts if retry else 1) self._connected = True log.debug("Host %s: Connection to controller was opened successfully.", self.host) diff --git a/tests/unit/test_devices/test_iosxr_device.py b/tests/unit/test_devices/test_iosxr_device.py index 6a953549..1ef5ade8 100644 --- a/tests/unit/test_devices/test_iosxr_device.py +++ b/tests/unit/test_devices/test_iosxr_device.py @@ -1,6 +1,7 @@ import unittest import mock +from netmiko.exceptions import AuthenticationException, SSHException from pyntc.devices import IOSXRDevice, supported_devices from pyntc.devices import iosxr_device as iosxr_module @@ -363,6 +364,46 @@ def test_wait_for_device_reboot_timeout(self, mock_close, mock_open, mock_time, with self.assertRaises(iosxr_module.RebootTimeoutError): self.device._wait_for_device_reboot(timeout=3600) + # --- connection retry (eXR SSH rate-limit / banner race) --- + + @mock.patch("pyntc.devices.iosxr_device.time.sleep") + @mock.patch("pyntc.devices.iosxr_device.ConnectHandler") + def test_connect_retries_transient_then_succeeds(self, mock_connect, mock_sleep): + conn = mock.MagicMock() + mock_connect.side_effect = [SSHException("Error reading SSH protocol banner"), conn] + self.assertIs(self.device._connect(self.device._connect_attempts), conn) + self.assertEqual(mock_connect.call_count, 2) + mock_sleep.assert_called_once_with(self.device._connect_retry_delay) + + @mock.patch("pyntc.devices.iosxr_device.time.sleep") + @mock.patch("pyntc.devices.iosxr_device.ConnectHandler") + def test_connect_does_not_retry_auth_failure(self, mock_connect, mock_sleep): + mock_connect.side_effect = AuthenticationException("bad creds") + with self.assertRaises(AuthenticationException): + self.device._connect(self.device._connect_attempts) + self.assertEqual(mock_connect.call_count, 1) + mock_sleep.assert_not_called() + + @mock.patch("pyntc.devices.iosxr_device.time.sleep") + @mock.patch("pyntc.devices.iosxr_device.ConnectHandler") + def test_connect_raises_after_exhausting_attempts(self, mock_connect, mock_sleep): + mock_connect.side_effect = SSHException("Error reading SSH protocol banner") + with self.assertRaises(SSHException): + self.device._connect(3) + self.assertEqual(mock_connect.call_count, 3) + self.assertEqual(mock_sleep.call_count, 2) + + @mock.patch("pyntc.devices.iosxr_device.time.sleep") + @mock.patch("pyntc.devices.iosxr_device.ConnectHandler") + def test_open_retry_false_makes_single_attempt(self, mock_connect, mock_sleep): + # The reboot-wait loop is its own retry; each probe must fail fast. + mock_connect.side_effect = OSError("down") + self.device._connected = False + with self.assertRaises(OSError): + self.device.open(retry=False) + self.assertEqual(mock_connect.call_count, 1) + mock_sleep.assert_not_called() + # --- install_os orchestration --- @mock.patch.object(IOSXRDevice, "_install_commit") From 78e8fbfb80e1b424b512fb35d6e9ef6390c0f4c6 Mon Sep 17 00:00:00 2001 From: Gary Snider <75227981+gsnider2195@users.noreply.github.com> Date: Wed, 24 Jun 2026 14:34:07 -0700 Subject: [PATCH 5/7] fix failing tests due to outdated F5 sdk --- pyproject.toml | 2 +- tests/unit/conftest.py | 11 +---------- tests/unit/test_devices/test_f5_device.py | 6 ++++++ tests/unit/test_infra.py | 9 +++++++-- 4 files changed, 15 insertions(+), 13 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 118397d2..af9e003f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -167,7 +167,7 @@ python_paths = "./" testpaths = [ "tests/" ] -addopts = "-vv --doctest-modules -p no:warnings --ignore-glob='*mock*'" +addopts = "-vv --doctest-modules -p no:warnings -p no:f5sdk_fixtures --ignore-glob='*mock*'" [tool.towncrier] package = "pyntc" diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index f3f974df..fe9fbede 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, supported_devices +from pyntc.devices import AIREOSDevice, ASADevice, EOSDevice, IOSDevice, IOSXEWLCDevice def get_side_effects(mock_path, side_effects): @@ -59,15 +59,6 @@ def _mock(side_effects, existing_device=None, device=eos_device): return _mock -def pytest_generate_tests(metafunc): - if metafunc.function.__name__ == "test_device_creation": - metafunc.parametrize( - "device_type,expected", - ((device_type, device_class) for device_type, device_class in supported_devices.items()), - ids=(device_type for device_type in supported_devices), - ) - - @pytest.fixture def aireos_boot_image(): return "8.2.170.0" diff --git a/tests/unit/test_devices/test_f5_device.py b/tests/unit/test_devices/test_f5_device.py index 6822d805..f533b2ea 100644 --- a/tests/unit/test_devices/test_f5_device.py +++ b/tests/unit/test_devices/test_f5_device.py @@ -1,3 +1,4 @@ +import sys from unittest import mock import pytest @@ -6,6 +7,11 @@ from pyntc.devices.f5_device import F5Device, FileTransferError from pyntc.errors import NTCFileNotFoundError +pytestmark = pytest.mark.skipif( + sys.version_info >= (3, 12), + reason="The F5 SDK is only compatible with Python 3.11 or lower", +) + BOOT_IMAGE = "BIGIP-11.3.0.2806.0.iso" VOLUME = "HD1.1" diff --git a/tests/unit/test_infra.py b/tests/unit/test_infra.py index 97e9ca79..0be3ff1a 100644 --- a/tests/unit/test_infra.py +++ b/tests/unit/test_infra.py @@ -1,10 +1,11 @@ import os +import sys import mock import pytest from pyntc import ntc_device, ntc_device_by_name -from pyntc.devices import EOSDevice, IOSDevice, NXOSDevice +from pyntc.devices import EOSDevice, IOSDevice, NXOSDevice, supported_devices from pyntc.errors import ConfFileNotFoundError, UnsupportedDeviceError BAD_DEVICE_TYPE = "238nzsvkn3981" @@ -12,7 +13,7 @@ @mock.patch("pyntc.devices.aireos_device.AIREOSDevice.open") -@mock.patch("pyntc.devices.f5_device.ManagementRoot") +@mock.patch("pyntc.devices.f5_device.ManagementRoot", create=True) @mock.patch("pyntc.devices.asa_device.ASADevice.open") @mock.patch("pyntc.devices.ios_device.IOSDevice.open") @mock.patch("pyntc.devices.iosxr_device.IOSXRDevice.open") @@ -20,9 +21,13 @@ @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 ): + # Skip f5 on python >3.11 + if sys.version_info >= (3, 12) and device_type == "f5_tmos_icontrol": + pytest.skip(f"F5 not supported in Python {sys.version}") device = ntc_device(device_type, "host", "user", "pass") assert isinstance(device, expected) From 6443af38112e59d5c45df4ddc6514a26120795a4 Mon Sep 17 00:00:00 2001 From: Gary Snider <75227981+gsnider2195@users.noreply.github.com> Date: Wed, 24 Jun 2026 15:42:23 -0700 Subject: [PATCH 6/7] address most of the PR feedback. Still TODO: checksum verification and wait_for_reboot updates --- pyntc/devices/iosxr_device.py | 171 ++++++++++++------- tests/unit/test_devices/test_iosxr_device.py | 78 +++++---- 2 files changed, 159 insertions(+), 90 deletions(-) diff --git a/pyntc/devices/iosxr_device.py b/pyntc/devices/iosxr_device.py index 943e8ffc..53564da5 100644 --- a/pyntc/devices/iosxr_device.py +++ b/pyntc/devices/iosxr_device.py @@ -21,10 +21,16 @@ from pyntc import log from pyntc.devices.base_device import BaseDevice, fix_docs -from pyntc.errors import CommandError, CommandListError, FileTransferError, OSInstallError, RebootTimeoutError +from pyntc.errors import ( + CommandError, + CommandListError, + FileSystemNotFoundError, + FileTransferError, + OSInstallError, + RebootTimeoutError, +) from pyntc.utils.models import FileCopyModel -DEFAULT_FILE_SYSTEM = "harddisk:" # A freshly reloaded eXR node — or one hit by the upgrade workflow's rapid, short-lived # sessions — refuses new SSH connections once it exceeds its `ssh server rate-limit`, # closing the socket before the version exchange (the client sees "Error reading SSH @@ -34,22 +40,9 @@ # kwargs. DEFAULT_SSH_CONNECT_ATTEMPTS = 5 DEFAULT_SSH_CONNECT_RETRY_DELAY = 15 -# Parse the operation id from an "install add" response, e.g. "Install operation 17 started". -RE_INSTALL_OP = re.compile(r"[Ii]nstall operation (\d+)") -# Parse the active boot package and its version from "show install active", e.g. "ncs5k-xr-7.11.2". -RE_XR_BOOT_IMAGE = re.compile(r"(?P\S*xr-(?P\d+\.\d+\.\d+\w*))") + # Parse the running version from "show version", e.g. "Version 7.11.2". RE_XR_VERSION = re.compile(r"Version\s+(\d+\.\d+\.\d+\w*)") -# Success / error markers emitted by the IOS-XR "copy" command (eXR uses -# "Successfully copied ... Bytes" / "Copy operation success"). -RE_COPY_SUCCESS = re.compile( - r"Successfully copied|Copy operation success|bytes copied|copied in|\[OK\]|Download Complete|transfer successful", - re.IGNORECASE, -) -RE_COPY_ERROR = re.compile( - r"%Error|Error opening|Invalid input|Failed|Aborted|denied|No such file|Connection refused|timed out|could not", - re.IGNORECASE, -) @fix_docs @@ -59,15 +52,29 @@ class IOSXRDevice(BaseDevice): vendor = "cisco" # pylint: disable=too-many-arguments, too-many-positional-arguments - def __init__(self, host, username, password, secret="", port=None, **kwargs): # noqa: D403 # nosec + def __init__( + self, + host, + username, + password, + secret="", + port=None, + read_timeout_override=None, + ssh_connect_attempts=DEFAULT_SSH_CONNECT_ATTEMPTS, + ssh_connect_retry_delay=DEFAULT_SSH_CONNECT_RETRY_DELAY, + **kwargs, + ): # noqa: D403 # nosec """PyNTC Device implementation for Cisco IOS-XR (eXR). 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 port to use to establish the connection. Defaults to 22. + secret (str, optional): The password to escalate privilege on the device. + port (int, optional): The port to use to establish the connection. Defaults to 22. + read_timeout_override (int, optional): If supplied, overrides all timeouts for netmiko send_command calls. + ssh_connect_attempts (int, optional): Number of times to try to connect to the device before giving up. Defaults to 5. + ssh_connect_retry_delay (int, optional): Number of seconds to wait between retries when ssh_connect_attempts is >1. Defaults to 15. kwargs (dict): Additional arguments to pass to the Netmiko ConnectHandler. """ super().__init__(host, username, password, device_type="cisco_iosxr_ssh") @@ -75,9 +82,9 @@ def __init__(self, host, username, password, secret="", port=None, **kwargs): # self.native = None self.secret = secret self.port = int(port) if port else 22 - self.read_timeout_override = kwargs.get("read_timeout_override") - self._connect_attempts = int(kwargs.get("ssh_connect_attempts", DEFAULT_SSH_CONNECT_ATTEMPTS)) - self._connect_retry_delay = int(kwargs.get("ssh_connect_retry_delay", DEFAULT_SSH_CONNECT_RETRY_DELAY)) + self.read_timeout_override = read_timeout_override + self._connect_attempts = ssh_connect_attempts + self._connect_retry_delay = ssh_connect_retry_delay self._connected = False self.open() log.init(host=host) @@ -90,13 +97,44 @@ def _send_command(self, command, expect_string=None, **kwargs): response = self.native.send_command(**command_args) - if "% " in response or "Error:" in response: + if re.search(r"^\s*%", response, flags=re.MULTILINE) or "Error:" in response: log.error("Host %s: Error in %s with response: %s", self.host, command, response) raise CommandError(command, response) log.info("Host %s: Command %s was executed successfully.", self.host, command) return response + def _get_file_system(self): + """Determine the default file system or directory for device. + + Returns: + (str): The name of the default file system or directory for the device. + + Raises: + FileSystemNotFound: When the module is unable to determine the default file system. + """ + raw_data = self.show("show filesystem location all") + + try: + found_filesystems = set() + fs_list = raw_data.split("File Systems:")[1].strip().split("\n")[1:] + for fs in fs_list: + size_bytes, free_bytes, fs_type, fs_flags, fs_name = fs.split() + if "disk" in fs_type: + found_filesystems.add(fs_name) + log.debug("Host %s: Found filesystem %s.", self.host, fs_name) + + # Prefer harddisk: then disk0: + # TODO: Do we need to support more than these? + for fs in ["harddisk:", "disk0:"]: + if fs in found_filesystems: + return fs + except (AttributeError, IndexError, ValueError): + pass + + log.error("host %s: Unable to determine the device's default filesystem.") + raise FileSystemNotFoundError(hostname=self.hostname, command="show filesystem location all") + def _uptime_components(self, uptime_full_string): match_weeks = re.search(r"(\d+) weeks?", uptime_full_string) match_days = re.search(r"(\d+) days?", uptime_full_string) @@ -142,44 +180,47 @@ def _install_add(self, source, image_name): command = f"install add source {source} {image_name}" response = self.native.send_command(command, read_timeout=120) - match = RE_INSTALL_OP.search(response) + # Parse the operation id from the response, e.g. "Install operation 17 started". + match = re.search(r"[Ii]nstall operation (\d+)", response) if match is None: log.error("Host %s: Unable to parse install operation id from response: %s", self.host, response) raise OSInstallError(hostname=self.host, desired_boot=image_name) - op_id = int(match.group(1)) - log.info("Host %s: install add started operation %s.", self.host, op_id) - return op_id + operation_id = int(match.group(1)) + log.info("Host %s: install add started operation %s.", self.host, operation_id) + return operation_id - def _wait_for_install_op(self, op_id, timeout=3600, interval=30): + def _wait_for_install_operation(self, operation_id, timeout=3600, interval=30): """Poll ``show install log `` until the operation reaches a terminal state. Args: - op_id (int): The install operation id to track. + operation_id (int): The install operation id to track. timeout (int): Maximum seconds to wait for a terminal state. Defaults to 3600. interval (int): Seconds to wait between polls. Defaults to 30. Raises: OSInstallError: When the operation aborts/fails or the timeout is exceeded. """ - success = re.compile(rf"operation\s+{op_id}\b.*(completed successfully|succeeded)", re.IGNORECASE | re.DOTALL) - failure = re.compile(rf"operation\s+{op_id}\b.*(aborted|failed)", re.IGNORECASE | re.DOTALL) + success = re.compile( + rf"operation\s+{operation_id}\b.*(completed successfully|succeeded)", re.IGNORECASE | re.DOTALL + ) + failure = re.compile(rf"operation\s+{operation_id}\b.*(aborted|failed)", re.IGNORECASE | re.DOTALL) start = time.time() while time.time() - start < timeout: - output = self.native.send_command(f"show install log {op_id}", read_timeout=120) + output = self.native.send_command(f"show install log {operation_id}", read_timeout=120) if failure.search(output): - log.error("Host %s: install operation %s aborted/failed.", self.host, op_id) - raise OSInstallError(hostname=self.host, desired_boot=f"operation {op_id}") + log.error("Host %s: install operation %s aborted/failed.", self.host, operation_id) + raise OSInstallError(hostname=self.host, desired_boot=f"operation {operation_id}") if success.search(output): - log.info("Host %s: install operation %s completed successfully.", self.host, op_id) + log.info("Host %s: install operation %s completed successfully.", self.host, operation_id) return time.sleep(interval) - log.error("Host %s: install operation %s timed out after %s seconds.", self.host, op_id, timeout) - raise OSInstallError(hostname=self.host, desired_boot=f"operation {op_id}") + log.error("Host %s: install operation %s timed out after %s seconds.", self.host, operation_id, timeout) + raise OSInstallError(hostname=self.host, desired_boot=f"operation {operation_id}") - def _install_activate(self, op_id, poll_interval=60, timeout=3600): + def _install_activate(self, operation_id, poll_interval=60, timeout=3600): """Activate a staged install operation and track it to completion. The activation is issued with ``noprompt`` (so eXR does not wait on the interactive @@ -195,14 +236,14 @@ def _install_activate(self, op_id, poll_interval=60, timeout=3600): marker or the session drops, and raises if the operation reports an abort/error. Args: - op_id (int): The staged ``install add`` operation id to activate. + operation_id (int): The staged ``install add`` operation id to activate. poll_interval (int): Seconds between status polls. Defaults to 60. timeout (int): Maximum seconds to wait for the activation to finish. Defaults to 3600. Raises: OSInstallError: When the activation operation aborts/fails or does not finish in time. """ - command = f"install activate id {op_id} noprompt" + command = f"install activate id {operation_id} noprompt" log.info("Host %s: issuing activation: %s", self.host, command) try: self.native.send_command_timing(command, read_timeout=180) @@ -220,16 +261,16 @@ def _install_activate(self, op_id, poll_interval=60, timeout=3600): return log.info("Host %s: polled activation status.", self.host) if re.search(r"abort|Error[:!]", request, re.IGNORECASE): - log.error("Host %s: activation of operation %s failed: %s", self.host, op_id, request) - raise OSInstallError(hostname=self.host, desired_boot=f"operation {op_id}") + log.error("Host %s: activation of operation %s failed: %s", self.host, operation_id, request) + raise OSInstallError(hostname=self.host, desired_boot=f"operation {operation_id}") if re.search( r"completed, pending reload|finished successfully|completed successfully", request, re.IGNORECASE ): log.info("Host %s: activation completed; reload imminent.", self.host) return - log.error("Host %s: activation of operation %s did not finish within %ss.", self.host, op_id, timeout) - raise OSInstallError(hostname=self.host, desired_boot=f"operation {op_id}") + log.error("Host %s: activation of operation %s did not finish within %ss.", self.host, operation_id, timeout) + raise OSInstallError(hostname=self.host, desired_boot=f"operation {operation_id}") def _install_commit(self, retries=3, retry_delay=30, read_timeout=120): """Persist the activated software so it survives future reloads. @@ -330,7 +371,10 @@ def boot_options(self): both values are ``None`` when the output cannot be parsed. """ show_install_active = self.show("show install active") - match = RE_XR_BOOT_IMAGE.search(show_install_active) + + # Parse the active boot package and its version, e.g. "ncs5k-xr-7.11.2". + match = re.search(r"(?P\S*xr-(?P\d+\.\d+\.\d+\w*))", show_install_active) + # The regex's named groups are exactly "sys" and "version". boot_options = match.groupdict() if match else {"sys": None, "version": None} @@ -370,13 +414,13 @@ def check_file_exists(self, filename, file_system=None): Args: filename (str): The filename to look for. - file_system (str, optional): Filesystem to inspect. Defaults to ``harddisk:``. + file_system (str, optional): Filesystem to inspect. Automatically retrieves the default filesystem if not provided. Returns: (bool): True if the file is present, False otherwise. """ if file_system is None: - file_system = DEFAULT_FILE_SYSTEM + file_system = self._get_file_system() result = self.native.send_command(f"dir {file_system}/{filename}", read_timeout=30) if re.search(r"No such file|No files matched|not found|Path does not exist|Error", result, re.IGNORECASE): @@ -393,15 +437,14 @@ def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kw """Copy a file from a remote URL onto the device filesystem. Pulls the file specified by ``src`` from a remote server (FTP/TFTP/SCP/HTTP/HTTPS) - using the IOS-XR ``copy`` command and saves it to ``file_system`` (default - ``harddisk:``). The transfer is verified by confirming the file exists after - copy. **Checksum verification is not performed** on IOS-XR in this release; the - ``checksum`` on ``src`` is not validated. + using the IOS-XR ``copy`` command and saves it to ``file_system``. The transfer is + verified by confirming the file exists after copy. **Checksum verification is not performed** + on IOS-XR in this release; the ``checksum`` on ``src`` is not validated. Args: src (FileCopyModel): The source specification (URL, credentials, timeout). dest (str, optional): Destination filename. Defaults to ``src.file_name``. - file_system (str, optional): Target filesystem. Defaults to ``harddisk:``. + file_system (str, optional): Target filesystem. Automatically retrieves the default filesystem if not provided. kwargs (dict): Additional keyword arguments (unused). Raises: @@ -412,7 +455,7 @@ def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kw raise TypeError("src must be an instance of FileCopyModel") if file_system is None: - file_system = DEFAULT_FILE_SYSTEM + file_system = self._get_file_system() if dest is None: dest = src.file_name @@ -445,10 +488,18 @@ def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kw # so the post-copy existence check (below) is the authoritative success signal; this # loop only answers prompts and surfaces explicit error markers early. for _ in range(10): - if RE_COPY_SUCCESS.search(output): + if re.search( + r"Successfully copied|Copy operation success|bytes copied|copied in|\[OK\]|Download Complete|transfer successful", + output, + flags=re.IGNORECASE, + ): log.info("Host %s: File %s transfer reported success.", self.host, dest) break - if RE_COPY_ERROR.search(output): + if re.search( + r"%Error|Error opening|Invalid input|Failed|Aborted|denied|No such file|Connection refused|timed out|could not", + output, + flags=re.IGNORECASE, + ): log.error("Host %s: File transfer error for %s: %s", self.host, dest, output) raise FileTransferError for prompt, answer in prompt_answers.items(): @@ -516,7 +567,7 @@ def _get_free_space(self, file_system=None): """Return free bytes on ``file_system`` as reported by ``dir`` output. Args: - file_system (str, optional): Target filesystem. Defaults to ``harddisk:``. + file_system (str, optional): Target filesystem. Automatically retrieves the default filesystem if not provided. Returns: (int): Free bytes available on ``file_system``. @@ -525,7 +576,7 @@ def _get_free_space(self, file_system=None): CommandError: When the free space cannot be parsed from ``dir`` output. """ if file_system is None: - file_system = DEFAULT_FILE_SYSTEM + file_system = self._get_file_system() raw_data = self.show(f"dir {file_system}") # eXR reports the trailer in kbytes (e.g. "9948012 kbytes total (9396256 kbytes free)"); @@ -556,7 +607,7 @@ def install_os(self, image_name, reboot=True, **vendor_specifics): single file. Installing a base ISO plus separate feature RPMs is not supported. Args: - image_name (str): The golden ISO filename already staged on ``harddisk:``. + image_name (str): The golden ISO filename already staged on the device. reboot (bool): Must be ``True``; activation reloads the device automatically. vendor_specifics (dict, optional): Supports ``timeout`` (default 3600) for the install-operation and reboot waits. @@ -582,8 +633,8 @@ def install_os(self, image_name, reboot=True, **vendor_specifics): "the reboot argument cannot be set to False." ) - add_id = self._install_add(f"{DEFAULT_FILE_SYSTEM}/", image_name) - self._wait_for_install_op(add_id, timeout=timeout) + add_id = self._install_add(f"{self._get_file_system()}/", image_name) + self._wait_for_install_operation(add_id, timeout=timeout) self._install_activate(add_id, timeout=timeout) self._wait_for_device_reboot(timeout=timeout) self._install_commit() diff --git a/tests/unit/test_devices/test_iosxr_device.py b/tests/unit/test_devices/test_iosxr_device.py index 1ef5ade8..1b26417b 100644 --- a/tests/unit/test_devices/test_iosxr_device.py +++ b/tests/unit/test_devices/test_iosxr_device.py @@ -70,6 +70,22 @@ INSTALL_ADD_NO_OP_ID = "Mon Jun 15 12:00:00.000 UTC\n% Unexpected output without an operation id\n" +SHOW_FILESYSTEM_LOCATION_ALL = """ +Tue Jun 23 21:22:51.023 UTC + + node: node0_RP0_CPU0 +------------------------------------------------------------------ +File Systems: + + Size(b) Free(b) Type Flags Prefixes + 2358312960 2347773952 flash-disk rw disk0: + 480907264 479154176 flash rw /misc/config + 10186764288 7176835072 harddisk rw harddisk: + 0 0 network rw ftp: + 0 0 network rw tftp: + 3962216448 3926454272 flash-disk rw apphost: +""" + SHOW_INSTALL_LOG_INPROGRESS = ( "Install operation 17: 'install add source harddisk:/ ...' started\nAction 17 in progress\n" ) @@ -239,20 +255,18 @@ def test_image_booted_false(self): # --- _get_free_space --- - def test_get_free_space(self): + @mock.patch.object(IOSXRDevice, "_get_file_system", return_value="harddisk:") + def test_get_free_space(self, *_mocks): self.device.native.send_command.return_value = DIR_HARDDISK self.assertEqual(self.device._get_free_space(), 2000000000) - def test_get_free_space_default_file_system_is_harddisk(self): - self.device.native.send_command.return_value = DIR_HARDDISK - self.device._get_free_space() - self.device.native.send_command.assert_any_call(command_string="dir harddisk:") - - def test_get_free_space_kbytes_units(self): + @mock.patch.object(IOSXRDevice, "_get_file_system", return_value="harddisk:") + def test_get_free_space_kbytes_units(self, *_mocks): self.device.native.send_command.return_value = DIR_HARDDISK_KBYTES self.assertEqual(self.device._get_free_space(), 9396256 * 1024) - def test_get_free_space_unparsable_raises(self): + @mock.patch.object(IOSXRDevice, "_get_file_system", return_value="harddisk:") + def test_get_free_space_unparsable_raises(self, *_mocks): self.device.native.send_command.return_value = "garbage output" with self.assertRaises(iosxr_module.CommandError): self.device._get_free_space() @@ -269,27 +283,27 @@ def test_install_add_no_op_id_raises(self): self.device._install_add("harddisk:/", ISO) @mock.patch("pyntc.devices.iosxr_device.time.sleep") - def test_wait_for_install_op_success(self, mock_sleep): + def test_wait_for_install_operation_success(self, mock_sleep): self.device.native.send_command.side_effect = [ SHOW_INSTALL_LOG_INPROGRESS, SHOW_INSTALL_LOG_INPROGRESS, SHOW_INSTALL_LOG_SUCCESS, ] - self.device._wait_for_install_op(17) + self.device._wait_for_install_operation(17) self.assertEqual(self.device.native.send_command.call_count, 3) @mock.patch("pyntc.devices.iosxr_device.time.sleep") - def test_wait_for_install_op_abort_raises(self, mock_sleep): + def test_wait_for_install_operation_abort_raises(self, mock_sleep): self.device.native.send_command.return_value = SHOW_INSTALL_LOG_ABORT with self.assertRaises(iosxr_module.OSInstallError): - self.device._wait_for_install_op(17) + self.device._wait_for_install_operation(17) @mock.patch("pyntc.devices.iosxr_device.time.sleep") @mock.patch("pyntc.devices.iosxr_device.time.time", side_effect=_fake_clock([0, 0])) - def test_wait_for_install_op_timeout_raises(self, mock_time, mock_sleep): + def test_wait_for_install_operation_timeout_raises(self, mock_time, mock_sleep): self.device.native.send_command.return_value = SHOW_INSTALL_LOG_INPROGRESS with self.assertRaises(iosxr_module.OSInstallError): - self.device._wait_for_install_op(17, timeout=3600) + self.device._wait_for_install_operation(17, timeout=3600) @mock.patch("pyntc.devices.iosxr_device.time.sleep") def test_install_activate_issues_async_and_returns_on_pending_reload(self, mock_sleep): @@ -406,16 +420,15 @@ def test_open_retry_false_makes_single_attempt(self, mock_connect, mock_sleep): # --- install_os orchestration --- + @mock.patch.object(IOSXRDevice, "uptime", new_callable=mock.PropertyMock, return_value=1000) + @mock.patch.object(IOSXRDevice, "_image_booted", side_effect=[False, True]) + @mock.patch.object(IOSXRDevice, "_get_file_system", return_value="harddisk:") @mock.patch.object(IOSXRDevice, "_install_commit") @mock.patch.object(IOSXRDevice, "_wait_for_device_reboot") @mock.patch.object(IOSXRDevice, "_install_activate", return_value=18) - @mock.patch.object(IOSXRDevice, "_wait_for_install_op") + @mock.patch.object(IOSXRDevice, "_wait_for_install_operation") @mock.patch.object(IOSXRDevice, "_install_add", return_value=17) - @mock.patch.object(IOSXRDevice, "uptime", new_callable=mock.PropertyMock, return_value=1000) - @mock.patch.object(IOSXRDevice, "_image_booted", side_effect=[False, True]) - def test_install_os( - self, mock_booted, mock_uptime, mock_add, mock_wait_op, mock_activate, mock_wait_reboot, mock_commit - ): + def test_install_os(self, mock_add, mock_wait_op, mock_activate, mock_wait_reboot, mock_commit, *_mocks): result = self.device.install_os(ISO) self.assertTrue(result) mock_add.assert_called_once_with("harddisk:/", ISO) @@ -439,23 +452,24 @@ def test_install_os_reboot_false_raises(self, mock_booted): @mock.patch.object(IOSXRDevice, "_install_commit") @mock.patch.object(IOSXRDevice, "_wait_for_device_reboot") @mock.patch.object(IOSXRDevice, "_install_activate", return_value=18) - @mock.patch.object(IOSXRDevice, "_wait_for_install_op") + @mock.patch.object(IOSXRDevice, "_wait_for_install_operation") @mock.patch.object(IOSXRDevice, "_install_add", return_value=17) @mock.patch.object(IOSXRDevice, "uptime", new_callable=mock.PropertyMock, return_value=1000) @mock.patch.object(IOSXRDevice, "_image_booted", side_effect=[False, False]) - def test_install_os_verify_failure_raises( - self, mock_booted, mock_uptime, mock_add, mock_wait_op, mock_activate, mock_wait_reboot, mock_commit - ): + @mock.patch.object(IOSXRDevice, "_get_file_system", return_value="harddisk:") + def test_install_os_verify_failure_raises(self, *_mocks): with self.assertRaises(iosxr_module.OSInstallError): self.device.install_os(ISO) # --- check_file_exists --- - def test_check_file_exists_true(self): + @mock.patch.object(IOSXRDevice, "_get_file_system", return_value="harddisk:") + def test_check_file_exists_true(self, *_mocks): self.device.native.send_command.return_value = DIR_FILE_PRESENT self.assertTrue(self.device.check_file_exists(ISO)) - def test_check_file_exists_false(self): + @mock.patch.object(IOSXRDevice, "_get_file_system", return_value="harddisk:") + def test_check_file_exists_false(self, *_mocks): self.device.native.send_command.return_value = DIR_FILE_ABSENT self.assertFalse(self.device.check_file_exists(ISO)) @@ -466,7 +480,8 @@ def test_remote_file_copy_requires_model(self): self.device.remote_file_copy(ISO_URL) @mock.patch.object(IOSXRDevice, "check_file_exists", side_effect=[False, True]) - def test_remote_file_copy_success(self, mock_exists): + @mock.patch.object(IOSXRDevice, "_get_file_system", return_value="harddisk:") + def test_remote_file_copy_success(self, *_mocks): self.device.native.find_prompt.return_value = PROMPT self.device.native.send_command.return_value = COPY_SUCCESS src = FileCopyModel(download_url=ISO_URL, checksum="", file_name=ISO) @@ -481,8 +496,9 @@ def test_remote_file_copy_success(self, mock_exists): self.assertTrue(copy_calls) self.assertEqual(copy_calls[0].args[0], f"copy {ISO_URL} harddisk:/{ISO}") + @mock.patch.object(IOSXRDevice, "_get_file_system", return_value="harddisk:") @mock.patch.object(IOSXRDevice, "check_file_exists", return_value=True) - def test_remote_file_copy_idempotent_when_present(self, mock_exists): + def test_remote_file_copy_idempotent_when_present(self, *_mocks): self.device.native.find_prompt.return_value = PROMPT src = FileCopyModel(download_url=ISO_URL, checksum="", file_name=ISO) @@ -495,8 +511,9 @@ def test_remote_file_copy_idempotent_when_present(self, mock_exists): ] self.assertEqual(copy_calls, []) + @mock.patch.object(IOSXRDevice, "_get_file_system", return_value="harddisk:") @mock.patch.object(IOSXRDevice, "check_file_exists", side_effect=[False, True]) - def test_remote_file_copy_success_exr_output(self, mock_exists): + def test_remote_file_copy_success_exr_output(self, mock_exists, *_mocks): # Real eXR success output (no trailing prompt, "Successfully copied"/"Copy operation success"). self.device.native.find_prompt.return_value = PROMPT self.device.native.send_command.return_value = COPY_SUCCESS_EXR @@ -507,7 +524,8 @@ def test_remote_file_copy_success_exr_output(self, mock_exists): self.assertEqual(mock_exists.call_count, 2) # idempotency check + post-copy verify @mock.patch.object(IOSXRDevice, "check_file_exists", side_effect=[False]) - def test_remote_file_copy_error_raises(self, mock_exists): + @mock.patch.object(IOSXRDevice, "_get_file_system", return_value="harddisk:") + def test_remote_file_copy_error_raises(self, *_mocks): self.device.native.find_prompt.return_value = PROMPT self.device.native.send_command.return_value = COPY_ERROR src = FileCopyModel(download_url=ISO_URL, checksum="", file_name=ISO) From d16590a39ea51e5cd229a6b308eb693eed33af14 Mon Sep 17 00:00:00 2001 From: Gary Snider <75227981+gsnider2195@users.noreply.github.com> Date: Wed, 24 Jun 2026 16:08:35 -0700 Subject: [PATCH 7/7] pylint --- pyntc/devices/iosxr_device.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyntc/devices/iosxr_device.py b/pyntc/devices/iosxr_device.py index 53564da5..5c6323e5 100644 --- a/pyntc/devices/iosxr_device.py +++ b/pyntc/devices/iosxr_device.py @@ -119,7 +119,7 @@ def _get_file_system(self): found_filesystems = set() fs_list = raw_data.split("File Systems:")[1].strip().split("\n")[1:] for fs in fs_list: - size_bytes, free_bytes, fs_type, fs_flags, fs_name = fs.split() + _size_bytes, _free_bytes, fs_type, _fs_flags, fs_name = fs.split() if "disk" in fs_type: found_filesystems.add(fs_name) log.debug("Host %s: Found filesystem %s.", self.host, fs_name)