diff --git a/changes/368.changed b/changes/368.changed new file mode 100644 index 00000000..494cf9b1 --- /dev/null +++ b/changes/368.changed @@ -0,0 +1,8 @@ +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 new file mode 100644 index 00000000..406ca955 --- /dev/null +++ b/changes/368.housekeeping @@ -0,0 +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 2727eec3..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 @@ -26,6 +25,8 @@ from pyntc.utils import convert_list_by_key from pyntc.utils.models import FileCopyModel +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", @@ -135,13 +136,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 @@ -440,6 +441,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) @@ -481,6 +483,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). """ + 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)}" + ) + + self.open() file_system = kwargs.get("file_system") if file_system is None: file_system = self._get_file_system() @@ -511,11 +520,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,123 +533,98 @@ 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): + @staticmethod + 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 + + @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).""" - return f"copy {src.download_url} {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): + 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 - 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 "" + netloc = self._netloc(src) + path = self._source_path(src, dest) - # 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}" + if src.scheme in ("http", "https"): + command = f"copy {src.scheme}://{src.username}:{src.token}@{netloc}{path} {file_system}" 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}" + # SCP/FTP/SFTP — password provided at the interactive prompt + command = f"copy {src.scheme}://{src.username}@{netloc}{path} {file_system}" detect_prompt = True return command, detect_prompt - def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, include_username=False, **kwargs): + 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: str | None = None, file_system: str | None = 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. """ - # Validate input if not isinstance(src, FileCopyModel): raise TypeError("src must be an instance of FileCopyModel") - # Determine file system + if src.scheme not in EOS_SUPPORTED_SCHEMES: + raise ValueError(f"Unsupported scheme: {src.scheme}") + + # 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}") + 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() - # 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: + 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() + 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) - 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: - # 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) - 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}") + self._check_copy_output_for_errors(output) - # Verify transfer success verification_result = self.verify_file( src.checksum, dest, hashing_algorithm=src.hashing_algorithm, file_system=file_system ) @@ -676,11 +660,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/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}" diff --git a/tests/integration/test_eos_device.py b/tests/integration/test_eos_device.py new file mode 100644 index 00000000..70547241 --- /dev/null +++ b/tests/integration/test_eos_device.py @@ -0,0 +1,193 @@ +"""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.""" + assert device.hostname + assert device.os_version + + +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( diff --git a/tests/unit/test_devices/test_eos_device.py b/tests/unit/test_devices/test_eos_device.py index 9848cf5a..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") @@ -438,174 +436,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 +452,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 +497,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 +506,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 +519,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 +533,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 +546,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 +593,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 +604,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 +616,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 +628,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 +643,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,130 +655,39 @@ 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 - - -# Property-based tests for Task 7: Pre-transfer verification - - -@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 - - 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/file.bin", - checksum="abc123def456", - file_name="file.bin", - ) - - # Call remote_file_copy - device.remote_file_copy(src) - - # Verify that verify_file was called - mock_verify.assert_called() - - # 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_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 any file that exists on the device but has a mismatched checksum, - the `remote_file_copy()` method should proceed with the file transfer. - - Validates: Requirements 5.3 - """ - from pyntc.utils.models import FileCopyModel - - mock_get_fs.return_value = "/mnt/flash" - # Verification fails (file doesn't exist or checksum mismatches) - mock_verify.return_value = False - - 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", - file_name="file.bin", - ) - - # Call remote_file_copy - should raise FileTransferError because verification fails - with pytest.raises(FileTransferError): - device.remote_file_copy(src) - - # 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() - + 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, + ) -# Tests for Task 8: Command Execution + self.device.remote_file_copy(src) - -class TestRemoteFileCopyCommandExecution(unittest.TestCase): - """Tests for command execution flow in remote_file_copy.""" + 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_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 - + 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 = True + mock_verify.return_value = False - 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", @@ -941,387 +695,103 @@ def test_command_execution_with_http(self, mock_get_fs, mock_open, mock_enable, file_name="file.bin", ) - device.remote_file_copy(src) - - # Verify open() was called - mock_open.assert_called_once() - - # Verify enable() was called - mock_enable.assert_called_once() + with self.assertRaises(FileTransferError): + self.device.remote_file_copy(src) - # 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", call_args[0][0].lower()) @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.""" - from pyntc.utils.models import FileCopyModel - + 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 - device = EOSDevice("host", "user", "pass") mock_ssh = mock.MagicMock() - mock_ssh.send_command_timing.return_value = "Copy completed successfully" - device.native_ssh = mock_ssh + mock_ssh.send_command.return_value = "Copy completed successfully" + self.device.native_ssh = mock_ssh + + 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, + ) + self.device.remote_file_copy(src) + mock_verify.assert_called() + + def test_remote_file_copy_unsupported_scheme(self): + """Test remote_file_copy raises ValueError for unsupported scheme.""" 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", + checksum="abc123", + file_name="file.bin", ) + # Override scheme to something unsupported + src.scheme = "gopher" - 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 + with self.assertRaises(ValueError) as ctx: + self.device.remote_file_copy(src) + self.assertIn("Unsupported scheme", 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.""" - from pyntc.utils.models import FileCopyModel - - 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 - 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", + download_url="http://example.com/file.bin", + checksum="abc123", file_name="file.bin", - timeout=600, + token="some_token", ) - device.remote_file_copy(src) + self.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. - - 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" + 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="http://server.example.com/file.bin", - checksum="abc123def456", + download_url="http://example.com/file.bin?token=abc", + checksum="abc123", file_name="file.bin", ) - with pytest.raises(FileTransferError): - device.remote_file_copy(src) + 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_logging_on_transfer_success(self, mock_get_fs, mock_open, mock_enable, mock_verify): + def test_remote_file_copy_logging_on_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" + 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", @@ -1330,103 +800,97 @@ def test_logging_on_transfer_success(self, mock_get_fs, mock_open, mock_enable, ) 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. + 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) + ) - 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 +class TestFileCopyModelValidation(unittest.TestCase): + """Tests for FileCopyModel defaults and validation.""" - # Should accept case-insensitive algorithms - for algorithm in ["MD5", "md5", "Md5", "SHA256", "sha256", "Sha256"]: + 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", - hashing_algorithm=algorithm, ) - - # 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.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="ftp://admin:password@ftp.example.com/images/eos.swi", + checksum="abc123def456", + file_name="eos.swi", + ) + 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="scp://url_user:url_pass@server.com/path", + checksum="abc123def456", + file_name="file.bin", + username="explicit_user", + token="explicit_pass", + ) + self.assertEqual(src.username, "explicit_user") + self.assertEqual(src.token, "explicit_pass")