From f82937e71a8a3d676b08deb5f29c95f4cc84e361 Mon Sep 17 00:00:00 2001 From: James Williams Date: Tue, 14 Apr 2026 10:47:54 -0500 Subject: [PATCH 1/8] Updates to the EOS driver based on PR comments from #365 --- pyntc/devices/eos_device.py | 95 +-- tests/unit/test_devices/test_eos_device.py | 946 ++++++--------------- 2 files changed, 279 insertions(+), 762 deletions(-) diff --git a/pyntc/devices/eos_device.py b/pyntc/devices/eos_device.py index 2727eec3..34ad3769 100644 --- a/pyntc/devices/eos_device.py +++ b/pyntc/devices/eos_device.py @@ -26,6 +26,7 @@ from pyntc.utils import convert_list_by_key from pyntc.utils.models import FileCopyModel +EOS_SUPPORTED_HASHING_ALGORITHMS = {"md5", "sha1", "sha256", "sha512"} BASIC_FACTS_KM = {"model": "modelName", "os_version": "internalVersion", "serial_number": "serialNumber"} INTERFACES_KM = { "speed": "bandwidth", @@ -481,6 +482,13 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5", **kwargs): Raises: CommandError: If the verify command fails (but not if file doesn't exist). """ + # Validate hashing algorithm against EOS-supported values + if hashing_algorithm.lower() not in EOS_SUPPORTED_HASHING_ALGORITHMS: + raise ValueError( + f"Unsupported hashing algorithm '{hashing_algorithm}' for EOS. " + f"Supported algorithms: {sorted(EOS_SUPPORTED_HASHING_ALGORITHMS)}" + ) + file_system = kwargs.get("file_system") if file_system is None: file_system = self._get_file_system() @@ -511,11 +519,11 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5", **kwargs): # Parse the checksum from the output # Expected format: verify /sha512 (flash:nautobot.png) = - match = re.search(r"=\s*([a-fA-F0-9]+)", result) - if match: - remote_checksum = match.group(1).lower() - log.debug("Host %s: Remote checksum for %s: %s", self.host, filename, remote_checksum) - return remote_checksum + if "=" in result: + remote_checksum = result.split("=")[-1].strip().lower() + if remote_checksum: + log.debug("Host %s: Remote checksum for %s: %s", self.host, filename, remote_checksum) + return remote_checksum log.error("Host %s: Could not parse checksum from verify output: %s", self.host, result) raise CommandError(command, f"Could not parse checksum from verify output: {result}") @@ -524,51 +532,39 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5", **kwargs): log.error("Host %s: Error getting remote checksum: %s", self.host, str(e)) raise CommandError(command, f"Error getting remote checksum: {str(e)}") - def _build_url_copy_command_simple(self, src, file_system): + def _build_url_copy_command_simple(self, src, file_system, dest): """Build copy command for simple URL-based transfers (TFTP, HTTP, HTTPS without credentials).""" - return f"copy {src.download_url} {file_system}", False + return f"copy {src.download_url} {file_system}/{dest}", False - def _build_url_copy_command_with_creds(self, src, 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).""" parsed = urlparse(src.download_url) - hostname = parsed.hostname + netloc = f"{parsed.hostname}:{parsed.port}" if parsed.port else parsed.hostname path = parsed.path - # Determine port based on scheme - if parsed.port: - port = parsed.port - elif src.scheme == "https": - port = "443" - elif src.scheme in ["http"]: - port = "80" - else: - port = "" - - port_str = f":{port}" if port else "" - # For HTTP/HTTPS, include both username and token if src.scheme in ["http", "https"]: - command = f"copy {src.scheme}://{src.username}:{src.token}@{hostname}{port_str}{path} {file_system}" + command = f"copy {src.scheme}://{src.username}:{src.token}@{netloc}{path} {file_system}/{dest}" detect_prompt = False # For SCP/FTP/SFTP, include only username (password via prompt) else: - command = f"copy {src.scheme}://{src.username}@{hostname}{port_str}{path} {file_system}" + command = f"copy {src.scheme}://{src.username}@{netloc}{path} {file_system}/{dest}" detect_prompt = True return command, detect_prompt - def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, include_username=False, **kwargs): + def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kwargs): """Copy a file from remote source to device. Args: src (FileCopyModel): The source file model with transfer parameters. dest (str): Destination filename (defaults to src.file_name). file_system (str): Device filesystem (auto-detected if not provided). - include_username (bool): Whether to include username in the copy command. Defaults to False. **kwargs (Any): Passible parameters such as file_system. Raises: TypeError: If src is not a FileCopyModel. + ValueError: If the URL scheme is unsupported or URL contains query strings. FileTransferError: If transfer or verification fails. FileSystemNotFoundError: If filesystem cannot be determined. """ @@ -576,6 +572,16 @@ def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, incl if not isinstance(src, FileCopyModel): raise TypeError("src must be an instance of FileCopyModel") + # Validate scheme before connecting to the device (fail fast) + supported_schemes = ["http", "https", "scp", "ftp", "sftp", "tftp"] + if src.scheme not in supported_schemes: + raise ValueError(f"Unsupported scheme: {src.scheme}") + + # Reject URLs with query strings since EOS CLI cannot handle '?' + parsed = urlparse(src.download_url) + if parsed.query: + raise ValueError(f"URLs with query strings are not supported on EOS: {src.download_url}") + # Determine file system if file_system is None: file_system = self._get_file_system() @@ -590,31 +596,11 @@ def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, incl self.open() self.enable() - # Validate scheme - supported_schemes = ["http", "https", "scp", "ftp", "sftp", "tftp"] - if src.scheme not in supported_schemes: - raise ValueError(f"Unsupported scheme: {src.scheme}") - # Build command based on scheme and credentials - command_builders = { - ("tftp", False): lambda: self._build_url_copy_command_simple(src, file_system), - ("http", False): lambda: self._build_url_copy_command_simple(src, file_system), - ("https", False): lambda: self._build_url_copy_command_simple(src, file_system), - ("http", True): lambda: self._build_url_copy_command_with_creds(src, file_system), - ("https", True): lambda: self._build_url_copy_command_with_creds(src, file_system), - ("scp", False): lambda: self._build_url_copy_command_with_creds(src, file_system), - ("scp", True): lambda: self._build_url_copy_command_with_creds(src, file_system), - ("ftp", False): lambda: self._build_url_copy_command_with_creds(src, file_system), - ("ftp", True): lambda: self._build_url_copy_command_with_creds(src, file_system), - ("sftp", False): lambda: self._build_url_copy_command_with_creds(src, file_system), - ("sftp", True): lambda: self._build_url_copy_command_with_creds(src, file_system), - } - - builder_key = (src.scheme, include_username and src.username is not None) - if builder_key not in command_builders: - raise ValueError(f"Unable to construct copy command for scheme {src.scheme} with provided credentials") - - command, detect_prompt = command_builders[builder_key]() + if src.scheme == "tftp" or (src.username is None and src.token is None): + command, detect_prompt = self._build_url_copy_command_simple(src, file_system, dest) + else: + command, detect_prompt = self._build_url_copy_command_with_creds(src, file_system, dest) log.debug("Host %s: Preparing copy command for %s", self.host, src.scheme) # Execute copy command @@ -676,11 +662,12 @@ def verify_file(self, checksum, filename, hashing_algorithm="md5", **kwargs): Returns: (bool): True if the file is verified successfully, False otherwise. """ - exists = self.check_file_exists(filename, **kwargs) - device_checksum = ( - self.get_remote_checksum(filename, hashing_algorithm=hashing_algorithm, **kwargs) if exists else None - ) - if checksum == device_checksum: + if not self.check_file_exists(filename, **kwargs): + log.debug("Host %s: File %s not found on device", self.host, filename) + return False + + device_checksum = self.get_remote_checksum(filename, hashing_algorithm=hashing_algorithm, **kwargs) + if checksum.lower() == device_checksum.lower(): log.debug("Host %s: Checksum verification successful for file %s", self.host, filename) return True diff --git a/tests/unit/test_devices/test_eos_device.py b/tests/unit/test_devices/test_eos_device.py index 9848cf5a..8786dfa1 100644 --- a/tests/unit/test_devices/test_eos_device.py +++ b/tests/unit/test_devices/test_eos_device.py @@ -438,174 +438,6 @@ def test_init_pass_port_and_timeout(mock_eos_connect): ) -# Property-based tests for file system normalization -try: - from hypothesis import given - from hypothesis import strategies as st -except ImportError: - # Create dummy decorators if hypothesis is not available - def given(*args, **kwargs): - def decorator(func): - return func - - return decorator - - class _ST: - @staticmethod - def just(value): - return value - - @staticmethod - def one_of(*args): - return args[0] - - st = _ST() - - -@given( - src=st.just("not_a_filecopymodel"), -) -def test_property_type_validation(src): - """Feature: arista-remote-file-copy, Property 1: Type Validation. - - For any non-FileCopyModel object passed as `src`, the `remote_file_copy()` - method should raise a `TypeError`. - - Validates: Requirements 1.2, 15.1 - """ - device = EOSDevice("host", "user", "pass") - - with pytest.raises(TypeError) as exc_info: - device.remote_file_copy(src) - - assert "src must be an instance of FileCopyModel" in str(exc_info.value) - - -@mock.patch.object(EOSDevice, "_get_file_system") -def test_property_file_system_auto_detection(mock_get_fs): - """Feature: arista-remote-file-copy, Property 26: File System Auto-Detection. - - For any `remote_file_copy()` call without an explicit `file_system` parameter, - the method should call `_get_file_system()` to determine the default file system. - - Validates: Requirements 11.1 - """ - mock_get_fs.return_value = "/mnt/flash" - device = EOSDevice("host", "user", "pass") - - src = FileCopyModel( - download_url="http://server.example.com/file.bin", - checksum="abc123", - file_name="file.bin", - ) - - # Call remote_file_copy without file_system parameter - try: - device.remote_file_copy(src) - except Exception: - # We expect it to fail later, but we just want to verify _get_file_system was called - pass - - # Verify _get_file_system was called - mock_get_fs.assert_called() - - -@mock.patch.object(EOSDevice, "_get_file_system") -def test_property_explicit_file_system_usage(mock_get_fs): - """Feature: arista-remote-file-copy, Property 27: Explicit File System Usage. - - For any `remote_file_copy()` call with an explicit `file_system` parameter, - that value should be used instead of auto-detection. - - Validates: Requirements 11.2 - """ - device = EOSDevice("host", "user", "pass") - - src = FileCopyModel( - download_url="http://server.example.com/file.bin", - checksum="abc123", - file_name="file.bin", - ) - - # Call remote_file_copy with explicit file_system parameter - try: - device.remote_file_copy(src, file_system="/mnt/flash") - except Exception: - # We expect it to fail later, but we just want to verify _get_file_system was NOT called - pass - - # Verify _get_file_system was NOT called - mock_get_fs.assert_not_called() - - -@mock.patch.object(EOSDevice, "verify_file") -@mock.patch.object(EOSDevice, "enable") -@mock.patch.object(EOSDevice, "open") -@mock.patch.object(EOSDevice, "_get_file_system") -def test_property_default_destination_from_filecopymodel(mock_get_fs, mock_open, mock_enable, mock_verify): - """Feature: arista-remote-file-copy, Property 28: Default Destination from FileCopyModel. - - For any `remote_file_copy()` call without an explicit `dest` parameter, - the destination should default to `src.file_name`. - - Validates: Requirements 12.1 - """ - mock_get_fs.return_value = "/mnt/flash" - mock_verify.return_value = True - - device = EOSDevice("host", "user", "pass") - device.native_ssh = mock.MagicMock() - device.native_ssh.send_command.return_value = "Copy completed successfully" - - src = FileCopyModel( - download_url="http://server.example.com/myfile.bin", - checksum="abc123", - file_name="myfile.bin", - ) - - # Call remote_file_copy without explicit dest - device.remote_file_copy(src) - - # Verify verify_file was called with the default destination - mock_verify.assert_called() - call_args = mock_verify.call_args - assert call_args[0][1] == "myfile.bin" # dest should be file_name - - -@mock.patch.object(EOSDevice, "verify_file") -@mock.patch.object(EOSDevice, "enable") -@mock.patch.object(EOSDevice, "open") -@mock.patch.object(EOSDevice, "_get_file_system") -def test_property_explicit_destination_usage(mock_get_fs, mock_open, mock_enable, mock_verify): - """Feature: arista-remote-file-copy, Property 29: Explicit Destination Usage. - - For any `remote_file_copy()` call with an explicit `dest` parameter, - that value should be used as the destination filename. - - Validates: Requirements 12.2 - """ - mock_get_fs.return_value = "/mnt/flash" - mock_verify.return_value = True - - device = EOSDevice("host", "user", "pass") - device.native_ssh = mock.MagicMock() - device.native_ssh.send_command.return_value = "Copy completed successfully" - - src = FileCopyModel( - download_url="http://server.example.com/myfile.bin", - checksum="abc123", - file_name="myfile.bin", - ) - - # Call remote_file_copy with explicit dest - device.remote_file_copy(src, dest="different_name.bin") - - # Verify verify_file was called with the explicit destination - mock_verify.assert_called() - call_args = mock_verify.call_args - assert call_args[0][1] == "different_name.bin" # dest should be the explicit value - - class TestRemoteFileCopy(unittest.TestCase): """Tests for remote_file_copy method.""" @@ -622,22 +454,41 @@ def tearDown(self): def test_remote_file_copy_invalid_src_type(self): """Test remote_file_copy raises TypeError for invalid src type.""" - with pytest.raises(TypeError) as exc_info: + with self.assertRaises(TypeError) as ctx: self.device.remote_file_copy("not_a_model") - assert "src must be an instance of FileCopyModel" in str(exc_info.value) + self.assertIn("src must be an instance of FileCopyModel", str(ctx.exception)) @mock.patch.object(EOSDevice, "verify_file") @mock.patch.object(EOSDevice, "enable") @mock.patch.object(EOSDevice, "open") @mock.patch.object(EOSDevice, "_get_file_system") - def test_remote_file_copy_skip_transfer_on_checksum_match(self, mock_get_fs, mock_open, mock_enable, mock_verify): - """Test remote_file_copy skips transfer when file exists with matching checksum.""" - from pyntc.utils.models import FileCopyModel + def test_remote_file_copy_file_system_auto_detection(self, mock_get_fs, mock_open, mock_enable, mock_verify): + """Test remote_file_copy calls _get_file_system when file_system is not provided.""" + mock_get_fs.return_value = "/mnt/flash" + mock_verify.return_value = True + mock_ssh = mock.MagicMock() + mock_ssh.send_command.return_value = "Copy completed successfully" + self.device.native_ssh = mock_ssh + + src = FileCopyModel( + download_url="http://server.example.com/file.bin", + checksum="abc123", + file_name="file.bin", + ) + + self.device.remote_file_copy(src) + mock_get_fs.assert_called() + + @mock.patch.object(EOSDevice, "verify_file") + @mock.patch.object(EOSDevice, "enable") + @mock.patch.object(EOSDevice, "open") + @mock.patch.object(EOSDevice, "_get_file_system") + def test_remote_file_copy_skip_transfer_on_checksum_match(self, mock_get_fs, mock_open, mock_enable, mock_verify): + """Test remote_file_copy completes when file exists with matching checksum.""" mock_get_fs.return_value = "flash:" mock_verify.return_value = True - # Mock netmiko connection mock_ssh = mock.MagicMock() mock_ssh.send_command.return_value = "Copy completed successfully" self.device.native_ssh = mock_ssh @@ -648,10 +499,7 @@ def test_remote_file_copy_skip_transfer_on_checksum_match(self, mock_get_fs, moc file_name="file.bin", ) - # Should return without raising exception self.device.remote_file_copy(src) - - # Verify that verify_file was called mock_verify.assert_called() @mock.patch.object(EOSDevice, "verify_file") @@ -660,12 +508,9 @@ def test_remote_file_copy_skip_transfer_on_checksum_match(self, mock_get_fs, moc @mock.patch.object(EOSDevice, "_get_file_system") def test_remote_file_copy_http_transfer(self, mock_get_fs, mock_open, mock_enable, mock_verify): """Test remote_file_copy executes HTTP transfer correctly.""" - from pyntc.utils.models import FileCopyModel - mock_get_fs.return_value = "flash:" - mock_verify.return_value = True # Verification passes + mock_verify.return_value = True - # Mock netmiko connection mock_ssh = mock.MagicMock() mock_ssh.send_command.return_value = "Copy completed successfully" self.device.native_ssh = mock_ssh @@ -676,15 +521,13 @@ def test_remote_file_copy_http_transfer(self, mock_get_fs, mock_open, mock_enabl file_name="file.bin", ) - # Should not raise exception self.device.remote_file_copy(src) - # Verify open and enable were called mock_open.assert_called_once() mock_enable.assert_called_once() - - # Verify send_command was called with correct command mock_ssh.send_command.assert_called() + call_args = mock_ssh.send_command.call_args + self.assertIn("copy http://", call_args[0][0]) @mock.patch.object(EOSDevice, "verify_file") @mock.patch.object(EOSDevice, "enable") @@ -692,12 +535,9 @@ def test_remote_file_copy_http_transfer(self, mock_get_fs, mock_open, mock_enabl @mock.patch.object(EOSDevice, "_get_file_system") def test_remote_file_copy_verification_failure(self, mock_get_fs, mock_open, mock_enable, mock_verify): """Test remote_file_copy raises FileTransferError when verification fails.""" - from pyntc.utils.models import FileCopyModel - mock_get_fs.return_value = "flash:" - mock_verify.return_value = False # Verification fails + mock_verify.return_value = False - # Mock netmiko connection mock_ssh = mock.MagicMock() mock_ssh.send_command.return_value = "Copy completed successfully" self.device.native_ssh = mock_ssh @@ -708,22 +548,43 @@ def test_remote_file_copy_verification_failure(self, mock_get_fs, mock_open, moc file_name="file.bin", ) - # Should raise FileTransferError - with pytest.raises(FileTransferError): + with self.assertRaises(FileTransferError): self.device.remote_file_copy(src) + @mock.patch.object(EOSDevice, "verify_file") + @mock.patch.object(EOSDevice, "enable") + @mock.patch.object(EOSDevice, "open") + @mock.patch.object(EOSDevice, "_get_file_system") + def test_remote_file_copy_with_default_dest(self, mock_get_fs, mock_open, mock_enable, mock_verify): + """Test remote_file_copy defaults dest to src.file_name.""" + mock_get_fs.return_value = "/mnt/flash" + mock_verify.return_value = True + + mock_ssh = mock.MagicMock() + mock_ssh.send_command.return_value = "Copy completed successfully" + self.device.native_ssh = mock_ssh + + src = FileCopyModel( + download_url="http://server.example.com/myfile.bin", + checksum="abc123", + file_name="myfile.bin", + ) + + self.device.remote_file_copy(src) + + mock_verify.assert_called() + call_args = mock_verify.call_args + self.assertEqual(call_args[0][1], "myfile.bin") + @mock.patch.object(EOSDevice, "verify_file") @mock.patch.object(EOSDevice, "enable") @mock.patch.object(EOSDevice, "open") @mock.patch.object(EOSDevice, "_get_file_system") def test_remote_file_copy_with_explicit_dest(self, mock_get_fs, mock_open, mock_enable, mock_verify): """Test remote_file_copy uses explicit dest parameter.""" - from pyntc.utils.models import FileCopyModel - mock_get_fs.return_value = "flash:" mock_verify.return_value = True - # Mock netmiko connection mock_ssh = mock.MagicMock() mock_ssh.send_command.return_value = "Copy completed successfully" self.device.native_ssh = mock_ssh @@ -734,12 +595,10 @@ def test_remote_file_copy_with_explicit_dest(self, mock_get_fs, mock_open, mock_ file_name="file.bin", ) - # Call with explicit dest self.device.remote_file_copy(src, dest="custom_name.bin") - # Verify verify_file was called with custom dest call_args = mock_verify.call_args - assert call_args[0][1] == "custom_name.bin" + self.assertEqual(call_args[0][1], "custom_name.bin") @mock.patch.object(EOSDevice, "verify_file") @mock.patch.object(EOSDevice, "enable") @@ -747,11 +606,8 @@ def test_remote_file_copy_with_explicit_dest(self, mock_get_fs, mock_open, mock_ @mock.patch.object(EOSDevice, "_get_file_system") def test_remote_file_copy_with_explicit_file_system(self, mock_get_fs, mock_open, mock_enable, mock_verify): """Test remote_file_copy uses explicit file_system parameter.""" - from pyntc.utils.models import FileCopyModel - mock_verify.return_value = True - # Mock netmiko connection mock_ssh = mock.MagicMock() mock_ssh.send_command.return_value = "Copy completed successfully" self.device.native_ssh = mock_ssh @@ -762,15 +618,11 @@ def test_remote_file_copy_with_explicit_file_system(self, mock_get_fs, mock_open file_name="file.bin", ) - # Call with explicit file_system self.device.remote_file_copy(src, file_system="flash:") - # Verify _get_file_system was NOT called mock_get_fs.assert_not_called() - - # Verify send_command was called with correct file_system call_args = mock_ssh.send_command.call_args - assert "flash:" in call_args[0][0] + self.assertIn("flash:", call_args[0][0]) @mock.patch.object(EOSDevice, "verify_file") @mock.patch.object(EOSDevice, "enable") @@ -778,12 +630,9 @@ def test_remote_file_copy_with_explicit_file_system(self, mock_get_fs, mock_open @mock.patch.object(EOSDevice, "_get_file_system") def test_remote_file_copy_scp_with_credentials(self, mock_get_fs, mock_open, mock_enable, mock_verify): """Test remote_file_copy constructs SCP command with username only.""" - from pyntc.utils.models import FileCopyModel - mock_get_fs.return_value = "flash:" mock_verify.return_value = True - # Mock netmiko connection mock_ssh = mock.MagicMock() mock_ssh.send_command_timing.return_value = "Copy completed successfully" self.device.native_ssh = mock_ssh @@ -796,13 +645,11 @@ def test_remote_file_copy_scp_with_credentials(self, mock_get_fs, mock_open, moc self.device.remote_file_copy(src) - # Verify send_command_timing was called with SCP command containing username only - # Token is provided at the Arista "Password:" prompt call_args = mock_ssh.send_command_timing.call_args command = call_args[0][0] - assert "scp://" in command - assert "user@" in command - assert "pass@" not in command # Password should not be in command + self.assertIn("scp://", command) + self.assertIn("user@", command) + self.assertNotIn("pass@", command) @mock.patch.object(EOSDevice, "verify_file") @mock.patch.object(EOSDevice, "enable") @@ -810,110 +657,130 @@ def test_remote_file_copy_scp_with_credentials(self, mock_get_fs, mock_open, moc @mock.patch.object(EOSDevice, "_get_file_system") def test_remote_file_copy_timeout_applied(self, mock_get_fs, mock_open, mock_enable, mock_verify): """Test remote_file_copy applies timeout to send_command.""" - from pyntc.utils.models import FileCopyModel - mock_get_fs.return_value = "flash:" mock_verify.return_value = True - # Mock netmiko connection mock_ssh = mock.MagicMock() mock_ssh.send_command.return_value = "Copy completed successfully" self.device.native_ssh = mock_ssh - src = FileCopyModel( - download_url="http://example.com/file.bin", - checksum="abc123", - file_name="file.bin", - timeout=1800, - ) - - self.device.remote_file_copy(src) - - # Verify send_command was called with correct timeout - call_args = mock_ssh.send_command.call_args - assert call_args[1]["read_timeout"] == 1800 - + for timeout in [300, 600, 900, 1800]: + with self.subTest(timeout=timeout): + src = FileCopyModel( + download_url="http://example.com/file.bin", + checksum="abc123", + file_name="file.bin", + timeout=timeout, + ) -# Property-based tests for Task 7: Pre-transfer verification + self.device.remote_file_copy(src) + call_args = mock_ssh.send_command.call_args + self.assertEqual(call_args[1]["read_timeout"], timeout) -@mock.patch.object(EOSDevice, "verify_file") -@mock.patch.object(EOSDevice, "enable") -@mock.patch.object(EOSDevice, "open") -@mock.patch.object(EOSDevice, "_get_file_system") -def test_property_skip_transfer_on_checksum_match(mock_get_fs, mock_open, mock_enable, mock_verify): - """Feature: arista-remote-file-copy, Property 14: Skip Transfer on Checksum Match. - - For any file that already exists on the device with a matching checksum, - the `remote_file_copy()` method should return successfully after verification. - - Validates: Requirements 5.2 - """ - from pyntc.utils.models import FileCopyModel - - mock_get_fs.return_value = "/mnt/flash" - mock_verify.return_value = True # File exists with matching checksum + @mock.patch.object(EOSDevice, "verify_file") + @mock.patch.object(EOSDevice, "enable") + @mock.patch.object(EOSDevice, "open") + @mock.patch.object(EOSDevice, "_get_file_system") + def test_remote_file_copy_checksum_mismatch_raises_error(self, mock_get_fs, mock_open, mock_enable, mock_verify): + """Test remote_file_copy raises FileTransferError on checksum mismatch after transfer.""" + mock_get_fs.return_value = "/mnt/flash" + mock_verify.return_value = False - device = EOSDevice("host", "user", "pass") - device.native_ssh = mock.MagicMock() - device.native_ssh.send_command.return_value = "Copy completed successfully" + mock_ssh = mock.MagicMock() + mock_ssh.send_command.return_value = "Copy completed successfully" + self.device.native_ssh = mock_ssh - src = FileCopyModel( - download_url="http://server.example.com/file.bin", - checksum="abc123def456", - file_name="file.bin", - ) + src = FileCopyModel( + download_url="http://server.example.com/file.bin", + checksum="abc123def456", + file_name="file.bin", + ) - # Call remote_file_copy - device.remote_file_copy(src) + with self.assertRaises(FileTransferError): + self.device.remote_file_copy(src) - # Verify that verify_file was called - mock_verify.assert_called() + mock_ssh.send_command.assert_called() + call_args = mock_ssh.send_command.call_args + self.assertIn("copy", call_args[0][0].lower()) - # Verify that send_command was called (transfer always occurs) - device.native_ssh.send_command.assert_called() + @mock.patch.object(EOSDevice, "verify_file") + @mock.patch.object(EOSDevice, "enable") + @mock.patch.object(EOSDevice, "open") + @mock.patch.object(EOSDevice, "_get_file_system") + def test_remote_file_copy_post_transfer_verification(self, mock_get_fs, mock_open, mock_enable, mock_verify): + """Test remote_file_copy calls verify_file with correct algorithm after transfer.""" + mock_get_fs.return_value = "/mnt/flash" + mock_verify.return_value = True + mock_ssh = mock.MagicMock() + mock_ssh.send_command.return_value = "Copy completed successfully" + self.device.native_ssh = mock_ssh -@mock.patch.object(EOSDevice, "verify_file") -@mock.patch.object(EOSDevice, "enable") -@mock.patch.object(EOSDevice, "open") -@mock.patch.object(EOSDevice, "_get_file_system") -def test_property_proceed_on_checksum_mismatch(mock_get_fs, mock_open, mock_enable, mock_verify): - """Feature: arista-remote-file-copy, Property 15: Proceed on Checksum Mismatch. + for checksum, algorithm in [("abc123def456", "md5"), ("abc123def456789", "sha256")]: + with self.subTest(algorithm=algorithm): + src = FileCopyModel( + download_url="http://server.example.com/file.bin", + checksum=checksum, + file_name="file.bin", + hashing_algorithm=algorithm, + ) - For any file that exists on the device but has a mismatched checksum, - the `remote_file_copy()` method should proceed with the file transfer. + self.device.remote_file_copy(src) + mock_verify.assert_called() - Validates: Requirements 5.3 - """ - from pyntc.utils.models import FileCopyModel + def test_remote_file_copy_unsupported_scheme(self): + """Test remote_file_copy raises ValueError for unsupported scheme.""" + src = FileCopyModel( + download_url="http://example.com/file.bin", + checksum="abc123", + file_name="file.bin", + ) + # Override scheme to something unsupported + src.scheme = "gopher" - mock_get_fs.return_value = "/mnt/flash" - # Verification fails (file doesn't exist or checksum mismatches) - mock_verify.return_value = False + with self.assertRaises(ValueError) as ctx: + self.device.remote_file_copy(src) + self.assertIn("Unsupported scheme", str(ctx.exception)) - device = EOSDevice("host", "user", "pass") - mock_ssh = mock.MagicMock() - mock_ssh.send_command.return_value = "Copy completed successfully" - device.native_ssh = mock_ssh + def test_remote_file_copy_query_string_rejected(self): + """Test remote_file_copy raises ValueError for URLs with query strings.""" + src = FileCopyModel( + download_url="http://example.com/file.bin", + checksum="abc123", + file_name="file.bin", + ) + # Override download_url to include a query string + src.download_url = "http://example.com/file.bin?token=abc" - src = FileCopyModel( - download_url="http://server.example.com/file.bin", - checksum="abc123def456", - file_name="file.bin", - ) + with self.assertRaises(ValueError) as ctx: + self.device.remote_file_copy(src) + self.assertIn("query strings are not supported", str(ctx.exception)) - # Call remote_file_copy - should raise FileTransferError because verification fails - with pytest.raises(FileTransferError): - device.remote_file_copy(src) + @mock.patch.object(EOSDevice, "verify_file") + @mock.patch.object(EOSDevice, "enable") + @mock.patch.object(EOSDevice, "open") + @mock.patch.object(EOSDevice, "_get_file_system") + def test_remote_file_copy_logging_on_success(self, mock_get_fs, mock_open, mock_enable, mock_verify): + """Test that transfer success is logged.""" + mock_get_fs.return_value = "/mnt/flash" + mock_verify.return_value = True - # Verify that send_command was called with a copy command - mock_ssh.send_command.assert_called() - call_args = mock_ssh.send_command.call_args - assert "copy" in call_args[0][0].lower() + mock_ssh = mock.MagicMock() + mock_ssh.send_command.return_value = "Copy completed successfully" + self.device.native_ssh = mock_ssh + src = FileCopyModel( + download_url="http://server.example.com/file.bin", + checksum="abc123def456", + file_name="file.bin", + ) -# Tests for Task 8: Command Execution + with mock.patch("pyntc.devices.eos_device.log") as mock_log: + self.device.remote_file_copy(src) + self.assertTrue( + any("transferred and verified successfully" in str(call) for call in mock_log.info.call_args_list) + ) class TestRemoteFileCopyCommandExecution(unittest.TestCase): @@ -925,8 +792,6 @@ class TestRemoteFileCopyCommandExecution(unittest.TestCase): @mock.patch.object(EOSDevice, "_get_file_system") def test_command_execution_with_http(self, mock_get_fs, mock_open, mock_enable, mock_verify): """Test command execution for HTTP transfer.""" - from pyntc.utils.models import FileCopyModel - mock_get_fs.return_value = "/mnt/flash" mock_verify.return_value = True @@ -943,16 +808,11 @@ def test_command_execution_with_http(self, mock_get_fs, mock_open, mock_enable, device.remote_file_copy(src) - # Verify open() was called mock_open.assert_called_once() - - # Verify enable() was called mock_enable.assert_called_once() - - # Verify send_command was called with HTTP copy command mock_ssh.send_command.assert_called() call_args = mock_ssh.send_command.call_args - assert "copy http://" in call_args[0][0] + self.assertIn("copy http://", call_args[0][0]) @mock.patch.object(EOSDevice, "verify_file") @mock.patch.object(EOSDevice, "enable") @@ -960,8 +820,6 @@ def test_command_execution_with_http(self, mock_get_fs, mock_open, mock_enable, @mock.patch.object(EOSDevice, "_get_file_system") def test_command_execution_with_scp_credentials(self, mock_get_fs, mock_open, mock_enable, mock_verify): """Test command execution for SCP transfer with username only.""" - from pyntc.utils.models import FileCopyModel - mock_get_fs.return_value = "/mnt/flash" mock_verify.return_value = True @@ -980,13 +838,11 @@ def test_command_execution_with_scp_credentials(self, mock_get_fs, mock_open, mo device.remote_file_copy(src) - # Verify send_command_timing was called with SCP copy command including username only - # Token is provided at the Arista "Password:" prompt mock_ssh.send_command_timing.assert_called() call_args = mock_ssh.send_command_timing.call_args - assert "copy scp://" in call_args[0][0] - assert "admin@" in call_args[0][0] - assert "password@" not in call_args[0][0] # Password should not be in command + self.assertIn("copy scp://", call_args[0][0]) + self.assertIn("admin@", call_args[0][0]) + self.assertNotIn("password@", call_args[0][0]) @mock.patch.object(EOSDevice, "verify_file") @mock.patch.object(EOSDevice, "enable") @@ -994,8 +850,6 @@ def test_command_execution_with_scp_credentials(self, mock_get_fs, mock_open, mo @mock.patch.object(EOSDevice, "_get_file_system") def test_timeout_applied_to_send_command(self, mock_get_fs, mock_open, mock_enable, mock_verify): """Test that timeout is applied to send_command calls.""" - from pyntc.utils.models import FileCopyModel - mock_get_fs.return_value = "/mnt/flash" mock_verify.return_value = True @@ -1013,420 +867,96 @@ def test_timeout_applied_to_send_command(self, mock_get_fs, mock_open, mock_enab device.remote_file_copy(src) - # Verify send_command was called with the specified timeout mock_ssh.send_command.assert_called() call_args = mock_ssh.send_command.call_args - assert call_args[1]["read_timeout"] == 600 - - -# Tests for Task 9: Post-transfer Verification - - -@pytest.mark.parametrize( - "checksum,algorithm", - [ - ("abc123def456", "md5"), - ("abc123def456789", "sha256"), - ], -) -def test_property_post_transfer_verification(checksum, algorithm): - """Feature: arista-remote-file-copy, Property 20: Post-Transfer Verification. - - For any completed file transfer, the method should verify the file exists - on the device and compute its checksum using the specified algorithm. - - Validates: Requirements 9.1, 9.2 - """ - from pyntc.utils.models import FileCopyModel - - device = EOSDevice("host", "user", "pass") - device.native_ssh = mock.MagicMock() - - with mock.patch.object(device, "verify_file") as mock_verify: - with mock.patch.object(device, "_get_file_system") as mock_get_fs: - with mock.patch.object(device, "open"): - with mock.patch.object(device, "enable"): - mock_get_fs.return_value = "/mnt/flash" - mock_verify.return_value = True - device.native_ssh.send_command.return_value = "Copy completed successfully" - - src = FileCopyModel( - download_url="http://server.example.com/file.bin", - checksum=checksum, - file_name="file.bin", - hashing_algorithm=algorithm, - ) - - device.remote_file_copy(src) - - # Verify that verify_file was called - mock_verify.assert_called() - - -@pytest.mark.parametrize( - "checksum,algorithm", - [ - ("abc123def456", "md5"), - ("abc123def456789", "sha256"), - ], -) -def test_property_checksum_match_verification(checksum, algorithm): - """Feature: arista-remote-file-copy, Property 21: Checksum Match Verification. - - For any transferred file where the computed checksum matches the expected checksum, - the method should consider the transfer successful. - - Validates: Requirements 9.3 - """ - from pyntc.utils.models import FileCopyModel - - device = EOSDevice("host", "user", "pass") - device.native_ssh = mock.MagicMock() - - with mock.patch.object(device, "verify_file") as mock_verify: - with mock.patch.object(device, "_get_file_system") as mock_get_fs: - with mock.patch.object(device, "open"): - with mock.patch.object(device, "enable"): - mock_get_fs.return_value = "/mnt/flash" - # Verification passes - mock_verify.return_value = True - device.native_ssh.send_command.return_value = "Copy completed successfully" - - src = FileCopyModel( - download_url="http://server.example.com/file.bin", - checksum=checksum, - file_name="file.bin", - hashing_algorithm=algorithm, - ) - - # Should not raise an exception - device.remote_file_copy(src) - - -def test_property_checksum_mismatch_error(): - """Feature: arista-remote-file-copy, Property 22: Checksum Mismatch Error. - - For any transferred file where the computed checksum does not match the expected checksum, - the method should raise a FileTransferError. - - Validates: Requirements 9.4 - """ - from pyntc.utils.models import FileCopyModel - - device = EOSDevice("host", "user", "pass") - device.native_ssh = mock.MagicMock() - - with mock.patch.object(device, "verify_file") as mock_verify: - with mock.patch.object(device, "_get_file_system") as mock_get_fs: - with mock.patch.object(device, "open"): - with mock.patch.object(device, "enable"): - mock_get_fs.return_value = "/mnt/flash" - # First call: file doesn't exist (False) - # Second call: checksum mismatch (False) - mock_verify.side_effect = [False, False] - device.native_ssh.send_command.return_value = "Copy completed successfully" - - src = FileCopyModel( - download_url="http://server.example.com/file.bin", - checksum="abc123def456", - file_name="file.bin", - ) - - # Should raise FileTransferError - with pytest.raises(FileTransferError): - device.remote_file_copy(src) - - -def test_property_missing_file_after_transfer_error(): - """Feature: arista-remote-file-copy, Property 23: Missing File After Transfer Error. - - For any transfer that completes but the file does not exist on the device afterward, - the method should raise a FileTransferError. - - Validates: Requirements 9.5 - """ - from pyntc.utils.models import FileCopyModel - - device = EOSDevice("host", "user", "pass") - device.native_ssh = mock.MagicMock() - - with mock.patch.object(device, "verify_file") as mock_verify: - with mock.patch.object(device, "_get_file_system") as mock_get_fs: - with mock.patch.object(device, "open"): - with mock.patch.object(device, "enable"): - mock_get_fs.return_value = "/mnt/flash" - # First call: file doesn't exist (False) - # Second call: file still doesn't exist (False) - mock_verify.side_effect = [False, False] - device.native_ssh.send_command.return_value = "Copy completed successfully" - - src = FileCopyModel( - download_url="http://server.example.com/file.bin", - checksum="abc123def456", - file_name="file.bin", - ) - - # Should raise FileTransferError - with pytest.raises(FileTransferError): - device.remote_file_copy(src) - - -# Tests for Task 10: Timeout and FTP Support - - -@pytest.mark.parametrize("timeout", [300, 600, 900, 1800]) -def test_property_timeout_application(timeout): - """Feature: arista-remote-file-copy, Property 24: Timeout Application. - - For any FileCopyModel with a specified timeout value, that timeout should be used - when sending commands to the device during transfer. - - Validates: Requirements 10.1, 10.3 - """ - from pyntc.utils.models import FileCopyModel - - device = EOSDevice("host", "user", "pass") - device.native_ssh = mock.MagicMock() - - with mock.patch.object(device, "verify_file") as mock_verify: - with mock.patch.object(device, "_get_file_system") as mock_get_fs: - with mock.patch.object(device, "open"): - with mock.patch.object(device, "enable"): - mock_get_fs.return_value = "/mnt/flash" - mock_verify.return_value = True - device.native_ssh.send_command.return_value = "Copy completed successfully" - - src = FileCopyModel( - download_url="http://server.example.com/file.bin", - checksum="abc123def456", - file_name="file.bin", - timeout=timeout, - ) - - device.remote_file_copy(src) - - # Verify send_command was called with the correct timeout - call_args = device.native_ssh.send_command.call_args - assert call_args[1]["read_timeout"] == timeout - - -def test_property_default_timeout_value(): - """Feature: arista-remote-file-copy, Property 25: Default Timeout Value. - - For any FileCopyModel without an explicit timeout, the default timeout should be 900 seconds. - - Validates: Requirements 10.2 - """ - from pyntc.utils.models import FileCopyModel - - src = FileCopyModel( - download_url="http://server.example.com/file.bin", - checksum="abc123def456", - file_name="file.bin", - ) - - # Verify default timeout is 900 - assert src.timeout == 900 - - -@pytest.mark.parametrize("ftp_passive", [True, False]) -def test_property_ftp_passive_mode_configuration(ftp_passive): - """Feature: arista-remote-file-copy, Property 30/31: FTP Passive Mode Configuration. - - For any FileCopyModel with ftp_passive flag, the FTP transfer should use the specified mode. - - Validates: Requirements 19.1, 19.2 - """ - from pyntc.utils.models import FileCopyModel - - src = FileCopyModel( - download_url="ftp://admin:password@ftp.example.com/images/eos.swi", - checksum="abc123def456", - file_name="eos.swi", - ftp_passive=ftp_passive, - ) - - # Verify ftp_passive is set correctly - assert src.ftp_passive == ftp_passive - - -def test_property_default_ftp_passive_mode(): - """Feature: arista-remote-file-copy, Property 32: Default FTP Passive Mode. + self.assertEqual(call_args[1]["read_timeout"], 600) - For any FileCopyModel without an explicit ftp_passive parameter, the default should be True. - Validates: Requirements 19.3 - """ - from pyntc.utils.models import FileCopyModel - - src = FileCopyModel( - download_url="ftp://admin:password@ftp.example.com/images/eos.swi", - checksum="abc123def456", - file_name="eos.swi", - ) - - # Verify default ftp_passive is True - assert src.ftp_passive is True - - -# Tests for Task 11: Error Handling and Logging - - -class TestRemoteFileCopyErrorHandling(unittest.TestCase): - """Tests for error handling in remote_file_copy.""" - - def test_invalid_src_type_raises_typeerror(self): - """Test that invalid src type raises TypeError.""" - device = EOSDevice("host", "user", "pass") - - with pytest.raises(TypeError) as exc_info: - device.remote_file_copy("not a FileCopyModel") - - assert "src must be an instance of FileCopyModel" in str(exc_info.value) - - @mock.patch.object(EOSDevice, "verify_file") - @mock.patch.object(EOSDevice, "enable") - @mock.patch.object(EOSDevice, "open") - @mock.patch.object(EOSDevice, "_get_file_system") - def test_transfer_failure_raises_filetransfererror(self, mock_get_fs, mock_open, mock_enable, mock_verify): - """Test that transfer failure raises FileTransferError.""" - from pyntc.utils.models import FileCopyModel - - mock_get_fs.return_value = "/mnt/flash" - mock_verify.side_effect = [False, False] # Post-transfer verification fails - - device = EOSDevice("host", "user", "pass") - device.native_ssh = mock.MagicMock() - device.native_ssh.send_command.return_value = "Copy completed successfully" +class TestFileCopyModelValidation(unittest.TestCase): + """Tests for FileCopyModel defaults and validation.""" + def test_default_timeout_value(self): + """Test FileCopyModel default timeout is 900 seconds.""" src = FileCopyModel( download_url="http://server.example.com/file.bin", checksum="abc123def456", file_name="file.bin", ) - - with pytest.raises(FileTransferError): - device.remote_file_copy(src) - - @mock.patch.object(EOSDevice, "verify_file") - @mock.patch.object(EOSDevice, "enable") - @mock.patch.object(EOSDevice, "open") - @mock.patch.object(EOSDevice, "_get_file_system") - def test_logging_on_transfer_success(self, mock_get_fs, mock_open, mock_enable, mock_verify): - """Test that transfer success is logged.""" - from pyntc.utils.models import FileCopyModel - - mock_get_fs.return_value = "/mnt/flash" - mock_verify.return_value = True - - device = EOSDevice("host", "user", "pass") - device.native_ssh = mock.MagicMock() - device.native_ssh.send_command.return_value = "Copy completed successfully" - + self.assertEqual(src.timeout, 900) + + def test_ftp_passive_mode_configuration(self): + """Test FileCopyModel ftp_passive flag is set correctly.""" + for ftp_passive in [True, False]: + with self.subTest(ftp_passive=ftp_passive): + src = FileCopyModel( + download_url="ftp://admin:password@ftp.example.com/images/eos.swi", + checksum="abc123def456", + file_name="eos.swi", + ftp_passive=ftp_passive, + ) + self.assertEqual(src.ftp_passive, ftp_passive) + + def test_default_ftp_passive_mode(self): + """Test FileCopyModel default ftp_passive is True.""" src = FileCopyModel( - download_url="http://server.example.com/file.bin", + download_url="ftp://admin:password@ftp.example.com/images/eos.swi", checksum="abc123def456", - file_name="file.bin", + file_name="eos.swi", ) - - with mock.patch("pyntc.devices.eos_device.log") as mock_log: - device.remote_file_copy(src) - - # Verify that info log was called for successful transfer - assert any("transferred and verified successfully" in str(call) for call in mock_log.info.call_args_list) - - -# Tests for Task 12: FileCopyModel Validation - - -@pytest.mark.parametrize("algorithm", ["md5", "sha256", "sha512"]) -def test_property_hashing_algorithm_validation(algorithm): - """Feature: arista-remote-file-copy, Property 10: Hashing Algorithm Validation. - - For any unsupported hashing algorithm, FileCopyModel initialization should raise a ValueError. - - Validates: Requirements 6.3, 17.1, 17.2 - """ - from pyntc.utils.models import FileCopyModel - - # Should not raise for supported algorithms - src = FileCopyModel( - download_url="http://server.example.com/file.bin", - checksum="abc123def456", - file_name="file.bin", - hashing_algorithm=algorithm, - ) - - assert src.hashing_algorithm == algorithm - - -def test_property_case_insensitive_algorithm_validation(): - """Feature: arista-remote-file-copy, Property 11: Case-Insensitive Algorithm Validation. - - For any hashing algorithm specified in different cases, the FileCopyModel should accept it as valid. - - Validates: Requirements 17.3 - """ - from pyntc.utils.models import FileCopyModel - - # Should accept case-insensitive algorithms - for algorithm in ["MD5", "md5", "Md5", "SHA256", "sha256", "Sha256"]: + self.assertTrue(src.ftp_passive) + + def test_hashing_algorithm_validation(self): + """Test FileCopyModel accepts supported hashing algorithms.""" + for algorithm in ["md5", "sha256", "sha512"]: + with self.subTest(algorithm=algorithm): + src = FileCopyModel( + download_url="http://server.example.com/file.bin", + checksum="abc123def456", + file_name="file.bin", + hashing_algorithm=algorithm, + ) + self.assertEqual(src.hashing_algorithm, algorithm) + + def test_case_insensitive_algorithm_validation(self): + """Test FileCopyModel accepts algorithms in different cases.""" + for algorithm in ["MD5", "md5", "Md5", "SHA256", "sha256", "Sha256"]: + with self.subTest(algorithm=algorithm): + src = FileCopyModel( + download_url="http://server.example.com/file.bin", + checksum="abc123def456", + file_name="file.bin", + hashing_algorithm=algorithm, + ) + self.assertIn(src.hashing_algorithm.lower(), ["md5", "sha256"]) + + +class TestFileCopyModelCredentials(unittest.TestCase): + """Tests for FileCopyModel credential extraction.""" + + def test_url_credential_extraction(self): + """Test FileCopyModel extracts credentials from URL.""" + test_cases = [ + ("scp://admin:password@server.com/path", "admin", "password"), + ("ftp://user:pass123@ftp.example.com/file", "user", "pass123"), + ] + for url, expected_username, expected_token in test_cases: + with self.subTest(url=url): + src = FileCopyModel( + download_url=url, + checksum="abc123def456", + file_name="file.bin", + ) + self.assertEqual(src.username, expected_username) + self.assertEqual(src.token, expected_token) + + def test_explicit_credentials_override(self): + """Test explicit credentials take precedence over URL-embedded credentials.""" src = FileCopyModel( - download_url="http://server.example.com/file.bin", + download_url="scp://url_user:url_pass@server.com/path", checksum="abc123def456", file_name="file.bin", - hashing_algorithm=algorithm, + username="explicit_user", + token="explicit_pass", ) - - # Verify it was accepted (no exception raised) - assert src.hashing_algorithm.lower() in ["md5", "sha256"] - - -@pytest.mark.parametrize( - "url,expected_username,expected_token", - [ - ("scp://admin:password@server.com/path", "admin", "password"), - ("ftp://user:pass123@ftp.example.com/file", "user", "pass123"), - ], -) -def test_property_url_credential_extraction(url, expected_username, expected_token): - """Feature: arista-remote-file-copy, Property 12: URL Credential Extraction. - - For any URL containing embedded credentials, FileCopyModel should extract username and password. - - Validates: Requirements 3.1, 16.1, 16.2, 16.3, 16.4, 16.5, 16.6 - """ - from pyntc.utils.models import FileCopyModel - - src = FileCopyModel( - download_url=url, - checksum="abc123def456", - file_name="file.bin", - ) - - # Verify credentials were extracted - assert src.username == expected_username - assert src.token == expected_token - - -def test_property_explicit_credentials_override(): - """Feature: arista-remote-file-copy, Property 13: Explicit Credentials Override. - - For any FileCopyModel where both URL-embedded credentials and explicit fields are provided, - the explicit fields should take precedence. - - Validates: Requirements 3.2 - """ - from pyntc.utils.models import FileCopyModel - - src = FileCopyModel( - download_url="scp://url_user:url_pass@server.com/path", - checksum="abc123def456", - file_name="file.bin", - username="explicit_user", - token="explicit_pass", - ) - - # Verify explicit credentials take precedence - assert src.username == "explicit_user" - assert src.token == "explicit_pass" + self.assertEqual(src.username, "explicit_user") + self.assertEqual(src.token, "explicit_pass") From f96205489c1209eecd7f34a9dbb0e74b642118ed Mon Sep 17 00:00:00 2001 From: James Williams Date: Tue, 14 Apr 2026 10:57:13 -0500 Subject: [PATCH 2/8] simplify and remove redundancy --- pyntc/devices/eos_device.py | 50 ++++----- tests/unit/test_devices/test_eos_device.py | 118 +++++---------------- 2 files changed, 45 insertions(+), 123 deletions(-) diff --git a/pyntc/devices/eos_device.py b/pyntc/devices/eos_device.py index 34ad3769..b2bb50cf 100644 --- a/pyntc/devices/eos_device.py +++ b/pyntc/devices/eos_device.py @@ -26,7 +26,8 @@ from pyntc.utils import convert_list_by_key from pyntc.utils.models import FileCopyModel -EOS_SUPPORTED_HASHING_ALGORITHMS = {"md5", "sha1", "sha256", "sha512"} +EOS_SUPPORTED_HASHING_ALGORITHMS = {"md5", "sha1", "sha256", "sha512"} # Subset of HASHING_ALGORITHMS for EOS verify +EOS_SUPPORTED_SCHEMES = {"http", "https", "scp", "ftp", "sftp", "tftp"} BASIC_FACTS_KM = {"model": "modelName", "os_version": "internalVersion", "serial_number": "serialNumber"} INTERFACES_KM = { "speed": "bandwidth", @@ -482,7 +483,6 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5", **kwargs): Raises: CommandError: If the verify command fails (but not if file doesn't exist). """ - # Validate hashing algorithm against EOS-supported values if hashing_algorithm.lower() not in EOS_SUPPORTED_HASHING_ALGORITHMS: raise ValueError( f"Unsupported hashing algorithm '{hashing_algorithm}' for EOS. " @@ -534,25 +534,30 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5", **kwargs): def _build_url_copy_command_simple(self, src, file_system, dest): """Build copy command for simple URL-based transfers (TFTP, HTTP, HTTPS without credentials).""" - return f"copy {src.download_url} {file_system}/{dest}", False + return f"copy {src.clean_url} {file_system}/{dest}", False 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).""" - parsed = urlparse(src.download_url) + parsed = urlparse(src.clean_url) netloc = f"{parsed.hostname}:{parsed.port}" if parsed.port else parsed.hostname path = parsed.path - # For HTTP/HTTPS, include both username and token - if src.scheme in ["http", "https"]: + if src.scheme in ("http", "https"): command = f"copy {src.scheme}://{src.username}:{src.token}@{netloc}{path} {file_system}/{dest}" detect_prompt = False - # For SCP/FTP/SFTP, include only username (password via prompt) else: + # SCP/FTP/SFTP — password provided at the interactive prompt command = f"copy {src.scheme}://{src.username}@{netloc}{path} {file_system}/{dest}" detect_prompt = True return command, detect_prompt + def _check_copy_output_for_errors(self, output): + """Raise FileTransferError if copy command output contains error indicators.""" + if any(error in output.lower() for error in ["error", "invalid", "failed"]): + log.error("Host %s: Error detected in copy command output: %s", self.host, output) + raise FileTransferError(f"Error detected in copy command output: {output}") + def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kwargs): """Copy a file from remote source to device. @@ -568,65 +573,48 @@ def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kw FileTransferError: If transfer or verification fails. FileSystemNotFoundError: If filesystem cannot be determined. """ - # Validate input if not isinstance(src, FileCopyModel): raise TypeError("src must be an instance of FileCopyModel") - # Validate scheme before connecting to the device (fail fast) - supported_schemes = ["http", "https", "scp", "ftp", "sftp", "tftp"] - if src.scheme not in supported_schemes: + if src.scheme not in EOS_SUPPORTED_SCHEMES: raise ValueError(f"Unsupported scheme: {src.scheme}") - # Reject URLs with query strings since EOS CLI cannot handle '?' - parsed = urlparse(src.download_url) - if parsed.query: + # EOS CLI cannot handle '?' in URLs + if "?" in src.clean_url: raise ValueError(f"URLs with query strings are not supported on EOS: {src.download_url}") - # Determine file system if file_system is None: file_system = self._get_file_system() - # Determine destination if dest is None: dest = src.file_name log.debug("Host %s: Starting remote file copy for %s to %s/%s", self.host, src.file_name, file_system, dest) - # Open SSH connection and enable self.open() self.enable() - # Build command based on scheme and credentials - if src.scheme == "tftp" or (src.username is None and src.token is None): + if src.scheme == "tftp" or src.username is None: command, detect_prompt = self._build_url_copy_command_simple(src, file_system, dest) else: command, detect_prompt = self._build_url_copy_command_with_creds(src, file_system, dest) log.debug("Host %s: Preparing copy command for %s", self.host, src.scheme) - # Execute copy command if detect_prompt and src.token: - # Use send_command_timing for interactive password prompt output = self.native_ssh.send_command_timing(command, read_timeout=src.timeout, cmd_verify=False) log.debug("Host %s: Copy command (with timing) output: %s", self.host, output) if "password:" in output.lower(): self.native_ssh.write_channel(src.token + "\n") - # Read the response after sending password output += self.native_ssh.read_channel() log.debug("Host %s: Output after password entry: %s", self.host, output) - elif any(error in output.lower() for error in ["error", "invalid", "failed"]): - log.error("Host %s: Error detected in copy command output: %s", self.host, output) - raise FileTransferError(f"Error detected in copy command output: {output}") + else: + self._check_copy_output_for_errors(output) else: - # Use regular send_command for non-interactive transfers output = self.native_ssh.send_command(command, read_timeout=src.timeout) log.debug("Host %s: Copy command output: %s", self.host, output) + self._check_copy_output_for_errors(output) - if any(error in output.lower() for error in ["error", "invalid", "failed"]): - log.error("Host %s: Error detected in copy command output: %s", self.host, output) - raise FileTransferError(f"Error detected in copy command output: {output}") - - # Verify transfer success verification_result = self.verify_file( src.checksum, dest, hashing_algorithm=src.hashing_algorithm, file_system=file_system ) diff --git a/tests/unit/test_devices/test_eos_device.py b/tests/unit/test_devices/test_eos_device.py index 8786dfa1..a9a78e70 100644 --- a/tests/unit/test_devices/test_eos_device.py +++ b/tests/unit/test_devices/test_eos_device.py @@ -279,8 +279,6 @@ def test_file_copy_fail(self, mock_open, mock_close, mock_ssh, mock_ft): with self.assertRaises(FileTransferError): self.device.file_copy("source_file") - # TODO: unit test for remote_file_copy - def test_reboot(self): self.device.reboot() self.device.native.enable.assert_called_with(["reload now"], encoding="json") @@ -743,27 +741,13 @@ def test_remote_file_copy_unsupported_scheme(self): self.device.remote_file_copy(src) self.assertIn("Unsupported scheme", str(ctx.exception)) - def test_remote_file_copy_query_string_rejected(self): - """Test remote_file_copy raises ValueError for URLs with query strings.""" - src = FileCopyModel( - download_url="http://example.com/file.bin", - checksum="abc123", - file_name="file.bin", - ) - # Override download_url to include a query string - src.download_url = "http://example.com/file.bin?token=abc" - - with self.assertRaises(ValueError) as ctx: - self.device.remote_file_copy(src) - self.assertIn("query strings are not supported", str(ctx.exception)) - @mock.patch.object(EOSDevice, "verify_file") @mock.patch.object(EOSDevice, "enable") @mock.patch.object(EOSDevice, "open") @mock.patch.object(EOSDevice, "_get_file_system") - def test_remote_file_copy_logging_on_success(self, mock_get_fs, mock_open, mock_enable, mock_verify): - """Test that transfer success is logged.""" - mock_get_fs.return_value = "/mnt/flash" + def test_remote_file_copy_token_only_uses_simple_builder(self, mock_get_fs, mock_open, mock_enable, mock_verify): + """Test remote_file_copy uses simple command when token is provided but username is None.""" + mock_get_fs.return_value = "flash:" mock_verify.return_value = True mock_ssh = mock.MagicMock() @@ -771,105 +755,55 @@ def test_remote_file_copy_logging_on_success(self, mock_get_fs, mock_open, mock_ self.device.native_ssh = mock_ssh src = FileCopyModel( - download_url="http://server.example.com/file.bin", - checksum="abc123def456", - file_name="file.bin", - ) - - with mock.patch("pyntc.devices.eos_device.log") as mock_log: - self.device.remote_file_copy(src) - self.assertTrue( - any("transferred and verified successfully" in str(call) for call in mock_log.info.call_args_list) - ) - - -class TestRemoteFileCopyCommandExecution(unittest.TestCase): - """Tests for command execution flow in remote_file_copy.""" - - @mock.patch.object(EOSDevice, "verify_file") - @mock.patch.object(EOSDevice, "enable") - @mock.patch.object(EOSDevice, "open") - @mock.patch.object(EOSDevice, "_get_file_system") - def test_command_execution_with_http(self, mock_get_fs, mock_open, mock_enable, mock_verify): - """Test command execution for HTTP transfer.""" - mock_get_fs.return_value = "/mnt/flash" - mock_verify.return_value = True - - device = EOSDevice("host", "user", "pass") - mock_ssh = mock.MagicMock() - mock_ssh.send_command.return_value = "Copy completed successfully" - device.native_ssh = mock_ssh - - src = FileCopyModel( - download_url="http://server.example.com/file.bin", - checksum="abc123def456", + download_url="http://example.com/file.bin", + checksum="abc123", file_name="file.bin", + token="some_token", ) - device.remote_file_copy(src) + self.device.remote_file_copy(src) - mock_open.assert_called_once() - mock_enable.assert_called_once() - mock_ssh.send_command.assert_called() call_args = mock_ssh.send_command.call_args - self.assertIn("copy http://", call_args[0][0]) - - @mock.patch.object(EOSDevice, "verify_file") - @mock.patch.object(EOSDevice, "enable") - @mock.patch.object(EOSDevice, "open") - @mock.patch.object(EOSDevice, "_get_file_system") - def test_command_execution_with_scp_credentials(self, mock_get_fs, mock_open, mock_enable, mock_verify): - """Test command execution for SCP transfer with username only.""" - mock_get_fs.return_value = "/mnt/flash" - mock_verify.return_value = True - - device = EOSDevice("host", "user", "pass") - mock_ssh = mock.MagicMock() - mock_ssh.send_command_timing.return_value = "Copy completed successfully" - device.native_ssh = mock_ssh + command = call_args[0][0] + self.assertNotIn("None", command) + self.assertIn("copy http://", command) + def test_remote_file_copy_query_string_rejected(self): + """Test remote_file_copy raises ValueError for URLs with query strings.""" src = FileCopyModel( - download_url="scp://admin:password@backup.example.com/configs/startup-config", - checksum="abc123def456", - file_name="startup-config", - username="admin", - token="password", + download_url="http://example.com/file.bin?token=abc", + checksum="abc123", + file_name="file.bin", ) - device.remote_file_copy(src) - - mock_ssh.send_command_timing.assert_called() - call_args = mock_ssh.send_command_timing.call_args - self.assertIn("copy scp://", call_args[0][0]) - self.assertIn("admin@", call_args[0][0]) - self.assertNotIn("password@", call_args[0][0]) + with self.assertRaises(ValueError) as ctx: + self.device.remote_file_copy(src) + self.assertIn("query strings are not supported", str(ctx.exception)) @mock.patch.object(EOSDevice, "verify_file") @mock.patch.object(EOSDevice, "enable") @mock.patch.object(EOSDevice, "open") @mock.patch.object(EOSDevice, "_get_file_system") - def test_timeout_applied_to_send_command(self, mock_get_fs, mock_open, mock_enable, mock_verify): - """Test that timeout is applied to send_command calls.""" + def test_remote_file_copy_logging_on_success(self, mock_get_fs, mock_open, mock_enable, mock_verify): + """Test that transfer success is logged.""" mock_get_fs.return_value = "/mnt/flash" mock_verify.return_value = True - device = EOSDevice("host", "user", "pass") mock_ssh = mock.MagicMock() mock_ssh.send_command.return_value = "Copy completed successfully" - device.native_ssh = mock_ssh + self.device.native_ssh = mock_ssh src = FileCopyModel( download_url="http://server.example.com/file.bin", checksum="abc123def456", file_name="file.bin", - timeout=600, ) - device.remote_file_copy(src) - - mock_ssh.send_command.assert_called() - call_args = mock_ssh.send_command.call_args - self.assertEqual(call_args[1]["read_timeout"], 600) + with mock.patch("pyntc.devices.eos_device.log") as mock_log: + self.device.remote_file_copy(src) + self.assertTrue( + any("transferred and verified successfully" in str(call) for call in mock_log.info.call_args_list) + ) class TestFileCopyModelValidation(unittest.TestCase): From 2b9d9707b0faf4b5091da1eae27c9158c4ea6190 Mon Sep 17 00:00:00 2001 From: James Williams Date: Tue, 14 Apr 2026 10:59:02 -0500 Subject: [PATCH 3/8] Add changelog fragments for PR #368. Co-Authored-By: Claude Opus 4.6 (1M context) --- changes/368.changed | 5 +++++ changes/368.housekeeping | 2 ++ 2 files changed, 7 insertions(+) create mode 100644 changes/368.changed create mode 100644 changes/368.housekeeping diff --git a/changes/368.changed b/changes/368.changed new file mode 100644 index 00000000..d637187a --- /dev/null +++ b/changes/368.changed @@ -0,0 +1,5 @@ +Improved EOS remote file copy to validate scheme and query strings before connecting, use `clean_url` to prevent credential leakage, simplify credential routing, and include destination filename in copy commands. +Added EOS-specific hashing algorithm validation in `get_remote_checksum` for unsupported algorithms. +Changed `verify_file` to return early when file does not exist and use case-insensitive checksum comparison. +Simplified checksum parsing in `get_remote_checksum` to use string splitting instead of regex. +Removed `include_username` parameter from `remote_file_copy` in favor of automatic credential routing based on scheme and username presence. diff --git a/changes/368.housekeeping b/changes/368.housekeeping new file mode 100644 index 00000000..db320a5f --- /dev/null +++ b/changes/368.housekeeping @@ -0,0 +1,2 @@ +Converted EOS remote file copy tests from hypothesis/pytest standalone functions to unittest TestCase with `self.assertRaises` and `subTest` for consistency with the rest of the codebase. +Removed duplicate test class `TestRemoteFileCopyCommandExecution` and consolidated into `TestRemoteFileCopy`. From 800644aac48ef45e8dff8a06e1e1760477fc5f16 Mon Sep 17 00:00:00 2001 From: James Williams Date: Tue, 14 Apr 2026 12:17:52 -0500 Subject: [PATCH 4/8] Fix EOS remote file copy bugs found during integration testing Fixed copy command builders to include the source file path in the URL and use `flash:` as destination, matching EOS CLI conventions. Fixed password-prompt handling to wait for transfer completion. Fixed `_uptime_to_string` integer division, and `check_file_exists` / `get_remote_checksum` to open SSH before use. Added integration tests and updated changelog fragments. Co-Authored-By: Claude Opus 4.6 (1M context) --- changes/368.changed | 9 +- changes/368.housekeeping | 1 + pyntc/devices/eos_device.py | 37 +++-- tests/integration/test_eos_device.py | 194 +++++++++++++++++++++++++++ tests/unit/conftest.py | 44 +++++- 5 files changed, 268 insertions(+), 17 deletions(-) create mode 100644 tests/integration/test_eos_device.py diff --git a/changes/368.changed b/changes/368.changed index d637187a..494cf9b1 100644 --- a/changes/368.changed +++ b/changes/368.changed @@ -1,5 +1,8 @@ -Improved EOS remote file copy to validate scheme and query strings before connecting, use `clean_url` to prevent credential leakage, simplify credential routing, and include destination filename in copy commands. -Added EOS-specific hashing algorithm validation in `get_remote_checksum` for unsupported algorithms. -Changed `verify_file` to return early when file does not exist and use case-insensitive checksum comparison. +Improved EOS remote file copy to validate scheme and query strings before connecting, use `clean_url` to prevent credential leakage, and simplify credential routing. +Changed copy command builders to include the source file path in the URL and use `flash:` as the destination, matching EOS CLI conventions. +Fixed `_uptime_to_string` to use integer division, preventing `ValueError` on format specifiers. +Fixed `check_file_exists` and `get_remote_checksum` to open the SSH connection before use, preventing `AttributeError` when called standalone. +Fixed password-prompt handling in `remote_file_copy` to wait for the transfer to complete before proceeding to verification. Simplified checksum parsing in `get_remote_checksum` to use string splitting instead of regex. +Changed `verify_file` to return early when file does not exist and use case-insensitive checksum comparison. Removed `include_username` parameter from `remote_file_copy` in favor of automatic credential routing based on scheme and username presence. diff --git a/changes/368.housekeeping b/changes/368.housekeeping index db320a5f..406ca955 100644 --- a/changes/368.housekeeping +++ b/changes/368.housekeeping @@ -1,2 +1,3 @@ Converted EOS remote file copy tests from hypothesis/pytest standalone functions to unittest TestCase with `self.assertRaises` and `subTest` for consistency with the rest of the codebase. Removed duplicate test class `TestRemoteFileCopyCommandExecution` and consolidated into `TestRemoteFileCopy`. +Added integration tests for EOS device connectivity and remote file copy across FTP, TFTP, SCP, HTTP, HTTPS, and SFTP protocols. diff --git a/pyntc/devices/eos_device.py b/pyntc/devices/eos_device.py index b2bb50cf..f3f3c172 100644 --- a/pyntc/devices/eos_device.py +++ b/pyntc/devices/eos_device.py @@ -137,13 +137,13 @@ def _parse_response(self, response, raw_text): return list(x["result"] for x in response) def _uptime_to_string(self, uptime): - days = uptime / (24 * 60 * 60) + days = uptime // (24 * 60 * 60) uptime = uptime % (24 * 60 * 60) - hours = uptime / (60 * 60) + hours = uptime // (60 * 60) uptime = uptime % (60 * 60) - mins = uptime / 60 + mins = uptime // 60 uptime = uptime % 60 seconds = uptime @@ -442,6 +442,7 @@ def check_file_exists(self, filename, file_system=None): """ exists = False + self.open() file_system = file_system or self._get_file_system() command = f"dir {file_system}/{filename}" result = self.native_ssh.send_command(command, read_timeout=30) @@ -489,6 +490,7 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5", **kwargs): f"Supported algorithms: {sorted(EOS_SUPPORTED_HASHING_ALGORITHMS)}" ) + self.open() file_system = kwargs.get("file_system") if file_system is None: file_system = self._get_file_system() @@ -532,22 +534,32 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5", **kwargs): log.error("Host %s: Error getting remote checksum: %s", self.host, str(e)) raise CommandError(command, f"Error getting remote checksum: {str(e)}") + @staticmethod + def _parse_copy_url_parts(clean_url, dest): + """Parse a clean URL into (scheme, netloc, path) for EOS copy commands. + + If the URL has no file path, falls back to using dest as the filename. + """ + parsed = urlparse(clean_url) + netloc = f"{parsed.hostname}:{parsed.port}" if parsed.port else parsed.hostname + path = parsed.path if parsed.path and parsed.path != "/" else f"/{dest}" + return parsed.scheme, netloc, path + def _build_url_copy_command_simple(self, src, file_system, dest): """Build copy command for simple URL-based transfers (TFTP, HTTP, HTTPS without credentials).""" - return f"copy {src.clean_url} {file_system}/{dest}", False + scheme, netloc, path = self._parse_copy_url_parts(src.clean_url, dest) + return f"copy {scheme}://{netloc}{path} {file_system}", False 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).""" - parsed = urlparse(src.clean_url) - netloc = f"{parsed.hostname}:{parsed.port}" if parsed.port else parsed.hostname - path = parsed.path + _, netloc, path = self._parse_copy_url_parts(src.clean_url, dest) if src.scheme in ("http", "https"): - command = f"copy {src.scheme}://{src.username}:{src.token}@{netloc}{path} {file_system}/{dest}" + command = f"copy {src.scheme}://{src.username}:{src.token}@{netloc}{path} {file_system}" detect_prompt = False else: # SCP/FTP/SFTP — password provided at the interactive prompt - command = f"copy {src.scheme}://{src.username}@{netloc}{path} {file_system}/{dest}" + command = f"copy {src.scheme}://{src.username}@{netloc}{path} {file_system}" detect_prompt = True return command, detect_prompt @@ -606,14 +618,13 @@ def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kw if "password:" in output.lower(): self.native_ssh.write_channel(src.token + "\n") - output += self.native_ssh.read_channel() + output = self.native_ssh.send_command_timing("", read_timeout=src.timeout, cmd_verify=False) log.debug("Host %s: Output after password entry: %s", self.host, output) - else: - self._check_copy_output_for_errors(output) else: output = self.native_ssh.send_command(command, read_timeout=src.timeout) log.debug("Host %s: Copy command output: %s", self.host, output) - self._check_copy_output_for_errors(output) + + self._check_copy_output_for_errors(output) verification_result = self.verify_file( src.checksum, dest, hashing_algorithm=src.hashing_algorithm, file_system=file_system diff --git a/tests/integration/test_eos_device.py b/tests/integration/test_eos_device.py new file mode 100644 index 00000000..1bc3a34b --- /dev/null +++ b/tests/integration/test_eos_device.py @@ -0,0 +1,194 @@ +"""Integration tests for EOSDevice.remote_file_copy. + +These tests connect to an actual Arista EOS device in the lab and are run manually. +They are NOT part of the CI unit test suite. + +Usage (from project root): + export EOS_HOST= + export EOS_USER= + export EOS_PASS= + export FTP_URL=ftp://:@/ + export TFTP_URL=tftp:/// + export SCP_URL=scp://:@/ + export HTTP_URL=http://:@:8081/ + export HTTPS_URL=https://:@:8443/ + export SFTP_URL=sftp://:@/ + export FILE_CHECKSUM= + poetry run pytest tests/integration/test_eos_device.py -v + +Set only the protocol URL vars for the servers you have available; each +protocol test will skip automatically if its URL is not set. + +Environment variables: + EOS_HOST - IP address or hostname of the lab EOS device + EOS_USER - SSH / eAPI username + EOS_PASS - SSH / eAPI password + FTP_URL - FTP URL of the file to transfer + TFTP_URL - TFTP URL of the file to transfer + SCP_URL - SCP URL of the file to transfer + HTTP_URL - HTTP URL of the file to transfer + HTTPS_URL - HTTPS URL of the file to transfer + SFTP_URL - SFTP URL of the file to transfer + FILE_NAME - Destination filename on the device (default: basename of URL path) + FILE_CHECKSUM - Expected sha512 checksum of the file (shared across all protocols) +""" + +import os +import posixpath + +import pytest + +from pyntc.devices import EOSDevice +from pyntc.utils.models import FileCopyModel + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_PROTOCOL_URL_VARS = { + "ftp": "FTP_URL", + "tftp": "TFTP_URL", + "scp": "SCP_URL", + "http": "HTTP_URL", + "https": "HTTPS_URL", + "sftp": "SFTP_URL", +} + + +def _make_model(url_env_var): + """Build a FileCopyModel from a per-protocol URL env var. + + Calls pytest.skip if the URL or FILE_CHECKSUM is not set. + """ + url = os.environ.get(url_env_var) + checksum = os.environ.get("FILE_CHECKSUM") + file_name = os.environ.get("FILE_NAME") or (posixpath.basename(url.split("?")[0]) if url else None) + + if not all([url, checksum, file_name]): + pytest.skip(f"{url_env_var} / FILE_CHECKSUM environment variables not set") + + return FileCopyModel( + download_url=url, + checksum=checksum, + file_name=file_name, + hashing_algorithm="sha512", + timeout=900, + ) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def device(): + """Connect to the lab EOS device. Skips all tests if credentials are not set.""" + host = os.environ.get("EOS_HOST") + user = os.environ.get("EOS_USER") + password = os.environ.get("EOS_PASS") + + if not all([host, user, password]): + pytest.skip("EOS_HOST / EOS_USER / EOS_PASS environment variables not set") + + dev = EOSDevice(host, user, password) + yield dev + dev.close() + + +@pytest.fixture(scope="module") +def any_file_copy_model(): + """Return a FileCopyModel using the first available protocol URL. + + Used by tests that only need a file reference (existence checks, checksum + verification) without caring about the transfer protocol. Skips if no + protocol URL and FILE_CHECKSUM are set. + """ + checksum = os.environ.get("FILE_CHECKSUM") + for env_var in _PROTOCOL_URL_VARS.values(): + url = os.environ.get(env_var) + if url and checksum: + file_name = os.environ.get("FILE_NAME") or posixpath.basename(url.split("?")[0]) + return FileCopyModel( + download_url=url, + checksum=checksum, + file_name=file_name, + hashing_algorithm="sha512", + timeout=900, + ) + pytest.skip("No protocol URL / FILE_CHECKSUM environment variables not set") + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +def test_device_connects(device): + """Verify the device is reachable and responds to show commands.""" + facts = device.facts() + assert "hostname" in facts + assert "os_version" in facts + + +def test_check_file_exists_false(device, any_file_copy_model): + """Before the copy, the file should not exist (or this test is a no-op if it does).""" + result = device.check_file_exists(any_file_copy_model.file_name) + assert isinstance(result, bool) + + +def test_get_remote_checksum_after_exists(device, any_file_copy_model): + """If the file already exists, verify get_remote_checksum returns a non-empty string.""" + if not device.check_file_exists(any_file_copy_model.file_name): + pytest.skip("File does not exist on device; run test_remote_file_copy_* first") + checksum = device.get_remote_checksum(any_file_copy_model.file_name, hashing_algorithm="sha512") + assert checksum and len(checksum) > 0 + + +def test_remote_file_copy_ftp(device): + """Transfer the file using FTP and verify it exists on the device.""" + model = _make_model("FTP_URL") + device.remote_file_copy(model) + assert device.check_file_exists(model.file_name) + + +def test_remote_file_copy_tftp(device): + """Transfer the file using TFTP and verify it exists on the device.""" + model = _make_model("TFTP_URL") + device.remote_file_copy(model) + assert device.check_file_exists(model.file_name) + + +def test_remote_file_copy_scp(device): + """Transfer the file using SCP and verify it exists on the device.""" + model = _make_model("SCP_URL") + device.remote_file_copy(model) + assert device.check_file_exists(model.file_name) + + +def test_remote_file_copy_http(device): + """Transfer the file using HTTP and verify it exists on the device.""" + model = _make_model("HTTP_URL") + device.remote_file_copy(model) + assert device.check_file_exists(model.file_name) + + +def test_remote_file_copy_https(device): + """Transfer the file using HTTPS and verify it exists on the device.""" + model = _make_model("HTTPS_URL") + device.remote_file_copy(model) + assert device.check_file_exists(model.file_name) + + +def test_remote_file_copy_sftp(device): + """Transfer the file using SFTP and verify it exists on the device.""" + model = _make_model("SFTP_URL") + device.remote_file_copy(model) + assert device.check_file_exists(model.file_name) + + +def test_verify_file_after_copy(device, any_file_copy_model): + """After a successful copy the file should verify cleanly.""" + if not device.check_file_exists(any_file_copy_model.file_name): + pytest.skip("File does not exist on device; run a copy test first") + assert device.verify_file(any_file_copy_model.checksum, any_file_copy_model.file_name, hashing_algorithm="sha512") diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index 37acf4f5..f3f974df 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -3,7 +3,7 @@ import pytest -from pyntc.devices import AIREOSDevice, ASADevice, IOSDevice, IOSXEWLCDevice, supported_devices +from pyntc.devices import AIREOSDevice, ASADevice, EOSDevice, IOSDevice, IOSXEWLCDevice, supported_devices def get_side_effects(mock_path, side_effects): @@ -17,6 +17,48 @@ def get_side_effects(mock_path, side_effects): return effects +# EOS fixtures + + +@pytest.fixture +def eos_device(): + with mock.patch("pyeapi.client.Node", autospec=True) as mock_node: + device = EOSDevice("host", "user", "password") + device.native = mock_node + yield device + + +@pytest.fixture +def eos_mock_path(mock_path): + return f"{mock_path}/eos" + + +@pytest.fixture +def eos_send_command(eos_device, eos_mock_path): + def _mock(side_effects, existing_device=None, device=eos_device): + if existing_device is not None: + device = existing_device + device.native_ssh = mock.MagicMock() + device.native_ssh.send_command.side_effect = get_side_effects(f"{eos_mock_path}/send_command", side_effects) + return device + + return _mock + + +@pytest.fixture +def eos_send_command_timing(eos_device, eos_mock_path): + def _mock(side_effects, existing_device=None, device=eos_device): + if existing_device is not None: + device = existing_device + device.native_ssh = mock.MagicMock() + device.native_ssh.send_command_timing.side_effect = get_side_effects( + f"{eos_mock_path}/send_command", side_effects + ) + return device + + return _mock + + def pytest_generate_tests(metafunc): if metafunc.function.__name__ == "test_device_creation": metafunc.parametrize( From 31c4fb3c2e1ed6be8252a8561fd2630e587e85cc Mon Sep 17 00:00:00 2001 From: James Williams Date: Tue, 14 Apr 2026 12:19:33 -0500 Subject: [PATCH 5/8] Fix test_device_connects to use individual properties instead of deprecated facts() Co-Authored-By: Claude Opus 4.6 (1M context) --- tests/integration/test_eos_device.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/tests/integration/test_eos_device.py b/tests/integration/test_eos_device.py index 1bc3a34b..70547241 100644 --- a/tests/integration/test_eos_device.py +++ b/tests/integration/test_eos_device.py @@ -126,9 +126,8 @@ def any_file_copy_model(): def test_device_connects(device): """Verify the device is reachable and responds to show commands.""" - facts = device.facts() - assert "hostname" in facts - assert "os_version" in facts + assert device.hostname + assert device.os_version def test_check_file_exists_false(device, any_file_copy_model): From 3324643a8cfca98b8ce4d5143fbb44a5dcf89c4e Mon Sep 17 00:00:00 2001 From: James Williams Date: Tue, 14 Apr 2026 12:22:25 -0500 Subject: [PATCH 6/8] Add type hints to remote_file_copy signature Addresses review feedback from jeffkala. Co-Authored-By: Claude Opus 4.6 (1M context) --- pyntc/devices/eos_device.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyntc/devices/eos_device.py b/pyntc/devices/eos_device.py index f3f3c172..a6bbc43c 100644 --- a/pyntc/devices/eos_device.py +++ b/pyntc/devices/eos_device.py @@ -570,7 +570,7 @@ def _check_copy_output_for_errors(self, output): log.error("Host %s: Error detected in copy command output: %s", self.host, output) raise FileTransferError(f"Error detected in copy command output: {output}") - def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kwargs): + def remote_file_copy(self, src: FileCopyModel, dest: str | None = None, file_system: str | None = None, **kwargs): """Copy a file from remote source to device. Args: From 6723ed4e1205608658ca17fde2becae57c0c86ff Mon Sep 17 00:00:00 2001 From: James Williams Date: Tue, 14 Apr 2026 12:51:04 -0500 Subject: [PATCH 7/8] Simplify password handling to single send_command_timing call Replaces write_channel + separate send_command_timing with a single send_command_timing(src.token) that sends the password and waits for transfer completion in one step. Co-Authored-By: Claude Opus 4.6 (1M context) --- pyntc/devices/eos_device.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/pyntc/devices/eos_device.py b/pyntc/devices/eos_device.py index a6bbc43c..f1e4ebd2 100644 --- a/pyntc/devices/eos_device.py +++ b/pyntc/devices/eos_device.py @@ -617,8 +617,7 @@ def remote_file_copy(self, src: FileCopyModel, dest: str | None = None, file_sys log.debug("Host %s: Copy command (with timing) output: %s", self.host, output) if "password:" in output.lower(): - self.native_ssh.write_channel(src.token + "\n") - output = self.native_ssh.send_command_timing("", read_timeout=src.timeout, cmd_verify=False) + output = self.native_ssh.send_command_timing(src.token, read_timeout=src.timeout, cmd_verify=False) log.debug("Host %s: Output after password entry: %s", self.host, output) else: output = self.native_ssh.send_command(command, read_timeout=src.timeout) From 661eb79360240ace829156730859ee3410e340d7 Mon Sep 17 00:00:00 2001 From: James Williams Date: Tue, 14 Apr 2026 12:55:34 -0500 Subject: [PATCH 8/8] Add hostname, port, and path fields to FileCopyModel Moves URL parsing into the model so it happens once instead of on every command builder call. The EOS driver now reads src.hostname, src.port, and src.path directly, eliminating urlparse calls from the copy command builders. Other drivers can adopt the new fields incrementally. Co-Authored-By: Claude Opus 4.6 (1M context) --- pyntc/devices/eos_device.py | 24 ++++++++++++------------ pyntc/utils/models.py | 20 ++++++++++++-------- 2 files changed, 24 insertions(+), 20 deletions(-) diff --git a/pyntc/devices/eos_device.py b/pyntc/devices/eos_device.py index f1e4ebd2..1c3e74ab 100644 --- a/pyntc/devices/eos_device.py +++ b/pyntc/devices/eos_device.py @@ -3,7 +3,6 @@ import os import re import time -from urllib.parse import urlparse from netmiko import ConnectHandler, FileTransfer from pyeapi import connect as eos_connect @@ -535,24 +534,25 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5", **kwargs): raise CommandError(command, f"Error getting remote checksum: {str(e)}") @staticmethod - def _parse_copy_url_parts(clean_url, dest): - """Parse a clean URL into (scheme, netloc, path) for EOS copy commands. + def _netloc(src: FileCopyModel) -> str: + """Return host:port or just host from a FileCopyModel.""" + return f"{src.hostname}:{src.port}" if src.port else src.hostname - If the URL has no file path, falls back to using dest as the filename. - """ - parsed = urlparse(clean_url) - netloc = f"{parsed.hostname}:{parsed.port}" if parsed.port else parsed.hostname - path = parsed.path if parsed.path and parsed.path != "/" else f"/{dest}" - return parsed.scheme, netloc, path + @staticmethod + def _source_path(src: FileCopyModel, dest: str) -> str: + """Return the file path from the URL, falling back to dest if empty.""" + return src.path if src.path and src.path != "/" else f"/{dest}" def _build_url_copy_command_simple(self, src, file_system, dest): """Build copy command for simple URL-based transfers (TFTP, HTTP, HTTPS without credentials).""" - scheme, netloc, path = self._parse_copy_url_parts(src.clean_url, dest) - return f"copy {scheme}://{netloc}{path} {file_system}", False + netloc = self._netloc(src) + path = self._source_path(src, dest) + return f"copy {src.scheme}://{netloc}{path} {file_system}", False 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).""" - _, netloc, path = self._parse_copy_url_parts(src.clean_url, dest) + netloc = self._netloc(src) + path = self._source_path(src, dest) if src.scheme in ("http", "https"): command = f"copy {src.scheme}://{src.username}:{src.token}@{netloc}{path} {file_system}" diff --git a/pyntc/utils/models.py b/pyntc/utils/models.py index 3d047a81..18c55e31 100644 --- a/pyntc/utils/models.py +++ b/pyntc/utils/models.py @@ -36,17 +36,18 @@ class FileCopyModel: vrf: Optional[str] = None ftp_passive: bool = True - # This field is calculated, so we don't pass it in the constructor + # Computed fields derived from download_url — not passed to the constructor clean_url: str = field(init=False) scheme: str = field(init=False) + hostname: str = field(init=False) + port: Optional[int] = field(init=False) + path: str = field(init=False) def __post_init__(self): """Validate the input and prepare the clean URL after initialization.""" - # 1. Validate the hashing algorithm choice if self.hashing_algorithm.lower() not in HASHING_ALGORITHMS: raise ValueError(f"Unsupported algorithm. Choose from: {HASHING_ALGORITHMS}") - # Parse the url to extract components parsed = urlparse(self.download_url) # Extract username/password from URL if not already provided as arguments @@ -55,13 +56,16 @@ def __post_init__(self): if parsed.password and not self.token: self.token = parsed.password - # 3. Create the 'clean_url' (URL without the credentials) - # This is what you actually send to the device if using ip http client - port = f":{parsed.port}" if parsed.port else "" - self.clean_url = f"{parsed.scheme}://{parsed.hostname}{port}{parsed.path}" + # Store parsed URL components self.scheme = parsed.scheme + self.hostname = parsed.hostname + self.port = parsed.port + self.path = parsed.path + + # Create the 'clean_url' (URL without credentials) + port_str = f":{parsed.port}" if parsed.port else "" + self.clean_url = f"{parsed.scheme}://{parsed.hostname}{port_str}{parsed.path}" - # Handle query params if they exist (though we're avoiding '?' for Cisco) if parsed.query: self.clean_url += f"?{parsed.query}"