Bug Fix NXOS

changes/387.fixed

@@ -0,0 +1,3 @@
+Fixed an issue where nx-api commands were mixed with ssh commands.
+Fixed an issue where a tuple was returning instead of a string.
+Fixed an issue where a prompt was not being caught.

pyntc/devices/nxos_device.py

@@ -328,7 +328,9 @@ def _get_file_system(self):
         Raises:
             FileSystemNotFoundError: When the module is unable to determine the default file system.
         """
-        raw_data = self.show("dir", raw_text=True)
+        self.open()
+        raw_data = self.native_ssh.send_command("dir", read_timeout=30)
+
         try:
             file_system = re.search(r"bootflash:", raw_data).group(0)
         except AttributeError:
@@ -343,7 +345,8 @@ def _get_free_space(self, file_system=None):
         if file_system is None:
             file_system = self._get_file_system()
 
-        raw_data = self.show(f"dir {file_system}", raw_text=True)
+        self.open()
+        raw_data = self.native_ssh.send_command(f"dir {file_system}", read_timeout=30)
         # Example NXOS dir output: 47171194880 bytes free
         match = re.search(r"(\d+)\s+bytes\s+free", raw_data)
         if match is None:
@@ -368,7 +371,7 @@ def _build_url_copy_command_simple(self, src, file_system, dest):
         """Build copy command for simple URL-based transfers (TFTP, HTTP, HTTPS without credentials)."""
         netloc = self._netloc(src)
         path = self._source_path(src, dest)
-        return f"copy {src.scheme}://{netloc}{path} {file_system}", False
+        return f"copy {src.scheme}://{netloc}{path} {file_system}"
 
     def _build_url_copy_command_with_creds(self, src, file_system, dest):
         """Build copy command for URL-based transfers with credentials (HTTP/HTTPS/SCP/FTP/SFTP)."""
@@ -537,6 +540,7 @@ def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kw
                 r"Source username": src.username or "",
                 r"yes/no|Are you sure you want to continue connecting": "yes",
                 r"(confirm|Address or name of remote host|Source filename|Destination filename)": "",
+                r"Enter vrf.*:": src.vrf or "",
             }
             keys = list(prompt_answers.keys()) + [current_prompt]
             expect_regex = f"({'|'.join(keys)})"

tests/unit/test_devices/test_nxos_device.py

@@ -1,6 +1,8 @@
 import unittest
 
 import mock
+from hypothesis import given
+from hypothesis import strategies as st
 
 from pyntc.devices.base_device import RollbackError
 from pyntc.devices.nxos_device import NXOSDevice
@@ -286,22 +288,21 @@ def test_refresh(self):
         self.assertIsNone(self.device._uptime)
         self.assertFalse(hasattr(self.device.native, "_facts"))
 
-    @mock.patch.object(NXOSDevice, "show", return_value="bootflash:")
-    def test_get_file_system(self, mock_show):
+    def test_get_file_system(self):
+        self.device.native_ssh.send_command.return_value = "bootflash:"
         self.assertEqual(self.device._get_file_system(), "bootflash:")
-        mock_show.assert_called_with("dir", raw_text=True)
+        self.device.native_ssh.send_command.assert_called_with("dir", read_timeout=30)
 
-    @mock.patch.object(NXOSDevice, "show", return_value="no filesystems here")
-    def test_get_file_system_not_found(self, mock_show):
+    def test_get_file_system_not_found(self):
+        self.device.native_ssh.send_command.return_value = "no filesystems here"
         with self.assertRaises(FileSystemNotFoundError):
             self.device._get_file_system()
-        mock_show.assert_called_with("dir", raw_text=True)
+        self.device.native_ssh.send_command.assert_called_with("dir", read_timeout=30)
 
-    @mock.patch.object(NXOSDevice, "show")
-    def test_get_free_space(self, mock_show):
+    def test_get_free_space(self):
         """Test _get_free_space parses NXOS dir output correctly."""
         # NXOS dir output format with free space at the end
-        mock_show.return_value = """Directory of bootflash:/
+        self.device.native_ssh.send_command.return_value = """Directory of bootflash:/
 4096         Mar 03 22:47:15 2026  .rpmstore/
 4733329408   bytes used
 47171194880  bytes free
@@ -310,25 +311,25 @@ def test_get_free_space(self, mock_show):
 """
         result = self.device._get_free_space()
         self.assertEqual(result, 47171194880)
-        mock_show.assert_called_with("dir bootflash:", raw_text=True)
+        # Should call _get_file_system (which uses SSH) and then dir command via SSH
+        ssh_calls = self.device.native_ssh.send_command.call_args_list
+        self.assertTrue(any("dir" in str(call) for call in ssh_calls))
 
-    @mock.patch.object(NXOSDevice, "show")
-    def test_get_free_space_with_custom_filesystem(self, mock_show):
+    def test_get_free_space_with_custom_filesystem(self):
         """Test _get_free_space uses custom file system when provided."""
-        mock_show.return_value = """Directory of disk0:/
+        self.device.native_ssh.send_command.return_value = """Directory of disk0:/
 1000000      bytes used
 2000000      bytes free
 3000000      bytes total
 
 """
         result = self.device._get_free_space("disk0:")
         self.assertEqual(result, 2000000)
-        mock_show.assert_called_with("dir disk0:", raw_text=True)
+        self.device.native_ssh.send_command.assert_called_with("dir disk0:", read_timeout=30)
 
-    @mock.patch.object(NXOSDevice, "show")
-    def test_get_free_space_raises_on_parse_error(self, mock_show):
+    def test_get_free_space_raises_on_parse_error(self):
         """Test _get_free_space raises CommandError when output can't be parsed."""
-        mock_show.return_value = "Directory of bootflash:/\nNo free space info here\n"
+        self.device.native_ssh.send_command.return_value = "Directory of bootflash:/\nNo free space info here\n"
         with self.assertRaises(CommandError):
             self.device._get_free_space()
 
@@ -397,10 +398,14 @@ def test_remote_file_copy_transfer_success(self):
             timeout=30,
         )
         self.device.native_ssh.find_prompt.return_value = "host#"
-        self.device.native_ssh.send_command.return_value = "Copy complete"
+        # Mock send_command to return success message that includes the prompt
+        self.device.native_ssh.send_command.return_value = "Copy complete\nhost#"
         with mock.patch.object(NXOSDevice, "verify_file", side_effect=[False, True]):
             self.device.remote_file_copy(src, file_system="bootflash:")
+        # Verify send_command was called with expect_string parameter
         self.device.native_ssh.send_command.assert_called_once()
+        call_args = self.device.native_ssh.send_command.call_args
+        self.assertIn("expect_string", call_args.kwargs)
 
     def test_remote_file_copy_transfer_fails_verification(self):
         src = FileCopyModel(
@@ -411,7 +416,8 @@ def test_remote_file_copy_transfer_fails_verification(self):
             timeout=30,
         )
         self.device.native_ssh.find_prompt.return_value = "host#"
-        self.device.native_ssh.send_command.return_value = "Copy complete"
+        # Mock send_command to return success message that includes the prompt
+        self.device.native_ssh.send_command.return_value = "Copy complete\nhost#"
         with mock.patch.object(NXOSDevice, "verify_file", side_effect=[False, False]):
             with self.assertRaises(FileTransferError):
                 self.device.remote_file_copy(src, file_system="bootflash:")
@@ -433,6 +439,58 @@ def test_remote_file_copy_raises_not_enough_free_space(self, mock_get_free_space
             self.device.remote_file_copy(src, file_system="bootflash:")
         self.device.native_ssh.send_command.assert_not_called()
 
+    def test_remote_file_copy_with_vrf_prompt_handling(self):
+        """Test remote_file_copy handles VRF prompts correctly."""
+        src = FileCopyModel(
+            download_url="ftp://example.com/nxos.bin",
+            checksum="abc123",
+            file_name="nxos.bin",
+            hashing_algorithm="md5",
+            timeout=30,
+            username="testuser",
+            token="testpass",
+            vrf="management",  # VRF specified for prompt response
+        )
+        self.device.native_ssh.find_prompt.return_value = "host#"
+        # Mock send_command to return success message that includes the prompt
+        self.device.native_ssh.send_command.return_value = "Copy complete\nhost#"
+        with mock.patch.object(NXOSDevice, "verify_file", side_effect=[False, True]):
+            self.device.remote_file_copy(src, file_system="bootflash:")
+
+        # Verify send_command was called with VRF prompt handling
+        self.device.native_ssh.send_command.assert_called_once()
+        call_args = self.device.native_ssh.send_command.call_args
+        self.assertIn("expect_string", call_args.kwargs)
+        # Verify the expect_string contains VRF prompt pattern
+        expect_string = call_args.kwargs["expect_string"]
+        self.assertIn("Enter vrf", expect_string)
+
+    def test_remote_file_copy_with_no_vrf_specified(self):
+        """Test remote_file_copy handles VRF prompts when no VRF is specified."""
+        src = FileCopyModel(
+            download_url="ftp://example.com/nxos.bin",
+            checksum="abc123",
+            file_name="nxos.bin",
+            hashing_algorithm="md5",
+            timeout=30,
+            username="testuser",
+            token="testpass",
+            # No VRF specified - should respond with empty string to VRF prompt
+        )
+        self.device.native_ssh.find_prompt.return_value = "host#"
+        # Mock send_command to return success message that includes the prompt
+        self.device.native_ssh.send_command.return_value = "Copy complete\nhost#"
+        with mock.patch.object(NXOSDevice, "verify_file", side_effect=[False, True]):
+            self.device.remote_file_copy(src, file_system="bootflash:")
+
+        # Verify send_command was called with VRF prompt handling
+        self.device.native_ssh.send_command.assert_called_once()
+        call_args = self.device.native_ssh.send_command.call_args
+        self.assertIn("expect_string", call_args.kwargs)
+        # Verify the expect_string contains VRF prompt pattern
+        expect_string = call_args.kwargs["expect_string"]
+        self.assertIn("Enter vrf", expect_string)
+
     def test_remote_file_copy_invalid_scheme(self):
         src = FileCopyModel(
             download_url="smtp://example.com/nxos.bin",
@@ -455,6 +513,132 @@ def test_remote_file_copy_query_string_not_supported(self):
         with self.assertRaises(ValueError):
             self.device.remote_file_copy(src, file_system="bootflash:")
 
+    @given(
+        scheme=st.sampled_from(["http", "https", "scp", "sftp", "ftp", "tftp"]),
+        hostname=st.text(min_size=1, max_size=20, alphabet=st.characters(whitelist_categories=("Ll", "Lu", "Nd"))),
+        filename=st.text(
+            min_size=1, max_size=20, alphabet=st.characters(whitelist_categories=("Ll", "Lu", "Nd", "Pd"))
+        ),
+        checksum=st.text(min_size=32, max_size=32, alphabet=st.characters(whitelist_categories=("Ll", "Nd"))),
+    )
+    def test_remote_file_copy_uses_ssh_for_filesystem_detection(self, scheme, hostname, filename, checksum):
+        """Property-based test: remote_file_copy should use SSH for _get_file_system calls.
+
+        This test verifies that the SSH/HTTP protocol mismatch bug is fixed by ensuring
+        that _get_file_system always uses SSH for file system operations.
+        """
+        src = FileCopyModel(
+            download_url=f"{scheme}://{hostname}/{filename}",
+            checksum=checksum,
+            file_name=filename,
+            hashing_algorithm="md5",
+            timeout=30,
+        )
+
+        # Mock SSH operations to simulate successful file system detection
+        self.device.native_ssh.send_command.return_value = "Directory of bootflash:/\n47171194880 bytes free"
+        self.device.native_ssh.find_prompt.return_value = "host#"
+
+        # Mock verify_file to return True (file already exists and verified)
+        with mock.patch.object(NXOSDevice, "verify_file", return_value=True):
+            # This should complete without attempting HTTP connections
+            self.device.remote_file_copy(src)
+
+        # Verify that SSH was used for directory command (filesystem detection)
+        ssh_calls = self.device.native_ssh.send_command.call_args_list
+        self.assertTrue(
+            any("dir" in str(call) for call in ssh_calls), "Expected SSH 'dir' command for filesystem detection"
+        )
+
+    @mock.patch("pyntc.devices.nxos_device.ConnectHandler", create=True)
+    @mock.patch("pyntc.devices.nxos_device.NXOSNative", autospec=True)
+    def test_port_default(self, mock_device, mock_connect_handler):
+        """Test that port defaults to None when not specified."""
+        _ = NXOSDevice("host", "user", "pass")
+
+        # Verify NXOSNative was called with default port (None)
+        mock_device.assert_called_with(
+            "host",
+            "user",
+            "pass",
+            transport="http",
+            timeout=30,
+            port=None,  # Default port
+            verify=True,
+        )
+
+    @mock.patch("pyntc.devices.nxos_device.ConnectHandler", create=True)
+    @mock.patch("pyntc.devices.nxos_device.NXOSNative", autospec=True)
+    def test_port_custom(self, mock_device, mock_connect_handler):
+        """Test that custom port is passed to NXOSNative."""
+        _ = NXOSDevice("host", "user", "pass", port=8080)
+
+        # Verify NXOSNative was called with custom port
+        mock_device.assert_called_with(
+            "host",
+            "user",
+            "pass",
+            transport="http",
+            timeout=30,
+            port=8080,  # Custom port
+            verify=True,
+        )
+
+    @mock.patch("pyntc.devices.nxos_device.ConnectHandler", create=True)
+    @mock.patch("pyntc.devices.nxos_device.NXOSNative", autospec=True)
+    def test_port_with_https(self, mock_device, mock_connect_handler):
+        """Test that port works with HTTPS transport."""
+        _ = NXOSDevice("host", "user", "pass", transport="https", port=8443)
+
+        # Verify NXOSNative was called with HTTPS and custom port
+        mock_device.assert_called_with(
+            "host",
+            "user",
+            "pass",
+            transport="https",
+            timeout=30,
+            port=8443,  # Custom HTTPS port
+            verify=True,
+        )
+
+    @mock.patch("pyntc.devices.nxos_device.ConnectHandler", create=True)
+    @mock.patch("pyntc.devices.nxos_device.NXOSNative", autospec=True)
+    def test_port_parameter_stored(self, mock_device, mock_connect_handler):
+        """Test that the port parameter is stored and used for NX-API connection."""
+        device = NXOSDevice("host", "user", "pass", port=8080)
+
+        # Verify port is used for NXOSNative (NX-API)
+        mock_device.assert_called_with(
+            "host",
+            "user",
+            "pass",
+            transport="http",
+            timeout=30,
+            port=8080,  # port for NX-API
+            verify=True,
+        )
+
+        # Verify port parameter is stored
+        self.assertEqual(device.port, 8080)
+
+    @mock.patch("pyntc.devices.nxos_device.ConnectHandler", create=True)
+    @mock.patch("pyntc.devices.nxos_device.NXOSNative", autospec=True)
+    def test_backward_compatibility_no_port(self, mock_device, mock_connect_handler):
+        """Test backward compatibility when port is not specified."""
+        # Create device without specifying port
+        _ = NXOSDevice("host", "user", "pass", transport="http")
+
+        # Should default to port None
+        mock_device.assert_called_with(
+            "host",
+            "user",
+            "pass",
+            transport="http",
+            timeout=30,
+            port=None,  # Default port
+            verify=True,
+        )
+
 
 if __name__ == "__main__":
     unittest.main()