Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions docs/admin/release_notes/version_3.2.md
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,12 @@ This document describes all new features and changes in the release. The format
- Fixed Arista EOS reboot detection when waiting for a device to reload.

<!-- towncrier release notes start -->
## [v3.2.2 (2026-08-04)](https://github.com/networktocode/pyntc/releases/tag/v3.2.2)

### Fixed

- [#410](https://github.com/networktocode/pyntc/issues/410) - Fixed checksum verification timing out on large OS images — `get_remote_checksum` and `verify_file` now accept a `read_timeout` argument and its default was raised from 300s to 900s for IOS, ASA and IOS-XR devices.

## [v3.2.1 (2026-07-27)](https://github.com/networktocode/pyntc/releases/tag/v3.2.1)

### Fixed
Expand Down
6 changes: 5 additions & 1 deletion pyntc/devices/asa_device.py
Original file line number Diff line number Diff line change
Expand Up @@ -657,6 +657,8 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5", **kwargs: Any):

Keyword Args:
file_system (str): The file system where the file resides. Defaults to ``_get_file_system()``.
read_timeout (int): Maximum time in seconds to wait for the checksum command to
complete. Hashing large files can take several minutes (default: 900).

Returns:
(str): The checksum of the file.
Expand All @@ -674,7 +676,7 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5", **kwargs: Any):

file_system = kwargs.get("file_system") or self._get_file_system()
cmd = f"verify /{asa_algorithm} {file_system}{filename}"
result = self.native.send_command_timing(cmd, read_timeout=300)
result = self.native.send_command_timing(cmd, read_timeout=kwargs.get("read_timeout", 900))

if match := re.search(r"=\s+(\S+)", result):
log.debug(
Expand Down Expand Up @@ -1326,6 +1328,8 @@ def verify_file(self, checksum, filename, hashing_algorithm="md5", **kwargs: Any

Keyword Args:
file_system (str): The file system where the file resides. Defaults to ``_get_file_system()``.
read_timeout (int): Maximum time in seconds to wait for the checksum command to
complete. Hashing large files can take several minutes (default: 900).

Returns:
(bool): True if the file exists and the checksum matches, False otherwise.
Expand Down
2 changes: 2 additions & 0 deletions pyntc/devices/base_device.py
Original file line number Diff line number Diff line change
Expand Up @@ -410,6 +410,8 @@ def compare_file_checksum(self, checksum, filename, hashing_algorithm="md5", **k
file_system (str): Supported only for IOS and NXOS. The file system for the
remote file. If no file_system is provided, then the ``get_file_system``
method is used to determine the correct file system to use.
read_timeout (int): Supported only for IOS, ASA and IOS-XR. Maximum time in
seconds to wait for the checksum command to complete (default: 900).

Returns:
(bool): True if the checksums match, False otherwise.
Expand Down
39 changes: 30 additions & 9 deletions pyntc/devices/ios_device.py
Original file line number Diff line number Diff line change
Expand Up @@ -641,7 +641,7 @@ def config_register(self):
log.debug("Host %s: Config register %s", self.host, self._config_register)
return self._config_register

def get_remote_checksum(self, filename, hashing_algorithm="md5", file_system=None):
def get_remote_checksum(self, filename, hashing_algorithm="md5", file_system=None, read_timeout=900):
"""Get the checksum of a remote file.

Args:
Expand All @@ -650,6 +650,8 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5", file_system=Non
file_system (str): Supported only for IOS and NXOS. The file system for the
remote file. If no file_system is provided, then the ``get_file_system``
method is used to determine the correct file system to use.
read_timeout (int): Maximum time in seconds to wait for the checksum command to
complete. Hashing large files can take several minutes (default: 900).

Returns:
(str): The checksum of the remote file.
Expand All @@ -663,7 +665,7 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5", file_system=Non
if file_system is None:
file_system = self._get_file_system()
cmd = f"verify /{hashing_algorithm} {file_system}/{filename}"
result = self.native.send_command_timing(cmd, read_timeout=300)
result = self.native.send_command_timing(cmd, read_timeout=read_timeout)

patterns = [r"=\s+(\S+)", r"^([a-fA-F0-9]+)$"]
for pattern in patterns:
Expand All @@ -675,7 +677,7 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5", file_system=Non
hashing_algorithm,
match[1],
)
return match[1]
return match[1]
log.error(
"Host %s: Unable to get remote checksum for file %s with hashing algorithm %s",
self.host,
Expand Down Expand Up @@ -718,7 +720,7 @@ def check_file_exists(self, filename, file_system=None):
return True
raise CommandError(cmd, f"Unable to determine if file {filename} exists on remote: {result}")

def verify_file(self, checksum, filename, hashing_algorithm="md5", file_system=None):
def verify_file(self, checksum, filename, hashing_algorithm="md5", file_system=None, read_timeout=900):
"""Verify a file on the remote device by and validate the checksums.

Args:
Expand All @@ -728,12 +730,14 @@ def verify_file(self, checksum, filename, hashing_algorithm="md5", file_system=N
file_system (str): Supported only for IOS and NXOS. The file system for the
remote file. If no file_system is provided, then the ``get_file_system``
method is used to determine the correct file system to use.
read_timeout (int): Maximum time in seconds to wait for the checksum command to
complete. Hashing large files can take several minutes (default: 900).

Returns:
(bool): True if the file is verified successfully, False otherwise.
"""
return self.check_file_exists(filename, file_system=file_system) and self.compare_file_checksum(
checksum, filename, hashing_algorithm, file_system=file_system
checksum, filename, hashing_algorithm, file_system=file_system, read_timeout=read_timeout
)

def file_copy(self, src, dest=None, file_system=None):
Expand Down Expand Up @@ -900,7 +904,9 @@ def _resolve_install_mode(self, install_mode):
)
return install_mode

def install_os(self, image_name, reboot=True, install_mode=None, read_timeout=2000, **vendor_specifics):
def install_os( # pylint: disable=too-many-branches
self, image_name, reboot=True, install_mode=None, read_timeout=2000, **vendor_specifics
):
"""Installs the prescribed Network OS, which must be present before issuing this command.

Args:
Expand Down Expand Up @@ -951,12 +957,27 @@ def install_os(self, image_name, reboot=True, install_mode=None, read_timeout=20
install_message = self.show(command, read_timeout=read_timeout)
if install_message.startswith("FAILED:"):
log.error("Host %s: OS install error for image %s", self.host, image_name)
raise OSInstallError(hostname=self.hostname, desired_boot=image_name)
raise OSInstallError(
hostname=self.hostname, desired_boot=image_name, detail=install_message
)
except IOError:
log.error("Host %s: IO error for image %s", self.host, image_name)
except CommandError:
except CommandError as original_error:
log.warning(
"Host %s: install command failed (%s); falling back to legacy "
"'request platform software package install' command.",
self.host,
original_error.cli_error_msg,
)
command = f"request platform software package install switch all file {self._get_file_system()}{image_name} auto-copy"
self.show(command, read_timeout=read_timeout)
try:
self.show(command, read_timeout=read_timeout)
except CommandError as fallback_error:
raise CommandError(
command,
f"{fallback_error.cli_error_msg} (legacy fallback; original install "
f"error: {original_error.cli_error_msg})",
) from original_error
self.reboot()
else:
self.set_boot_options(image_name, **vendor_specifics)
Expand Down
12 changes: 8 additions & 4 deletions pyntc/devices/iosxr_device.py
Original file line number Diff line number Diff line change
Expand Up @@ -408,7 +408,7 @@ def enable(self):
"""
log.debug("Host %s: enable() is a no-op on IOS-XR.", self.host)

def get_remote_checksum(self, filename, hashing_algorithm="md5", file_system=None):
def get_remote_checksum(self, filename, hashing_algorithm="md5", file_system=None, read_timeout=900):
"""Get the checksum of a remote file.

Args:
Expand All @@ -417,6 +417,8 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5", file_system=Non
file_system (str): The file system for the remote file.
If no file_system is provided, then the ``get_file_system``
method is used to determine the correct file system to use.
read_timeout (int): Maximum time in seconds to wait for the checksum command to
complete. Hashing large files can take several minutes (default: 900).

Returns:
(str): The checksum of the remote file.
Expand All @@ -434,7 +436,7 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5", file_system=Non
if not file_system.startswith("/"):
file_system = "/" + file_system
cmd = f"run {hashing_algorithm}sum {file_system}/{filename}"
result = self._send_command(cmd, read_timeout=300)
result = self._send_command(cmd, read_timeout=read_timeout)

match = re.search(r"^([a-fA-F0-9]+)\s", result, flags=re.MULTILINE)
if match:
Expand Down Expand Up @@ -481,7 +483,7 @@ def check_file_exists(self, filename, file_system=None):
log.debug("Host %s: File %s not found in 'dir' output on %s.", self.host, filename, file_system)
return False

def verify_file(self, checksum, filename, hashing_algorithm="md5", file_system=None):
def verify_file(self, checksum, filename, hashing_algorithm="md5", file_system=None, read_timeout=900):
"""Verify a file on the remote device exists and its checksum matches.

Args:
Expand All @@ -491,12 +493,14 @@ def verify_file(self, checksum, filename, hashing_algorithm="md5", file_system=N
file_system (str): The file system for the remote file. If no file_system
is provided, then the ``_get_file_system`` method is used to determine
the correct file system to use.
read_timeout (int): Maximum time in seconds to wait for the checksum command to
complete. Hashing large files can take several minutes (default: 900).

Returns:
(bool): True if the file is verified successfully, False otherwise.
"""
return self.check_file_exists(filename, file_system=file_system) and self.compare_file_checksum(
checksum, filename, hashing_algorithm, file_system=file_system
checksum, filename, hashing_algorithm, file_system=file_system, read_timeout=read_timeout
)

def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kwargs):
Expand Down
5 changes: 4 additions & 1 deletion pyntc/errors.py
Original file line number Diff line number Diff line change
Expand Up @@ -229,15 +229,18 @@ def __init__(self, hostname, min_space=None, *, required=None, available=None, f
class OSInstallError(NTCError):
"""Error for failing to install an OS on a device."""

def __init__(self, hostname, desired_boot):
def __init__(self, hostname, desired_boot, detail=None):
"""
Error for failing to install an OS on a device.

Args:
hostname (str): The hostname of the device that failed to install OS.
desired_boot (str): The OS that was attempted to be installed.
detail (str, optional): The error output reported by the device.
"""
message = f"{hostname} was unable to boot into {desired_boot}"
if detail:
message = f"{message}: {detail}"
super().__init__(message)


Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[tool.poetry]
name = "pyntc"
version = "3.2.1"
version = "3.2.2"
description = "Python library focused on tasks related to device level and OS management."
authors = ["Network to Code, LLC <opensource@networktocode.com>"]
readme = "README.md"
Expand Down
16 changes: 13 additions & 3 deletions tests/unit/test_devices/test_asa_device.py
Original file line number Diff line number Diff line change
Expand Up @@ -969,7 +969,7 @@ def test_get_remote_checksum_md5(mock_fs, asa_device):
)
result = asa_device.get_remote_checksum("asa.bin")
assert result == MD5_CHECKSUM
asa_device.native.send_command_timing.assert_called_with("verify /md5 disk0:asa.bin", read_timeout=300)
asa_device.native.send_command_timing.assert_called_with("verify /md5 disk0:asa.bin", read_timeout=900)


@mock.patch.object(ASADevice, "_get_file_system", return_value="disk0:")
Expand All @@ -979,7 +979,7 @@ def test_get_remote_checksum_sha512(mock_fs, asa_device):
)
result = asa_device.get_remote_checksum("asa.bin", hashing_algorithm="sha512")
assert result == SHA512_CHECKSUM
asa_device.native.send_command_timing.assert_called_with("verify /sha-512 disk0:asa.bin", read_timeout=300)
asa_device.native.send_command_timing.assert_called_with("verify /sha-512 disk0:asa.bin", read_timeout=900)


@mock.patch.object(ASADevice, "_get_file_system", return_value="disk0:")
Expand All @@ -989,10 +989,20 @@ def test_get_remote_checksum_uses_provided_file_system(mock_fs, asa_device):
)
result = asa_device.get_remote_checksum("asa.bin", file_system="flash:")
assert result == MD5_CHECKSUM
asa_device.native.send_command_timing.assert_called_with("verify /md5 flash:asa.bin", read_timeout=300)
asa_device.native.send_command_timing.assert_called_with("verify /md5 flash:asa.bin", read_timeout=900)
mock_fs.assert_not_called()


@mock.patch.object(ASADevice, "_get_file_system", return_value="disk0:")
def test_get_remote_checksum_custom_read_timeout(mock_fs, asa_device):
asa_device.native.send_command_timing.return_value = (
f"!!!!!!!!!!!!!!!!!!!!!!!!Done!\nverify /MD5 (disk0:/asa.bin) = {MD5_CHECKSUM}"
)
result = asa_device.get_remote_checksum("asa.bin", read_timeout=1800)
assert result == MD5_CHECKSUM
asa_device.native.send_command_timing.assert_called_with("verify /md5 disk0:asa.bin", read_timeout=1800)


def test_get_remote_checksum_invalid_algorithm(asa_device):
with pytest.raises(ValueError, match="hashing_algorithm must be"):
asa_device.get_remote_checksum("asa.bin", hashing_algorithm="sha256")
Expand Down
Loading
Loading