diff --git a/docs/admin/release_notes/version_3.0.md b/docs/admin/release_notes/version_3.0.md index 336193ee..dd258994 100644 --- a/docs/admin/release_notes/version_3.0.md +++ b/docs/admin/release_notes/version_3.0.md @@ -7,6 +7,20 @@ This document describes all new features and changes in the release. The format - Pyntc now requires the `PYNTC_LOG_FILE` environment variable to output logging to a file. The new default behavior is to only log to stderr. +## [v3.0.2 (2026-05-28)](https://github.com/networktocode/pyntc/releases/tag/v3.0.2) + +### Added + +- [#395](https://github.com/networktocode/pyntc/issues/395) - Added an ``install_mode`` property to ``BaseDevice`` (abstract) and ``IOSDevice``; the IOS implementation returns ``True`` when the device boots from ``packages.conf``. + +### Deprecated + +- [#395](https://github.com/networktocode/pyntc/issues/395) - Deprecated the ``install_mode`` argument to ``IOSDevice.install_os``; install mode is now derived from the device's ``boot_options`` via the new ``install_mode`` property and will be removed in a future release. + +### Fixed + +- [#394](https://github.com/networktocode/pyntc/issues/394) - Fixed Junos reboots not being detected when waiting for the device to reload. +- [#394](https://github.com/networktocode/pyntc/issues/394) - Increased the default Junos reboot wait timeout from 1 hour to 2 hours. ## [v3.0.1 (2026-05-26)](https://github.com/networktocode/pyntc/releases/tag/v3.0.1) diff --git a/pyntc/devices/base_device.py b/pyntc/devices/base_device.py index b24a7496..b01c8ff6 100644 --- a/pyntc/devices/base_device.py +++ b/pyntc/devices/base_device.py @@ -450,6 +450,20 @@ def verify_file(self, checksum, filename, hashing_algorithm="md5", **kwargs): """ raise NotImplementedError + @property + def install_mode(self): + """Indicate whether the device is operating in install mode. + + Drivers override this to derive the value from the device's current boot + configuration. Used by ``install_os`` to choose between install-mode and + legacy upgrade procedures. + + Returns: + (bool): True when the device boots from an install-mode bundle + (e.g., ``packages.conf`` on IOS-XE), False otherwise. + """ + raise NotImplementedError + def install_os(self, image_name, reboot=True, **vendor_specifics): """Install the OS from specified image_name. diff --git a/pyntc/devices/ios_device.py b/pyntc/devices/ios_device.py index 471c53c5..5baeb4e3 100644 --- a/pyntc/devices/ios_device.py +++ b/pyntc/devices/ios_device.py @@ -3,6 +3,7 @@ import os import re import time +import warnings from netmiko import ConnectHandler, FileTransfer from netmiko.exceptions import ReadTimeout @@ -319,6 +320,17 @@ def boot_options(self): log.debug("Host %s: the boot options are {dict(sys=boot_image)}", self.host) return {"sys": boot_image} + @property + def install_mode(self): + """Return whether the device is currently booted in install mode. + + Returns: + (bool): True when the current boot image equals + :data:`INSTALL_MODE_FILE_NAME` (i.e., ``packages.conf``), + False otherwise. + """ + return self.boot_options.get("sys") == INSTALL_MODE_FILE_NAME + def checkpoint(self, checkpoint_file): """Create checkpoint file. @@ -877,13 +889,28 @@ def file_copy_remote_exists(self, src, dest=None, file_system=None): log.debug("Host %s: File %s does not already exist on remote.", self.host, src) return False - def install_os(self, image_name, reboot=True, install_mode=False, read_timeout=2000, **vendor_specifics): + def _resolve_install_mode(self, install_mode): + """Return the effective install_mode flag, warning if the caller passed it explicitly.""" + if install_mode is None: + return self.install_mode + warnings.warn( + "The install_mode argument to install_os is deprecated; install mode is now " + "derived from the device's boot_options via the install_mode property.", + DeprecationWarning, + ) + return install_mode + + def install_os(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: image_name (str): Name of the IOS image to boot into reboot (bool): Whether to reboot the device after setting the boot options. Defaults to true. - install_mode (bool, optional): Uses newer install method on devices. Defaults to False. + install_mode (bool, optional): **Deprecated.** Whether to use the newer install-mode + upgrade procedure. When omitted (the default), the value is derived from + :attr:`install_mode`, which reads the device's current boot configuration. + Passing the argument explicitly still works but emits a ``DeprecationWarning`` + and will be removed in a future release. read_timeout (int, optional): Netmiko timeout when waiting for device prompt. Default 2000. vendor_specifics (dict, optional): Vendor specific arguments to pass to the install command. @@ -893,14 +920,15 @@ def install_os(self, image_name, reboot=True, install_mode=False, read_timeout=2 Returns: (bool): False if no install is needed, true if the install completes successfully """ + use_install_mode = self._resolve_install_mode(install_mode) timeout = vendor_specifics.get("timeout", 3600) if not self._image_booted(image_name): - if install_mode and not reboot: + if use_install_mode and not reboot: raise ValueError( "IOS devices automatically reboot after installation when using install mode but the reboot argument was set to false." ) - if install_mode: + if use_install_mode: # Change boot statement to be boot system :packages.conf self.set_boot_options(INSTALL_MODE_FILE_NAME, **vendor_specifics) @@ -942,7 +970,7 @@ def install_os(self, image_name, reboot=True, install_mode=False, read_timeout=2 self._wait_for_device_reboot(timeout=timeout) # Set FastCLI back to originally set when using install mode - if install_mode: + if use_install_mode: image_name = INSTALL_MODE_FILE_NAME # Verify the OS level if not self._image_booted(image_name): diff --git a/pyntc/devices/jnpr_device.py b/pyntc/devices/jnpr_device.py index 70c8fcad..b6431204 100644 --- a/pyntc/devices/jnpr_device.py +++ b/pyntc/devices/jnpr_device.py @@ -228,18 +228,53 @@ def _uptime_to_string(self, uptime_full_string): days, hours, minutes, seconds = self._uptime_components(uptime_full_string) return f"{days:02d}:{hours:02d}:{minutes:02d}:{seconds:02d}" - def _wait_for_device_reboot(self, timeout=3600): + def _wait_for_device_reboot(self, original_uptime, timeout=7200): + """Block until the device reboots and accepts a fresh connection. + + Drops the existing NETCONF session and polls for the device to come back. + The reboot is considered complete when a new connection succeeds and the + device reports an uptime lower than ``original_uptime`` (i.e., it has + booted since the reboot was issued). + + The pre-reboot session must be discarded first: once the device restarts, + PyEZ still reports it as connected even though the transport is dead, so it + is closed here to force each probe to establish a fresh connection. + + Args: + original_uptime (int): Device uptime in seconds captured before the reboot. + timeout (int, optional): Max seconds to wait for the device to return. Defaults to 2 hours. + """ start = time.time() - disconnected = False + + # Drop the pre-reboot NETCONF session so subsequent probes can't read from + # a stale connection PyEZ still reports as connected. + try: + self.close() + except Exception as close_exc: # pylint: disable=broad-exception-caught + log.debug("Host %s: Pre-reboot disconnect raised %s (ignored).", self.host, close_exc) + while time.time() - start < timeout: - if disconnected: - try: - self.open() + try: + self.open() + self._uptime = None + current_uptime = self.uptime + if current_uptime is not None and current_uptime < original_uptime: + log.info( + "Host %s: Device rebooted (uptime %ss < pre-reboot %ss).", + self.host, + current_uptime, + original_uptime, + ) return - except: # noqa E722 # nosec # pylint: disable=bare-except - pass - elif not self.connected: - disconnected = True + log.debug( + "Host %s: Reachable but uptime %ss >= pre-reboot %ss; still waiting.", + self.host, + current_uptime, + original_uptime, + ) + except Exception as exc: # pylint: disable=broad-exception-caught + log.debug("Host %s: Reboot probe failed (%s); will retry.", self.host, exc) + self.native.connected = False time.sleep(10) raise RebootTimeoutError(hostname=self.hostname, wait_time=timeout) @@ -318,12 +353,14 @@ def uptime(self): Returns: (int): Device uptime in seconds. """ - try: - native_uptime_string = self.native.facts["RE0"]["up_time"] - except (AttributeError, TypeError): - native_uptime_string = None - if self._uptime is None: + try: + # Bust PyEZ's cached facts so a cold cache always reflects the live device. + self.native.facts_refresh(keys="RE0") + native_uptime_string = self.native.facts["RE0"]["up_time"] + except (AttributeError, TypeError, KeyError): + native_uptime_string = None + if native_uptime_string is not None: self._uptime = self._uptime_to_seconds(native_uptime_string) @@ -337,13 +374,16 @@ def uptime_string(self): Returns: (str): Device uptime. """ - try: - native_uptime_string = self.native.facts["RE0"]["up_time"] - except (AttributeError, TypeError): - native_uptime_string = None - if self._uptime_string is None: - self._uptime_string = self._uptime_to_string(native_uptime_string) + try: + # Bust PyEZ's cached facts so a cold cache always reflects the live device. + self.native.facts_refresh(keys="RE0") + native_uptime_string = self.native.facts["RE0"]["up_time"] + except (AttributeError, TypeError, KeyError): + native_uptime_string = None + + if native_uptime_string is not None: + self._uptime_string = self._uptime_to_string(native_uptime_string) return self._uptime_string @@ -505,13 +545,13 @@ def open(self): if not self.connected: self.native.open() - def reboot(self, wait_for_reload=False, timeout=3600, confirm=None): + def reboot(self, wait_for_reload=False, timeout=7200, confirm=None): """ Reload the controller or controller pair. Args: wait_for_reload (bool): Whether the reboot method should wait for the device to come back up before returning. Defaults to False. - timeout (int, optional): Time in seconds to wait for the device to return after reboot. Defaults to 1 hour. + timeout (int, optional): Time in seconds to wait for the device to return after reboot. Defaults to 2 hours. confirm (None): Not used. Deprecated since v0.17.0. Example: @@ -522,9 +562,16 @@ def reboot(self, wait_for_reload=False, timeout=3600, confirm=None): if confirm is not None: warnings.warn("Passing 'confirm' to reboot method is deprecated.", DeprecationWarning) + self._uptime = None + original_uptime = self.uptime + if original_uptime is None: + raise CommandError( + command="reboot", + message="Could not determine pre-reboot uptime; refusing to wait for reload.", + ) self.sw.reboot(in_min=0) if wait_for_reload: - self._wait_for_device_reboot(timeout=timeout) + self._wait_for_device_reboot(original_uptime, timeout=timeout) def rollback(self, filename): """Rollback to a specific configuration file. diff --git a/pyproject.toml b/pyproject.toml index f8bc23fa..978fdc61 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "pyntc" -version = "3.0.1" +version = "3.0.2" description = "Python library focused on tasks related to device level and OS management." authors = ["Network to Code, LLC "] readme = "README.md" diff --git a/tests/unit/test_devices/test_base_device.py b/tests/unit/test_devices/test_base_device.py new file mode 100644 index 00000000..4e3c6ecb --- /dev/null +++ b/tests/unit/test_devices/test_base_device.py @@ -0,0 +1,15 @@ +"""Tests for the abstract :class:`pyntc.devices.base_device.BaseDevice` contract.""" + +import pytest + +from pyntc.devices.base_device import BaseDevice + + +@pytest.fixture +def base_device(): + return BaseDevice(host="host", username="user", password="pass") + + +def test_install_mode_raises_not_implemented(base_device): + with pytest.raises(NotImplementedError): + _ = base_device.install_mode diff --git a/tests/unit/test_devices/test_ios_device.py b/tests/unit/test_devices/test_ios_device.py index ccc141e0..6e7b9b47 100644 --- a/tests/unit/test_devices/test_ios_device.py +++ b/tests/unit/test_devices/test_ios_device.py @@ -392,22 +392,26 @@ def test_enable_from_config(self): self.device.native.check_config_mode.assert_called() self.device.native.exit_config_mode.assert_called() + @mock.patch.object(IOSDevice, "install_mode", new_callable=mock.PropertyMock, return_value=False) @mock.patch.object(IOSDevice, "_image_booted", side_effect=[False, True]) @mock.patch.object(IOSDevice, "set_boot_options") @mock.patch.object(IOSDevice, "reboot") @mock.patch.object(IOSDevice, "_wait_for_device_reboot") - def test_install_os(self, mock_wait, mock_reboot, mock_set_boot, mock_image_booted): + def test_install_os(self, mock_wait, mock_reboot, mock_set_boot, mock_image_booted, mock_install_mode): state = self.device.install_os(BOOT_IMAGE) mock_set_boot.assert_called() mock_reboot.assert_called() mock_wait.assert_called() self.assertEqual(state, True) + @mock.patch.object(IOSDevice, "install_mode", new_callable=mock.PropertyMock, return_value=False) @mock.patch.object(IOSDevice, "_image_booted", side_effect=[True]) @mock.patch.object(IOSDevice, "set_boot_options") @mock.patch.object(IOSDevice, "reboot") @mock.patch.object(IOSDevice, "_wait_for_device_reboot") - def test_install_os_already_installed(self, mock_wait, mock_reboot, mock_set_boot, mock_image_booted): + def test_install_os_already_installed( + self, mock_wait, mock_reboot, mock_set_boot, mock_image_booted, mock_install_mode + ): state = self.device.install_os(BOOT_IMAGE) mock_image_booted.assert_called_once() mock_set_boot.assert_not_called() @@ -415,15 +419,19 @@ def test_install_os_already_installed(self, mock_wait, mock_reboot, mock_set_boo mock_wait.assert_not_called() self.assertEqual(state, False) + @mock.patch.object(IOSDevice, "install_mode", new_callable=mock.PropertyMock, return_value=False) @mock.patch.object(IOSDevice, "_image_booted", side_effect=[False, False]) @mock.patch.object(IOSDevice, "set_boot_options") @mock.patch.object(IOSDevice, "reboot") @mock.patch.object(IOSDevice, "_wait_for_device_reboot") @mock.patch.object(IOSDevice, "_raw_version_data") - def test_install_os_error(self, mock_wait, mock_reboot, mock_set_boot, mock_image_booted, mock_raw_version_data): + def test_install_os_error( + self, mock_wait, mock_reboot, mock_set_boot, mock_image_booted, mock_raw_version_data, mock_install_mode + ): mock_raw_version_data.return_value = DEVICE_FACTS self.assertRaises(ios_module.OSInstallError, self.device.install_os, BOOT_IMAGE) + @mock.patch.object(IOSDevice, "install_mode", new_callable=mock.PropertyMock, return_value=True) @mock.patch.object(IOSDevice, "os_version", new_callable=mock.PropertyMock) @mock.patch.object(IOSDevice, "_image_booted", side_effect=[False, True]) @mock.patch.object(IOSDevice, "set_boot_options") @@ -440,11 +448,12 @@ def test_install_os_not_enough_space( mock_set_boot, mock_image_booted, mock_os_version, + mock_install_mode, ): mock_raw_version_data.return_value = DEVICE_FACTS mock_os_version.return_value = "17.4.3" mock_show.return_value = "FAILED: There is not enough free disk available to perform this operation on switch 1. At least 1276287 KB of free disk is required" - self.assertRaises(ios_module.OSInstallError, self.device.install_os, image_name=BOOT_IMAGE, install_mode=True) + self.assertRaises(ios_module.OSInstallError, self.device.install_os, image_name=BOOT_IMAGE) mock_wait.assert_not_called() mock_reboot.assert_not_called() @@ -1260,12 +1269,36 @@ def test_set_boot_options_image_packages_conf_file( mock_boot_options.assert_called_once() +# +# TESTS FOR IOS INSTALL MODE PROPERTY +# + + +@mock.patch.object(IOSDevice, "boot_options", new_callable=mock.PropertyMock) +def test_install_mode_true_when_boot_image_is_packages_conf(mock_boot_options, ios_device): + mock_boot_options.return_value = {"sys": ios_module.INSTALL_MODE_FILE_NAME} + assert ios_device.install_mode is True + + +@mock.patch.object(IOSDevice, "boot_options", new_callable=mock.PropertyMock) +def test_install_mode_false_for_image_bin(mock_boot_options, ios_device): + mock_boot_options.return_value = {"sys": "cat9k_iosxe.16.12.04.SPA.bin"} + assert ios_device.install_mode is False + + +@mock.patch.object(IOSDevice, "boot_options", new_callable=mock.PropertyMock) +def test_install_mode_false_for_missing_sys_key(mock_boot_options, ios_device): + mock_boot_options.return_value = {} + assert ios_device.install_mode is False + + # # TESTS FOR IOS INSTALL MODE METHOD # # Test install mode upgrade for install mode with latest method +@mock.patch.object(IOSDevice, "install_mode", new_callable=mock.PropertyMock) @mock.patch.object(IOSDevice, "os_version", new_callable=mock.PropertyMock) @mock.patch.object(IOSDevice, "_image_booted") @mock.patch.object(IOSDevice, "set_boot_options") @@ -1281,16 +1314,18 @@ def test_install_os_install_mode( mock_set_boot_options, mock_image_booted, mock_os_version, + mock_install_mode, ios_device, ): image_name = "cat9k_iosxe.16.12.04.SPA.bin" file_system = "flash:" + mock_install_mode.return_value = True mock_get_file_system.return_value = file_system mock_os_version.return_value = "16.12.03a" mock_image_booted.side_effect = [False, True] mock_show.side_effect = [IOError("Search pattern never detected in send_command")] # Call the install os function - actual = ios_device.install_os(image_name, install_mode=True) + actual = ios_device.install_os(image_name) # Check the results mock_set_boot_options.assert_called_with("packages.conf") @@ -1305,6 +1340,7 @@ def test_install_os_install_mode( # Test install mode upgrade fail +@mock.patch.object(IOSDevice, "install_mode", new_callable=mock.PropertyMock) @mock.patch.object(IOSDevice, "os_version", new_callable=mock.PropertyMock) @mock.patch.object(IOSDevice, "_image_booted") @mock.patch.object(IOSDevice, "set_boot_options") @@ -1322,8 +1358,10 @@ def test_install_os_install_mode_failed( mock_set_boot_options, mock_image_booted, mock_os_version, + mock_install_mode, ios_device, ): + mock_install_mode.return_value = True mock_hostname.return_value = "ntc-rtr01" image_name = "cat9k_iosxe.16.12.04.SPA.bin" file_system = "flash:" @@ -1333,7 +1371,7 @@ def test_install_os_install_mode_failed( mock_show.side_effect = [IOError("Search pattern never detected in send_command")] # Call the install os function with pytest.raises(ios_module.OSInstallError) as err: - ios_device.install_os(image_name, install_mode=True) + ios_device.install_os(image_name) assert err.value.message == "ntc-rtr01 was unable to boot into packages.conf" @@ -1349,6 +1387,7 @@ def test_install_os_install_mode_failed( # Test install mode upgrade for install mode with latest method +@mock.patch.object(IOSDevice, "install_mode", new_callable=mock.PropertyMock) @mock.patch.object(IOSDevice, "os_version", new_callable=mock.PropertyMock) @mock.patch.object(IOSDevice, "_image_booted") @mock.patch.object(IOSDevice, "set_boot_options") @@ -1364,16 +1403,18 @@ def test_install_os_install_mode_no_upgrade( mock_set_boot_options, mock_image_booted, mock_os_version, + mock_install_mode, ios_device, ): image_name = "cat9k_iosxe.16.12.04.SPA.bin" file_system = "flash:" + mock_install_mode.return_value = True mock_get_file_system.return_value = file_system mock_os_version.return_value = "16.12.03a" mock_image_booted.side_effect = [True, True] mock_show.side_effect = [IOError("Search pattern never detected in send_command")] # Call the install os function - actual = ios_device.install_os(image_name, install_mode=True) + actual = ios_device.install_os(image_name) # Check the results mock_set_boot_options.assert_not_called() @@ -1391,6 +1432,7 @@ def test_install_os_install_mode_no_upgrade( # Test install mode upgrade for install mode with interim method on OS Version +@mock.patch.object(IOSDevice, "install_mode", new_callable=mock.PropertyMock) @mock.patch.object(IOSDevice, "os_version", new_callable=mock.PropertyMock) @mock.patch.object(IOSDevice, "_image_booted") @mock.patch.object(IOSDevice, "set_boot_options") @@ -1406,15 +1448,17 @@ def test_install_os_install_mode_from_everest( mock_set_boot_options, mock_image_booted, mock_os_version, + mock_install_mode, ios_device, ): image_name = "cat9k_iosxe.16.12.04.SPA.bin" file_system = "flash:" + mock_install_mode.return_value = True mock_get_file_system.return_value = file_system mock_os_version.return_value = "16.6.1" mock_image_booted.side_effect = [False, True] # Call the install_os - actual = ios_device.install_os(image_name, install_mode=True) + actual = ios_device.install_os(image_name) # Test the results mock_set_boot_options.assert_called_with("packages.conf") @@ -1430,6 +1474,7 @@ def test_install_os_install_mode_from_everest( # Test install mode upgrade for install mode with interim method on OS Version with error unable to complete +@mock.patch.object(IOSDevice, "install_mode", new_callable=mock.PropertyMock) @mock.patch.object(IOSDevice, "os_version", new_callable=mock.PropertyMock) @mock.patch.object(IOSDevice, "_image_booted") @mock.patch.object(IOSDevice, "set_boot_options") @@ -1448,8 +1493,10 @@ def test_install_os_install_mode_from_everest_failed( mock_set_boot_options, mock_image_booted, mock_os_version, + mock_install_mode, ios_device, ): + mock_install_mode.return_value = True mock_hostname.return_value = "ntc-rtr01" image_name = "cat9k_iosxe.16.12.04.SPA.bin" file_system = "flash:" @@ -1458,7 +1505,7 @@ def test_install_os_install_mode_from_everest_failed( mock_image_booted.side_effect = [False, False] # Call the install_os with pytest.raises(ios_module.OSInstallError) as err: - ios_device.install_os(image_name, install_mode=True) + ios_device.install_os(image_name) assert err.value.message == "ntc-rtr01 was unable to boot into packages.conf" @@ -1475,6 +1522,7 @@ def test_install_os_install_mode_from_everest_failed( # Test install mode upgrade for install mode with interim method on OS Version with error unable to complete +@mock.patch.object(IOSDevice, "install_mode", new_callable=mock.PropertyMock) @mock.patch.object(IOSDevice, "os_version", new_callable=mock.PropertyMock) @mock.patch.object(IOSDevice, "_image_booted") @mock.patch.object(IOSDevice, "set_boot_options") @@ -1493,8 +1541,10 @@ def test_install_os_install_mode_from_everest_to_everest( mock_set_boot_options, mock_image_booted, mock_os_version, + mock_install_mode, ios_device, ): + mock_install_mode.return_value = True mock_hostname.return_value = "ntc-rtr01" image_name = "cat9k_iosxe.16.05.01a.SPA.bin" file_system = "flash:" @@ -1502,7 +1552,7 @@ def test_install_os_install_mode_from_everest_to_everest( mock_os_version.return_value = "16.5.1" mock_image_booted.side_effect = [True, True] # Call the install_os - actual = ios_device.install_os(image_name, install_mode=True) + actual = ios_device.install_os(image_name) # Test the results mock_set_boot_options.assert_not_called() @@ -1514,6 +1564,7 @@ def test_install_os_install_mode_from_everest_to_everest( assert actual is False +@mock.patch.object(IOSDevice, "install_mode", new_callable=mock.PropertyMock) @mock.patch.object(IOSDevice, "_has_reload_happened_recently") @mock.patch.object(IOSDevice, "os_version", new_callable=mock.PropertyMock) @mock.patch.object(IOSDevice, "_image_booted") @@ -1531,10 +1582,12 @@ def test_install_os_install_mode_with_retries( mock_image_booted, mock_os_version, mock_has_reload_happened_recently, + mock_install_mode, ios_device, ): image_name = "cat9k_iosxe.16.12.04.SPA.bin" file_system = "flash:" + mock_install_mode.return_value = True mock_get_file_system.return_value = file_system mock_os_version.return_value = "16.12.03a" mock_has_reload_happened_recently.side_effect = [False, False, True] @@ -1542,7 +1595,7 @@ def test_install_os_install_mode_with_retries( mock_sleep.return_value = None mock_show.return_value = "show must go on" # Call the install os function - actual = ios_device.install_os(image_name, install_mode=True) + actual = ios_device.install_os(image_name) # Check the results mock_set_boot_options.assert_called_with("packages.conf") @@ -1556,6 +1609,113 @@ def test_install_os_install_mode_with_retries( assert actual is True +# +# TESTS FOR install_os PROPERTY-DRIVEN BEHAVIOR & DEPRECATION +# + + +@mock.patch.object(IOSDevice, "install_mode", new_callable=mock.PropertyMock) +@mock.patch.object(IOSDevice, "_image_booted") +@mock.patch.object(IOSDevice, "set_boot_options") +@mock.patch.object(IOSDevice, "show") +@mock.patch.object(IOSDevice, "_wait_for_device_reboot") +@mock.patch.object(IOSDevice, "_get_file_system") +@mock.patch.object(IOSDevice, "reboot") +def test_install_os_uses_install_mode_property_true( + mock_reboot, + mock_get_file_system, + mock_wait_for_reboot, + mock_show, + mock_set_boot_options, + mock_image_booted, + mock_install_mode, + ios_device, +): + image_name = "cat9k_iosxe.16.12.04.SPA.bin" + file_system = "flash:" + mock_install_mode.return_value = True + mock_get_file_system.return_value = file_system + mock_image_booted.side_effect = [False, True] + mock_show.side_effect = [IOError("Search pattern never detected in send_command")] + with mock.patch.object(IOSDevice, "os_version", new_callable=mock.PropertyMock) as mock_os_version: + mock_os_version.return_value = "16.12.03a" + actual = ios_device.install_os(image_name) + + mock_set_boot_options.assert_called_with("packages.conf") + mock_show.assert_called_with( + f"install add file {file_system}{image_name} activate commit prompt-level none", read_timeout=2000 + ) + mock_reboot.assert_not_called() + mock_wait_for_reboot.assert_called() + assert actual is True + + +@mock.patch.object(IOSDevice, "install_mode", new_callable=mock.PropertyMock) +@mock.patch.object(IOSDevice, "_image_booted") +@mock.patch.object(IOSDevice, "set_boot_options") +@mock.patch.object(IOSDevice, "show") +@mock.patch.object(IOSDevice, "_wait_for_device_reboot") +@mock.patch.object(IOSDevice, "reboot") +def test_install_os_uses_install_mode_property_false( + mock_reboot, + mock_wait_for_reboot, + mock_show, + mock_set_boot_options, + mock_image_booted, + mock_install_mode, + ios_device, +): + image_name = "cat9k_iosxe.16.12.04.SPA.bin" + mock_install_mode.return_value = False + mock_image_booted.side_effect = [False, True] + actual = ios_device.install_os(image_name) + + mock_set_boot_options.assert_called_once_with(image_name) + # The install-mode "install add file" command must NOT be issued on the legacy path. + for call in mock_show.call_args_list: + args, _ = call + if args: + assert "install add file" not in args[0] + mock_reboot.assert_called_once() + mock_wait_for_reboot.assert_called() + assert actual is True + + +@mock.patch.object(IOSDevice, "install_mode", new_callable=mock.PropertyMock) +@mock.patch.object(IOSDevice, "os_version", new_callable=mock.PropertyMock) +@mock.patch.object(IOSDevice, "_image_booted") +@mock.patch.object(IOSDevice, "set_boot_options") +@mock.patch.object(IOSDevice, "show") +@mock.patch.object(IOSDevice, "_wait_for_device_reboot") +@mock.patch.object(IOSDevice, "_get_file_system") +@mock.patch.object(IOSDevice, "reboot") +def test_install_os_install_mode_kwarg_emits_deprecation_warning( + mock_reboot, + mock_get_file_system, + mock_wait_for_reboot, + mock_show, + mock_set_boot_options, + mock_image_booted, + mock_os_version, + mock_install_mode, + ios_device, +): + image_name = "cat9k_iosxe.16.12.04.SPA.bin" + file_system = "flash:" + # Property would say False, but the explicit kwarg must override. + mock_install_mode.return_value = False + mock_get_file_system.return_value = file_system + mock_os_version.return_value = "16.12.03a" + mock_image_booted.side_effect = [False, True] + mock_show.side_effect = [IOError("Search pattern never detected in send_command")] + + with pytest.warns(DeprecationWarning, match="install_mode argument"): + actual = ios_device.install_os(image_name, install_mode=True) + + mock_set_boot_options.assert_called_with("packages.conf") + assert actual is True + + def test_show(ios_send_command): command = "show_ip_arp" device = ios_send_command([f"{command}.txt"]) diff --git a/tests/unit/test_devices/test_jnpr_device.py b/tests/unit/test_devices/test_jnpr_device.py index 5b9d325d..28271a64 100644 --- a/tests/unit/test_devices/test_jnpr_device.py +++ b/tests/unit/test_devices/test_jnpr_device.py @@ -244,18 +244,28 @@ def test_reboot(self): @mock.patch("pyntc.devices.jnpr_device.time.sleep") def test_wait_for_device_to_reboot(self, mock_sleep): - with mock.patch.object(self.device, "open") as mock_open: - # Emulate the device disconnected and reconnecting - type(self.device.native).connected = mock.PropertyMock(side_effect=[True, False, True]) - mock_open.side_effect = [Exception, Exception, True] - self.device.reboot(wait_for_reload=True, timeout=3) - mock_open.assert_has_calls([mock.call()] * 3) + """Reboot completes when the device returns with a lower uptime than the baseline.""" + with ( + mock.patch.object(self.device, "open") as mock_open, + mock.patch.object(self.device, "close"), + mock.patch.object(type(self.device), "uptime", new_callable=mock.PropertyMock) as mock_uptime, + ): + # First read is the pre-reboot baseline; second is the post-reboot uptime. + mock_uptime.side_effect = [455, 30] + self.device.reboot(wait_for_reload=True, timeout=30) + + self.device.sw.reboot.assert_called_with(in_min=0) + mock_open.assert_called() @mock.patch("pyntc.devices.jnpr_device.time.sleep") def test_wait_for_device_to_reboot_error(self, mock_sleep): - with mock.patch.object(self.device, "open") as mock_open: - type(self.device.native).connected = mock.PropertyMock(side_effect=[True, False]) - mock_open.side_effect = Exception + """Raise RebootTimeoutError when the device never reports a lower uptime within the timeout.""" + with ( + mock.patch.object(self.device, "open"), + mock.patch.object(self.device, "close"), + mock.patch.object(type(self.device), "uptime", new_callable=mock.PropertyMock) as mock_uptime, + ): + mock_uptime.return_value = 455 with pytest.raises(RebootTimeoutError): self.device.reboot(wait_for_reload=True, timeout=1) @@ -290,12 +300,62 @@ def test_checkpoint(self, mock_scp): self.device.show.assert_called_with("show config") def test_uptime(self): + """Cold cache (_uptime is None) refreshes facts and parses the uptime.""" + self.assertIsNone(self.device._uptime) + uptime = self.device.uptime + self.assertEqual(uptime, 455) + self.device.native.facts_refresh.assert_called_once_with(keys="RE0") + + def test_uptime_cached(self): + """A populated cache is returned as-is, with no device round-trip.""" + self.device._uptime = 1234 uptime = self.device.uptime - assert uptime == 455 + self.assertEqual(uptime, 1234) + self.device.native.facts_refresh.assert_not_called() + + def test_uptime_refreshes_after_cache_cleared(self): + """Clearing the cache forces a fresh read.""" + self.assertEqual(self.device.uptime, 455) + + self.device._uptime = None + self.device.native.facts = {"RE0": {"up_time": "30 seconds"}} + + self.assertEqual(self.device.uptime, 30) + + def test_uptime_none_when_facts_unavailable(self): + """Missing/unavailable facts return None gracefully instead of raising an Exception.""" + self.device._uptime = None + self.device.native.facts = {} + self.assertIsNone(self.device.uptime) def test_uptime_string(self): + """Cold cache (_uptime_string is None) refreshes facts and formats the uptime.""" + self.assertIsNone(self.device._uptime_string) + uptime_string = self.device.uptime_string + self.assertEqual(uptime_string, "00:00:07:35") + self.device.native.facts_refresh.assert_called_once_with(keys="RE0") + + def test_uptime_string_cached(self): + """A populated cache is returned as-is, with no device round-trip.""" + self.device._uptime_string = "01:02:03:04" uptime_string = self.device.uptime_string - assert uptime_string == "00:00:07:35" + self.assertEqual(uptime_string, "01:02:03:04") + self.device.native.facts_refresh.assert_not_called() + + def test_uptime_string_refreshes_after_cache_cleared(self): + """Clearing the cache forces a fresh read.""" + self.assertEqual(self.device.uptime_string, "00:00:07:35") + + self.device._uptime_string = None + self.device.native.facts = {"RE0": {"up_time": "30 seconds"}} + + self.assertEqual(self.device.uptime_string, "00:00:00:30") + + def test_uptime_string_none_when_facts_unavailable(self): + """Missing/unavailable facts return None gracefully instead of raising an Exception.""" + self.device._uptime_string = None + self.device.native.facts = {} + self.assertIsNone(self.device.uptime_string) def test_vendor(self): vendor = self.device.vendor