diff --git a/.gitignore b/.gitignore index 8141a6a9..5130735a 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,6 @@ +# Local working notes / drafts (not committed) +ideas/ + # Ansible Retry Files *.retry diff --git a/changes/388.changed b/changes/388.changed new file mode 100644 index 00000000..b2f5617a --- /dev/null +++ b/changes/388.changed @@ -0,0 +1 @@ +Migrated `NXOSDevice` methods from NX-API to Netmiko SSH: `running_config`, `set_timeout`, `install_os`, `reboot`, `backup_running_config`, `checkpoint`, `rollback`, `redundancy_state`, `file_copy_remote_exists`, and `file_copy`. \ No newline at end of file diff --git a/changes/388.deprecated b/changes/388.deprecated new file mode 100644 index 00000000..9252bfd6 --- /dev/null +++ b/changes/388.deprecated @@ -0,0 +1 @@ +Deprecated the `transport`, `verify`, and `port` constructor kwargs on `NXOSDevice`. Supplying any of these now emits a `DeprecationWarning`. These kwargs will be removed in a future release as `NXOSDevice` migrates to Netmiko SSH exclusively. \ No newline at end of file diff --git a/docs/user/lib_overview.md b/docs/user/lib_overview.md index 30f29070..47a7acab 100644 --- a/docs/user/lib_overview.md +++ b/docs/user/lib_overview.md @@ -15,7 +15,28 @@ 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 NX-OS - uses pynxos (NX-API) +- Cisco NX-OS - migrating from pynxos (NX-API) to netmiko (SSH); see [NXOS transport change](#nxos-transport-change) below - Arista EOS - uses pyeapi (eAPI) - Juniper Junos - uses PyEz (NETCONF) - F5 Networks - uses f5-sdk (ReST) + +## NXOS transport change + +`NXOSDevice` is migrating from the pynxos (NX-API) transport to Netmiko SSH. The migration is being delivered in phases: + +- The following methods have been reimplemented on Netmiko SSH: `save()`, `running_config`, `set_timeout()`, `install_os()` (the `terminal dont-ask` step), `reboot()`, `backup_running_config()`, `checkpoint()`, `rollback()`, `redundancy_state`, `file_copy_remote_exists()`, and `file_copy()`. +- `file_copy()` now uses Netmiko's `file_transfer()` (SCP over the existing SSH session) instead of pynxos. +- `reboot()` now catches `netmiko.exceptions.ReadTimeout` (raised by Netmiko when the SSH session drops during reload) instead of `requests.exceptions.ReadTimeout`. +- `redundancy_state` now falls back to `"active"` when the underlying SSH command raises any `netmiko.exceptions.NetmikoBaseException`. +- `file_copy_remote_exists()` now compares the local file's checksum against the remote file's checksum via `verify_file()`; it no longer delegates to pynxos. +- `refresh()` now also invalidates the cached `redundancy_state`. +- The constructor still accepts the `transport`, `verify`, and `port` kwargs for backwards compatibility, but supplying any of them now emits a `DeprecationWarning`. These kwargs will be removed in a future release and will be ignored once the migration completes. +- The remaining NX-API call sites (`show`, `config`, `facts`-derived properties such as `hostname`/`uptime`/`os_version`, `boot_options`, `set_boot_options`, etc.) are still wired through pynxos and will be migrated in follow-up releases. + +### Behavioral differences to expect once migration completes + +- `show(..., raw_text=False)` will return TextFSM-parsed structures (via `ntc-templates`) rather than NX-API JSON. The result shape may differ for some commands; verify against `ntc-templates` output for the relevant `show` command. +- Properties derived from facts (`uptime`, `hostname`, `model`, `os_version`, etc.) will issue an SSH round trip on first access rather than reading from an NX-API JSON payload, and may not be populated until called. +- Niche `show` commands without an `ntc-templates` parser will return raw text; callers that relied on NX-API structured output may need to switch to `raw_text=True` and parse themselves. + +See the corresponding entry in the release notes for the change-tracking fragment. diff --git a/pyntc/devices/nxos_device.py b/pyntc/devices/nxos_device.py index 8a453beb..00afaa8e 100644 --- a/pyntc/devices/nxos_device.py +++ b/pyntc/devices/nxos_device.py @@ -3,15 +3,17 @@ import os import re import time +import warnings -from netmiko import ConnectHandler -from requests.exceptions import ConnectTimeout, ReadTimeout +from netmiko import ConnectHandler, file_transfer +from netmiko.exceptions import NetmikoBaseException, ReadTimeout +from requests.exceptions import ConnectTimeout +from requests.exceptions import ReadTimeout as RequestsReadTimeout from pyntc import log from pyntc.devices.base_device import BaseDevice, RollbackError, fix_docs from pyntc.devices.pynxos.device import Device as NXOSNative from pyntc.devices.pynxos.errors import CLIError -from pyntc.devices.pynxos.features.file_copy import FileTransferError as NXOSFileTransferError from pyntc.errors import ( CommandError, CommandListError, @@ -49,6 +51,21 @@ def __init__(self, host, username, password, transport="http", timeout=30, port= """ super().__init__(host, username, password, device_type="cisco_nxos_nxapi") + deprecated_kwargs = [] + if transport != "http": + deprecated_kwargs.append("transport") + if port is not None: + deprecated_kwargs.append("port") + if verify is not True: + deprecated_kwargs.append("verify") + if deprecated_kwargs: + warnings.warn( + f"NXOSDevice kwargs {deprecated_kwargs} are deprecated and will be removed in a future release. " + "NXOSDevice is migrating to Netmiko SSH exclusively; these NX-API-only kwargs will no longer " + "be honored once the migration is complete.", + DeprecationWarning, + stacklevel=2, + ) self.transport = transport self.timeout = timeout self.port = port @@ -88,6 +105,7 @@ def refresh(self): """Refresh caches on device instance.""" if hasattr(self.native, "_facts"): delattr(self.native, "_facts") + self._redundancy_state = None super().refresh() def backup_running_config(self, filename): @@ -96,8 +114,11 @@ def backup_running_config(self, filename): Args: filename (str): Name of backup file. """ - self.native.backup_running_config(filename) - log.debug("Host %s: Running config backed up.", self.host) + self.open() + output = self.native_ssh.send_command("show running-config") + with open(filename, "w", encoding="utf-8") as backup_file: + backup_file.write(output) + log.debug("Host %s: Running config backed up to %s.", self.host, filename) @property def boot_options(self): @@ -115,9 +136,17 @@ def checkpoint(self, filename): Args: filename (str): The filename to save the checkpoint on the remote device. + + Raises: + CommandError: If the device rejects the checkpoint command. """ + self.open() + command = f"checkpoint file {filename}" log.debug("Host %s: checkpoint is %s.", self.host, filename) - return self.native.checkpoint(filename) + output = self.native_ssh.send_command(command) + if re.search(r"%\s*Error|ERROR:", output): + log.error("Host %s: Checkpoint failed for %s: %s", self.host, filename, output) + raise CommandError(command, output.strip()) def close(self): """Disconnect from device.""" @@ -267,40 +296,45 @@ def serial_number(self): return self._serial_number def file_copy(self, src, dest=None, file_system="bootflash:"): - """Send a local file to the device. + """Send a local file to the device via SCP over the SSH session. Args: src (str): Path to the local file to send. dest (str, optional): The destination file path. Defaults to basename of source path. - file_system (str, optional): [The file system for the remote file. Defaults to "bootflash:". + file_system (str, optional): The file system for the remote file. Defaults to "bootflash:". Raises: FileTransferError: Error if transfer of file cannot be verified. """ + dest = dest or os.path.basename(src) + if self.file_copy_remote_exists(src, dest, file_system): + return + self._check_free_space(os.path.getsize(src), file_system=file_system) + self.open() + try: + file_transfer( + self.native_ssh, + source_file=src, + dest_file=dest, + file_system=file_system, + direction="put", + overwrite_file=True, + ) + except Exception as err: # noqa: BLE001 + log.error("Host %s: SCP file transfer error %s", self.host, str(err)) + raise FileTransferError from err + log.info("Host %s: File %s transferred successfully.", self.host, src) if not self.file_copy_remote_exists(src, dest, file_system): - dest = dest or os.path.basename(src) - self._check_free_space(os.path.getsize(src), file_system=file_system) - try: - file_copy = self.native.file_copy( # pylint: disable=assignment-from-no-return - src, dest, file_system=file_system - ) # pylint: disable=assignment-from-no-return - log.info("Host %s: File %s transferred successfully.", self.host, src) - if not self.file_copy_remote_exists(src, dest, file_system): - log.error( - "Host %s: Attempted file copy, but could not validate file existed after transfer %s", - self.host, - FileTransferError.default_message, - ) - raise FileTransferError - return file_copy - - except NXOSFileTransferError as err: - log.error("Host %s: NXOS file transfer error %s", self.host, str(err)) - raise FileTransferError + log.error( + "Host %s: Attempted file copy, but could not validate file existed after transfer %s", + self.host, + FileTransferError.default_message, + ) + raise FileTransferError # TODO: Make this an internal method since exposing file_copy should be sufficient def file_copy_remote_exists(self, src, dest=None, file_system="bootflash:"): - """Check if a remote file exists. + """Check if a remote file exists and matches the local file's checksum. Args: src (str): Path to the local file to send. @@ -308,16 +342,13 @@ def file_copy_remote_exists(self, src, dest=None, file_system="bootflash:"): file_system (str, optional): The file system for the remote file. Defaults to "bootflash:". Returns: - (bool): True if the remote file exists. Otherwise, false. + (bool): True if the remote file exists and its checksum matches the local file. Otherwise, false. """ dest = dest or os.path.basename(src) - log.debug( - "Host %s: File %s exists on remote %s.", - self.host, - src, - self.native.file_copy_remote_exists(src, dest, file_system=file_system), - ) - return self.native.file_copy_remote_exists(src, dest, file_system=file_system) + local_checksum = self.get_local_checksum(src) + result = self.verify_file(local_checksum, dest, file_system=file_system) + log.debug("Host %s: File %s exists on remote and matches local: %s", self.host, dest, result) + return result def _get_file_system(self): """Determine the default file system or directory for device. @@ -470,7 +501,6 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5", **kwargs): command, result, ) - print(f"result: {result}") remote_checksum = result return remote_checksum @@ -632,7 +662,7 @@ def install_os(self, image_name, reboot=True, **vendor_specifics): Returns: (bool): True if new image is boot option on device. Otherwise, false. """ - self.native.show("terminal dont-ask") + self.native_ssh.send_command("terminal dont-ask") timeout = vendor_specifics.get("timeout", 3600) if not self._image_booted(image_name): log.info("Host %s: Setting Image %s in boot options.", self.host, image_name) @@ -674,8 +704,8 @@ def redundancy_state(self): """ if self._redundancy_state is None: try: - output = self.native.show("show redundancy state", raw_text=True) - # Parse the redundancy state from output + self.open() + output = self.native_ssh.send_command("show redundancy state") # Example output: "Redundancy state = active" match = re.search(r"Redundancy\s+state\s*=\s*(\w+)", output, re.IGNORECASE) if match: @@ -683,8 +713,8 @@ def redundancy_state(self): else: # If no redundancy info, device may not support HA self._redundancy_state = "active" - except CLIError: - # If command fails, assume active (non-HA or error condition) + except NetmikoBaseException: + # If the SSH command fails, assume active (non-HA or error condition) self._redundancy_state = "active" return self._redundancy_state @@ -749,9 +779,8 @@ def reboot(self, wait_for_reload=False, **kwargs): log.warning("Passing 'confirm' to reboot method is deprecated.") raise DeprecationWarning("Passing 'confirm' to reboot method is deprecated.") try: - self.native.show_list(["terminal dont-ask", "reload"]) - # The native reboot is not always properly disabling confirmation. Above is more consistent. - # self.native.reboot(confirm=True) + self.native_ssh.send_command("terminal dont-ask") + self.native_ssh.send_command("reload") except ReadTimeout as expected_exception: log.info("Host %s: Device rebooted.", self.host) log.info("Hit expected exception during reload: %s", expected_exception.__class__) @@ -769,12 +798,13 @@ def rollback(self, filename): Raises: RollbackError: Error if rollback command is unsuccesfull. """ - try: - self.native.rollback(filename) - log.info("Host %s: Rollback to %s.", self.host, filename) - except CLIError: - log.error("Host %s: Rollback unsuccessful. %s may not exist.", self.host, filename) + self.open() + command = f"rollback running-config file {filename}" + output = self.native_ssh.send_command(command) + if re.search(r"%\s*Error|ERROR:|Rollback failed|does not exist", output, re.IGNORECASE): + log.error("Host %s: Rollback unsuccessful. %s may not exist. Output: %s", self.host, filename, output) raise RollbackError(f"Rollback unsuccessful, {filename} may not exist.") + log.info("Host %s: Rollback to %s.", self.host, filename) @property def running_config(self): @@ -784,7 +814,8 @@ def running_config(self): (str): Running configuration of device. """ log.debug("Host %s: Show running config.", self.host) - return self.native.running_config + self.open() + return self.native_ssh.send_command("show running-config") def save(self, filename="startup-config"): """Save a device's running configuration. @@ -795,8 +826,13 @@ def save(self, filename="startup-config"): Returns: (bool): True if configuration is saved. """ + self.open() + command = f"copy running-config {filename}" + self.native_ssh.send_command_timing(command) + self.native_ssh.send_command_timing("\n", read_timeout=200) + self.native_ssh.find_prompt() log.debug("Host %s: Copy running config with name %s.", self.host, filename) - return self.native.save(filename=filename) + return True def set_boot_options(self, image_name, kickstart=None, reboot=True, **vendor_specifics): """Set boot variables. @@ -829,7 +865,7 @@ def set_boot_options(self, image_name, kickstart=None, reboot=True, **vendor_spe image_name = file_system + image_name try: self.native.set_boot_options(image_name, kickstart=kickstart, reboot=reboot) - except (ReadTimeout, ConnectTimeout): + except (RequestsReadTimeout, ConnectTimeout): pass log.info("Host %s: boot options have been set to %s", self.host, image_name) @@ -840,7 +876,8 @@ def set_timeout(self, timeout): timeout (int): Timeout value. """ log.debug("Host %s: Timeout set to %s.", self.host, timeout) - self.native.timeout = timeout + self.timeout = timeout + self.native_ssh.timeout = timeout def show(self, command, raw_text=False): """Send a non-configuration command. diff --git a/tests/unit/test_devices/test_nxos_device.py b/tests/unit/test_devices/test_nxos_device.py index 0fb30aa1..3d224098 100644 --- a/tests/unit/test_devices/test_nxos_device.py +++ b/tests/unit/test_devices/test_nxos_device.py @@ -1,4 +1,5 @@ import unittest +import warnings import mock @@ -117,57 +118,95 @@ def test_bad_show_list(self): def test_save(self): result = self.device.save() - self.device.native.save.return_value = True self.assertTrue(result) - self.device.native.save.assert_called_with(filename="startup-config") + self.mock_native_ssh.send_command_timing.assert_any_call("copy running-config startup-config") + self.mock_native_ssh.send_command_timing.assert_any_call("\n", read_timeout=200) + self.mock_native_ssh.find_prompt.assert_called() - def test_file_copy_remote_exists(self): - self.device.native.file_copy_remote_exists.return_value = True + def test_save_custom_filename(self): + result = self.device.save("my-backup") + + self.assertTrue(result) + self.mock_native_ssh.send_command_timing.assert_any_call("copy running-config my-backup") + + def test_save_reopens_when_disconnected(self): + self.device._connected = False + + with mock.patch.object(NXOSDevice, "open") as mock_open: + self.device.save() + + mock_open.assert_called() + + @mock.patch.object(NXOSDevice, "verify_file", return_value=True) + @mock.patch.object(NXOSDevice, "get_local_checksum", return_value="abc123") + def test_file_copy_remote_exists(self, mock_local_checksum, mock_verify): result = self.device.file_copy_remote_exists("source_file", "dest_file") self.assertTrue(result) - self.device.native.file_copy_remote_exists.assert_called_with( - "source_file", "dest_file", file_system=FILE_SYSTEM - ) + mock_local_checksum.assert_called_with("source_file") + mock_verify.assert_called_with("abc123", "dest_file", file_system=FILE_SYSTEM) - def test_file_copy_remote_exists_failure(self): - self.device.native.file_copy_remote_exists.return_value = False + @mock.patch.object(NXOSDevice, "verify_file", return_value=False) + @mock.patch.object(NXOSDevice, "get_local_checksum", return_value="abc123") + def test_file_copy_remote_exists_failure(self, mock_local_checksum, mock_verify): result = self.device.file_copy_remote_exists("source_file", "dest_file") self.assertFalse(result) - self.device.native.file_copy_remote_exists.assert_called_with( - "source_file", "dest_file", file_system=FILE_SYSTEM - ) + mock_verify.assert_called_with("abc123", "dest_file", file_system=FILE_SYSTEM) + @mock.patch("pyntc.devices.nxos_device.file_transfer") @mock.patch.object(NXOSDevice, "_get_free_space", return_value=1024 * 1024 * 1024) @mock.patch("pyntc.devices.nxos_device.os.path.getsize", return_value=1024) @mock.patch.object(NXOSDevice, "file_copy_remote_exists", side_effect=[False, True]) - def test_file_copy(self, mock_fcre, mock_getsize, mock_get_free_space): + def test_file_copy(self, mock_fcre, mock_getsize, mock_get_free_space, mock_transfer): self.device.file_copy("source_file", "dest_file") - self.device.native.file_copy.assert_called_with("source_file", "dest_file", file_system=FILE_SYSTEM) - self.device.native.file_copy.assert_called() + mock_transfer.assert_called_with( + self.mock_native_ssh, + source_file="source_file", + dest_file="dest_file", + file_system=FILE_SYSTEM, + direction="put", + overwrite_file=True, + ) + @mock.patch("pyntc.devices.nxos_device.file_transfer") @mock.patch.object(NXOSDevice, "_get_free_space", return_value=1024 * 1024 * 1024) @mock.patch("pyntc.devices.nxos_device.os.path.getsize", return_value=1024) @mock.patch.object(NXOSDevice, "file_copy_remote_exists", side_effect=[False, True]) - def test_file_copy_no_dest(self, mock_fcre, mock_getsize, mock_get_free_space): + def test_file_copy_no_dest(self, mock_fcre, mock_getsize, mock_get_free_space, mock_transfer): self.device.file_copy("source_file") - self.device.native.file_copy.assert_called_with("source_file", "source_file", file_system=FILE_SYSTEM) - self.device.native.file_copy.assert_called() + mock_transfer.assert_called_with( + self.mock_native_ssh, + source_file="source_file", + dest_file="source_file", + file_system=FILE_SYSTEM, + direction="put", + overwrite_file=True, + ) + @mock.patch("pyntc.devices.nxos_device.file_transfer") @mock.patch.object(NXOSDevice, "file_copy_remote_exists", side_effect=[True]) - def test_file_copy_file_exists(self, mock_fcre): + def test_file_copy_file_exists(self, mock_fcre, mock_transfer): self.device.file_copy("source_file", "dest_file") - self.device.native.file_copy.assert_not_called() + mock_transfer.assert_not_called() + @mock.patch("pyntc.devices.nxos_device.file_transfer") @mock.patch.object(NXOSDevice, "_get_free_space", return_value=1024 * 1024 * 1024) @mock.patch("pyntc.devices.nxos_device.os.path.getsize", return_value=1024) @mock.patch.object(NXOSDevice, "file_copy_remote_exists", side_effect=[False, False]) - def test_file_copy_fail(self, mock_fcre, mock_getsize, mock_get_free_space): + def test_file_copy_fail(self, mock_fcre, mock_getsize, mock_get_free_space, mock_transfer): + with self.assertRaises(FileTransferError): + self.device.file_copy("source_file") + mock_transfer.assert_called() + + @mock.patch("pyntc.devices.nxos_device.file_transfer", side_effect=OSError("scp broken")) + @mock.patch.object(NXOSDevice, "_get_free_space", return_value=1024 * 1024 * 1024) + @mock.patch("pyntc.devices.nxos_device.os.path.getsize", return_value=1024) + @mock.patch.object(NXOSDevice, "file_copy_remote_exists", side_effect=[False]) + def test_file_copy_raises_on_transfer_error(self, mock_fcre, mock_getsize, mock_get_free_space, mock_transfer): with self.assertRaises(FileTransferError): self.device.file_copy("source_file") - self.device.native.file_copy.assert_called() @mock.patch.object(NXOSDevice, "_get_free_space", return_value=1024) # Only 1KB free @mock.patch("pyntc.devices.nxos_device.os.path.getsize", return_value=1024 * 1024) # Trying to copy 1MB @@ -179,8 +218,19 @@ def test_file_copy_raises_not_enough_free_space(self, mock_fcre, mock_getsize, m def test_reboot(self): self.device.reboot() - self.device.native.show_list.assert_called_with(["terminal dont-ask", "reload"]) - # self.device.native.reboot.assert_called_with(confirm=True) + self.mock_native_ssh.send_command.assert_any_call("terminal dont-ask") + self.mock_native_ssh.send_command.assert_any_call("reload") + + def test_reboot_handles_read_timeout(self): + from netmiko.exceptions import ReadTimeout as NetmikoReadTimeout + + self.mock_native_ssh.send_command.side_effect = [None, NetmikoReadTimeout("reload killed the session")] + self.device.reboot() + self.mock_native_ssh.send_command.assert_any_call("reload") + + def test_reboot_rejects_confirm_kwarg(self): + with self.assertRaises(DeprecationWarning): + self.device.reboot(confirm=True) def test_boot_options(self): expected = {"sys": "my_sys", "boot": "my_boot"} @@ -220,23 +270,34 @@ def test_set_boot_options_no_kickstart(self, mock_show): def test_backup_running_config(self): filename = "local_running_config" - self.device.backup_running_config(filename) + expected = "!\nhostname n9k1\n!\n" + self.mock_native_ssh.send_command.return_value = expected + with mock.patch("builtins.open", mock.mock_open()) as mock_file: + self.device.backup_running_config(filename) - self.device.native.backup_running_config.assert_called_with(filename) + self.mock_native_ssh.send_command.assert_called_with("show running-config") + mock_file.assert_called_with(filename, "w", encoding="utf-8") + mock_file().write.assert_called_with(expected) def test_rollback(self): + self.mock_native_ssh.send_command.return_value = "" self.device.rollback("good_checkpoint") - self.device.native.rollback.assert_called_with("good_checkpoint") + self.mock_native_ssh.send_command.assert_called_with("rollback running-config file good_checkpoint") def test_bad_rollback(self): - self.device.native.rollback.side_effect = CLIError("rollback", "bad rollback command") - + self.mock_native_ssh.send_command.return_value = "Rollback failed: file bad_checkpoint does not exist" with self.assertRaises(RollbackError): self.device.rollback("bad_checkpoint") - def test_checkpiont(self): + def test_checkpoint(self): + self.mock_native_ssh.send_command.return_value = "" self.device.checkpoint("good_checkpoint") - self.device.native.checkpoint.assert_called_with("good_checkpoint") + self.mock_native_ssh.send_command.assert_called_with("checkpoint file good_checkpoint") + + def test_checkpoint_failure(self): + self.mock_native_ssh.send_command.return_value = "ERROR: Checkpoint named good_checkpoint already exists" + with self.assertRaises(CommandError): + self.device.checkpoint("good_checkpoint") def test_uptime(self): uptime = self.device.uptime @@ -270,11 +331,17 @@ def test_model(self): model = self.device.model assert model == "Nexus9000 C9396PX Chassis" - @mock.patch("pyntc.devices.pynxos.device.Device.running_config", new_callable=mock.PropertyMock) - def test_running_config(self, mock_rc): - type(self.device.native).running_config = mock_rc - self.device.running_config() - self.device.native.running_config.assert_called_with() + def test_running_config(self): + expected = "!\nhostname n9k1\n!\n" + self.mock_native_ssh.send_command.return_value = expected + result = self.device.running_config + self.assertEqual(result, expected) + self.mock_native_ssh.send_command.assert_called_with("show running-config") + + def test_set_timeout(self): + self.device.set_timeout(120) + self.assertEqual(self.device.timeout, 120) + self.assertEqual(self.mock_native_ssh.timeout, 120) def test_starting_config(self): expected = self.device.show("show startup-config", raw_text=True) @@ -286,6 +353,46 @@ def test_refresh(self): self.assertIsNone(self.device._uptime) self.assertFalse(hasattr(self.device.native, "_facts")) + def test_refresh_clears_redundancy_state(self): + self.mock_native_ssh.send_command.return_value = "Redundancy state = active" + self.device.redundancy_state # noqa: B018 # populate cache + self.assertEqual(self.device._redundancy_state, "active") + self.device.refresh() + self.assertIsNone(self.device._redundancy_state) + + def test_redundancy_state_active(self): + self.mock_native_ssh.send_command.return_value = "Redundancy state = active" + self.assertEqual(self.device.redundancy_state, "active") + self.mock_native_ssh.send_command.assert_called_with("show redundancy state") + + def test_redundancy_state_standby(self): + self.mock_native_ssh.send_command.return_value = "Redundancy state = standby" + self.assertEqual(self.device.redundancy_state, "standby") + + def test_redundancy_state_no_match_falls_back_to_active(self): + self.mock_native_ssh.send_command.return_value = "% Invalid command at marker" + self.assertEqual(self.device.redundancy_state, "active") + + def test_redundancy_state_session_error_falls_back_to_active(self): + from netmiko.exceptions import NetmikoBaseException + + self.mock_native_ssh.send_command.side_effect = NetmikoBaseException("session dropped") + self.assertEqual(self.device.redundancy_state, "active") + + def test_redundancy_state_is_memoized(self): + self.mock_native_ssh.send_command.return_value = "Redundancy state = active" + self.device.redundancy_state # noqa: B018 # trigger memoization + self.device.redundancy_state # noqa: B018 # second access should not re-send + self.assertEqual(self.mock_native_ssh.send_command.call_count, 1) + + def test_is_active_returns_true_when_active(self): + self.mock_native_ssh.send_command.return_value = "Redundancy state = active" + self.assertTrue(self.device.is_active()) + + def test_is_active_returns_false_when_standby(self): + self.mock_native_ssh.send_command.return_value = "Redundancy state = standby" + self.assertFalse(self.device.is_active()) + @mock.patch.object(NXOSDevice, "show", return_value="bootflash:") def test_get_file_system(self, mock_show): self.assertEqual(self.device._get_file_system(), "bootflash:") @@ -456,5 +563,58 @@ def test_remote_file_copy_query_string_not_supported(self): self.device.remote_file_copy(src, file_system="bootflash:") +class TestNXOSDeviceDeprecationWarnings(unittest.TestCase): + @mock.patch("pyntc.devices.nxos_device.ConnectHandler", create=True) + @mock.patch("pyntc.devices.nxos_device.NXOSNative", autospec=True) + def test_no_warning_on_default_kwargs(self, _mock_native, _mock_connect_handler): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + NXOSDevice("host", "user", "pass") + deprecation_warnings = [w for w in caught if issubclass(w.category, DeprecationWarning)] + self.assertEqual(deprecation_warnings, []) + + @mock.patch("pyntc.devices.nxos_device.ConnectHandler", create=True) + @mock.patch("pyntc.devices.nxos_device.NXOSNative", autospec=True) + def test_warning_on_non_default_transport(self, _mock_native, _mock_connect_handler): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + NXOSDevice("host", "user", "pass", transport="https") + deprecation_warnings = [w for w in caught if issubclass(w.category, DeprecationWarning)] + self.assertEqual(len(deprecation_warnings), 1) + self.assertIn("transport", str(deprecation_warnings[0].message)) + + @mock.patch("pyntc.devices.nxos_device.ConnectHandler", create=True) + @mock.patch("pyntc.devices.nxos_device.NXOSNative", autospec=True) + def test_warning_on_non_default_port(self, _mock_native, _mock_connect_handler): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + NXOSDevice("host", "user", "pass", port=8443) + deprecation_warnings = [w for w in caught if issubclass(w.category, DeprecationWarning)] + self.assertEqual(len(deprecation_warnings), 1) + self.assertIn("port", str(deprecation_warnings[0].message)) + + @mock.patch("pyntc.devices.nxos_device.ConnectHandler", create=True) + @mock.patch("pyntc.devices.nxos_device.NXOSNative", autospec=True) + def test_warning_on_non_default_verify(self, _mock_native, _mock_connect_handler): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + NXOSDevice("host", "user", "pass", verify=False) + deprecation_warnings = [w for w in caught if issubclass(w.category, DeprecationWarning)] + self.assertEqual(len(deprecation_warnings), 1) + self.assertIn("verify", str(deprecation_warnings[0].message)) + + @mock.patch("pyntc.devices.nxos_device.ConnectHandler", create=True) + @mock.patch("pyntc.devices.nxos_device.NXOSNative", autospec=True) + def test_warning_lists_multiple_kwargs(self, _mock_native, _mock_connect_handler): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + NXOSDevice("host", "user", "pass", transport="https", port=8443, verify=False) + deprecation_warnings = [w for w in caught if issubclass(w.category, DeprecationWarning)] + self.assertEqual(len(deprecation_warnings), 1) + message = str(deprecation_warnings[0].message) + for kwarg in ("transport", "port", "verify"): + self.assertIn(kwarg, message) + + if __name__ == "__main__": unittest.main()