diff --git a/changes/370.added b/changes/370.added new file mode 100644 index 00000000..4b93036f --- /dev/null +++ b/changes/370.added @@ -0,0 +1,2 @@ +Added a pre-transfer free-space check to EOS ``file_copy`` and ``remote_file_copy`` that raises ``NotEnoughFreeSpaceError`` when the target filesystem lacks room for the image. +Added ``file_size_unit`` (``bytes``, ``megabytes``, or ``gigabytes``; default ``bytes``) and a computed ``file_size_bytes`` to ``FileCopyModel`` so ``remote_file_copy`` can verify free space against a caller-supplied size; when ``file_size`` is omitted the pre-transfer check is skipped. diff --git a/docs/user/lib_getting_started.md b/docs/user/lib_getting_started.md index 2bf2bc4e..f84227f5 100644 --- a/docs/user/lib_getting_started.md +++ b/docs/user/lib_getting_started.md @@ -264,7 +264,12 @@ from pyntc.utils.models import FileCopyModel ... checksum='abc123def456', ... hashing_algorithm='md5', ... file_name='newconfig.cfg', - vrf='Mgmt-vrf' +... # file_size is optional. When supplied, remote_file_copy verifies +... # the target device has room before starting the transfer. When +... # omitted, the pre-transfer space check is skipped. +... file_size=512, +... file_size_unit='megabytes', +... vrf='Mgmt-vrf', ... ) >>> for device in devices: ... device.remote_file_copy(source_file) diff --git a/pyntc/devices/base_device.py b/pyntc/devices/base_device.py index abe02237..eb4da066 100644 --- a/pyntc/devices/base_device.py +++ b/pyntc/devices/base_device.py @@ -2,11 +2,14 @@ import hashlib import importlib +import logging import warnings -from pyntc.errors import FeatureNotFoundError, NTCError +from pyntc.errors import FeatureNotFoundError, NotEnoughFreeSpaceError, NTCError from pyntc.utils.models import FileCopyModel +log = logging.getLogger(__name__) + def fix_docs(cls): """Create docstring at runtime. @@ -52,6 +55,90 @@ def __init__(self, host, username, password, device_type=None, **kwargs): # noq self._model = None self._vlans = None + def _check_free_space(self, required_bytes, file_system=None): + """Raise NotEnoughFreeSpaceError when the target filesystem lacks room for a transfer. + + Drivers call this from ``file_copy`` and ``remote_file_copy`` before starting a + transfer. The concrete per-platform probe is ``_get_free_space``; this helper + centralises the comparison, logging, and error shape so every driver behaves the + same way. + + Args: + required_bytes (int): Number of bytes the pending transfer needs. + file_system (str, optional): Target filesystem passed through to + ``_get_free_space``. Drivers that auto-detect a filesystem should resolve + it before calling. + + Raises: + NotEnoughFreeSpaceError: When the device reports fewer free bytes than + ``required_bytes``. + """ + available = self._get_free_space(file_system) + log.debug( + "Host %s: filesystem %s has %s bytes free; %s bytes required.", + self.host, + file_system, + available, + required_bytes, + ) + if available < required_bytes: + log.error( + "Host %s: insufficient free space on %s (%s free, %s required).", + self.host, + file_system, + available, + required_bytes, + ) + raise NotEnoughFreeSpaceError( + hostname=self.host, + required=required_bytes, + available=available, + file_system=file_system, + ) + + def _get_free_space(self, file_system=None): + """Return the number of free bytes on ``file_system``. + + Drivers override this to issue a platform-specific command (e.g., ``dir`` on + EOS/IOS/NXOS, ``show system storage`` on JUNOS) and parse the result. Drivers + whose platform exposes only one practical filesystem (e.g., EOS) may safely + ignore ``file_system``. + + Args: + file_system (str, optional): The target filesystem to inspect. Drivers + that support multiple filesystems should treat ``None`` as "auto-detect + the default filesystem". + + Returns: + int: Free bytes available on ``file_system``. + + Raises: + NotImplementedError: When a driver has not yet implemented the probe. + """ + raise NotImplementedError(f"{self.__class__.__name__} does not implement _get_free_space") + + def _pre_transfer_space_check(self, src, file_system=None): + """Run the free-space check if ``src.file_size_bytes`` is populated. + + Drivers call this from ``remote_file_copy`` so the check is skipped + (fail-open) when the caller omits ``FileCopyModel.file_size``; when set, + ``_check_free_space`` raises ``NotEnoughFreeSpaceError`` if the device + lacks room. Lives on ``BaseDevice`` so every driver inherits the same + shape without duplicating the if/else. + + Args: + src (FileCopyModel): The source specification for the pending transfer. + file_system (str, optional): Target filesystem; passed through to + ``_check_free_space`` (and from there to ``_get_free_space``). + """ + if src.file_size_bytes is not None: + self._check_free_space(src.file_size_bytes, file_system=file_system) + else: + log.debug( + "Host %s: no file_size on FileCopyModel; skipping pre-transfer space check.", + self.host, + ) + def _image_booted(self, image_name, **vendor_specifics): """Determine if a particular image is serving as the active OS. diff --git a/pyntc/devices/eos_device.py b/pyntc/devices/eos_device.py index 1c3e74ab..36a5be80 100644 --- a/pyntc/devices/eos_device.py +++ b/pyntc/devices/eos_device.py @@ -88,6 +88,34 @@ def _file_copy_instance(self, src, dest=None, file_system="/mnt/flash"): log.debug("Host %s: File copy instance %s.", self.host, file_copy) return file_copy + def _get_free_space(self, file_system=None): # pylint: disable=unused-argument + """Return free bytes on ``file_system`` as reported by Arista's ``dir`` output. + + EOS only exposes a single flash filesystem in practice, and ``dir`` always + prints `` bytes total ( bytes free)`` as the last line, so the + ``file_system`` argument is accepted for API parity with other drivers but is + otherwise unused. + + Args: + file_system (str, optional): Ignored; retained for BaseDevice API parity. + + Returns: + int: Free bytes available on the target filesystem. + + Raises: + CommandError: When the ``dir`` output does not contain a parseable + ``(N bytes free)`` trailer. + """ + raw_data = self.show("dir", raw_text=True) + # Example: 3634421760 bytes total (1951289344 bytes free) + match = re.search(r"\((\d+)\s+bytes\s+free\)", raw_data) + if match is None: + log.error("Host %s: could not parse free space from 'dir' output.", self.host) + raise CommandError(command="dir", message="Unable to parse free space from dir output.") + free_bytes = int(match.group(1)) + log.debug("Host %s: %s bytes free on flash.", self.host, free_bytes) + return free_bytes + def _get_file_system(self): """Determine the default file system or directory for device. @@ -371,6 +399,8 @@ def file_copy(self, src, dest=None, file_system=None): Raises: FileTransferError: raise exception if there is an error + NotEnoughFreeSpaceError: When ``file_system`` has fewer free bytes than + ``src`` requires. """ self.open() self.enable() @@ -379,6 +409,7 @@ def file_copy(self, src, dest=None, file_system=None): file_system = self._get_file_system() if not self.file_copy_remote_exists(src, dest, file_system): + self._check_free_space(os.path.getsize(src), file_system=file_system) file_copy = self._file_copy_instance(src, dest, file_system=file_system) try: @@ -584,6 +615,10 @@ def remote_file_copy(self, src: FileCopyModel, dest: str | None = None, file_sys ValueError: If the URL scheme is unsupported or URL contains query strings. FileTransferError: If transfer or verification fails. FileSystemNotFoundError: If filesystem cannot be determined. + NotEnoughFreeSpaceError: If ``src.file_size_bytes`` is set and + ``file_system`` has fewer free bytes than ``src.file_size_bytes``. + When ``file_size`` is omitted from ``src``, the pre-transfer space + check is skipped entirely. """ if not isinstance(src, FileCopyModel): raise TypeError("src must be an instance of FileCopyModel") @@ -606,6 +641,8 @@ def remote_file_copy(self, src: FileCopyModel, dest: str | None = None, file_sys self.open() self.enable() + self._pre_transfer_space_check(src, file_system) + if src.scheme == "tftp" or src.username is None: command, detect_prompt = self._build_url_copy_command_simple(src, file_system, dest) else: diff --git a/pyntc/errors.py b/pyntc/errors.py index a262533d..4fd5bab9 100644 --- a/pyntc/errors.py +++ b/pyntc/errors.py @@ -206,15 +206,23 @@ def __init__(self, hostname, wait_time): class NotEnoughFreeSpaceError(NTCError): """Error for not having enough free space to transfer a file.""" - def __init__(self, hostname, min_space): + def __init__(self, hostname, min_space=None, *, required=None, available=None, file_system=None): """ Error for not having enough free space to transfer a file. Args: - hostname (str): The hostname of the device that did not boot back up. - min_space (str): The minimum amount of space required to transfer the file. - """ - message = f"{hostname} does not meet the minimum disk space requirements of {min_space}" + hostname (str): The hostname of the device being checked. + min_space (str, optional): The minimum amount of space required. Retained for + backward compatibility with callers that only know the required value. + required (int, optional): Required bytes for the pending transfer. + available (int, optional): Free bytes currently available on the target filesystem. + file_system (str, optional): The target filesystem that was checked. + """ + if required is not None and available is not None: + location = f"{file_system} " if file_system else "" + message = f"{hostname}: {location}has {available} bytes free; {required} bytes required for transfer" + else: + message = f"{hostname} does not meet the minimum disk space requirements of {min_space}" super().__init__(message) diff --git a/pyntc/utils/models.py b/pyntc/utils/models.py index 18c55e31..8501367c 100644 --- a/pyntc/utils/models.py +++ b/pyntc/utils/models.py @@ -7,6 +7,14 @@ # Use Hashing algorithms from Nautobot's supported list. HASHING_ALGORITHMS = {"md5", "sha1", "sha224", "sha384", "sha256", "sha512", "sha3", "blake2", "blake3"} +# Supported units for FileCopyModel.file_size, mapped to their multiplier in bytes. +# Conversions use binary units (1 MB = 1024**2 bytes) to match network-device reporting. +FILE_SIZE_UNITS = { + "bytes": 1, + "megabytes": 1024**2, + "gigabytes": 1024**3, +} + @dataclass class FileCopyModel: @@ -16,9 +24,10 @@ class FileCopyModel: download_url (str): The URL to download the file from. Can include credentials, but it's recommended to use the username and token fields instead for security reasons. checksum (str): The expected checksum of the file. file_name (str): The name of the file to be saved on the device. + file_size (int, optional): The expected size of the file. When supplied, ``remote_file_copy`` verifies the target device has room before starting the transfer. When omitted, the pre-transfer space check is skipped (callers can probe the source URL themselves and populate this field). Defaults to ``None``. + file_size_unit (str, optional): Unit that ``file_size`` is expressed in. One of ``"bytes"``, ``"megabytes"``, ``"gigabytes"``. Only consulted when ``file_size`` is supplied. Defaults to ``"bytes"``. hashing_algorithm (str, optional): The hashing algorithm to use for checksum verification. Defaults to "md5". timeout (int, optional): The timeout for the download operation in seconds. Defaults to 900. - file_size (int, optional): The expected size of the file in bytes. Optional but can be used for an additional layer of verification. username (str, optional): The username for authentication if required by the URL. Optional if credentials are included in the URL. token (str, optional): The password or token for authentication if required by the URL. Optional if credentials are included in the URL. vrf (str, optional): The VRF to use for the download if the device supports VRFs. Optional. @@ -28,26 +37,40 @@ class FileCopyModel: download_url: str checksum: str file_name: str + file_size: Optional[int] = None + file_size_unit: str = "bytes" hashing_algorithm: str = "md5" timeout: int = 900 # Timeout for the download operation in seconds - file_size: Optional[int] = None # Size in bytes username: Optional[str] = None token: Optional[str] = None # Password/Token vrf: Optional[str] = None ftp_passive: bool = True - # Computed fields derived from download_url — not passed to the constructor + # Computed fields derived from download_url and file_size — 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) + file_size_bytes: Optional[int] = field(init=False) def __post_init__(self): """Validate the input and prepare the clean URL after initialization.""" if self.hashing_algorithm.lower() not in HASHING_ALGORITHMS: raise ValueError(f"Unsupported algorithm. Choose from: {HASHING_ALGORITHMS}") + unit = self.file_size_unit.lower() + if unit not in FILE_SIZE_UNITS: + raise ValueError(f"Unsupported file_size_unit. Choose from: {sorted(FILE_SIZE_UNITS)}") + self.file_size_unit = unit + + if self.file_size is None: + self.file_size_bytes = None + else: + if self.file_size < 0: + raise ValueError("file_size must be a non-negative integer.") + self.file_size_bytes = self.file_size * FILE_SIZE_UNITS[unit] + parsed = urlparse(self.download_url) # Extract username/password from URL if not already provided as arguments diff --git a/tests/integration/_helpers.py b/tests/integration/_helpers.py new file mode 100644 index 00000000..24e33553 --- /dev/null +++ b/tests/integration/_helpers.py @@ -0,0 +1,57 @@ +"""Shared helpers for integration tests that drive ``remote_file_copy``.""" + +import os +import posixpath + +import pytest + +from pyntc.utils.models import FileCopyModel + +# Every protocol that ``remote_file_copy`` might transfer from. Individual device +# test modules can narrow this set when they only support a subset. +PROTOCOL_URL_VARS = { + "ftp": "FTP_URL", + "tftp": "TFTP_URL", + "scp": "SCP_URL", + "http": "HTTP_URL", + "https": "HTTPS_URL", + "sftp": "SFTP_URL", +} + + +def build_file_copy_model(url_env_var): + """Build a ``FileCopyModel`` from a per-protocol URL env var. + + Calls ``pytest.skip`` if the URL, ``FILE_CHECKSUM``, or ``FILE_SIZE`` env + vars are 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) + file_size = int(os.environ.get("FILE_SIZE", "0")) + file_size_unit = os.environ.get("FILE_SIZE_UNIT", "bytes") + + if not all([url, checksum, file_name, file_size]): + pytest.skip(f"{url_env_var} / FILE_CHECKSUM / FILE_SIZE environment variables not set") + + return FileCopyModel( + download_url=url, + checksum=checksum, + file_name=file_name, + file_size=file_size, + file_size_unit=file_size_unit, + hashing_algorithm="sha512", + timeout=900, + ) + + +def first_available_url(protocol_url_vars=None): + """Return ``(scheme, url)`` for the first configured protocol URL. + + ``(None, None)`` when none of the env vars in ``protocol_url_vars`` are set. + """ + for scheme, env_var in (protocol_url_vars or PROTOCOL_URL_VARS).items(): + url = os.environ.get(env_var) + if url: + return scheme, url + return None, None diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py new file mode 100644 index 00000000..0ded2a40 --- /dev/null +++ b/tests/integration/conftest.py @@ -0,0 +1,37 @@ +"""Shared fixtures for pyntc integration tests.""" + +import os +import posixpath + +import pytest + +from pyntc.utils.models import FileCopyModel + +from ._helpers import PROTOCOL_URL_VARS + + +@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 / ``FILE_CHECKSUM`` / ``FILE_SIZE`` env vars are set. + """ + checksum = os.environ.get("FILE_CHECKSUM") + file_size = int(os.environ.get("FILE_SIZE", "0")) + file_size_unit = os.environ.get("FILE_SIZE_UNIT", "bytes") + for env_var in PROTOCOL_URL_VARS.values(): + url = os.environ.get(env_var) + if url and checksum and file_size: + file_name = os.environ.get("FILE_NAME") or posixpath.basename(url.split("?")[0]) + return FileCopyModel( + download_url=url, + checksum=checksum, + file_name=file_name, + file_size=file_size, + file_size_unit=file_size_unit, + hashing_algorithm="sha512", + timeout=900, + ) + pytest.skip("No protocol URL / FILE_CHECKSUM / FILE_SIZE environment variables not set") diff --git a/tests/integration/test_asa_device.py b/tests/integration/test_asa_device.py index b72a21de..fc230b9e 100644 --- a/tests/integration/test_asa_device.py +++ b/tests/integration/test_asa_device.py @@ -34,46 +34,12 @@ """ import os -import posixpath import pytest from pyntc.devices import ASADevice -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", -} - - -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, - ) +from ._helpers import build_file_copy_model # --------------------------------------------------------------------------- # Fixtures @@ -96,29 +62,6 @@ def device(): 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 # --------------------------------------------------------------------------- @@ -146,35 +89,35 @@ def test_get_remote_checksum_after_exists(device, any_file_copy_model): def test_remote_file_copy_ftp(device): """Transfer the file using FTP and verify it exists on the device.""" - model = _make_model("FTP_URL") + model = build_file_copy_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") + model = build_file_copy_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") + model = build_file_copy_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") + model = build_file_copy_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") + model = build_file_copy_model("HTTPS_URL") device.remote_file_copy(model) assert device.check_file_exists(model.file_name) diff --git a/tests/integration/test_eos_device.py b/tests/integration/test_eos_device.py index 70547241..6bc3b551 100644 --- a/tests/integration/test_eos_device.py +++ b/tests/integration/test_eos_device.py @@ -14,67 +14,40 @@ export HTTPS_URL=https://:@:8443/ export SFTP_URL=sftp://:@/ export FILE_CHECKSUM= + export FILE_SIZE= + export FILE_SIZE_UNIT=megabytes # optional; defaults to "bytes" 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) + 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) + FILE_SIZE - Expected size of the file expressed in FILE_SIZE_UNIT units; used for + the pre-transfer free-space check + FILE_SIZE_UNIT - One of "bytes", "megabytes", or "gigabytes" (default: "bytes") """ import os -import posixpath +from unittest import mock 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, - ) +from pyntc.errors import NotEnoughFreeSpaceError +from pyntc.utils.models import FILE_SIZE_UNITS, FileCopyModel +from ._helpers import PROTOCOL_URL_VARS, build_file_copy_model, first_available_url # --------------------------------------------------------------------------- # Fixtures @@ -96,29 +69,6 @@ def device(): 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 # --------------------------------------------------------------------------- @@ -146,42 +96,42 @@ def test_get_remote_checksum_after_exists(device, any_file_copy_model): def test_remote_file_copy_ftp(device): """Transfer the file using FTP and verify it exists on the device.""" - model = _make_model("FTP_URL") + model = build_file_copy_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") + model = build_file_copy_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") + model = build_file_copy_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") + model = build_file_copy_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") + model = build_file_copy_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") + model = build_file_copy_model("SFTP_URL") device.remote_file_copy(model) assert device.check_file_exists(model.file_name) @@ -191,3 +141,124 @@ def test_verify_file_after_copy(device, any_file_copy_model): 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") + + +# --------------------------------------------------------------------------- +# Free-space / pre-transfer tests (NAPPS-1091) +# --------------------------------------------------------------------------- + + +def test_get_free_space_returns_positive_int(device): + """``_get_free_space`` parses the ``dir`` trailer into a positive int.""" + free = device._get_free_space() # pylint: disable=protected-access + assert isinstance(free, int) + assert free > 0 + + +def test_check_free_space_succeeds_for_small_request(device): + """A 1-byte request must always fit; ``_check_free_space`` returns ``None``.""" + # pylint: disable=protected-access + assert device._check_free_space(required_bytes=1) is None + + +def test_check_free_space_raises_when_required_exceeds_free(device): + """When required bytes exceed what the device reports, raise NotEnoughFreeSpaceError.""" + # pylint: disable=protected-access + free = device._get_free_space() + with pytest.raises(NotEnoughFreeSpaceError): + device._check_free_space(required_bytes=free + 1) + + +def test_file_size_unit_conversion_matches_device_free_space(device): + """A megabyte-denominated request converts through ``FILE_SIZE_UNITS`` correctly.""" + # pylint: disable=protected-access + free_bytes = device._get_free_space() + one_mb = FILE_SIZE_UNITS["megabytes"] + if free_bytes < one_mb: + pytest.skip("Device has less than 1 MB free; conversion sanity test not meaningful") + # 1 MB should always fit when free space is at least that large. + assert device._check_free_space(required_bytes=one_mb) is None + + +def test_remote_file_copy_rejects_oversized_transfer(device): + """remote_file_copy raises NotEnoughFreeSpaceError and never copies the file.""" + checksum = os.environ.get("FILE_CHECKSUM") + scheme, url = first_available_url() + if not (url and checksum): + pytest.skip("No protocol URL / FILE_CHECKSUM environment variables not set") + + # pylint: disable=protected-access + free_bytes = device._get_free_space() + free_gb = free_bytes // FILE_SIZE_UNITS["gigabytes"] + # Ask for ten times the currently-free capacity (minimum 10 GB), expressed in + # gigabytes so this also exercises the unit conversion end-to-end. + oversized_gb = max(free_gb * 10, 10) + + unique_name = f"pyntc_integration_space_check_{os.getpid()}_{scheme}.bin" + model = FileCopyModel( + download_url=url, + checksum=checksum, + file_name=unique_name, + file_size=oversized_gb, + file_size_unit="gigabytes", + hashing_algorithm="sha512", + timeout=60, + ) + + assert not device.check_file_exists(unique_name), "Unique filename unexpectedly exists before test" + + with pytest.raises(NotEnoughFreeSpaceError): + device.remote_file_copy(model) + + # The transfer must never have started — file should still be absent. + assert not device.check_file_exists(unique_name) + + +def test_remote_file_copy_accepts_declared_size_within_free_space(device): + """A correctly-sized FileCopyModel copies without the space check interfering.""" + scheme, _url = first_available_url() + if scheme is None: + pytest.skip("No protocol URL environment variables set") + model = build_file_copy_model(PROTOCOL_URL_VARS[scheme]) + # pylint: disable=protected-access + free_bytes = device._get_free_space() + assert model.file_size_bytes <= free_bytes, ( + "Configured FILE_SIZE/FILE_SIZE_UNIT exceeds device free space; update env vars" + ) + device.remote_file_copy(model) + assert device.check_file_exists(model.file_name) + + +def test_remote_file_copy_skips_space_check_when_file_size_omitted(device): + """When FileCopyModel has no file_size, _check_free_space is never called. + + Spies on ``EOSDevice._check_free_space`` for the duration of the + transfer and asserts it was not invoked. The transfer itself uses the + same canonical ``FILE_NAME`` that the other copy tests use (the EOS + copy command derives the source name from the destination filename + when the URL has no path, so source and destination share a name). + The file already existing from a prior test run is fine — the + assertion that matters is ``spy.assert_not_called()`` combined with + the transfer completing without raising ``FileTransferError``. + """ + checksum = os.environ.get("FILE_CHECKSUM") + file_name = os.environ.get("FILE_NAME") + _, url = first_available_url() + if not (url and checksum and file_name): + pytest.skip("URL / FILE_CHECKSUM / FILE_NAME environment variables not set") + + model = FileCopyModel( + download_url=url, + checksum=checksum, + file_name=file_name, + hashing_algorithm="sha512", + timeout=60, + ) # file_size intentionally omitted + assert model.file_size is None + assert model.file_size_bytes is None + + with mock.patch.object(EOSDevice, "_check_free_space") as spy: + device.remote_file_copy(model) + + spy.assert_not_called() + assert device.check_file_exists(model.file_name) diff --git a/tests/unit/test_devices/device_mocks/eos/enable_text/dir b/tests/unit/test_devices/device_mocks/eos/enable_text/dir index c5a8b92f..0361acbd 100644 --- a/tests/unit/test_devices/device_mocks/eos/enable_text/dir +++ b/tests/unit/test_devices/device_mocks/eos/enable_text/dir @@ -1,7 +1,7 @@ { "command": "dir", "result": { - "output": "Directory of flash:/\n1 -rw- 15183868 c2600-ik9o3s3-mz.122-15.T9.bin\n16777216 bytes total (1592488 bytes free)\n" + "output": "Directory of flash:/\n1 -rw- 15183868 c2600-ik9o3s3-mz.122-15.T9.bin\n1073741824 bytes total (536870912 bytes free)\n" }, "encoding": "text" } diff --git a/tests/unit/test_devices/test_eos_device.py b/tests/unit/test_devices/test_eos_device.py index a9a78e70..287025ba 100644 --- a/tests/unit/test_devices/test_eos_device.py +++ b/tests/unit/test_devices/test_eos_device.py @@ -9,7 +9,7 @@ from pyntc.devices.base_device import RollbackError from pyntc.devices.eos_device import FileTransferError from pyntc.devices.system_features.vlans.eos_vlans import EOSVlans -from pyntc.errors import CommandError, CommandListError +from pyntc.errors import CommandError, CommandListError, NotEnoughFreeSpaceError # noqa: F401 from pyntc.utils.models import FileCopyModel from .device_mocks.eos import config, enable, send_command, send_command_expect @@ -225,11 +225,12 @@ def test_file_copy_remote_not_exist(self, mock_open, mock_close, mock_ssh, mock_ self.assertFalse(result) + @mock.patch("pyntc.devices.eos_device.os.path.getsize", return_value=1024) @mock.patch("pyntc.devices.eos_device.FileTransfer", autospec=True) @mock.patch.object(EOSDevice, "open") @mock.patch.object(EOSDevice, "close") @mock.patch("netmiko.arista.arista.AristaSSH", autospec=True) - def test_file_copy(self, mock_open, mock_close, mock_ssh, mock_ft): + def test_file_copy(self, mock_open, mock_close, mock_ssh, mock_ft, mock_getsize): self.device.native_ssh = mock_open self.device.native_ssh.send_command_timing.side_effect = None self.device.native_ssh.send_command_timing.return_value = "flash: /dev/null" @@ -245,11 +246,12 @@ def test_file_copy(self, mock_open, mock_close, mock_ssh, mock_ft): mock_ft_instance.establish_scp_conn.assert_any_call() mock_ft_instance.transfer_file.assert_any_call() + @mock.patch("pyntc.devices.eos_device.os.path.getsize", return_value=1024) @mock.patch("pyntc.devices.eos_device.FileTransfer", autospec=True) @mock.patch.object(EOSDevice, "open") @mock.patch.object(EOSDevice, "close") @mock.patch("netmiko.arista.arista.AristaSSH", autospec=True) - def test_file_copy_different_dest(self, mock_open, mock_close, mock_ssh, mock_ft): + def test_file_copy_different_dest(self, mock_open, mock_close, mock_ssh, mock_ft, mock_getsize): self.device.native_ssh = mock_open self.device.native_ssh.send_command_timing.side_effect = None self.device.native_ssh.send_command_timing.return_value = "flash: /dev/null" @@ -263,11 +265,12 @@ def test_file_copy_different_dest(self, mock_open, mock_close, mock_ssh, mock_ft mock_ft_instance.establish_scp_conn.assert_any_call() mock_ft_instance.transfer_file.assert_any_call() + @mock.patch("pyntc.devices.eos_device.os.path.getsize", return_value=1024) @mock.patch("pyntc.devices.eos_device.FileTransfer", autospec=True) @mock.patch.object(EOSDevice, "open") @mock.patch.object(EOSDevice, "close") @mock.patch("netmiko.arista.arista.AristaSSH", autospec=True) - def test_file_copy_fail(self, mock_open, mock_close, mock_ssh, mock_ft): + def test_file_copy_fail(self, mock_open, mock_close, mock_ssh, mock_ft, mock_getsize): self.device.native_ssh = mock_open self.device.native_ssh.send_command_timing.side_effect = None self.device.native_ssh.send_command_timing.return_value = "flash: /dev/null" @@ -436,8 +439,8 @@ def test_init_pass_port_and_timeout(mock_eos_connect): ) -class TestRemoteFileCopy(unittest.TestCase): - """Tests for remote_file_copy method.""" +class EOSDeviceMockedTestCase(unittest.TestCase): + """Base test case wiring a mocked ``pyeapi`` node onto an ``EOSDevice``.""" @mock.patch("pyeapi.client.Node", autospec=True) def setUp(self, mock_node): @@ -450,6 +453,10 @@ def setUp(self, mock_node): def tearDown(self): self.device.native.reset_mock() + +class TestRemoteFileCopy(EOSDeviceMockedTestCase): + """Tests for remote_file_copy method.""" + def test_remote_file_copy_invalid_src_type(self): """Test remote_file_copy raises TypeError for invalid src type.""" with self.assertRaises(TypeError) as ctx: @@ -473,6 +480,7 @@ def test_remote_file_copy_file_system_auto_detection(self, mock_get_fs, mock_ope download_url="http://server.example.com/file.bin", checksum="abc123", file_name="file.bin", + file_size=1024, ) self.device.remote_file_copy(src) @@ -495,6 +503,7 @@ def test_remote_file_copy_skip_transfer_on_checksum_match(self, mock_get_fs, moc download_url="http://example.com/file.bin", checksum="abc123", file_name="file.bin", + file_size=1024, ) self.device.remote_file_copy(src) @@ -517,6 +526,7 @@ def test_remote_file_copy_http_transfer(self, mock_get_fs, mock_open, mock_enabl download_url="http://example.com/file.bin", checksum="abc123", file_name="file.bin", + file_size=1024, ) self.device.remote_file_copy(src) @@ -544,6 +554,7 @@ def test_remote_file_copy_verification_failure(self, mock_get_fs, mock_open, moc download_url="http://example.com/file.bin", checksum="abc123", file_name="file.bin", + file_size=1024, ) with self.assertRaises(FileTransferError): @@ -566,6 +577,7 @@ def test_remote_file_copy_with_default_dest(self, mock_get_fs, mock_open, mock_e download_url="http://server.example.com/myfile.bin", checksum="abc123", file_name="myfile.bin", + file_size=1024, ) self.device.remote_file_copy(src) @@ -591,6 +603,7 @@ def test_remote_file_copy_with_explicit_dest(self, mock_get_fs, mock_open, mock_ download_url="http://example.com/file.bin", checksum="abc123", file_name="file.bin", + file_size=1024, ) self.device.remote_file_copy(src, dest="custom_name.bin") @@ -614,6 +627,7 @@ def test_remote_file_copy_with_explicit_file_system(self, mock_get_fs, mock_open download_url="http://example.com/file.bin", checksum="abc123", file_name="file.bin", + file_size=1024, ) self.device.remote_file_copy(src, file_system="flash:") @@ -639,6 +653,7 @@ def test_remote_file_copy_scp_with_credentials(self, mock_get_fs, mock_open, moc download_url="scp://user:pass@server.com/file.bin", checksum="abc123", file_name="file.bin", + file_size=1024, ) self.device.remote_file_copy(src) @@ -668,6 +683,7 @@ def test_remote_file_copy_timeout_applied(self, mock_get_fs, mock_open, mock_ena download_url="http://example.com/file.bin", checksum="abc123", file_name="file.bin", + file_size=1024, timeout=timeout, ) @@ -693,6 +709,7 @@ def test_remote_file_copy_checksum_mismatch_raises_error(self, mock_get_fs, mock download_url="http://server.example.com/file.bin", checksum="abc123def456", file_name="file.bin", + file_size=1024, ) with self.assertRaises(FileTransferError): @@ -721,6 +738,7 @@ def test_remote_file_copy_post_transfer_verification(self, mock_get_fs, mock_ope download_url="http://server.example.com/file.bin", checksum=checksum, file_name="file.bin", + file_size=1024, hashing_algorithm=algorithm, ) @@ -733,6 +751,7 @@ def test_remote_file_copy_unsupported_scheme(self): download_url="http://example.com/file.bin", checksum="abc123", file_name="file.bin", + file_size=1024, ) # Override scheme to something unsupported src.scheme = "gopher" @@ -758,6 +777,7 @@ def test_remote_file_copy_token_only_uses_simple_builder(self, mock_get_fs, mock download_url="http://example.com/file.bin", checksum="abc123", file_name="file.bin", + file_size=1024, token="some_token", ) @@ -774,6 +794,7 @@ def test_remote_file_copy_query_string_rejected(self): download_url="http://example.com/file.bin?token=abc", checksum="abc123", file_name="file.bin", + file_size=1024, ) with self.assertRaises(ValueError) as ctx: @@ -797,6 +818,7 @@ def test_remote_file_copy_logging_on_success(self, mock_get_fs, mock_open, mock_ download_url="http://server.example.com/file.bin", checksum="abc123def456", file_name="file.bin", + file_size=1024, ) with mock.patch("pyntc.devices.eos_device.log") as mock_log: @@ -815,6 +837,7 @@ def test_default_timeout_value(self): download_url="http://server.example.com/file.bin", checksum="abc123def456", file_name="file.bin", + file_size=1024, ) self.assertEqual(src.timeout, 900) @@ -826,6 +849,7 @@ def test_ftp_passive_mode_configuration(self): download_url="ftp://admin:password@ftp.example.com/images/eos.swi", checksum="abc123def456", file_name="eos.swi", + file_size=1024, ftp_passive=ftp_passive, ) self.assertEqual(src.ftp_passive, ftp_passive) @@ -836,6 +860,7 @@ def test_default_ftp_passive_mode(self): download_url="ftp://admin:password@ftp.example.com/images/eos.swi", checksum="abc123def456", file_name="eos.swi", + file_size=1024, ) self.assertTrue(src.ftp_passive) @@ -847,6 +872,7 @@ def test_hashing_algorithm_validation(self): download_url="http://server.example.com/file.bin", checksum="abc123def456", file_name="file.bin", + file_size=1024, hashing_algorithm=algorithm, ) self.assertEqual(src.hashing_algorithm, algorithm) @@ -859,6 +885,7 @@ def test_case_insensitive_algorithm_validation(self): download_url="http://server.example.com/file.bin", checksum="abc123def456", file_name="file.bin", + file_size=1024, hashing_algorithm=algorithm, ) self.assertIn(src.hashing_algorithm.lower(), ["md5", "sha256"]) @@ -879,6 +906,7 @@ def test_url_credential_extraction(self): download_url=url, checksum="abc123def456", file_name="file.bin", + file_size=1024, ) self.assertEqual(src.username, expected_username) self.assertEqual(src.token, expected_token) @@ -889,8 +917,88 @@ def test_explicit_credentials_override(self): download_url="scp://url_user:url_pass@server.com/path", checksum="abc123def456", file_name="file.bin", + file_size=1024, username="explicit_user", token="explicit_pass", ) self.assertEqual(src.username, "explicit_user") self.assertEqual(src.token, "explicit_pass") + + +class TestFreeSpaceCheck(EOSDeviceMockedTestCase): + """Tests for EOS pre-transfer free-space verification on file transfers.""" + + def test_get_free_space_parses_dir_trailer(self): + """_get_free_space returns the bytes-free value from the dir trailer.""" + self.assertEqual(self.device._get_free_space(), 536870912) + + @mock.patch.object(EOSDevice, "show", return_value="Directory of flash:/\nno trailer here") + def test_get_free_space_raises_when_trailer_missing(self, _mock_show): + """_get_free_space raises CommandError when the trailer can't be parsed.""" + with self.assertRaises(CommandError): + self.device._get_free_space() + + @mock.patch("pyntc.devices.eos_device.os.path.getsize", return_value=10**12) + @mock.patch("pyntc.devices.eos_device.FileTransfer", autospec=True) + @mock.patch.object(EOSDevice, "open") + @mock.patch.object(EOSDevice, "close") + def test_file_copy_raises_not_enough_free_space(self, _close, _open, mock_ft, _getsize): + """file_copy raises NotEnoughFreeSpaceError and never touches FileTransfer.""" + self.device.native_ssh = mock.MagicMock() + self.device.native_ssh.send_command_timing.return_value = "flash: /dev/null" + mock_ft.return_value.check_file_exists.return_value = False + + with self.assertRaises(NotEnoughFreeSpaceError): + self.device.file_copy("source_file") + + mock_ft.return_value.establish_scp_conn.assert_not_called() + mock_ft.return_value.transfer_file.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", return_value="flash:") + def test_remote_file_copy_raises_not_enough_free_space(self, _fs, _open, _enable, _verify): + """remote_file_copy raises NotEnoughFreeSpaceError and never issues a copy command.""" + mock_ssh = mock.MagicMock() + self.device.native_ssh = mock_ssh + + oversized = FileCopyModel( + download_url="http://example.com/image.bin", + checksum="abc123", + file_name="image.bin", + file_size=2, + file_size_unit="gigabytes", + ) + + with self.assertRaises(NotEnoughFreeSpaceError): + self.device.remote_file_copy(oversized) + + mock_ssh.send_command.assert_not_called() + mock_ssh.send_command_timing.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", return_value="flash:") + @mock.patch.object(EOSDevice, "_check_free_space") + def test_remote_file_copy_skips_space_check_when_file_size_omitted( + self, mock_check, _fs, _open, _enable, mock_verify + ): + """When FileCopyModel has no file_size, _check_free_space is NOT called.""" + mock_verify.return_value = True + mock_ssh = mock.MagicMock() + mock_ssh.send_command.return_value = "Copy completed successfully" + self.device.native_ssh = mock_ssh + + model = FileCopyModel( + download_url="http://example.com/image.bin", + checksum="abc123", + file_name="image.bin", + ) # file_size intentionally omitted + assert model.file_size_bytes is None + + self.device.remote_file_copy(model) + + mock_check.assert_not_called() + mock_ssh.send_command.assert_called() # transfer still happens diff --git a/tests/unit/utils/__init__.py b/tests/unit/utils/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/unit/utils/test_models.py b/tests/unit/utils/test_models.py new file mode 100644 index 00000000..bc6f5f91 --- /dev/null +++ b/tests/unit/utils/test_models.py @@ -0,0 +1,87 @@ +"""Unit tests for ``FileCopyModel`` size/unit handling.""" + +import pytest + +from pyntc.utils.models import FILE_SIZE_UNITS, FileCopyModel + + +def _model(**overrides): + defaults = { + "download_url": "http://example.com/image.bin", + "checksum": "abc123", + "file_name": "image.bin", + "file_size": 1, + } + defaults.update(overrides) + return FileCopyModel(**defaults) + + +def test_file_size_unit_defaults_to_bytes(): + model = _model(file_size=42) + assert model.file_size_unit == "bytes" + assert model.file_size_bytes == 42 + + +@pytest.mark.parametrize( + "unit, multiplier", + [ + ("bytes", 1), + ("megabytes", 1024**2), + ("gigabytes", 1024**3), + ], +) +def test_file_size_bytes_converts_from_unit(unit, multiplier): + model = _model(file_size=3, file_size_unit=unit) + assert model.file_size_bytes == 3 * multiplier + assert model.file_size_bytes == 3 * FILE_SIZE_UNITS[unit] + + +def test_file_size_unit_is_normalised_to_lower_case(): + model = _model(file_size=5, file_size_unit="MegaBytes") + assert model.file_size_unit == "megabytes" + assert model.file_size_bytes == 5 * 1024**2 + + +def test_unknown_file_size_unit_raises(): + with pytest.raises(ValueError, match="Unsupported file_size_unit"): + _model(file_size=1, file_size_unit="terabytes") + + +def test_negative_file_size_raises(): + with pytest.raises(ValueError, match="non-negative"): + _model(file_size=-1) + + +def test_file_size_is_optional(): + model = FileCopyModel( + download_url="http://example.com/image.bin", + checksum="abc123", + file_name="image.bin", + ) + assert model.file_size is None + assert model.file_size_bytes is None + # Unit default survives even when size is omitted. + assert model.file_size_unit == "bytes" + + +def test_unknown_unit_raises_even_when_file_size_omitted(): + """``file_size_unit`` is always validated, whether ``file_size`` is set or not.""" + with pytest.raises(ValueError, match="Unsupported file_size_unit"): + FileCopyModel( + download_url="http://example.com/image.bin", + checksum="abc123", + file_name="image.bin", + file_size_unit="terabytes", + ) + + +def test_file_size_unit_default_is_bytes_when_file_size_omitted(): + """Default unit survives and stays lowercase even with no ``file_size``.""" + model = FileCopyModel( + download_url="http://example.com/image.bin", + checksum="abc123", + file_name="image.bin", + file_size_unit="MegaBytes", + ) + assert model.file_size_unit == "megabytes" + assert model.file_size_bytes is None